Add source plugin support

.goreleaser.plugin.yaml

@@ -15,3 +15,15 @@ builds:
     goarch:
       - amd64
       - arm64
+  - id: cm-watcher
+    binary: source_cm-watcher_{{ .Os }}_{{ .Arch }}
+    no_unique_dist_dir: true
+    main: cmd/source/cm-watcher/main.go
+    env:
+      - CGO_ENABLED=0
+    goos:
+      - linux
+      - darwin
+    goarch:
+      - amd64
+      - arm64

cmd/botkube/main.go

@@ -30,6 +30,7 @@ import (
 	"github.com/kubeshop/botkube/internal/analytics"
 	"github.com/kubeshop/botkube/internal/lifecycle"
 	"github.com/kubeshop/botkube/internal/plugin"
+	"github.com/kubeshop/botkube/internal/source"
 	"github.com/kubeshop/botkube/internal/storage"
 	"github.com/kubeshop/botkube/pkg/action"
 	"github.com/kubeshop/botkube/pkg/bot"
@@ -99,8 +100,8 @@ func run() error {
 
 	errGroup, ctx := errgroup.WithContext(ctx)
 
-	enabledPluginExecutors := plugin.GetAllEnabledAndUsedPluginExecutors(conf)
-	pluginManager := plugin.NewManager(logger, conf.Plugins, enabledPluginExecutors)
+	enabledPluginExecutors, enabledPluginSources := plugin.GetAllEnabledAndUsedPlugins(conf)
+	pluginManager := plugin.NewManager(logger, conf.Plugins, enabledPluginExecutors, enabledPluginSources)
 
 	err = pluginManager.Start(ctx)
 	if err != nil {
@@ -321,6 +322,13 @@ func run() error {
 	actionProvider := action.NewProvider(logger.WithField(componentLogFieldKey, "Action Provider"), conf.Actions, executorFactory)
 	router.AddEnabledActionBindings(conf.Actions)
 
+	// TODO: temporary dispatcher for events
+	sourcePluginDispatcher := source.NewDispatcher(logger, notifiers, pluginManager, enabledPluginSources)
+	err = sourcePluginDispatcher.Start(ctx)
+	if err != nil {
+		return fmt.Errorf("while starting source plugin event dispatcher: %w", err)
+	}
+
 	// Create and start controller
 	ctrl := controller.New(
 		logger.WithField(componentLogFieldKey, "Controller"),

cmd/source/cm-watcher/README.md

@@ -0,0 +1,11 @@
+# Echo executor
+
+Echo is the example Botkube executor used during [e2e tests](../../../test/e2e).
+
+## Configuration parameters
+
+The Echo configuration should be specified in YAML format. It accepts such parameters:
+
+```yaml
+changeResponseToUpperCase: true # default is 'false'.
+```

cmd/source/cm-watcher/main.go

@@ -0,0 +1,58 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+
+	"github.com/hashicorp/go-plugin"
+
+	"github.com/kubeshop/botkube/pkg/api/source"
+)
+
+const pluginName = "cm-watcher"
+
+// Config holds executor configuration.
+type Config struct {
+	ConfigMapName string
+}
+
+// CMWatcher implements Botkube source plugin.
+type CMWatcher struct{}
+
+// Stream returns a given command as response.
+func (CMWatcher) Stream(ctx context.Context) (source.StreamOutput, error) {
+	// TODO: in request we should receive the executor configuration.
+	cfg := Config{
+		ConfigMapName: "cm-watcher-trigger",
+	}
+
+	raw, err := json.Marshal(cfg)
+	if err != nil {
+		return source.StreamOutput{}, err
+	}
+	out := source.StreamOutput{
+		Output: make(chan []byte),
+	}
+
+	go func() {
+		for {
+			select {
+			case <-time.Tick(1 * time.Second):
+				out.Output <- raw
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	return out, nil
+}
+
+func main() {
+	source.Serve(map[string]plugin.Plugin{
+		pluginName: &source.Plugin{
+			Source: &CMWatcher{},
+		},
+	})
+}

go.mod

@@ -15,6 +15,7 @@ require (
 	github.com/gorilla/mux v1.8.0
 	github.com/hashicorp/go-multierror v1.1.1
 	github.com/hashicorp/go-plugin v1.4.3
+	github.com/hashicorp/go-version v1.6.0
 	github.com/infracloudio/msbotbuilder-go v0.2.5
 	github.com/knadh/koanf v1.4.4
 	github.com/mattermost/mattermost-server/v5 v5.39.3

go.sum

@@ -881,6 +881,8 @@ github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/b
 github.com/hashicorp/go-uuid v1.0.1/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
 github.com/hashicorp/go-version v1.1.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
 github.com/hashicorp/go-version v1.2.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
+github.com/hashicorp/go-version v1.6.0 h1:feTTfFNnjP967rlCxM/I9g701jU+RN74YKx2mOkIeek=
+github.com/hashicorp/go-version v1.6.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
 github.com/hashicorp/go.net v0.0.1/go.mod h1:hjKkEWcCURg++eb33jQU7oqQcI9XDCnUzHA0oac0k90=
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=

internal/plugin/bindings.go → internal/plugin/collector.go

@@ -4,16 +4,23 @@ import (
 	"github.com/kubeshop/botkube/pkg/config"
 )
 
-// GetAllEnabledAndUsedPluginExecutors returns the list of all plugins executors that are both enabled and bind to at
+// GetAllEnabledAndUsedPlugins returns the list of all plugins that are both enabled and bind to at
 // least one communicator that is also enabled.
-func GetAllEnabledAndUsedPluginExecutors(cfg *config.Config) []string {
-	// Collect all used executor bindings
-	bindExecutors := map[string]struct{}{}
+func GetAllEnabledAndUsedPlugins(cfg *config.Config) ([]string, []string) {
+	// Collect all used executor/sources
+	var (
+		bindExecutors = map[string]struct{}{}
+		bindSources   = map[string]struct{}{}
+	)
+
 	collect := func(channels config.IdentifiableMap[config.ChannelBindingsByName]) {
 		for _, bindings := range channels {
 			for _, name := range bindings.Bindings.Executors {
 				bindExecutors[name] = struct{}{}
 			}
+			for _, name := range bindings.Bindings.Sources {
+				bindSources[name] = struct{}{}
+			}
 		}
 	}
 
@@ -34,13 +41,19 @@ func GetAllEnabledAndUsedPluginExecutors(cfg *config.Config) []string {
 			for _, name := range commGroupCfg.Teams.Bindings.Executors {
 				bindExecutors[name] = struct{}{}
 			}
+			for _, name := range commGroupCfg.Teams.Bindings.Sources {
+				bindSources[name] = struct{}{}
+			}
 		}
 
 		if commGroupCfg.Discord.Enabled {
 			for _, bindings := range commGroupCfg.Discord.Channels {
 				for _, name := range bindings.Bindings.Executors {
 					bindExecutors[name] = struct{}{}
 				}
+				for _, name := range bindings.Bindings.Sources {
+					bindSources[name] = struct{}{}
+				}
 			}
 		}
 	}
@@ -61,5 +74,20 @@ func GetAllEnabledAndUsedPluginExecutors(cfg *config.Config) []string {
 		}
 	}
 
-	return usedExecutorPlugins
+	// Collect all sources that are both enabled and bind to at least one communicator that is enabled.
+	var usedSourcePlugins []string
+	for groupName, groupItems := range cfg.Sources {
+		for name, source := range groupItems.Plugins {
+			if !source.Enabled {
+				continue
+			}
+			_, found := bindSources[groupName]
+			if !found {
+				continue
+			}
+
+			usedSourcePlugins = append(usedSourcePlugins, name)
+		}
+	}
+	return usedExecutorPlugins, usedSourcePlugins
 }

internal/plugin/index.go

@@ -1,5 +1,12 @@
 package plugin
 
+import (
+	"fmt"
+	"strings"
+
+	semver "github.com/hashicorp/go-version"
+)
+
 // Type represents the plugin type.
 type Type string
 
@@ -24,3 +31,37 @@ type (
 		Links       []string
 	}
 )
+
+// byIndexEntryVersion implements sort.Interface based on the version field.
+type byIndexEntryVersion []IndexEntry
+
+func (a byIndexEntryVersion) Len() int      { return len(a) }
+func (a byIndexEntryVersion) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
+func (a byIndexEntryVersion) Less(i, j int) bool {
+	return semvVerGreater(a[i].Version, a[j].Version)
+}
+
+// semvVerGreater returns true if A is greater than B.
+func semvVerGreater(a, b string) bool {
+	a = strings.TrimPrefix(a, "v")
+	b = strings.TrimPrefix(b, "v")
+
+	verA, aErr := semver.NewVersion(a)
+	verB, bErr := semver.NewVersion(b)
+
+	return aErr == nil && bErr == nil && verA.GreaterThan(verB)
+}
+
+// BuildPluginKey returns plugin key.
+func BuildPluginKey(repo, plugin string) string {
+	return repo + "/" + plugin
+}
+
+// DecomposePluginKey extract details from plugin key.
+func DecomposePluginKey(key string) (string, string, error) {
+	repo, name, found := strings.Cut(key, "/")
+	if !found {
+		return "", "", fmt.Errorf("plugin %q doesn't follow required {repo_name}/{plugin_name} syntax", key)
+	}
+	return repo, name, nil
+}