Send telemetry events when a cluster agent mutates a remote config

.github/CODEOWNERS

@@ -220,6 +220,7 @@
 /pkg/obfuscate/                         @DataDog/agent-apm
 /pkg/trace/                             @DataDog/agent-apm
 /pkg/trace/api/otlp*.go                 @DataDog/opentelemetry
+/pkg/trace/telemetry/                   @DataDog/telemetry-and-analytics
 /pkg/autodiscovery/                     @DataDog/container-integrations @DataDog/agent-metrics-logs
 /pkg/autodiscovery/listeners/           @DataDog/container-integrations
 /pkg/autodiscovery/listeners/cloudfoundry*.go  @DataDog/platform-integrations
@@ -231,6 +232,7 @@
 /pkg/cloudfoundry                       @Datadog/platform-integrations
 /pkg/clusteragent/                      @DataDog/container-integrations
 /pkg/clusteragent/orchestrator/         @DataDog/container-app
+/pkg/clusteragent/telemetry/            @DataDog/telemetry-and-analytics
 /pkg/collector/                         @DataDog/agent-metrics-logs
 /pkg/collector/corechecks/cluster/      @DataDog/container-integrations
 /pkg/collector/corechecks/cluster/orchestrator  @DataDog/container-app

pkg/clusteragent/admission/patch/patch_request.go

@@ -12,6 +12,7 @@ import (
 	"fmt"
 
 	"github.com/DataDog/datadog-agent/pkg/clusteragent/admission/common"
+	"github.com/DataDog/datadog-agent/pkg/clusteragent/telemetry"
 )
 
 // TargetObjKind represents the supported k8s object kinds
