Webhook refactored after k8s plugin extraction

[huseyinbabal]

Description

Changes proposed in this pull request:

• Webhook sink is updated to accept new contract api.Message
• Introduced rawData param to send original data to sinks
• Collector is updated to enable plugins for also sinks

Testing

Build plugins

PLUGIN_DOWNLOAD_URL_BASE_PATH=http://localhost:8080/plugin-dist make gen-plugins-index

Run plugin server

npx serve --listen 8080

Run Botkube with the following config

ℹ️ You can use Webhook Site Tester for webhook URL

export BOTKUBE_CONFIG_PATHS=/tmp/botkube.yml
export BOTKUBE_SETTINGS_KUBECONFIG=${USER}/.kube/config

go run cmd/botkube/main.go

## /tmp/botkube.yml
communications:
  'default-group':
    webhook:
      enabled: true
      url: 'webhook_url'
      bindings:
        sources: 
          - k8s-events
          - k8s-updates


settings:
  clusterName: gke-playground
  configWatcher: true
  log:
    level: debug
  lifecycleServer:
    deployment:
      name: botkube
      namespace: botkube
    port: "2113"

configWatcher:
  enabled: false
  initialSyncTimeout: 0

sources:
  'k8s-events':
    displayName: "K8s recommendations"
    'botkube/kubernetes':
      enabled: true
      config:
        log:
          level: debug
        recommendations:
          pod:
            noLatestImageTag: true
            labelsSet: true
          ingress:
            backendServiceValid: false
            tlsSecretValid: false
        namespaces:
          include:
            - botkube
        event:
          types:
            - create
            - update

  'k8s-updates':
    displayName: "K8s ConfigMaps updates"
    'botkube/kubernetes':
      enabled: true
      config:
        log:
          level: debug
        namespaces:
          include:
            - default
        event:
          types:
            - create
            - update
            - delete
        resources:
          - type: v1/configmaps
            namespaces:
              include:
                - botkube
            event: # overrides top level `event` entry
              types:
                - update
            updateSetting:
              includeDiff: true
              fields:
                - data

plugins:
  repositories:
    botkube:
      url: http://localhost:8080/plugins-index.yaml

Create a pod to see a notification

kubectl run nginx_webhook --image=nginx:latest -n botkube

Related issue(s)

Fixes #999

[pkosiec]

Bots don't consume raw event, so, so can we specify a different interface for Sinks and Bots if they are different?

So for sinks we can still require such method:

type Sink interface {
    SendEvent(ctx context.Context, event any, eventSources []string) error
  // (...)
}

and for Bots:

type Bot interface {
  SendMessage(context.Context, interactive.CoreMessage, []string) error
  // (...)
}

And then in dispatcher, instead of iterating over notifiers, we can iterate over both bots and sinks and send the data properly (message + raw data).

Could you please update also Elasticsearch integration? It can be one PR if it will be easier for you 👍 Thank you!

BTW We can remove SendEvent methods from bots as they are not used and won't be in such setup 🙂

[huseyinbabal]

I need @mszostok 's input here, I don't want to break something :) After that I will introduce 2 interfaces.

[mszostok]

Yep, I agree with Paweł's comment 👍 For now, Bots don't need to have the raw event and I don't see any use case for that at least for the current features that we plan to have. Having that separated will be better for our code-base 👍

[huseyinbabal]

@pkosiec @mszostok I separated sink and bot notifiers as suggested

[pkosiec]

🤔 That changes the contract which were before. Can we try to keep the same shape of data? so basically we can inline data.

And why do we send just a first source binding which instead of all of them?

[huseyinbabal]

Actually we cannot keep the previous format since, that format was all about kubernetes events. I applied same strategy we did for telemetry. So, those information is called RawData and we send it to webhook and es. Notice that, those data is coming from plugins which can be anything

We haven't changed the signature for sources yet, it is still accepting list of sources, and that is what I get only first since there is only one source inside it. this will be removed soon, and we will have only single source.

[pkosiec]

Alright, fine with me - but please make sure we mention this change in the breaking changes 🙂

I understand that there is one item, but let's do this in the meantime it is not yet refactored to a single source:

Suggested change

	Source: sources[0],
	Data: rawData,
	Source: strings.Join(sources, ","),
	Data: rawData,

Thanks!

[huseyinbabal]

This is good suggestion :)

[pkosiec]

Works well! Just minor code-related suggestions.

[pkosiec]

Maybe it could stay as SendEvent, as we don't send message, but raw event data? What do you think?

[pkosiec]

botNotifiers []notifier.Bot is redundant aa we have this bot map. This map type can be changed to

bots = map[string]notifier.Bot{}

and essentially pass it everywhere instead of the slice. But we can do it later - can you change the type and add this TODO at least? Thanks!

