Fix closing streams

cmd/executor/gh/main.go

@@ -63,7 +63,7 @@ func (*GHExecutor) Metadata(context.Context) (api.MetadataOutput, error) {
 
 // Execute returns a given command as a response.
 func (e *GHExecutor) Execute(ctx context.Context, in executor.ExecuteInput) (executor.ExecuteOutput, error) {
-	if err := pluginx.CheckKubeConfigProvided(pluginName, in.Context.KubeConfig); err != nil {
+	if err := pluginx.ValidateKubeConfigProvided(pluginName, in.Context.KubeConfig); err != nil {
 		return executor.ExecuteOutput{}, err
 	}
 

internal/executor/helm/executor.go

@@ -84,7 +84,7 @@ func (e *Executor) Metadata(context.Context) (api.MetadataOutput, error) {
 // - history
 // - get [all|manifest|hooks|notes]
 func (e *Executor) Execute(ctx context.Context, in executor.ExecuteInput) (executor.ExecuteOutput, error) {
-	if err := pluginx.CheckKubeConfigProvided(PluginName, in.Context.KubeConfig); err != nil {
+	if err := pluginx.ValidateKubeConfigProvided(PluginName, in.Context.KubeConfig); err != nil {
 		return executor.ExecuteOutput{}, err
 	}
 

internal/executor/kubectl/executor.go

@@ -74,7 +74,7 @@ func (e *Executor) Metadata(context.Context) (api.MetadataOutput, error) {
 
 // Execute returns a given command as response.
 func (e *Executor) Execute(ctx context.Context, in executor.ExecuteInput) (executor.ExecuteOutput, error) {
-	if err := pluginx.CheckKubeConfigProvided(PluginName, in.Context.KubeConfig); err != nil {
+	if err := pluginx.ValidateKubeConfigProvided(PluginName, in.Context.KubeConfig); err != nil {
 		return executor.ExecuteOutput{}, err
 	}
 

internal/source/dispatcher.go

@@ -121,10 +121,16 @@ func (d *Dispatcher) Dispatch(dispatch PluginDispatch) error {
 	go func() {
 		for {
 			select {
-			case event := <-out.Output:
+			case event, ok := <-out.Output:
+				if !ok {
+					return
+				}
 				log.WithField("event", string(event)).Debug("Dispatching received event...")
 				d.dispatch(ctx, event, dispatch)
-			case msg := <-out.Event:
+			case msg, ok := <-out.Event:
+				if !ok {
+					return
+				}
 				log.WithField("message", msg).Debug("Dispatching received message...")
 				d.dispatchMsg(ctx, msg, dispatch)
 			case <-ctx.Done():

internal/source/kubernetes/source.go

@@ -65,7 +65,7 @@ func NewSource(version string) *Source {
 
 // Stream streams Kubernetes events
 func (*Source) Stream(ctx context.Context, input source.StreamInput) (source.StreamOutput, error) {
-	if err := pluginx.CheckKubeConfigProvided(PluginName, input.Context.KubeConfig); err != nil {
+	if err := pluginx.ValidateKubeConfigProvided(PluginName, input.Context.KubeConfig); err != nil {
 		return source.StreamOutput{}, err
 	}
 

pkg/api/source/grpc_adapter.go

@@ -5,10 +5,11 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
-	"log"
 
 	"github.com/hashicorp/go-plugin"
+	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/emptypb"
 
 	"github.com/kubeshop/botkube/pkg/api"
@@ -114,11 +115,13 @@ func (p *Plugin) GRPCServer(_ *plugin.GRPCBroker, s *grpc.Server) error {
 func (p *Plugin) GRPCClient(_ context.Context, _ *plugin.GRPCBroker, c *grpc.ClientConn) (interface{}, error) {
 	return &grpcClient{
 		client: NewSourceClient(c),
+		logger: NewLogger(),
 	}, nil
 }
 
 type grpcClient struct {
 	client SourceClient
+	logger logrus.FieldLogger
 }
 
 func (p *grpcClient) Stream(ctx context.Context, in StreamInput) (StreamOutput, error) {
@@ -152,20 +155,22 @@ func (p *grpcClient) Stream(ctx context.Context, in StreamInput) (StreamOutput,
 			// On any other error, the stream is aborted and the error contains the RPC
 			// status.
 			if err != nil {
-				log.Print(err)
+				p.logger.Errorf("canceling streaming: %s", status.Convert(err).Message())
 				// TODO: we should consider adding error feedback channel to StreamOutput.
 				return
 			}
 			var event Event
 			if len(feature.Event) != 0 && string(feature.Event) != "" {
 				if err := json.Unmarshal(feature.Event, &event); err != nil {
-					log.Printf("while unmarshalling message from JSON: %s", err.Error())
+					p.logger.Errorf("canceling streaming: cannot unmarshal JSON message: %s", err.Error())
 					return
 				}
 			}
 			out.Output <- feature.Output
 			out.Event <- event
 		}
+		close(out.Output)
+		close(out.Event)
 	}()
 
 	return out, nil

pkg/api/source/logger.go

@@ -0,0 +1,15 @@
+package source
+
+import (
+	"os"
+
+	"github.com/sirupsen/logrus"
+)
+
+// NewLogger returns a new logger used internally. We should replace it in the near future, as we shouldn't be so opinionated.
+func NewLogger() logrus.FieldLogger {
+	logger := logrus.New()
+	logger.SetOutput(os.Stdout)
+	logger.SetLevel(logrus.InfoLevel)
+	return logger
+}

pkg/pluginx/kubeconfig.go

@@ -35,8 +35,8 @@ func PersistKubeConfig(_ context.Context, kc []byte) (string, func(context.Conte
 	return abs, deleteFn, nil
 }
 
-// CheckKubeConfigProvided returns an error if a given kubeconfig is empty or nil.
-func CheckKubeConfigProvided(pluginName string, kubeconfig []byte) error {
+// ValidateKubeConfigProvided returns an error if a given kubeconfig is empty or nil.
+func ValidateKubeConfigProvided(pluginName string, kubeconfig []byte) error {
 	if len(kubeconfig) != 0 {
 		return nil
 	}