@@ -38,6 +39,7 @@ const (
 type PatchRequest struct {
 	ID            string `json:"id"`
 	Revision      int64  `json:"revision"`
+	RcVersion     uint64 `json:"rc_version"`
 	SchemaVersion string `json:"schema_version"`
 	Action        Action `json:"action"`
 
@@ -59,6 +61,37 @@ func (pr PatchRequest) Validate(clusterName string) error {
 	return pr.K8sTarget.validate(clusterName)
 }
 
+func (pr PatchRequest) getApmRemoteConfigEvent(err error, errorCode int) telemetry.ApmRemoteConfigEvent {
+	env := ""
+	if pr.LibConfig.Env != nil {
+		env = *pr.LibConfig.Env
+	}
+	errorMessage := ""
+	if err != nil {
+		errorMessage = err.Error()
+	}
+	return telemetry.ApmRemoteConfigEvent{
+		RequestType: "apm-remote-config-event",
+		ApiVersion:  "v2",
+		Payload: telemetry.ApmRemoteConfigEventPayload{
+			Tags: telemetry.ApmRemoteConfigEventTags{
+				Env:                 env,
+				RcId:                pr.ID,
+				RcRevision:          pr.Revision,
+				RcVersion:           pr.RcVersion,
+				KubernetesCluster:   pr.K8sTarget.Cluster,
+				KubernetesNamespace: pr.K8sTarget.Namespace,
+				KubernetesKind:      string(pr.K8sTarget.Kind),
+				KubernetesName:      pr.K8sTarget.Name,
+			},
+			Error: telemetry.ApmRemoteConfigEventError{
+				Code:    errorCode,
+				Message: errorMessage,
+			},
+		},
+	}
+}
+
 // K8sTarget represent the targetet k8s object
 type K8sTarget struct {
 	Cluster   string        `json:"cluster"`

pkg/clusteragent/admission/patch/patcher.go

@@ -11,9 +11,9 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-
 	"github.com/DataDog/datadog-agent/pkg/clusteragent/admission/common"
 	"github.com/DataDog/datadog-agent/pkg/clusteragent/admission/metrics"
+	"github.com/DataDog/datadog-agent/pkg/clusteragent/telemetry"
 	k8sutil "github.com/DataDog/datadog-agent/pkg/util/kubernetes"
 	"github.com/DataDog/datadog-agent/pkg/util/log"
 
@@ -25,16 +25,18 @@ import (
 )
 
 type patcher struct {
-	k8sClient        kubernetes.Interface
-	isLeader         func() bool
-	deploymentsQueue chan PatchRequest
+	k8sClient          kubernetes.Interface
+	isLeader           func() bool
+	deploymentsQueue   chan PatchRequest
+	telemetryCollector telemetry.TelemetryCollector
 }
 
-func newPatcher(k8sClient kubernetes.Interface, isLeaderFunc func() bool, pp patchProvider) *patcher {
+func newPatcher(k8sClient kubernetes.Interface, isLeaderFunc func() bool, telemetryCollector telemetry.TelemetryCollector, pp patchProvider) *patcher {
 	return &patcher{
-		k8sClient:        k8sClient,
-		isLeader:         isLeaderFunc,
-		deploymentsQueue: pp.subscribe(KindDeployment),
+		k8sClient:          k8sClient,
+		isLeader:           isLeaderFunc,
+		deploymentsQueue:   pp.subscribe(KindDeployment),
+		telemetryCollector: telemetryCollector,
 	}
 }
 
@@ -102,8 +104,10 @@ func (p *patcher) patchDeployment(req PatchRequest) error {
 	}
 	log.Infof("Patching %s with patch %s", req.K8sTarget, string(patch))
 	if _, err = p.k8sClient.AppsV1().Deployments(req.K8sTarget.Namespace).Patch(context.TODO(), req.K8sTarget.Name, types.StrategicMergePatchType, patch, metav1.PatchOptions{}); err != nil {
+		p.telemetryCollector.SendRemoteConfigMutateEvent(req.getApmRemoteConfigEvent(err, telemetry.FailedToMutateConfig))
 		return err
 	}
+	p.telemetryCollector.SendRemoteConfigMutateEvent(req.getApmRemoteConfigEvent(nil, telemetry.Success))
 	metrics.PatchCompleted.Inc()
 	return nil
 }

pkg/clusteragent/admission/patch/patcher_test.go

@@ -12,6 +12,7 @@ import (
 	"testing"
 
 	"github.com/DataDog/datadog-agent/pkg/clusteragent/admission/common"
+	"github.com/DataDog/datadog-agent/pkg/clusteragent/telemetry"
 
 	"github.com/stretchr/testify/require"
 	appsv1 "k8s.io/api/apps/v1"
@@ -34,8 +35,9 @@ func TestPatchDeployment(t *testing.T) {
 
 	// Create patcher
 	p := patcher{
-		k8sClient: client,
-		isLeader:  func() bool { return true },
+		k8sClient:          client,
+		isLeader:           func() bool { return true },
+		telemetryCollector: telemetry.NewNoopCollector(),
 	}
 
 	// Create request skeleton

pkg/clusteragent/admission/patch/provider.go

@@ -10,6 +10,7 @@ package patch
 import (
 	"errors"
 
+	"github.com/DataDog/datadog-agent/pkg/clusteragent/telemetry"
 	"github.com/DataDog/datadog-agent/pkg/config"
 	"github.com/DataDog/datadog-agent/pkg/config/remote"
 )
@@ -19,9 +20,9 @@ type patchProvider interface {
 	subscribe(kind TargetObjKind) chan PatchRequest
 }
 
-func newPatchProvider(rcClient *remote.Client, isLeaderNotif <-chan struct{}, clusterId string, clusterName string) (patchProvider, error) {
+func newPatchProvider(rcClient *remote.Client, isLeaderNotif <-chan struct{}, telemetryCollector telemetry.TelemetryCollector, clusterName string) (patchProvider, error) {
 	if config.Datadog.GetBool("remote_configuration.enabled") {
-		return newRemoteConfigProvider(rcClient, isLeaderNotif, clusterId, clusterName)
+		return newRemoteConfigProvider(rcClient, isLeaderNotif, telemetryCollector, clusterName)
 	}
 	if config.Datadog.GetBool("admission_controller.auto_instrumentation.patcher.fallback_to_file_provider") {
 		// Use the file config provider for e2e testing only (it replaces RC as a source of configs)

pkg/clusteragent/admission/patch/rc_provider.go

@@ -23,24 +23,22 @@ type remoteConfigProvider struct {
 	client             *remote.Client
 	isLeaderNotif      <-chan struct{}
 	subscribers        map[TargetObjKind]chan PatchRequest
-	clusterId          string
 	clusterName        string
 	telemetryCollector telemetry.TelemetryCollector
 }
 
 var _ patchProvider = &remoteConfigProvider{}
 
-func newRemoteConfigProvider(client *remote.Client, isLeaderNotif <-chan struct{}, clusterId string, clusterName string) (*remoteConfigProvider, error) {
+func newRemoteConfigProvider(client *remote.Client, isLeaderNotif <-chan struct{}, telemetryCollector telemetry.TelemetryCollector, clusterName string) (*remoteConfigProvider, error) {
 	if client == nil {
 		return nil, errors.New("remote config client not initialized")
 	}
 	return &remoteConfigProvider{
 		client:             client,
 		isLeaderNotif:      isLeaderNotif,
 		subscribers:        make(map[TargetObjKind]chan PatchRequest),
-		clusterId:          clusterId,
 		clusterName:        clusterName,
-		telemetryCollector: telemetry.NewCollector(),
+		telemetryCollector: telemetryCollector,
 	}, nil
 }
 
@@ -77,52 +75,26 @@ func (rcp *remoteConfigProvider) process(update map[string]state.APMTracingConfi
 		err := json.Unmarshal(config.Config, &req)
 		if err != nil {
 			invalid++
+			rcp.telemetryCollector.SendRemoteConfigPatchEvent(req.getApmRemoteConfigEvent(err, telemetry.ConfigParseFailure))
 			log.Errorf("Error while parsing config: %v", err)
 			continue
 		}
+		req.RcVersion = config.Metadata.Version
 		log.Debugf("Patch request parsed %+v", req)
 		if err := req.Validate(rcp.clusterName); err != nil {
 			invalid++
+			rcp.telemetryCollector.SendRemoteConfigPatchEvent(req.getApmRemoteConfigEvent(err, telemetry.InvalidPatchRequest))
 			log.Errorf("Skipping invalid patch request: %s", err)
 			continue
 		}
 		if ch, found := rcp.subscribers[req.K8sTarget.Kind]; found {
 			valid++
 			// Log a telemetry event indicating a remote config patch to the Datadog backend
-			patchEvent := rcp.getRemoteConfigPatchEvent(config, req)
-			rcp.telemetryCollector.SendEvent(&patchEvent)
+			rcp.telemetryCollector.SendRemoteConfigPatchEvent(req.getApmRemoteConfigEvent(nil, telemetry.Success))
 			log.Debugf("Publishing patch request for target %s", req.K8sTarget)
 			ch <- req
 		}
 	}
 	metrics.RemoteConfigs.Set(valid)
 	metrics.InvalidRemoteConfigs.Set(invalid)
 }
-
-// getRemoteConfigPatchEvent fills out the fields of a telemetry event that can be sent
-// to the Datadog backend to indicate that a remote config has been successfully patched
-func (rcp *remoteConfigProvider) getRemoteConfigPatchEvent(config state.APMTracingConfig, req PatchRequest) telemetry.ApmRemoteConfigEvent {
-	env := ""
-	if req.LibConfig.Env != nil {
-		env = *req.LibConfig.Env
-	}
-	return telemetry.ApmRemoteConfigEvent{
-		RequestType: "apm-remote-config-event",
-		ApiVersion:  "v2",
-		Payload: telemetry.ApmRemoteConfigEventPayload{
-			EventName: "agent.k8s.patch",
-			Tags: telemetry.ApmRemoteConfigEventTags{
-				Env:                 env,
-				RcId:                req.ID,
-				RcClientId:          rcp.client.ID,
-				RcRevision:          req.Revision,
-				RcVersion:           config.Metadata.Version,
-				KubernetesClusterId: rcp.clusterId,
-				KubernetesCluster:   req.K8sTarget.Cluster,
-				KubernetesNamespace: req.K8sTarget.Namespace,
-				KubernetesKind:      string(req.K8sTarget.Kind),
-				KubernetesName:      req.K8sTarget.Name,
-			},
-		},
-	}
-}

pkg/clusteragent/admission/patch/rc_provider_test.go

@@ -9,6 +9,7 @@ package patch
 
 import (
 	"fmt"
+	"github.com/DataDog/datadog-agent/pkg/clusteragent/telemetry"
 	"testing"
 
 	"github.com/DataDog/datadog-agent/pkg/config/remote"
@@ -38,7 +39,7 @@ func TestProcess(t *testing.T) {
 `
 		return []byte(fmt.Sprintf(base, cluster, kind))
 	}
-	rcp, err := newRemoteConfigProvider(&remote.Client{}, make(chan struct{}), "0090c771-add5-4313-948f-d1ab99b471d6", "dev")
+	rcp, err := newRemoteConfigProvider(&remote.Client{}, make(chan struct{}), telemetry.NewNoopCollector(), "dev")
 	require.NoError(t, err)
 	notifs := rcp.subscribe(KindDeployment)
 	in := map[string]state.APMTracingConfig{

pkg/clusteragent/admission/patch/start.go

@@ -8,6 +8,7 @@
 package patch
 
 import (
+	"github.com/DataDog/datadog-agent/pkg/clusteragent/telemetry"
 	"github.com/DataDog/datadog-agent/pkg/config/remote"
 	"github.com/DataDog/datadog-agent/pkg/util/log"
 
@@ -28,11 +29,12 @@ type ControllerContext struct {
 // StartControllers starts the patch controllers
 func StartControllers(ctx ControllerContext) error {
 	log.Info("Starting patch controllers")
-	provider, err := newPatchProvider(ctx.RcClient, ctx.LeaderSubscribeFunc(), ctx.ClusterId, ctx.ClusterName)
+	telemetryCollector := telemetry.NewCollector(ctx.RcClient.ID, ctx.ClusterId)
+	provider, err := newPatchProvider(ctx.RcClient, ctx.LeaderSubscribeFunc(), telemetryCollector, ctx.ClusterName)
 	if err != nil {
 		return err
 	}
-	patcher := newPatcher(ctx.K8sClient, ctx.IsLeaderFunc, provider)
+	patcher := newPatcher(ctx.K8sClient, ctx.IsLeaderFunc, telemetryCollector, provider)
 	go provider.start(ctx.StopCh)
 	go patcher.start(ctx.StopCh)
 	return nil