Suggested change

	bots = map[string]bot.Bot{}
	// TODO: Use bots everywhere instead of `botNotifiers` slice
	bots = map[string]notifier.Bot{}

[huseyinbabal]

bot.Bot refers to bot implementation while bot.Notifier is for notification. When I compare with main, we have bots and notifiers

[pkosiec]

Already discussed f2f 🙂 tl;dr for others: we use bots map for sendHelp where we need only the SendMessageToAll

[pkosiec]

I just checked and this is duplicated code for bots. Can we unify it? At least the reporting part can be extracted to a separate method with all the error wrapping

For example something like this:

func (d *Dispatcher) dispatchMsg(ctx context.Context, event source.Event, dispatch PluginDispatch) {
	var (
		pluginName = dispatch.pluginName
		sources    = []string{dispatch.sourceName}
	)

	for _, n := range d.getBotNotifiers(dispatch) {
		go func(n notifier.Bot) {
			defer analytics.ReportPanicIfOccurs(d.log, d.reporter)
			msg := interactive.CoreMessage{
				Message: event.Message,
			}
			err := n.SendMessage(ctx, msg, sources)
			if err != nil {
				reportErr := d.reportError(err, n, pluginName, event)
				if reportErr != nil {
					err = multierror.Append(err, fmt.Errorf("while reporting error: %w", reportErr))
				}

				d.log.Errorf("while sending bot message: %s", err.Error())
				// TODO(Huseyin): Shouldn't we have return here? Currently we don't, which is weird...
			}

			reportErr := d.reportSuccess(ctx, n, pluginName, event, dispatch.sourceName)
			if reportErr != nil {
				d.log.Error(err)
			}
		}(n)
	}

	for _, n := range d.sinkNotifiers {
		go func(n notifier.Sink) {
			defer analytics.ReportPanicIfOccurs(d.log, d.reporter)
			err := n.SendMessage(ctx, event.RawObject, sources)
			if err != nil {
				reportErr := d.reportError(err, n, pluginName, event)
				if reportErr != nil {
					err = multierror.Append(err, fmt.Errorf("while reporting error: %w", reportErr))
				}

				d.log.Errorf("while sending sink message: %s", err.Error())
				// TODO(Huseyin): Shouldn't we have return here? Currently we don't, which is weird...
			}

			reportErr := d.reportSuccess(ctx, n, pluginName, event, dispatch.sourceName)
			if reportErr != nil {
				d.log.Error(err)
			}
		}(n)
	}

	// execute actions
	actions, err := d.actionProvider.RenderedActions(event.RawObject, sources)
	if err != nil {
		d.log.Errorf("while rendering automated actions: %s", err.Error())
		return
	}
	for _, act := range actions {
		d.log.Infof("Executing action %q (command: %q)...", act.DisplayName, act.Command)
		genericMsg := d.actionProvider.ExecuteAction(ctx, act)
		for _, n := range d.getBotNotifiers(dispatch) {
			go func(n notifier.Bot) {
				defer analytics.ReportPanicIfOccurs(d.log, d.reporter)
				err := n.SendMessage(ctx, genericMsg, sources)
				if err != nil {
					d.log.Errorf("while sending action result message: %s", err.Error())
				}
			}(n)
		}
	}
}

type genericNotifier interface {
	IntegrationName() config.CommPlatformIntegration
	Type() config.IntegrationType
}

func (d *Dispatcher) reportSuccess(ctx context.Context, n genericNotifier, pluginName string, event source.Event, sourceName string) error {
	errs := multierror.New()
	reportErr := d.reporter.ReportHandledEventSuccess(analytics.ReportEvent{
		IntegrationType:       n.Type(),
		Platform:              n.IntegrationName(),
		PluginName:            pluginName,
		AnonymizedEventFields: event.AnalyticsLabels,
	})
	if reportErr != nil {
		errs = multierror.Append(errs, fmt.Errorf("while reporting %s analytics: %w", n.Type(), reportErr))
	}
	if err := d.reportAudit(ctx, pluginName, fmt.Sprintf("%v", event.RawObject), sourceName); err != nil {
		errs = multierror.Append(errs, fmt.Errorf("while reporting %s audit event: %w", n.Type(), reportErr))
	}
	
	return errs.ErrorOrNil()
}

func (d *Dispatcher) reportError(err error, n genericNotifier, pluginName string, event source.Event) error {
	reportErr := d.reporter.ReportHandledEventError(analytics.ReportEvent{
		IntegrationType:       n.Type(),
		Platform:              n.IntegrationName(),
		PluginName:            pluginName,
		AnonymizedEventFields: event.AnalyticsLabels,
	}, err)
	if reportErr != nil {
		return fmt.Errorf("while reporting %s analytics: %w", n.Type(), reportErr)
	}
	
	return nil
}

