feat: report deployments to datadog (#1244)

When enabled in the helm chart `dogstatsdMetrics.eventsEnabled: true`, this sends datadog-events for each deployment. Both manual deployments and release trains are taken into account.

Reference: SRX-PLTW20

---------

Co-authored-by: Sven Urbanski <[email protected]>

charts/kuberpult/templates/cd-service.yaml

@@ -150,6 +150,8 @@ spec:
 {{- if .Values.dogstatsdMetrics.enabled }}
         - name: KUBERPULT_ENABLE_METRICS
           value: "{{ .Values.dogstatsdMetrics.enabled }}"
+        - name: KUBERPULT_ENABLE_EVENTS
+          value: "{{ .Values.dogstatsdMetrics.eventsEnabled }}"
         - name: KUBERPULT_DOGSTATSD_ADDR
           value: "{{ .Values.dogstatsdMetrics.address }}"
 {{- end }}

charts/kuberpult/values.yaml.tpl

@@ -155,7 +155,13 @@ datadogTracing:
   environment: "shared"
 
 dogstatsdMetrics:
+  # send metrics:
   enabled: false
+
+  # sends additional events for each deployments:
+  # dogstatsdMetrics.enabled must be true for this to have an effect.
+  eventsEnabled: false
+
   #  dogstatsD listens on port udp:8125 by default.
   #  https://docs.datadoghq.com/developers/dogstatsd/?tab=hostagent#agent
   #  datadog.dogstatsd.socketPath -- Path to the DogStatsD socket

services/cd-service/pkg/cmd/server.go

@@ -58,6 +58,7 @@ type Config struct {
 	DexRbacPolicyPath string        `split_words:"true"`
 	EnableTracing     bool          `default:"false" split_words:"true"`
 	EnableMetrics     bool          `default:"false" split_words:"true"`
+	EnableEvents      bool          `default:"false" split_words:"true"`
 	DogstatsdAddr     string        `default:"127.0.0.1:8125" split_words:"true"`
 	EnableSqlite      bool          `default:"true" split_words:"true"`
 	DexMock           bool          `default:"false" split_words:"true"`
@@ -164,6 +165,7 @@ func RunServer() {
 			ArgoWebhookUrl:         c.ArgoCdServer,
 			WebURL:                 c.GitWebUrl,
 			NetworkTimeout:         c.GitNetworkTimeout,
+			DogstatsdEvents:        c.EnableMetrics,
 		}
 		repo, repoQueue, err := repository.New2(ctx, cfg)
 		if err != nil {

services/cd-service/pkg/repository/repository.go

@@ -76,7 +76,7 @@ func defaultBackOffProvider() backoff.BackOff {
 }
 
 var (
-	ddMetrics *statsd.Client
+	ddMetrics statsd.ClientInterface
 )
 
 type StorageBackend int
@@ -136,7 +136,8 @@ type RepositoryConfig struct {
 	// if set, kuberpult will generate push events to argoCd whenever it writes to the manifest repo:
 	ArgoWebhookUrl string
 	// the url to the git repo, like the browser requires it (https protocol)
-	WebURL string
+	WebURL          string
+	DogstatsdEvents bool
 }
 
 func openOrCreate(path string, storageBackend StorageBackend) (*git.Repository, error) {
@@ -260,7 +261,7 @@ func New2(ctx context.Context, cfg RepositoryConfig) (Repository, setup.Backgrou
 
 	ddMetricsFromCtx := ctx.Value("ddMetrics")
 	if ddMetricsFromCtx != nil {
-		ddMetrics = ddMetricsFromCtx.(*statsd.Client)
+		ddMetrics = ddMetricsFromCtx.(statsd.ClientInterface)
 	}
 
 	if cfg.Branch == "" {
@@ -541,7 +542,7 @@ func (r *repository) ProcessQueueOnce(ctx context.Context, e element, callback P
 				return
 			}
 			// Apply the items
-			elements, err, _ = r.applyElements(elements, false)
+			elements, err, changes = r.applyElements(elements, false)
 			if err != nil || len(elements) == 0 {
 				return
 			}
@@ -562,6 +563,15 @@ func (r *repository) ProcessQueueOnce(ctx context.Context, e element, callback P
 	span, ctx := tracer.StartSpanFromContext(e.ctx, "PostPush")
 	defer span.Finish()
 
+	ddSpan, ctx := tracer.StartSpanFromContext(ctx, "SendMetrics")
+	if r.config.DogstatsdEvents {
+		ddError := UpdateDatadogMetrics(r.State(), changes)
+		if ddError != nil {
+			logger.Warn(fmt.Sprintf("Could not send datadog metrics/events %v", ddError))
+		}
+	}
+	ddSpan.Finish()
+
 	if r.config.ArgoWebhookUrl != "" {
 		r.sendWebhookToArgoCd(ctx, logger, changes)
 	}
@@ -778,8 +788,9 @@ func (r *repository) ApplyTransformersInternal(ctx context.Context, transformers
 }
 
 type AppEnv struct {
-	App string
-	Env string
+	App  string
+	Env  string
+	Team string
 }
 
 type RootApp struct {
@@ -798,10 +809,11 @@ type CommitIds struct {
 	Current  *git.Oid
 }
 
-func (r *TransformerResult) AddAppEnv(app string, env string) {
+func (r *TransformerResult) AddAppEnv(app string, env string, team string) {
 	r.ChangedApps = append(r.ChangedApps, AppEnv{
-		App: app,
-		Env: env,
+		App:  app,
+		Env:  env,
+		Team: team,
 	})
 }
 
@@ -817,7 +829,7 @@ func (r *TransformerResult) Combine(other *TransformerResult) {
 	}
 	for i := range other.ChangedApps {
 		a := other.ChangedApps[i]
-		r.AddAppEnv(a.App, a.Env)
+		r.AddAppEnv(a.App, a.Env, a.Team)
 	}
 	for i := range other.DeletedRootApps {
 		a := other.DeletedRootApps[i]
@@ -841,10 +853,6 @@ func (r *repository) ApplyTransformers(ctx context.Context, transformers ...Tran
 	if err != nil {
 		return nil, err
 	}
-	err = UpdateDatadogMetrics(state)
-	if err != nil {
-		return nil, err
-	}
 	if err := r.afterTransform(ctx, *state); err != nil {
 		return nil, grpc.InternalError(ctx, fmt.Errorf("%s: %w", "failure in afterTransform", err))
 	}

services/cd-service/pkg/repository/transformer.go

@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/DataDog/datadog-go/v5/statsd"
 	"github.com/freiheit-com/kuberpult/pkg/logger"
 	"io"
 	"io/fs"
@@ -111,8 +112,8 @@ func GaugeEnvAppLockMetric(fs billy.Filesystem, env, app string) {
 	}
 }
 
-func UpdateDatadogMetrics(state *State) error {
-	fs := state.Filesystem
+func UpdateDatadogMetrics(state *State, changes *TransformerResult) error {
+	filesystem := state.Filesystem
 	if ddMetrics == nil {
 		return nil
 	}
@@ -121,11 +122,37 @@ func UpdateDatadogMetrics(state *State) error {
 		return err
 	}
 	for env := range configs {
-		GaugeEnvLockMetric(fs, env)
-		appsDir := fs.Join(environmentDirectory(fs, env), "applications")
-		if entries, _ := fs.ReadDir(appsDir); entries != nil {
+		GaugeEnvLockMetric(filesystem, env)
+		appsDir := filesystem.Join(environmentDirectory(filesystem, env), "applications")
+		if entries, _ := filesystem.ReadDir(appsDir); entries != nil {
 			for _, app := range entries {
-				GaugeEnvAppLockMetric(fs, env, app.Name())
+				GaugeEnvAppLockMetric(filesystem, env, app.Name())
+			}
+		}
+	}
+	now := time.Now() // ensure all events have the same timestamp
+	if changes != nil && ddMetrics != nil {
+		for i := range changes.ChangedApps {
+			oneChange := changes.ChangedApps[i]
+			teamMessage := func() string {
+				if oneChange.Team != "" {
+					return fmt.Sprintf(" for team %s", oneChange.Team)
+				}
+				return ""
+			}()
+			event := statsd.Event{
+				Title:     "Kuberpult app deployed",
+				Text:      fmt.Sprintf("Kuberpult has deployed %s to %s%s", oneChange.App, oneChange.Env, teamMessage),
+				Timestamp: now,
+				Tags: []string{
+					"kuberpult.application:" + oneChange.App,
+					"kuberpult.environment:" + oneChange.Env,
+					"kuberpult.team:" + oneChange.Team,
+				},
+			}
+			err := ddMetrics.Event(&event)
+			if err != nil {
+				return err
 			}
 		}
 	}
@@ -144,7 +171,7 @@ func RegularlySendDatadogMetrics(repo Repository, interval time.Duration, callBa
 
 func GetRepositoryStateAndUpdateMetrics(repo Repository) {
 	repoState := repo.State()
-	if err := UpdateDatadogMetrics(repoState); err != nil {
+	if err := UpdateDatadogMetrics(repoState, nil); err != nil {
 		panic(err.Error())
 	}
 }
@@ -288,7 +315,11 @@ func (c *CreateApplicationVersion) Transform(ctx context.Context, state *State)
 		if err := util.WriteFile(fs, fs.Join(envDir, "manifests.yaml"), []byte(man), 0666); err != nil {
 			return "", nil, err
 		}
-		changes.AddAppEnv(c.Application, env)
+		teamOwner, err := state.GetApplicationTeamOwner(c.Application)
+		if err != nil {
+			return "", nil, err
+		}
+		changes.AddAppEnv(c.Application, env, teamOwner)
 		if hasUpstream && config.Upstream.Latest && isLatest {
 			d := &DeployApplicationVersion{
 				Environment:    env,
@@ -405,7 +436,11 @@ func (c *CreateUndeployApplicationVersion) Transform(ctx context.Context, state
 		if err := util.WriteFile(fs, fs.Join(envDir, "manifests.yaml"), []byte(" "), 0666); err != nil {
 			return "", nil, err
 		}
-		changes.AddAppEnv(c.Application, env)
+		teamOwner, err := state.GetApplicationTeamOwner(c.Application)
+		if err != nil {
+			return "", nil, err
+		}
+		changes.AddAppEnv(c.Application, env, teamOwner)
 		if hasUpstream && config.Upstream.Latest {
 			d := &DeployApplicationVersion{
 				Environment: env,
@@ -497,7 +532,11 @@ func (u *UndeployApplication) Transform(ctx context.Context, state *State) (stri
 	changes := &TransformerResult{}
 	for env := range configs {
 		appDir := environmentApplicationDirectory(fs, env, u.Application)
-		changes.AddAppEnv(u.Application, env)
+		teamOwner, err := state.GetApplicationTeamOwner(u.Application)
+		if err != nil {
+			return "", nil, err
+		}
+		changes.AddAppEnv(u.Application, env, teamOwner)
 		// remove environment application
 		if err := fs.Remove(appDir); err != nil && !errors.Is(err, os.ErrNotExist) {
 			return "", nil, fmt.Errorf("UndeployApplication: unexpected error application '%v' environment '%v': '%w'", u.Application, env, err)
@@ -1121,8 +1160,11 @@ func (c *DeployApplicationVersion) Transform(ctx context.Context, state *State)
 	if err := util.WriteFile(fs, manifestFilename, manifestContent, 0666); err != nil {
 		return "", nil, err
 	}
-	changes.AddAppEnv(c.Application, c.Environment)
-	logger.FromContext(ctx).Info(fmt.Sprintf("DeployApp: changes added: %v+", changes))
+	teamOwner, err := state.GetApplicationTeamOwner(c.Application)
+	if err != nil {
+		return "", nil, err
+	}
+	changes.AddAppEnv(c.Application, c.Environment, teamOwner)
 
 	user, err := auth.ReadUserFromContext(ctx)
 	if err != nil {