feat: cluster name for events

api/v1/testkube.yaml

@@ -4746,7 +4746,7 @@ components:
             app: "backend"
 
     Event:
-      description: CRD based executor data
+      description: Event data
       type: object
       required:
         - type
@@ -4768,6 +4768,9 @@ components:
           $ref: "#/components/schemas/Execution"
         testSuiteExecution:
           $ref: "#/components/schemas/TestSuiteExecution"
+        clusterName:
+          type: string
+          description: cluster name of event      
 
     EventResource:
       type: string

cmd/api-server/main.go

@@ -286,7 +286,7 @@ func main() {
 		log.DefaultLogger.Errorw("error creating NATS connection", "error", err)
 	}
 	eventBus := bus.NewNATSBus(nc)
-	eventsEmitter := event.NewEmitter(eventBus)
+	eventsEmitter := event.NewEmitter(eventBus, cfg.TestkubeClusterName)
 
 	metrics := metrics.NewMetrics()
 
@@ -401,7 +401,17 @@ func main() {
 	if mode == common.ModeAgent {
 		log.DefaultLogger.Info("starting agent service")
 
-		agentHandle, err := agent.NewAgent(log.DefaultLogger, api.Mux.Handler(), cfg.TestkubeCloudAPIKey, grpcClient, cfg.TestkubeCloudWorkerCount, cfg.TestkubeCloudLogStreamWorkerCount, api.GetLogsStream, clusterId)
+		agentHandle, err := agent.NewAgent(
+			log.DefaultLogger,
+			api.Mux.Handler(),
+			cfg.TestkubeCloudAPIKey,
+			grpcClient,
+			cfg.TestkubeCloudWorkerCount,
+			cfg.TestkubeCloudLogStreamWorkerCount,
+			api.GetLogsStream,
+			clusterId,
+			cfg.TestkubeClusterName,
+		)
 		if err != nil {
 			ui.ExitOnError("Starting agent", err)
 		}
@@ -560,7 +570,7 @@ func newSlackLoader(cfg *config.Config) (*slack.SlackLoader, error) {
 		return nil, err
 	}
 
-	return slack.NewSlackLoader(slackTemplate, slackConfig, testkube.AllEventTypes), nil
+	return slack.NewSlackLoader(slackTemplate, slackConfig, cfg.TestkubeClusterName, testkube.AllEventTypes), nil
 }
 
 func loadFromBase64StringOrFile(base64Val string, configDir, filename, configType string) (raw string, err error) {

docs/docs/articles/helm-chart.md

@@ -114,6 +114,7 @@ The following Helm defaults are used in the `testkube` chart:
 | testkube-api.logs.bucket             | no          | "testkube-logs"                      |
 | testkube-api.cdeventsTarget          | yes         | ""                                   |
 | testkube-api.dashboardUri            | yes         | ""                                   |
+| testkube-api.clusterName             | yes         | ""                                   |
 
 >For more configuration parameters of a `MongoDB` chart please visit: 
 <https://github.com/bitnami/charts/tree/master/bitnami/mongodb#parameters>

internal/config/config.go

@@ -61,6 +61,7 @@ type Config struct {
 	CDEventsTarget                    string        `envconfig:"CDEVENTS_TARGET" default:""`
 	TestkubeDashboardURI              string        `envconfig:"TESTKUBE_DASHBOARD_URI" default:""`
 	DisableReconciler                 bool          `envconfig:"DISABLE_RECONCILER" default:"false"`
+	TestkubeClusterName               string        `envconfig:"TESTKUBE_CLUSTER_NAME" default:""`
 }
 
 func Get() (*Config, error) {

pkg/agent/agent.go

@@ -66,7 +66,8 @@ type Agent struct {
 	receiveTimeout      time.Duration
 	healthcheckInterval time.Duration
 
-	clusterID string
+	clusterID   string
+	clusterName string
 }
 
 func NewAgent(logger *zap.SugaredLogger,
@@ -77,6 +78,7 @@ func NewAgent(logger *zap.SugaredLogger,
 	logStreamWorkerCount int,
 	logStreamFunc func(ctx context.Context, executionID string) (chan output.Output, error),
 	clusterID string,
+	clusterName string,
 ) (*Agent, error) {
 	return &Agent{
 		handler:                 handler,

pkg/agent/agent_test.go

@@ -56,7 +56,7 @@ func TestCommandExecution(t *testing.T) {
 	var logStreamFunc func(ctx context.Context, executionID string) (chan output.Output, error)
 
 	logger, _ := zap.NewDevelopment()
-	agent, err := agent.NewAgent(logger.Sugar(), m, "api-key", grpcClient, 5, 5, logStreamFunc, "")
+	agent, err := agent.NewAgent(logger.Sugar(), m, "api-key", grpcClient, 5, 5, logStreamFunc, "", "")
 	if err != nil {
 		t.Fatal(err)
 	}

pkg/agent/events.go

@@ -48,6 +48,7 @@ func (ag *Agent) Metadata() map[string]string {
 }
 
 func (ag *Agent) Notify(event testkube.Event) (result testkube.EventResult) {
+	event.ClusterName = ag.clusterName
 	// Non blocking send
 	select {
 	case ag.events <- event:

pkg/agent/events_test.go

@@ -52,7 +52,7 @@ func TestEventLoop(t *testing.T) {
 	grpcClient := cloud.NewTestKubeCloudAPIClient(grpcConn)
 
 	var logStreamFunc func(ctx context.Context, executionID string) (chan output.Output, error)
-	agent, err := agent.NewAgent(logger.Sugar(), nil, "api-key", grpcClient, 5, 5, logStreamFunc, "")
+	agent, err := agent.NewAgent(logger.Sugar(), nil, "api-key", grpcClient, 5, 5, logStreamFunc, "", "")
 	assert.NoError(t, err)
 	go func() {
 		l, err := agent.Load()

pkg/agent/logs_test.go

@@ -63,7 +63,7 @@ func TestLogStream(t *testing.T) {
 	}
 
 	logger, _ := zap.NewDevelopment()
-	agent, err := agent.NewAgent(logger.Sugar(), m, "api-key", grpcClient, 5, 5, logStreamFunc, "")
+	agent, err := agent.NewAgent(logger.Sugar(), m, "api-key", grpcClient, 5, 5, logStreamFunc, "", "")
 	if err != nil {
 		t.Fatal(err)
 	}

pkg/api/v1/testkube/model_event.go

@@ -9,7 +9,7 @@
  */
 package testkube
 
-// CRD based executor data
+// Event data
 type Event struct {
 	// UUID of event
 	Id       string         `json:"id"`
@@ -19,4 +19,6 @@ type Event struct {
 	Type_              *EventType          `json:"type"`
 	TestExecution      *Execution          `json:"testExecution,omitempty"`
 	TestSuiteExecution *TestSuiteExecution `json:"testSuiteExecution,omitempty"`
+	// cluster name of event
+	ClusterName string `json:"clusterName,omitempty"`
 }

pkg/event/emitter.go

@@ -20,24 +20,26 @@ const (
 )
 
 // NewEmitter returns new emitter instance
-func NewEmitter(eventBus bus.Bus) *Emitter {
+func NewEmitter(eventBus bus.Bus, clusterName string) *Emitter {
 	return &Emitter{
-		Results:   make(chan testkube.EventResult, eventsBuffer),
-		Log:       log.DefaultLogger,
-		Loader:    NewLoader(),
-		Bus:       eventBus,
-		Listeners: make(common.Listeners, 0),
+		Results:     make(chan testkube.EventResult, eventsBuffer),
+		Log:         log.DefaultLogger,
+		Loader:      NewLoader(),
+		Bus:         eventBus,
+		Listeners:   make(common.Listeners, 0),
+		ClusterName: clusterName,
 	}
 }
 
 // Emitter handles events emitting for webhooks
 type Emitter struct {
-	Results   chan testkube.EventResult
-	Listeners common.Listeners
-	Loader    *Loader
-	Log       *zap.SugaredLogger
-	mutex     sync.Mutex
-	Bus       bus.Bus
+	Results     chan testkube.EventResult
+	Listeners   common.Listeners
+	Loader      *Loader
+	Log         *zap.SugaredLogger
+	mutex       sync.Mutex
+	Bus         bus.Bus
+	ClusterName string
 }
 
 // Register adds new listener
@@ -124,6 +126,7 @@ func (e *Emitter) UpdateListeners(listeners common.Listeners) {
 
 // Notify notifies emitter with webhook
 func (e *Emitter) Notify(event testkube.Event) {
+	event.ClusterName = e.ClusterName
 	err := e.Bus.PublishTopic(event.Topic(), event)
 	e.Log.Infow("event published", append(event.Log(), "error", err)...)
 }

pkg/event/emitter_integration_test.go

@@ -24,7 +24,7 @@ func GetTestNATSEmitter() *Emitter {
 	if err != nil {
 		panic(err)
 	}
-	return NewEmitter(bus.NewNATSBus(nc))
+	return NewEmitter(bus.NewNATSBus(nc), "")
 }
 
 func TestEmitter_NATS_Register_Integration(t *testing.T) {

pkg/event/emitter_test.go

@@ -25,7 +25,7 @@ func TestEmitter_Register(t *testing.T) {
 		t.Parallel()
 		// given
 		eventBus := bus.NewEventBusMock()
-		emitter := NewEmitter(eventBus)
+		emitter := NewEmitter(eventBus, "")
 		// when
 		emitter.Register(&dummy.DummyListener{Id: "l1"})
 
@@ -43,7 +43,7 @@ func TestEmitter_Listen(t *testing.T) {
 		t.Parallel()
 		// given
 		eventBus := bus.NewEventBusMock()
-		emitter := NewEmitter(eventBus)
+		emitter := NewEmitter(eventBus, "")
 		// given listener with matching selector
 		listener1 := &dummy.DummyListener{Id: "l1", SelectorString: "type=listener1"}
 		// and listener with second matic selector
@@ -97,7 +97,7 @@ func TestEmitter_Notify(t *testing.T) {
 		t.Parallel()
 		// given
 		eventBus := bus.NewEventBusMock()
-		emitter := NewEmitter(eventBus)
+		emitter := NewEmitter(eventBus, "")
 		// and 2 listeners subscribed to the same queue
 		// * first on pod1
 		listener1 := &dummy.DummyListener{Id: "l3"}
@@ -131,7 +131,7 @@ func TestEmitter_Reconcile(t *testing.T) {
 		t.Parallel()
 		// given first reconciler loop was done
 		eventBus := bus.NewEventBusMock()
-		emitter := NewEmitter(eventBus)
+		emitter := NewEmitter(eventBus, "")
 		emitter.Loader.Register(&dummy.DummyLoader{IdPrefix: "dummy1"})
 		emitter.Loader.Register(&dummy.DummyLoader{IdPrefix: "dummy2"})
 
@@ -185,7 +185,7 @@ func TestEmitter_UpdateListeners(t *testing.T) {
 		t.Parallel()
 		// given
 		eventBus := bus.NewEventBusMock()
-		emitter := NewEmitter(eventBus)
+		emitter := NewEmitter(eventBus, "")
 		// given listener with matching selector
 		listener1 := &dummy.DummyListener{Id: "l1", SelectorString: "type=listener1"}
 		// and listener with second matching selector

pkg/event/kind/slack/loader.go

@@ -13,12 +13,12 @@ import (
 
 var _ common.ListenerLoader = (*SlackLoader)(nil)
 
-func NewSlackLoader(messageTemplate, configString string, events []testkube.EventType) *SlackLoader {
+func NewSlackLoader(messageTemplate, configString, clusterName string, events []testkube.EventType) *SlackLoader {
 	var config []slack.NotificationsConfig
 	if err := json.Unmarshal([]byte(configString), &config); err != nil {
 		log.DefaultLogger.Errorw("error unmarshalling slack config", "error", err)
 	}
-	slackNotifier := slack.NewNotifier(messageTemplate, config)
+	slackNotifier := slack.NewNotifier(messageTemplate, clusterName, config)
 	return &SlackLoader{
 		Log:           log.DefaultLogger,
 		events:        events,

pkg/event/kind/slack/loader_test.go

@@ -14,7 +14,7 @@ func TestSlackLoader_Load(t *testing.T) {
 		t.Parallel()
 		// given
 		// default slack notifier is not ready by default
-		l := NewSlackLoader("", "", testkube.AllEventTypes)
+		l := NewSlackLoader("", "", "", testkube.AllEventTypes)
 
 		// when
 		listeners, err := l.Load()

pkg/slack/slack.go

@@ -26,18 +26,20 @@ type MessageArgs struct {
 	StartTime     string
 	EndTime       string
 	Duration      string
+	ClusterName   string
 }
 
 type Notifier struct {
 	client          *slack.Client
 	timestamps      map[string]string
 	Ready           bool
 	messageTemplate string
+	clusterName     string
 	config          *Config
 }
 
-func NewNotifier(template string, config []NotificationsConfig) *Notifier {
-	notifier := Notifier{messageTemplate: template, config: NewConfig(config)}
+func NewNotifier(template, clusterName string, config []NotificationsConfig) *Notifier {
+	notifier := Notifier{messageTemplate: template, clusterName: clusterName, config: NewConfig(config)}
 	notifier.timestamps = make(map[string]string)
 	if token, ok := os.LookupEnv("SLACK_TOKEN"); ok {
 		log.DefaultLogger.Infow("initializing slack client", "SLACK_TOKEN", token)
@@ -182,6 +184,7 @@ func (s *Notifier) composeTestsuiteMessage(execution *testkube.TestSuiteExecutio
 		Duration:      execution.Duration,
 		TotalSteps:    len(execution.ExecuteStepResults),
 		FailedSteps:   execution.FailedStepsCount(),
+		ClusterName:   s.clusterName,
 	}
 
 	log.DefaultLogger.Infow("Execution changed", "status", execution.Status)
@@ -216,6 +219,7 @@ func (s *Notifier) composeTestMessage(execution *testkube.Execution, eventType t
 		Duration:      execution.Duration,
 		TotalSteps:    len(execution.ExecutionResult.Steps),
 		FailedSteps:   execution.ExecutionResult.FailedStepsCount(),
+		ClusterName:   s.clusterName,
 	}
 
 	log.DefaultLogger.Infow("Execution changed", "status", execution.ExecutionResult.Status)

pkg/triggers/executor_test.go

@@ -51,7 +51,7 @@ func TestExecute(t *testing.T) {
 
 	mockExecutor := client.NewMockExecutor(mockCtrl)
 
-	mockEventEmitter := event.NewEmitter(bus.NewEventBusMock())
+	mockEventEmitter := event.NewEmitter(bus.NewEventBusMock(), "")
 
 	mockTest := testsv3.Test{
 		ObjectMeta: metav1.ObjectMeta{Namespace: "testkube", Name: "some-test"},

pkg/triggers/service_test.go

@@ -57,7 +57,7 @@ func TestService_Run(t *testing.T) {
 
 	mockExecutor := client.NewMockExecutor(mockCtrl)
 
-	mockEventEmitter := event.NewEmitter(bus.NewEventBusMock())
+	mockEventEmitter := event.NewEmitter(bus.NewEventBusMock(), "")
 
 	mockTest := testsv3.Test{
 		ObjectMeta: metav1.ObjectMeta{Namespace: "testkube", Name: "some-test"},