[pkosiec]

Update: after discussion let's go with this snippet:

func (d *Dispatcher) dispatchMsg(ctx context.Context, event source.Event, dispatch PluginDispatch) {
	var (
		pluginName = dispatch.pluginName
		sources    = []string{dispatch.sourceName}
	)

	for _, n := range d.getBotNotifiers(dispatch) {
		go func(n notifier.Bot) {
			defer analytics.ReportPanicIfOccurs(d.log, d.reporter)
			msg := interactive.CoreMessage{
				Message: event.Message,
			}
			err := n.SendMessage(ctx, msg, sources)
			if err != nil {
				reportErr := d.reportError(ctx, err, n, pluginName, event, dispatch.sourceName
				if reportErr != nil {
					err = multierror.Append(err, fmt.Errorf("while reporting error: %w", reportErr))
				}

				d.log.Errorf("while sending bot message: %s", err.Error())
				return
			}

			reportErr := d.reportSuccess(ctx, n, pluginName, event, dispatch.sourceName)
			if reportErr != nil {
				d.log.Error(err)
			}
		}(n)
	}

	for _, n := range d.sinkNotifiers {
		go func(n notifier.Sink) {
			defer analytics.ReportPanicIfOccurs(d.log, d.reporter)
			err := n.SendMessage(ctx, event.RawObject, sources)
			if err != nil {
				reportErr := d.reportError(ctx, err, n, pluginName, event, dispatch.sourceName)
				if reportErr != nil {
					err = multierror.Append(err, fmt.Errorf("while reporting error: %w", reportErr))
				}

				d.log.Errorf("while sending sink message: %s", err.Error())
				return
			}

			reportErr := d.reportSuccess(ctx, n, pluginName, event, dispatch.sourceName)
			if reportErr != nil {
				d.log.Error(err)
			}
		}(n)
	}

	// execute actions
	actions, err := d.actionProvider.RenderedActions(event.RawObject, sources)
	if err != nil {
		d.log.Errorf("while rendering automated actions: %s", err.Error())
		return
	}
	for _, act := range actions {
		d.log.Infof("Executing action %q (command: %q)...", act.DisplayName, act.Command)
		genericMsg := d.actionProvider.ExecuteAction(ctx, act)
		for _, n := range d.getBotNotifiers(dispatch) {
			go func(n notifier.Bot) {
				defer analytics.ReportPanicIfOccurs(d.log, d.reporter)
				err := n.SendMessage(ctx, genericMsg, sources)
				if err != nil {
					d.log.Errorf("while sending action result message: %s", err.Error())
				}
			}(n)
		}
	}
}

type genericNotifier interface {
	IntegrationName() config.CommPlatformIntegration
	Type() config.IntegrationType
}

func (d *Dispatcher) reportSuccess(ctx context.Context, n genericNotifier, pluginName string, event source.Event, sourceName string) error {
	errs := multierror.New()
	reportErr := d.reporter.ReportHandledEventSuccess(analytics.ReportEvent{
		IntegrationType:       n.Type(),
		Platform:              n.IntegrationName(),
		PluginName:            pluginName,
		AnonymizedEventFields: event.AnalyticsLabels,
	})
	if reportErr != nil {
		errs = multierror.Append(errs, fmt.Errorf("while reporting %s analytics: %w", n.Type(), reportErr))
	}
	if err := d.reportAudit(ctx, pluginName, fmt.Sprintf("%v", event.RawObject), sourceName); err != nil {
		errs = multierror.Append(errs, fmt.Errorf("while reporting %s audit event: %w", n.Type(), reportErr))
	}

	return errs.ErrorOrNil()
}

func (d *Dispatcher) reportError(ctx context.Context, err error, n genericNotifier, pluginName string, event source.Event, sourceName string) error {
	errs := multierror.New()
	reportErr := d.reporter.ReportHandledEventError(analytics.ReportEvent{
		IntegrationType:       n.Type(),
		Platform:              n.IntegrationName(),
		PluginName:            pluginName,
		AnonymizedEventFields: event.AnalyticsLabels,
	}, err)
	if reportErr != nil {
		errs = multierror.Append(errs, fmt.Errorf("while reporting %s analytics: %w", n.Type(), reportErr))
	}
	// TODO: add additional metadata about event failed to send
	if err := d.reportAudit(ctx, pluginName, fmt.Sprintf("%v", event.RawObject), sourceName); err != nil {
		errs = multierror.Append(errs, fmt.Errorf("while reporting %s audit event: %w", n.Type(), reportErr))
	}

	return nil
}

[huseyinbabal]

@pkosiec addressed all suggestions and created a ticket for audit event metadata https://github.com/kubeshop/botkube-cloud/issues/203