Handle multiple configurations for multiple bots and sinks (#670)

- Spawn multiple bots and sinks for each communication group
- Support multiple notification channels for all bots
- Support multiple indices for Elasticsearch sink
- Support toggling notifications on / off by channel for all bots
- Do not fall back to sending notifications to default channel (as there are no "default channel") when annotation config is wrong; instead, report an error
- Pass executor bindings to Executor
- Refactor trimming message prefix for all bots
- Always return when running notifier commands (even if in not authorized channel)
- Add E2E tests
- Fix outdated E2E instructions
- Fix security vulnerability with yaml library

cmd/botkube/main.go

@@ -41,6 +41,7 @@ const (
 	componentLogFieldKey = "component"
 	botLogFieldKey       = "bot"
 	sinkLogFieldKey      = "sink"
+	commGroupFieldKey    = "commGroup"
 	printAPIKeyCharCount = 3
 )
 
@@ -142,71 +143,82 @@ func run() error {
 		reporter,
 	)
 
-	commCfg := conf.Communications.GetFirst()
+	commCfg := conf.Communications
 	var notifiers []controller.Notifier
 
-	// Run bots
-	if commCfg.Slack.Enabled {
-		sb, err := bot.NewSlack(logger.WithField(botLogFieldKey, "Slack"), conf, executorFactory, reporter)
-		if err != nil {
-			return reportFatalError("while creating Slack bot", err)
+	// TODO: Current limitation: Communication platform config should be separate inside every group:
+	//    For example, if in both communication groups there's a Slack configuration pointing to the same workspace,
+	//	  when user executes `kubectl` command, one Bot instance will execute the command and return response,
+	//	  and the second "Sorry, this channel is not authorized to execute kubectl command" error.
+	for commGroupName, commGroupCfg := range commCfg {
+		commGroupLogger := logger.WithField(commGroupFieldKey, commGroupName)
+
+		// Run bots
+		if commGroupCfg.Slack.Enabled {
+			sb, err := bot.NewSlack(commGroupLogger.WithField(botLogFieldKey, "Slack"), commGroupCfg.Slack, executorFactory, reporter)
+			if err != nil {
+				return reportFatalError("while creating Slack bot", err)
+			}
+			notifiers = append(notifiers, sb)
+			errGroup.Go(func() error {
+				defer analytics.ReportPanicIfOccurs(commGroupLogger, reporter)
+				return sb.Start(ctx)
+			})
 		}
-		notifiers = append(notifiers, sb)
-		errGroup.Go(func() error {
-			defer analytics.ReportPanicIfOccurs(logger, reporter)
-			return sb.Start(ctx)
-		})
-	}
 
-	if commCfg.Mattermost.Enabled {
-		mb, err := bot.NewMattermost(logger.WithField(botLogFieldKey, "Mattermost"), conf, executorFactory, reporter)
-		if err != nil {
-			return reportFatalError("while creating Mattermost bot", err)
+		if commGroupCfg.Mattermost.Enabled {
+			mb, err := bot.NewMattermost(commGroupLogger.WithField(botLogFieldKey, "Mattermost"), commGroupCfg.Mattermost, executorFactory, reporter)
+			if err != nil {
+				return reportFatalError("while creating Mattermost bot", err)
+			}
+			notifiers = append(notifiers, mb)
+			errGroup.Go(func() error {
+				defer analytics.ReportPanicIfOccurs(commGroupLogger, reporter)
+				return mb.Start(ctx)
+			})
 		}
-		notifiers = append(notifiers, mb)
-		errGroup.Go(func() error {
-			defer analytics.ReportPanicIfOccurs(logger, reporter)
-			return mb.Start(ctx)
-		})
-	}
 
-	if commCfg.Teams.Enabled {
-		tb := bot.NewTeams(logger.WithField(botLogFieldKey, "MS Teams"), conf, executorFactory, reporter)
-		notifiers = append(notifiers, tb)
-		errGroup.Go(func() error {
-			defer analytics.ReportPanicIfOccurs(logger, reporter)
-			return tb.Start(ctx)
-		})
-	}
-
-	if commCfg.Discord.Enabled {
-		db, err := bot.NewDiscord(logger.WithField(botLogFieldKey, "Discord"), conf, executorFactory, reporter)
-		if err != nil {
-			return reportFatalError("while creating Discord bot", err)
+		if commGroupCfg.Teams.Enabled {
+			tb, err := bot.NewTeams(commGroupLogger.WithField(botLogFieldKey, "MS Teams"), commGroupCfg.Teams, conf.Settings.ClusterName, executorFactory, reporter)
+			if err != nil {
+				return reportFatalError("while creating Teams bot", err)
+			}
+			notifiers = append(notifiers, tb)
+			errGroup.Go(func() error {
+				defer analytics.ReportPanicIfOccurs(commGroupLogger, reporter)
+				return tb.Start(ctx)
+			})
 		}
-		notifiers = append(notifiers, db)
-		errGroup.Go(func() error {
-			defer analytics.ReportPanicIfOccurs(logger, reporter)
-			return db.Start(ctx)
-		})
-	}
 
-	// Run sinks
-	if commCfg.Elasticsearch.Enabled {
-		es, err := sink.NewElasticsearch(logger.WithField(sinkLogFieldKey, "Elasticsearch"), commCfg.Elasticsearch, reporter)
-		if err != nil {
-			return reportFatalError("while creating Elasticsearch sink", err)
+		if commGroupCfg.Discord.Enabled {
+			db, err := bot.NewDiscord(commGroupLogger.WithField(botLogFieldKey, "Discord"), commGroupCfg.Discord, executorFactory, reporter)
+			if err != nil {
+				return reportFatalError("while creating Discord bot", err)
+			}
+			notifiers = append(notifiers, db)
+			errGroup.Go(func() error {
+				defer analytics.ReportPanicIfOccurs(commGroupLogger, reporter)
+				return db.Start(ctx)
+			})
 		}
-		notifiers = append(notifiers, es)
-	}
 
-	if commCfg.Webhook.Enabled {
-		wh, err := sink.NewWebhook(logger.WithField(sinkLogFieldKey, "Webhook"), commCfg.Webhook, reporter)
-		if err != nil {
-			return reportFatalError("while creating Webhook sink", err)
+		// Run sinks
+		if commGroupCfg.Elasticsearch.Enabled {
+			es, err := sink.NewElasticsearch(commGroupLogger.WithField(sinkLogFieldKey, "Elasticsearch"), commGroupCfg.Elasticsearch, reporter)
+			if err != nil {
+				return reportFatalError("while creating Elasticsearch sink", err)
+			}
+			notifiers = append(notifiers, es)
 		}
 
-		notifiers = append(notifiers, wh)
+		if commGroupCfg.Webhook.Enabled {
+			wh, err := sink.NewWebhook(commGroupLogger.WithField(sinkLogFieldKey, "Webhook"), commGroupCfg.Webhook, reporter)
+			if err != nil {
+				return reportFatalError("while creating Webhook sink", err)
+			}
+
+			notifiers = append(notifiers, wh)
+		}
 	}
 
 	// Start upgrade checker

go.mod

@@ -25,7 +25,7 @@ require (
 	github.com/vrischmann/envconfig v1.3.0
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
 	golang.org/x/text v0.3.7
-	gopkg.in/yaml.v3 v3.0.0
+	gopkg.in/yaml.v3 v3.0.1
 	gotest.tools/v3 v3.0.3
 	k8s.io/api v0.24.0
 	k8s.io/apimachinery v0.24.0

go.sum

@@ -1790,8 +1790,8 @@ gopkg.in/yaml.v3 v3.0.0-20191120175047-4206685974f2/go.mod h1:K4uyk7z7BCEPqu6E+C
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-gopkg.in/yaml.v3 v3.0.0 h1:hjy8E9ON/egN1tAYqKb61G10WtihqetD4sz2H+8nIeA=
-gopkg.in/yaml.v3 v3.0.0/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gotest.tools v2.2.0+incompatible/go.mod h1:DsYFclhRJ6vuDpmuTbkuFWG+y2sxOXAzmJt81HFBacw=
 gotest.tools/v3 v3.0.2/go.mod h1:3SzNCllyD9/Y+b5r9JIKQ474KzkZyqLqEfYqMsX94Bk=
 gotest.tools/v3 v3.0.3 h1:4AuOwCGf4lLR9u3YOe2awrHygurzhO/HeQ6laiA6Sx0=

helm/botkube/e2e-test-values.yaml

@@ -5,13 +5,22 @@ communications:
       token: "" # Provide a valid token for BotKube app
       channels:
         'default':
-          name: ""           # Test will override that
+          name: "" # Tests will override this temporarily
           bindings:
             executors:
               - kubectl-read-only
               - kubectl-wait-cmd
               - kubectl-exec-cmd
               - kubectl-allow-all
+            sources:
+              - k8s-events
+        'secondary':
+          name: "" # Tests will override this temporarily
+          bindings:
+            executors:
+              - kubectl-read-only
+            sources:
+              - k8s-events
 
 sources:
   'k8s-events':

helm/botkube/templates/tests/e2e-test.yaml

@@ -32,8 +32,10 @@ spec:
           value: "{{ .Values.e2eTest.deployment.waitTimeout }}"
         - name: DEPLOYMENT_ENVS_SLACK_ENABLED_NAME
           value: "BOTKUBE_COMMUNICATIONS_DEFAULT-GROUP_SLACK_ENABLED"
-        - name: DEPLOYMENT_ENVS_SLACK_CHANNEL_ID_NAME
+        - name: DEPLOYMENT_ENVS_DEFAULT_SLACK_CHANNEL_ID_NAME
           value: "BOTKUBE_COMMUNICATIONS_DEFAULT-GROUP_SLACK_CHANNELS_DEFAULT_NAME"
+        - name: DEPLOYMENT_ENVS_SECONDARY_SLACK_CHANNEL_ID_NAME
+          value: "BOTKUBE_COMMUNICATIONS_DEFAULT-GROUP_SLACK_CHANNELS_SECONDARY_NAME"
         - name: CLUSTER_NAME
           value: "{{ .Values.settings.clusterName }}"
         - name: SLACK_BOT_NAME

pkg/bot/bot.go

@@ -8,6 +8,10 @@ import (
 	"github.com/kubeshop/botkube/pkg/execute"
 )
 
+const (
+	defaultNotifyValue = true
+)
+
 // Bot connects to communication channels and reads/sends messages. It is a two-way integration.
 type Bot interface {
 	Start(ctx context.Context) error
@@ -17,7 +21,7 @@ type Bot interface {
 
 // ExecutorFactory facilitates creation of execute.Executor instances.
 type ExecutorFactory interface {
-	NewDefault(platform config.CommPlatformIntegration, notifierHandler execute.NotifierHandler, isAuthChannel bool, message string) execute.Executor
+	NewDefault(platform config.CommPlatformIntegration, notifierHandler execute.NotifierHandler, isAuthChannel bool, conversationID string, bindings []string, message string) execute.Executor
 }
 
 // AnalyticsReporter defines a reporter that collects analytics data.
@@ -36,3 +40,15 @@ type FatalErrorAnalyticsReporter interface {
 	// Close cleans up the reporter resources.
 	Close() error
 }
+
+type channelConfigByID struct {
+	config.ChannelBindingsByID
+
+	notify bool
+}
+
+type channelConfigByName struct {
+	config.ChannelBindingsByName
+
+	notify bool
+}