feat(server): support event properties header (#3179)

server/analytics/analytics.go

@@ -1,6 +1,7 @@
 package analytics
 
 import (
+	"encoding/json"
 	"fmt"
 	"os"
 )
@@ -67,3 +68,21 @@ func SendEvent(name, category, clientID string, data *map[string]string) error {
 
 	return err
 }
+
+func InjectProperties(data map[string]string, properties string) map[string]string {
+	if properties == "" {
+		return data
+	}
+
+	var p map[string]string
+	err := json.Unmarshal([]byte(properties), &p)
+	if err != nil {
+		return data
+	}
+
+	for k, v := range p {
+		data[k] = v
+	}
+
+	return data
+}

server/app/app.go

@@ -370,11 +370,15 @@ func analyticsMW(next http.Handler) http.Handler {
 		routeName := mux.CurrentRoute(r).GetName()
 		machineID := r.Header.Get("x-client-id")
 		source := r.Header.Get("x-source")
+		eventProperties := r.Header.Get("x-event-properties")
+
+		eventData := map[string]string{
+			"source": source,
+		}
+		eventData = analytics.InjectProperties(eventData, eventProperties)
 
 		if routeName != "" {
-			analytics.SendEvent(toWords(routeName), "test", machineID, &map[string]string{
-				"source": source,
-			})
+			analytics.SendEvent(toWords(routeName), "test", machineID, &eventData)
 		}
 
 		next.ServeHTTP(w, r)

server/executor/assertion_runner.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/kubeshop/tracetest/server/analytics"
 	"github.com/kubeshop/tracetest/server/expression"
+	"github.com/kubeshop/tracetest/server/http/middleware"
 	"github.com/kubeshop/tracetest/server/model/events"
 	"github.com/kubeshop/tracetest/server/pkg/maps"
 	"github.com/kubeshop/tracetest/server/pkg/pipeline"
@@ -80,6 +81,7 @@ func (e *defaultAssertionRunner) runAssertionsAndUpdateResult(ctx context.Contex
 		run = run.AssertionFailed(err)
 		analytics.SendEvent("test_run_finished", "error", "", &map[string]string{
 			"finalState": string(run.State),
+			"tenant_id":  middleware.TenantIDFromContext(ctx),
 		})
 
 		return test.Run{}, e.updater.Update(ctx, run)
@@ -139,6 +141,7 @@ func (e *defaultAssertionRunner) executeAssertions(ctx context.Context, req Job)
 
 	analytics.SendEvent("test_run_finished", "successful", "", &map[string]string{
 		"finalState": string(run.State),
+		"tenant_id":  middleware.TenantIDFromContext(ctx),
 	})
 
 	return run, nil

server/executor/linter_runner.go

@@ -6,6 +6,7 @@ import (
 	"log"
 
 	"github.com/kubeshop/tracetest/server/analytics"
+	"github.com/kubeshop/tracetest/server/http/middleware"
 	"github.com/kubeshop/tracetest/server/linter"
 	"github.com/kubeshop/tracetest/server/linter/analyzer"
 	"github.com/kubeshop/tracetest/server/model/events"
@@ -131,6 +132,7 @@ func (e *defaultLinterRunner) onError(ctx context.Context, job Job, run test.Run
 
 	analytics.SendEvent("test_run_finished", "error", "", &map[string]string{
 		"finalState": string(run.State),
+		"tenant_id":  middleware.TenantIDFromContext(ctx),
 	})
 
 	run = run.LinterError(err)

server/executor/trace_poller.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/kubeshop/tracetest/server/analytics"
 	"github.com/kubeshop/tracetest/server/executor/pollingprofile"
+	"github.com/kubeshop/tracetest/server/http/middleware"
 	"github.com/kubeshop/tracetest/server/model/events"
 	"github.com/kubeshop/tracetest/server/pkg/pipeline"
 	"github.com/kubeshop/tracetest/server/subscription"
@@ -204,6 +205,7 @@ func (tp tracePoller) handleTraceDBError(ctx context.Context, job Job, err error
 	run = run.TraceFailed(err)
 	analytics.SendEvent("test_run_finished", "error", "", &map[string]string{
 		"finalState": string(run.State),
+		"tenant_id":  middleware.TenantIDFromContext(ctx),
 	})
 
 	tp.handleDBError(tp.updater.Update(ctx, run))

server/executor/tracepollerworker/evaluator_worker.go

@@ -9,6 +9,7 @@ import (
 	"github.com/kubeshop/tracetest/server/analytics"
 	"github.com/kubeshop/tracetest/server/datastore"
 	"github.com/kubeshop/tracetest/server/executor"
+	"github.com/kubeshop/tracetest/server/http/middleware"
 	"github.com/kubeshop/tracetest/server/model/events"
 	"github.com/kubeshop/tracetest/server/pkg/pipeline"
 	"github.com/kubeshop/tracetest/server/resourcemanager"
@@ -98,6 +99,7 @@ func (w *tracePollerEvaluatorWorker) ProcessItem(ctx context.Context, job execut
 		run := job.Run.TraceFailed(err)
 		analytics.SendEvent("test_run_finished", "error", "", &map[string]string{
 			"finalState": string(run.State),
+			"tenant_id":  middleware.TenantIDFromContext(ctx),
 		})
 
 		handleDBError(w.state.updater.Update(ctx, run))

server/executor/trigger_result_processor_worker.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 
 	"github.com/kubeshop/tracetest/server/analytics"
+	"github.com/kubeshop/tracetest/server/http/middleware"
 	"github.com/kubeshop/tracetest/server/model"
 	"github.com/kubeshop/tracetest/server/model/events"
 	"github.com/kubeshop/tracetest/server/pkg/pipeline"
@@ -61,7 +62,7 @@ func (r triggerResultProcessorWorker) ProcessItem(ctx context.Context, job Job)
 	ctx, pollingSpan := r.tracer.Start(ctx, "Start processing trigger response")
 	defer pollingSpan.End()
 
-	job.Run = r.handleExecutionResult(job.Run)
+	job.Run = r.handleExecutionResult(job.Run, ctx)
 	triggerResult := job.Run.TriggerResult
 	if triggerResult.Error != nil {
 		err := triggerResult.Error.Error()
@@ -120,13 +121,14 @@ func (r triggerResultProcessorWorker) emitMismatchEndpointEvent(ctx context.Cont
 	}
 }
 
-func (r triggerResultProcessorWorker) handleExecutionResult(run test.Run) test.Run {
+func (r triggerResultProcessorWorker) handleExecutionResult(run test.Run, ctx context.Context) test.Run {
 	run = run.TriggerCompleted(run.TriggerResult)
 	if run.TriggerResult.Error != nil {
 		run = run.TriggerFailed(fmt.Errorf(run.TriggerResult.Error.ErrorMessage))
 
 		analytics.SendEvent("test_run_finished", "error", "", &map[string]string{
 			"finalState": string(run.State),
+			"tenant_id":  middleware.TenantIDFromContext(ctx),
 		})
 
 		return run