Clean up grpc (#1197)

cmd/executor/echo/main.go

@@ -59,7 +59,7 @@ func (*EchoExecutor) Execute(_ context.Context, in executor.ExecuteInput) (execu
 	}
 
 	return executor.ExecuteOutput{
-		Data: data,
+		Message: api.NewCodeBlockMessage(data, true),
 	}, nil
 }
 

cmd/executor/gh/main.go

@@ -81,7 +81,7 @@ func (e *GHExecutor) Execute(ctx context.Context, in executor.ExecuteInput) (exe
 
 	if cmd.Create == nil || cmd.Create.Issue == nil {
 		return executor.ExecuteOutput{
-			Data: fmt.Sprintf("Usage: %s create issue KIND/NAME", pluginName),
+			Message: api.NewCodeBlockMessage(fmt.Sprintf("Usage: %s create issue KIND/NAME", pluginName), false),
 		}, nil
 	}
 
@@ -114,7 +114,7 @@ func (e *GHExecutor) Execute(ctx context.Context, in executor.ExecuteInput) (exe
 	}
 
 	return executor.ExecuteOutput{
-		Data: fmt.Sprintf("New issue created successfully! 🎉\n\nIssue URL: %s", issueURL),
+		Message: api.NewCodeBlockMessage(fmt.Sprintf("New issue created successfully! 🎉\n\nIssue URL: %s", issueURL), false),
 	}, nil
 }
 

cmd/source/cm-watcher/main.go

@@ -72,15 +72,15 @@ func (CMWatcher) Stream(ctx context.Context, in source.StreamInput) (source.Stre
 	}
 
 	out := source.StreamOutput{
-		Output: make(chan []byte),
+		Event: make(chan source.Event),
 	}
 
-	go listenEvents(ctx, in.Context.KubeConfig, cfg.ConfigMap, out.Output)
+	go listenEvents(ctx, in.Context.KubeConfig, cfg.ConfigMap, out.Event)
 
 	return out, nil
 }
 
-func listenEvents(ctx context.Context, kubeConfig []byte, obj Object, sink chan<- []byte) {
+func listenEvents(ctx context.Context, kubeConfig []byte, obj Object, sink chan source.Event) {
 	config, err := clientcmd.RESTConfigFromKubeConfig(kubeConfig)
 	exitOnError(err)
 	clientset, err := kubernetes.NewForConfig(config)
@@ -102,7 +102,9 @@ func listenEvents(ctx context.Context, kubeConfig []byte, obj Object, sink chan<
 		if event.Type == obj.Event {
 			cm := event.Object.(*corev1.ConfigMap)
 			msg := fmt.Sprintf("Plugin %s detected `%s` event on `%s/%s`", pluginName, obj.Event, cm.Namespace, cm.Name)
-			sink <- []byte(msg)
+			sink <- source.Event{
+				Message: api.NewPlaintextMessage(msg, true),
+			}
 		}
 
 		// always continue - context will cancel this watch for us :)

hack/gen-grpc-resources.sh

@@ -11,9 +11,9 @@ REPO_ROOT_DIR=$(cd "${CURRENT_DIR}/.." && pwd)
 readonly CURRENT_DIR
 readonly REPO_ROOT_DIR
 
-readonly STABLE_PROTOC_VERSION=3.20.2
-readonly STABLE_PROTOC_GEN_GO_GRPC_VERSION=1.2.0
-readonly STABLE_PROTOC_GEN_GO_VERSION=v1.27.1
+readonly STABLE_PROTOC_VERSION=24.0
+readonly STABLE_PROTOC_GEN_GO_GRPC_VERSION=1.3.0
+readonly STABLE_PROTOC_GEN_GO_VERSION=v1.31.0
 
 readonly GREEN='\033[0;32m'
 readonly NC='\033[0m' # No Color

internal/executor/helm/executor_test.go

@@ -135,7 +135,6 @@ func TestExecutorHelmInstallFlagsErrors(t *testing.T) {
 			// then
 			require.EqualError(t, err, tc.expErrMsg)
 			assert.Empty(t, out.Message)
-			assert.Empty(t, out.Data)
 		})
 	}
 }

internal/source/dispatcher.go

@@ -13,7 +13,6 @@ import (
 	"github.com/kubeshop/botkube/internal/audit"
 	"github.com/kubeshop/botkube/internal/plugin"
 	"github.com/kubeshop/botkube/pkg/action"
-	"github.com/kubeshop/botkube/pkg/api"
 	"github.com/kubeshop/botkube/pkg/api/source"
 	"github.com/kubeshop/botkube/pkg/bot"
 	"github.com/kubeshop/botkube/pkg/bot/interactive"
@@ -123,12 +122,6 @@ func (d *Dispatcher) Dispatch(dispatch PluginDispatch) error {
 	go func() {
 		for {
 			select {
-			case event, ok := <-out.Output:
-				if !ok {
-					return
-				}
-				log.WithField("event", string(event)).Debug("Dispatching received event...")
-				d.dispatch(ctx, event, dispatch)
 			case msg, ok := <-out.Event:
 				if !ok {
 					return
@@ -232,20 +225,6 @@ func (d *Dispatcher) dispatchMsg(ctx context.Context, event source.Event, dispat
 	}
 }
 
-func (d *Dispatcher) dispatch(ctx context.Context, event []byte, dispatch PluginDispatch) {
-	if event == nil {
-		return
-	}
-
-	d.dispatchMsg(ctx, source.Event{
-		Message: api.Message{
-			BaseBody: api.Body{
-				Plaintext: string(event),
-			},
-		},
-	}, dispatch)
-}
-
 func (d *Dispatcher) reportAuditEvent(ctx context.Context, pluginName string, event any, sourceName, sourceDisplayName string) error {
 	eventBytes, err := json.Marshal(event)
 	if err != nil {

internal/source/kubernetes/event/event.go

@@ -17,14 +17,12 @@ import (
 type Event struct {
 	APIVersion      string
 	Kind            string
-	Code            string
 	Title           string
 	Name            string
 	Namespace       string
 	Messages        []string
 	Type            config.EventType
 	Reason          string
-	Error           string
 	Level           config.Level
 	Cluster         string
 	TimeStamp       time.Time
@@ -34,7 +32,6 @@ type Event struct {
 	Resource        string
 	Recommendations []string
 	Warnings        []string
-	Actions         []Action
 
 	// The following fields are ignored when marshalling the event by purpose.
 	// We send the whole Event struct via sink.Elasticsearch integration.