diff --git a/go.mod b/go.mod index 571f1c0e60..8c522aa10c 100644 --- a/go.mod +++ b/go.mod @@ -66,6 +66,7 @@ require ( github.com/samber/lo v1.52.0 golang.org/x/exp v0.0.0-20250106191152-7588d65b2ba8 golang.org/x/text v0.39.0 + gomodules.xyz/jsonpatch/v2 v2.5.0 google.golang.org/protobuf v1.36.11 helm.sh/helm/v3 v3.20.2 k8s.io/kubectl v0.35.3 @@ -280,7 +281,6 @@ require ( golang.org/x/time v0.14.0 // indirect golang.org/x/tools v0.47.0 // indirect golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect - gomodules.xyz/jsonpatch/v2 v2.5.0 // indirect google.golang.org/genproto v0.0.0-20240903143218-8af14fe29dc1 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect diff --git a/internal/controller/datadogagent/experiment.go b/internal/controller/datadogagent/experiment.go index 4f54d2166a..398bec5bbf 100644 --- a/internal/controller/datadogagent/experiment.go +++ b/internal/controller/datadogagent/experiment.go @@ -11,9 +11,9 @@ import ( "encoding/json" "fmt" "maps" - "strings" "time" + jsonpatch "gomodules.xyz/jsonpatch/v2" appsv1 "k8s.io/api/apps/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" @@ -21,6 +21,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" v2alpha1 "github.com/DataDog/datadog-operator/api/datadoghq/v2alpha1" + "github.com/DataDog/datadog-operator/pkg/kubernetes" ) // ExperimentDefaultTimeout is the duration after which a running experiment is automatically rolled back. @@ -347,28 +348,15 @@ func (r *Reconciler) processPromoteSignal( return true, nil } -// annotationToJSONPatchPath converts an annotation key to a JSON Patch path -// under /metadata/annotations, escaping "/" as "~1" per RFC 6901. -func annotationToJSONPatchPath(key string) string { - return "/metadata/annotations/" + strings.ReplaceAll(key, "/", "~1") -} - -// jsonPatchOp represents a single JSON Patch operation (RFC 6902). -type jsonPatchOp struct { - Op string `json:"op"` - Path string `json:"path"` - Value string `json:"value,omitempty"` -} - // clearExperimentAnnotations removes the experiment signal annotations from the // DDA using a conditional JSON Patch. The patch asserts the annotation ID matches // the one we just processed, preventing accidental removal of a newer signal // written concurrently by the daemon. func (r *Reconciler) clearExperimentAnnotations(ctx context.Context, instance *v2alpha1.DatadogAgent, expectedID string) error { - ops := []jsonPatchOp{ - {Op: "test", Path: annotationToJSONPatchPath(v2alpha1.AnnotationExperimentID), Value: expectedID}, - {Op: "remove", Path: annotationToJSONPatchPath(v2alpha1.AnnotationExperimentSignal)}, - {Op: "remove", Path: annotationToJSONPatchPath(v2alpha1.AnnotationExperimentID)}, + ops := []jsonpatch.Operation{ + {Operation: "test", Path: kubernetes.AnnotationJSONPatchPath(v2alpha1.AnnotationExperimentID), Value: expectedID}, + {Operation: "remove", Path: kubernetes.AnnotationJSONPatchPath(v2alpha1.AnnotationExperimentSignal)}, + {Operation: "remove", Path: kubernetes.AnnotationJSONPatchPath(v2alpha1.AnnotationExperimentID)}, } patch, err := json.Marshal(ops) if err != nil { diff --git a/pkg/fleet/daemon_operations.go b/pkg/fleet/daemon_operations.go index df08b42291..94bee0ae74 100644 --- a/pkg/fleet/daemon_operations.go +++ b/pkg/fleet/daemon_operations.go @@ -7,13 +7,14 @@ package fleet import ( "context" - "encoding/json" "fmt" + appsv1 "k8s.io/api/apps/v1" "k8s.io/apimachinery/pkg/types" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" + apicommon "github.com/DataDog/datadog-operator/api/datadoghq/common" v2alpha1 "github.com/DataDog/datadog-operator/api/datadoghq/v2alpha1" ) @@ -42,13 +43,26 @@ func (d *Daemon) resolveOperation(req remoteAPIRequest, signal experimentSignal) if len(cfg.Operations) != 1 { return resolvedOperation{}, fmt.Errorf("%s: config %s must have exactly 1 operation, got %d", signal, cfg.ID, len(cfg.Operations)) } - if cfg.Operations[0].Operation != OperationUpdate { - return resolvedOperation{}, fmt.Errorf("%s: invalid operation: %s", signal, cfg.Operations[0].Operation) + + op := cfg.Operations[0] + config := op.Config + switch op.Operation { + case OperationReplace: + spec, err := extractReplaceSpec(op.Config) + if err != nil { + return resolvedOperation{}, fmt.Errorf("%s: invalid replace config: %w", signal, err) + } + config = spec + case OperationUpdate: + // config is already op.Config. + default: + return resolvedOperation{}, fmt.Errorf("%s: invalid operation: %s", signal, op.Operation) } return resolvedOperation{ NamespacedName: req.Params.NamespacedName, - Config: cfg.Operations[0].Config, + Operation: op.Operation, + Config: config, }, nil } @@ -111,57 +125,32 @@ func (d *Daemon) guardPendingOperationSlot(annotations map[string]string, nsn ty } } -func (d *Daemon) applyOperation(ctx context.Context, nsn types.NamespacedName, signalLog string, pending *pendingOperation, patch []byte) (*pendingOperation, error) { - if pending == nil && len(patch) == 0 { +func (d *Daemon) applyOperation(ctx context.Context, signal experimentSignal, pending *pendingOperation, sp signalPatch) (*pendingOperation, error) { + if pending == nil { + // A nil pending always means there's nothing to patch. return nil, nil } - if pending != nil { - var patchMap map[string]any - if len(patch) != 0 { - if err := json.Unmarshal(patch, &patchMap); err != nil { - return nil, fmt.Errorf("%s: failed to unmarshal base patch: %w", signalLog, err) - } - } else { - patchMap = make(map[string]any) - } - - metadata, ok := patchMap["metadata"].(map[string]any) - if !ok { - metadata = make(map[string]any) - patchMap["metadata"] = metadata - } - annotations, ok := metadata["annotations"].(map[string]any) - if !ok { - annotations = make(map[string]any) - metadata["annotations"] = annotations - } - // Write the pending task in the same patch as the signal. If the daemon - // restarts, the worker can read these annotations and continue. - annotations[v2alpha1.AnnotationPendingTaskID] = pending.taskID - annotations[v2alpha1.AnnotationPendingAction] = string(pending.intent) - annotations[v2alpha1.AnnotationPendingExperimentID] = pending.experimentID - annotations[v2alpha1.AnnotationPendingPackage] = pending.packageName - if pending.resultVersion != "" { - annotations[v2alpha1.AnnotationPendingResultVersion] = pending.resultVersion - } else { - // Clear any old promote result version. Merge patch leaves keys alone - // when they are omitted. - annotations[v2alpha1.AnnotationPendingResultVersion] = nil - } - var err error - patch, err = json.Marshal(patchMap) - if err != nil { - return nil, fmt.Errorf("%s: failed to build pending operation patch: %w", signalLog, err) - } + // Write the pending task in the same patch as the signal, whatever patch type + // sp is. If the daemon restarts, the worker can read these annotations and + // continue. + patch, err := injectPendingAnnotations(sp, pending) + if err != nil { + return nil, fmt.Errorf("%s: %w", signal, err) } + dda := &v2alpha1.DatadogAgent{} - dda.Name = nsn.Name - dda.Namespace = nsn.Namespace + dda.Name = pending.nsn.Name + dda.Namespace = pending.nsn.Namespace if err := retryWithBackoff(ctx, func() error { - return d.client.Patch(ctx, dda, client.RawPatch(types.MergePatchType, patch), client.FieldOwner("fleet-daemon")) + // Strict field validation makes the API server reject unrecognized fields + // instead of silently pruning them. This matters most for replace: without + // it, a spec with fields the installed CRD doesn't know about (e.g. version + // skew between fleet and the operator) would get pruned down to an empty + // spec and accepted, wiping the resource without any error. + return d.client.Patch(ctx, dda, client.RawPatch(sp.Type, patch), client.FieldOwner("fleet-daemon"), client.FieldValidation("Strict")) }); err != nil { - return nil, fmt.Errorf("%s: failed to patch DatadogAgent: %w", signalLog, err) + return nil, fmt.Errorf("%s: failed to patch DatadogAgent: %w", signal, err) } ctrl.LoggerFrom(ctx).Info("Wrote signal") return pending, nil @@ -188,14 +177,14 @@ func (d *Daemon) startDatadogAgentExperiment(ctx context.Context, req remoteAPIR return nil, err } logger.Info("Prepared DatadogAgent experiment start signal") - return d.applyOperation(ctx, op.NamespacedName, "start DatadogAgent experiment", pending, patch) + return d.applyOperation(ctx, signalStartDatadogAgentExperiment, pending, patch) } // stopDatadogAgentExperiment writes a rollback signal annotation on the DDA. // If the phase is already terminal, the patch is skipped. After writing, the // status worker waits for any terminal phase before marking the task done. func (d *Daemon) stopDatadogAgentExperiment(ctx context.Context, req remoteAPIRequest) (*pendingOperation, error) { - op, err := d.resolveOperation(req, "stop DatadogAgent experiment") + op, err := d.resolveOperation(req, signalStopDatadogAgentExperiment) if err != nil { return nil, err } @@ -208,7 +197,7 @@ func (d *Daemon) stopDatadogAgentExperiment(ctx context.Context, req remoteAPIRe return nil, err } logger.Info("Prepared DatadogAgent experiment stop signal") - return d.applyOperation(ctx, op.NamespacedName, "stop DatadogAgent experiment", pending, patch) + return d.applyOperation(ctx, signalStopDatadogAgentExperiment, pending, patch) } // promoteDatadogAgentExperiment writes a promote signal annotation on the DDA. @@ -228,49 +217,83 @@ func (d *Daemon) promoteDatadogAgentExperiment(ctx context.Context, req remoteAP return nil, err } logger.Info("Prepared DatadogAgent experiment promote signal") - return d.applyOperation(ctx, op.NamespacedName, "promote DatadogAgent experiment", pending, patch) + return d.applyOperation(ctx, signalPromoteDatadogAgentExperiment, pending, patch) } -func (d *Daemon) planStart(ctx context.Context, req remoteAPIRequest, op resolvedOperation) (*pendingOperation, []byte, error) { +func (d *Daemon) planStart(ctx context.Context, req remoteAPIRequest, op resolvedOperation) (*pendingOperation, signalPatch, error) { experimentID := req.Params.Version pending := d.newPendingOperation(pendingIntentStart, req, op.NamespacedName, experimentID) dda := &v2alpha1.DatadogAgent{} if err := d.client.Get(ctx, op.NamespacedName, dda); err != nil { - return nil, nil, fmt.Errorf("start DatadogAgent experiment: failed to get DatadogAgent: %w", err) + return nil, signalPatch{}, fmt.Errorf("%s: failed to get DatadogAgent: %w", signalStartDatadogAgentExperiment, err) } if experimentHasPhase(dda, experimentID, v2alpha1.ExperimentPhaseRunning) { // The controller already started this experiment. Update RC now and let // handleTask mark the task done. stable, _ := d.getPackageConfigVersions(req.Package) d.setPackageConfigVersions(req.Package, stable, req.Params.Version) - return nil, nil, nil + return nil, signalPatch{}, nil } if dda.Annotations[v2alpha1.AnnotationExperimentID] == experimentID { // The start signal is already on the DDA. Keep the same signal, but make // sure the pending task annotations exist. if err := d.guardPendingOperationSlot(dda.Annotations, op.NamespacedName, *pending); err != nil { - return nil, nil, err + return nil, signalPatch{}, err } - return pending, nil, nil + return pending, mergePatch(nil), nil } if runningID := runningExperimentID(dda); runningID != "" { - return nil, nil, fmt.Errorf("start DatadogAgent experiment: experiment %q already running", runningID) + return nil, signalPatch{}, fmt.Errorf("%s: experiment %q already running", signalStartDatadogAgentExperiment, runningID) } // Do not overwrite another unfinished task. if err := d.guardPendingOperationSlot(dda.Annotations, op.NamespacedName, *pending); err != nil { - return nil, nil, err + return nil, signalPatch{}, err + } + if op.Operation == OperationReplace { + // If the operator has never reconciled this DDA, there's no + // ControllerRevision to roll back to, so stopping the experiment later + // would have nothing to restore. Require a baseline revision first. + hasBaseline, err := d.hasControllerRevision(ctx, dda) + if err != nil { + return nil, signalPatch{}, fmt.Errorf("%s: %w", signalStartDatadogAgentExperiment, err) + } + if !hasBaseline { + return nil, signalPatch{}, fmt.Errorf("%s: no baseline ControllerRevision exists yet for %s; wait for the operator to reconcile before starting a replace experiment", signalStartDatadogAgentExperiment, op.NamespacedName) + } + ops := buildReplaceSignalPatch(v2alpha1.ExperimentSignalStart, experimentID, op.Config, len(dda.Annotations) == 0) + return pending, jsonPatch(ops), nil } + patch, err := buildSignalPatch(v2alpha1.ExperimentSignalStart, experimentID, op.Config) if err != nil { - return nil, nil, fmt.Errorf("start DatadogAgent experiment: %w", err) + return nil, signalPatch{}, fmt.Errorf("%s: %w", signalStartDatadogAgentExperiment, err) + } + return pending, mergePatch(patch), nil +} + +// hasControllerRevision reports whether dda already owns a ControllerRevision. +func (d *Daemon) hasControllerRevision(ctx context.Context, dda *v2alpha1.DatadogAgent) (bool, error) { + revList := &appsv1.ControllerRevisionList{} + if err := d.client.List(ctx, revList, + client.InNamespace(dda.Namespace), + client.MatchingLabels{apicommon.DatadogAgentNameLabelKey: dda.Name}, + ); err != nil { + return false, fmt.Errorf("failed to list ControllerRevisions: %w", err) + } + for i := range revList.Items { + for _, ref := range revList.Items[i].OwnerReferences { + if ref.Controller != nil && *ref.Controller && ref.UID == dda.UID { + return true, nil + } + } } - return pending, patch, nil + return false, nil } -func (d *Daemon) planStop(ctx context.Context, req remoteAPIRequest, op resolvedOperation) (*pendingOperation, []byte, error) { +func (d *Daemon) planStop(ctx context.Context, req remoteAPIRequest, op resolvedOperation) (*pendingOperation, signalPatch, error) { dda := &v2alpha1.DatadogAgent{} if getErr := d.client.Get(ctx, op.NamespacedName, dda); getErr != nil { - return nil, nil, fmt.Errorf("stop DatadogAgent experiment: failed to get DatadogAgent: %w", getErr) + return nil, signalPatch{}, fmt.Errorf("%s: failed to get DatadogAgent: %w", signalStopDatadogAgentExperiment, getErr) } // Stop requests intentionally do not use params.version as the experiment @@ -289,47 +312,47 @@ func (d *Daemon) planStop(ctx context.Context, req remoteAPIRequest, op resolved if experimentID == "" || dda.Annotations[v2alpha1.AnnotationExperimentSignal] != v2alpha1.ExperimentSignalStart { // Nothing is running and there is no start signal to roll back. d.clearExperimentConfigVersion(req.Package) - return nil, nil, nil + return nil, signalPatch{}, nil } } else { if isTerminalPhase(dda.Status.Experiment.Phase) { // The experiment is already stopped/promoted/aborted. d.clearExperimentConfigVersion(req.Package) - return nil, nil, nil + return nil, signalPatch{}, nil } switch dda.Status.Experiment.Phase { case v2alpha1.ExperimentPhaseRunning: if experimentID == "" { - return nil, nil, fmt.Errorf("stop DatadogAgent experiment: running experiment is missing an ID") + return nil, signalPatch{}, fmt.Errorf("%s: running experiment is missing an ID", signalStopDatadogAgentExperiment) } case "": // Start was requested, but the reconciler has not written a phase yet. if experimentID == "" { - return nil, nil, fmt.Errorf("stop DatadogAgent experiment: current experiment is missing an ID") + return nil, signalPatch{}, fmt.Errorf("%s: current experiment is missing an ID", signalStopDatadogAgentExperiment) } default: - return nil, nil, fmt.Errorf("stop DatadogAgent experiment: cannot stop, current phase is %q", dda.Status.Experiment.Phase) + return nil, signalPatch{}, fmt.Errorf("%s: cannot stop, current phase is %q", signalStopDatadogAgentExperiment, dda.Status.Experiment.Phase) } } pending := d.newPendingOperation(pendingIntentStop, req, op.NamespacedName, experimentID) if err := d.guardPendingOperationSlot(dda.Annotations, op.NamespacedName, *pending); err != nil { - return nil, nil, err + return nil, signalPatch{}, err } patch, err := buildSignalPatch(v2alpha1.ExperimentSignalRollback, experimentID) if err != nil { - return nil, nil, fmt.Errorf("stop DatadogAgent experiment: %w", err) + return nil, signalPatch{}, fmt.Errorf("%s: %w", signalStopDatadogAgentExperiment, err) } - return pending, patch, nil + return pending, mergePatch(patch), nil } -func (d *Daemon) planPromote(ctx context.Context, req remoteAPIRequest, op resolvedOperation) (*pendingOperation, []byte, error) { +func (d *Daemon) planPromote(ctx context.Context, req remoteAPIRequest, op resolvedOperation) (*pendingOperation, signalPatch, error) { _, experiment := d.getPackageConfigVersions(req.Package) if experiment == "" { - return nil, nil, fmt.Errorf("promote DatadogAgent experiment: no experiment config version set") + return nil, signalPatch{}, fmt.Errorf("%s: no experiment config version set", signalPromoteDatadogAgentExperiment) } dda := &v2alpha1.DatadogAgent{} if err := d.client.Get(ctx, op.NamespacedName, dda); err != nil { - return nil, nil, fmt.Errorf("promote DatadogAgent experiment: failed to get DatadogAgent: %w", err) + return nil, signalPatch{}, fmt.Errorf("%s: failed to get DatadogAgent: %w", signalPromoteDatadogAgentExperiment, err) } // Promote requests intentionally do not use params.version as the experiment @@ -348,24 +371,24 @@ func (d *Daemon) planPromote(ctx context.Context, req remoteAPIRequest, op resol // Promotion already happened. Update RC now and let handleTask mark the // task done. d.setPackageConfigVersions(req.Package, experiment, "") - return nil, nil, nil + return nil, signalPatch{}, nil } if !experimentHasPhase(dda, experimentID, v2alpha1.ExperimentPhaseRunning) { currentPhase := "" if dda.Status.Experiment != nil { currentPhase = string(dda.Status.Experiment.Phase) } - return nil, nil, fmt.Errorf("promote DatadogAgent experiment: cannot promote, current phase is %q", currentPhase) + return nil, signalPatch{}, fmt.Errorf("%s: cannot promote, current phase is %q", signalPromoteDatadogAgentExperiment, currentPhase) } pending := d.newPendingOperation(pendingIntentPromote, req, op.NamespacedName, experimentID) // Promote makes the current experiment config the stable config on success. pending.resultVersion = experiment if err := d.guardPendingOperationSlot(dda.Annotations, op.NamespacedName, *pending); err != nil { - return nil, nil, err + return nil, signalPatch{}, err } patch, err := buildSignalPatch(v2alpha1.ExperimentSignalPromote, experimentID) if err != nil { - return nil, nil, fmt.Errorf("promote DatadogAgent experiment: %w", err) + return nil, signalPatch{}, fmt.Errorf("%s: %w", signalPromoteDatadogAgentExperiment, err) } - return pending, patch, nil + return pending, mergePatch(patch), nil } diff --git a/pkg/fleet/daemon_test.go b/pkg/fleet/daemon_test.go index 8b0c49ae36..cb083f782c 100644 --- a/pkg/fleet/daemon_test.go +++ b/pkg/fleet/daemon_test.go @@ -16,6 +16,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "google.golang.org/protobuf/proto" + appsv1 "k8s.io/api/apps/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" @@ -23,7 +24,9 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" + apicommon "github.com/DataDog/datadog-operator/api/datadoghq/common" v2alpha1 "github.com/DataDog/datadog-operator/api/datadoghq/v2alpha1" + "github.com/DataDog/datadog-operator/pkg/kubernetes" ) // --- Test helpers --- @@ -31,15 +34,19 @@ import ( func testFleetScheme() *runtime.Scheme { s := runtime.NewScheme() _ = v2alpha1.AddToScheme(s) + _ = appsv1.AddToScheme(s) return s } -func testDaemon(dda *v2alpha1.DatadogAgent, configs map[string]installerConfig) (*Daemon, client.Client) { +func testDaemon(dda *v2alpha1.DatadogAgent, configs map[string]installerConfig, extraObjs ...client.Object) (*Daemon, client.Client) { s := testFleetScheme() b := fake.NewClientBuilder().WithScheme(s).WithStatusSubresource(&v2alpha1.DatadogAgent{}) if dda != nil { b = b.WithObjects(dda) } + if len(extraObjs) > 0 { + b = b.WithObjects(extraObjs...) + } c := b.Build() return &Daemon{ client: c, @@ -60,11 +67,14 @@ var testDDANSN = types.NamespacedName{Namespace: "datadog", Name: "datadog-agent const testExperimentID = "test-config" +const testDDAUID = types.UID("test-dda-uid") + func testDDAObject(phase v2alpha1.ExperimentPhase) *v2alpha1.DatadogAgent { dda := &v2alpha1.DatadogAgent{ ObjectMeta: metav1.ObjectMeta{ Name: testDDANSN.Name, Namespace: testDDANSN.Namespace, + UID: testDDAUID, }, } if phase != "" { @@ -137,6 +147,22 @@ func testInstallerConfigWithDDA() map[string]installerConfig { } } +// testControllerRevisionForDDA builds a ControllerRevision owned by dda, as if +// the operator had already reconciled it once. +func testControllerRevisionForDDA(dda *v2alpha1.DatadogAgent) *appsv1.ControllerRevision { + isController := true + return &appsv1.ControllerRevision{ + ObjectMeta: metav1.ObjectMeta{ + Name: dda.Name + "-baseline", + Namespace: dda.Namespace, + Labels: map[string]string{apicommon.DatadogAgentNameLabelKey: dda.Name}, + OwnerReferences: []metav1.OwnerReference{ + {UID: dda.UID, Controller: &isController}, + }, + }, + } +} + func testStartRequest() remoteAPIRequest { return remoteAPIRequest{ ID: "exp-abc", @@ -810,6 +836,291 @@ func TestBuildSignalPatch_WithoutConfig(t *testing.T) { assert.Equal(t, "exp-123", annotations[v2alpha1.AnnotationExperimentID]) } +// --- operation:replace tests --- + +func testReplaceConfig(spec string) map[string]installerConfig { + return map[string]installerConfig{ + "test-config": { + ID: "test-config", + Operations: []fleetManagementOperation{ + { + Operation: OperationReplace, + Config: json.RawMessage(`{"spec":` + spec + `}`), + }, + }, + }, + } +} + +func TestResolveOperation_Replace_RejectsMissingSpec(t *testing.T) { + d, _ := testDaemon(testDDAObject(""), map[string]installerConfig{ + "test-config": { + ID: "test-config", + Operations: []fleetManagementOperation{ + {Operation: OperationReplace, Config: json.RawMessage(`{"other":{}}`)}, + }, + }, + }) + err := syncTaskErr(d.startDatadogAgentExperiment(context.Background(), testStartRequest())) + assert.ErrorContains(t, err, `"spec" key`) +} + +func TestResolveOperation_Replace_RejectsEmptySpec(t *testing.T) { + d, _ := testDaemon(testDDAObject(""), testReplaceConfig(`{}`)) + err := syncTaskErr(d.startDatadogAgentExperiment(context.Background(), testStartRequest())) + assert.ErrorContains(t, err, "must not be empty") +} + +func TestResolveOperation_Replace_RejectsWhitespaceOnlyEmptySpec(t *testing.T) { + // A spec that is empty modulo whitespace must be rejected the same way as + // "{}" — this used to bypass validation because the old check compared + // against the literal string "{}". + d, _ := testDaemon(testDDAObject(""), testReplaceConfig(`{ }`)) + err := syncTaskErr(d.startDatadogAgentExperiment(context.Background(), testStartRequest())) + assert.ErrorContains(t, err, "must not be empty") +} + +func TestResolveOperation_Replace_AllowsNonObjectSpec(t *testing.T) { + // extractReplaceSpec only rejects an empty spec; it doesn't type-check + // "spec" beyond that. The API server rejects a malformed spec on its own. + d, _ := testDaemon(testDDAObject(""), testReplaceConfig(`"not-an-object"`)) + op, err := d.resolveOperation(testStartRequest(), signalStartDatadogAgentExperiment) + require.NoError(t, err) + assert.Equal(t, OperationReplace, op.Operation) +} + +func TestResolveOperation_Replace_AllowsExtraTopLevelKeys(t *testing.T) { + // Extra top-level keys (e.g. "metadata") are ignored — only "spec" is used. + d, _ := testDaemon(testDDAObject(""), map[string]installerConfig{ + "test-config": { + ID: "test-config", + Operations: []fleetManagementOperation{ + { + Operation: OperationReplace, + Config: json.RawMessage(`{"spec":{"features":{}},"metadata":{"annotations":{"a":"b"}}}`), + }, + }, + }, + }) + op, err := d.resolveOperation(testStartRequest(), signalStartDatadogAgentExperiment) + require.NoError(t, err) + assert.Equal(t, OperationReplace, op.Operation) +} + +func TestBuildReplaceSignalPatch_Shape(t *testing.T) { + spec := json.RawMessage(`{"features":{"apm":{"enabled":true}}}`) + rawOps := buildReplaceSignalPatch(v2alpha1.ExperimentSignalStart, "exp-123", spec, false) + patch, err := json.Marshal(rawOps) + require.NoError(t, err) + + var ops []map[string]any + require.NoError(t, json.Unmarshal(patch, &ops)) + + // bootstrapAnnotations=false: just spec + signal + experimentID. + require.Len(t, ops, 3) + + // Replace uses no merging, so every op is an "add". + for _, op := range ops { + assert.Equal(t, "add", op["op"]) + } + + specOp := ops[0] + assert.Equal(t, "/spec", specOp["path"]) + assert.Equal(t, map[string]any{"features": map[string]any{"apm": map[string]any{"enabled": true}}}, specOp["value"]) + + assert.Equal(t, "/metadata/annotations/"+kubernetes.JSONPointerEscape(v2alpha1.AnnotationExperimentSignal), ops[1]["path"]) + assert.Equal(t, v2alpha1.ExperimentSignalStart, ops[1]["value"]) + assert.Equal(t, "/metadata/annotations/"+kubernetes.JSONPointerEscape(v2alpha1.AnnotationExperimentID), ops[2]["path"]) + assert.Equal(t, "exp-123", ops[2]["value"]) +} + +func TestBuildReplaceSignalPatch_BootstrapsAnnotations(t *testing.T) { + rawOps := buildReplaceSignalPatch(v2alpha1.ExperimentSignalStart, "exp-123", json.RawMessage(`{"features":{}}`), true) + patch, err := json.Marshal(rawOps) + require.NoError(t, err) + + var ops []map[string]any + require.NoError(t, json.Unmarshal(patch, &ops)) + require.Len(t, ops, 5) + + // "test" checks annotations are still absent before "add" creates them. + assert.Equal(t, "test", ops[0]["op"]) + assert.Equal(t, "/metadata/annotations", ops[0]["path"]) + assert.Nil(t, ops[0]["value"]) + + assert.Equal(t, "add", ops[1]["op"]) + assert.Equal(t, "/metadata/annotations", ops[1]["path"]) + assert.Equal(t, map[string]any{}, ops[1]["value"]) +} + +func TestInjectPendingAnnotations_JSONPatch(t *testing.T) { + base := buildReplaceSignalPatch(v2alpha1.ExperimentSignalStart, "exp-123", json.RawMessage(`{"features":{}}`), false) + pending := &pendingOperation{taskID: "task-1", intent: pendingIntentStart, experimentID: "exp-123", packageName: "datadog-operator"} + + patch, err := injectPendingAnnotations(jsonPatch(base), pending) + require.NoError(t, err) + + var ops []map[string]any + require.NoError(t, json.Unmarshal(patch, &ops)) + // 3 base ops (spec, signal, experimentID) + 4 pending annotation ops + + // 1 resultVersion op (written even though it's empty here). + require.Len(t, ops, 8) + + for _, op := range ops { + assert.Equal(t, "add", op["op"]) + } + + found := false + for _, op := range ops { + if op["path"] == "/metadata/annotations/"+kubernetes.JSONPointerEscape(v2alpha1.AnnotationPendingTaskID) { + found = true + assert.Equal(t, "task-1", op["value"]) + } + } + assert.True(t, found, "expected an op for the pending-task-id annotation") + + foundResultVersion := false + for _, op := range ops { + if op["path"] == "/metadata/annotations/"+kubernetes.JSONPointerEscape(v2alpha1.AnnotationPendingResultVersion) { + foundResultVersion = true + assert.Equal(t, "", op["value"]) + } + } + assert.True(t, foundResultVersion, "expected a resultVersion op even when resultVersion is unset, to clear any stale value") +} + +func TestInjectPendingAnnotations_JSONPatch_WithResultVersion(t *testing.T) { + pending := &pendingOperation{taskID: "task-1", intent: pendingIntentPromote, experimentID: "exp-123", packageName: "datadog-operator", resultVersion: "stable-2"} + + patch, err := injectPendingAnnotations(jsonPatch(nil), pending) + require.NoError(t, err) + + var ops []map[string]any + require.NoError(t, json.Unmarshal(patch, &ops)) + + found := false + for _, op := range ops { + if op["path"] == "/metadata/annotations/"+kubernetes.JSONPointerEscape(v2alpha1.AnnotationPendingResultVersion) { + found = true + assert.Equal(t, "stable-2", op["value"]) + } + } + assert.True(t, found, "expected an op for the pending-result-version annotation") +} + +func TestInjectPendingAnnotations_MergePatch(t *testing.T) { + pending := &pendingOperation{taskID: "task-1", intent: pendingIntentStart, experimentID: "exp-123", packageName: "datadog-operator"} + + patch, err := injectPendingAnnotations(mergePatch(nil), pending) + require.NoError(t, err) + + var patchMap map[string]any + require.NoError(t, json.Unmarshal(patch, &patchMap)) + annotations := patchMap["metadata"].(map[string]any)["annotations"].(map[string]any) + assert.Equal(t, "task-1", annotations[v2alpha1.AnnotationPendingTaskID]) + // Unset result version is written as an explicit null so the merge patch clears any stale value. + assert.Nil(t, annotations[v2alpha1.AnnotationPendingResultVersion]) +} + +func TestInjectPendingAnnotations_UnsupportedPatchType(t *testing.T) { + pending := &pendingOperation{taskID: "task-1", intent: pendingIntentStart, experimentID: "exp-123", packageName: "datadog-operator"} + + _, err := injectPendingAnnotations(signalPatch{Type: types.StrategicMergePatchType}, pending) + assert.ErrorContains(t, err, "unsupported patch type") +} + +func TestStartDatadogAgentExperiment_Replace_Success(t *testing.T) { + seedDDA := testDDAObject("") + d, c := testDaemon(seedDDA, testReplaceConfig(`{"features":{"apm":{"enabled":true}}}`), testControllerRevisionForDDA(seedDDA)) + req := testStartRequest() + requireStartQueued(t, d, req) + + dda := &v2alpha1.DatadogAgent{} + require.NoError(t, c.Get(context.Background(), testDDANSN, dda)) + require.NotNil(t, dda.Spec.Features) + require.NotNil(t, dda.Spec.Features.APM) + require.NotNil(t, dda.Spec.Features.APM.Enabled) + assert.True(t, *dda.Spec.Features.APM.Enabled) + + assert.Equal(t, req.Params.Version, dda.Annotations[v2alpha1.AnnotationExperimentID]) + assert.Equal(t, v2alpha1.ExperimentSignalStart, dda.Annotations[v2alpha1.AnnotationExperimentSignal]) + assert.Equal(t, req.ID, dda.Annotations[v2alpha1.AnnotationPendingTaskID]) + assert.Equal(t, string(pendingIntentStart), dda.Annotations[v2alpha1.AnnotationPendingAction]) + assert.Equal(t, req.Params.Version, dda.Annotations[v2alpha1.AnnotationPendingExperimentID]) + assert.Equal(t, req.Package, dda.Annotations[v2alpha1.AnnotationPendingPackage]) +} + +func TestStartDatadogAgentExperiment_Replace_NoBaselineRevision(t *testing.T) { + // Without a ControllerRevision, the operator has never reconciled this DDA: + // a replace has nothing to roll back to if the experiment is later stopped. + d, _ := testDaemon(testDDAObject(""), testReplaceConfig(`{"features":{"apm":{"enabled":true}}}`)) + _, err := d.startDatadogAgentExperiment(context.Background(), testStartRequest()) + assert.ErrorContains(t, err, "no baseline ControllerRevision") +} + +func TestStartDatadogAgentExperiment_Replace_WholesaleReplacesSpec(t *testing.T) { + dda := testDDAObject("") + clusterName := "old-cluster" + dda.Spec.Global = &v2alpha1.GlobalConfig{ClusterName: &clusterName} + d, c := testDaemon(dda, testReplaceConfig(`{"features":{"apm":{"enabled":true}}}`), testControllerRevisionForDDA(dda)) + + requireStartQueued(t, d, testStartRequest()) + + got := &v2alpha1.DatadogAgent{} + require.NoError(t, c.Get(context.Background(), testDDANSN, got)) + // The new config's spec has no "global" key — a merge would have kept the old + // Global config, but replace must wipe it since the whole spec is replaced. + assert.Nil(t, got.Spec.Global) + require.NotNil(t, got.Spec.Features) + require.NotNil(t, got.Spec.Features.APM) + assert.True(t, *got.Spec.Features.APM.Enabled) +} + +func TestStartDatadogAgentExperiment_Replace_BootstrapsAnnotations(t *testing.T) { + dda := testDDAObject("") + dda.Annotations = nil + d, c := testDaemon(dda, testReplaceConfig(`{"features":{"apm":{"enabled":true}}}`), testControllerRevisionForDDA(dda)) + req := testStartRequest() + requireStartQueued(t, d, req) + + got := &v2alpha1.DatadogAgent{} + require.NoError(t, c.Get(context.Background(), testDDANSN, got)) + assert.Equal(t, req.Params.Version, got.Annotations[v2alpha1.AnnotationExperimentID]) +} + +func TestStopDatadogAgentExperiment_AfterReplace(t *testing.T) { + // Stop/rollback always uses the merge-patch signal path regardless of how the + // experiment was started; replace only changes how the start patch is built. + dda := testDDAObject(v2alpha1.ExperimentPhaseRunning) + dda.Spec.Features = &v2alpha1.DatadogFeatures{APM: &v2alpha1.APMFeatureConfig{Enabled: proto.Bool(true)}} + d, c := testDaemon(dda, testReplaceConfig(`{"features":{"apm":{"enabled":true}}}`)) + requireStopQueued(t, d, testStopRequest()) + + got := &v2alpha1.DatadogAgent{} + require.NoError(t, c.Get(context.Background(), testDDANSN, got)) + // Stop only writes the rollback signal; the replaced spec must not be touched. + require.NotNil(t, got.Spec.Features) + require.NotNil(t, got.Spec.Features.APM) + assert.True(t, *got.Spec.Features.APM.Enabled) +} + +func TestPromoteDatadogAgentExperiment_AfterReplace(t *testing.T) { + dda := testDDAObject(v2alpha1.ExperimentPhaseRunning) + dda.Spec.Features = &v2alpha1.DatadogFeatures{APM: &v2alpha1.APMFeatureConfig{Enabled: proto.Bool(true)}} + d, c := testDaemon(dda, testReplaceConfig(`{"features":{"apm":{"enabled":true}}}`)) + d.rcClient = &mockRCClient{state: []*pbgo.PackageState{ + {Package: "datadog-operator", StableConfigVersion: "stable-1", ExperimentConfigVersion: testExperimentID}, + }} + requirePromoteQueued(t, d, testPromoteRequest()) + + got := &v2alpha1.DatadogAgent{} + require.NoError(t, c.Get(context.Background(), testDDANSN, got)) + // Promote only writes the promote signal; the replaced spec must not be touched. + require.NotNil(t, got.Spec.Features) + require.NotNil(t, got.Spec.Features.APM) + assert.True(t, *got.Spec.Features.APM.Enabled) +} + // --- Start idempotency with annotations already applied --- func TestStartDatadogAgentExperiment_Idempotent_AnnotationAlreadyApplied(t *testing.T) { diff --git a/pkg/fleet/experiment.go b/pkg/fleet/experiment.go index c9ad54e699..e249fb8f33 100644 --- a/pkg/fleet/experiment.go +++ b/pkg/fleet/experiment.go @@ -13,12 +13,14 @@ import ( "math" "time" + jsonpatch "gomodules.xyz/jsonpatch/v2" apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/util/retry" v2alpha1 "github.com/DataDog/datadog-operator/api/datadoghq/v2alpha1" + "github.com/DataDog/datadog-operator/pkg/kubernetes" ) // validateParams checks that experimentParams has the fields required to locate @@ -45,14 +47,40 @@ const ( signalPromoteDatadogAgentExperiment experimentSignal = "promote DatadogAgent experiment" ) -// resolvedOperation holds the resolved data needed to execute an experiment operation. +// resolvedOperation holds the data needed to execute an experiment operation. +// For OperationUpdate, Config is the full config, merge-patched as-is. For +// OperationReplace, Config is just the "spec" value (see extractReplaceSpec), +// which replaces the DatadogAgent's spec wholesale. type resolvedOperation struct { NamespacedName types.NamespacedName + Operation Operation Config json.RawMessage } -// experimentBackoff is the retry backoff for k8s operations during experiment signals. -// Retries start at 1s, doubling each attempt up to 10s, for up to 3 minutes total. +// signalPatch is a patch plus the type to apply it as: a JSON merge patch +// (RFC 7386, used by update) or a JSON Patch (RFC 6902, used by replace). +// Data holds the merge patch body; Ops holds the JSON Patch operations. +// +// Use mergePatch(nil) for "nothing to patch", not the zero value: an empty +// Type isn't valid and gets rejected by injectPendingAnnotations. +type signalPatch struct { + Type types.PatchType + Data []byte + Ops []jsonpatch.Operation +} + +// mergePatch wraps data as a JSON merge patch (RFC 7386). +func mergePatch(data []byte) signalPatch { + return signalPatch{Type: types.MergePatchType, Data: data} +} + +// jsonPatch wraps ops as a JSON Patch (RFC 6902). +func jsonPatch(ops []jsonpatch.Operation) signalPatch { + return signalPatch{Type: types.JSONPatchType, Ops: ops} +} + +// experimentBackoff retries k8s operations starting at 1s, doubling up to 10s, +// for up to 3 minutes total. var experimentBackoff = wait.Backoff{ Duration: 1 * time.Second, Factor: 2.0, @@ -69,8 +97,7 @@ func isRetryable(err error) bool { !apierrors.IsMethodNotSupported(err) } -// retryWithBackoff retries fn on transient errors with exponential backoff. -// The total retry window is bounded by a 3-minute context timeout. +// retryWithBackoff retries fn on transient errors, bounded by a 3-minute timeout. // Permanent errors (not-found, forbidden, invalid, method-not-supported) are not retried. func retryWithBackoff(ctx context.Context, fn func() error) error { ctx, cancel := context.WithTimeout(ctx, 3*time.Minute) @@ -81,8 +108,8 @@ func retryWithBackoff(ctx context.Context, fn func() error) error { } // buildSignalPatch creates a JSON merge patch that sets the experiment signal -// and ID annotations. If config is non-nil, spec fields from the config are -// merged into the patch so that the spec and annotations are written atomically. +// and ID annotations, merging in config's spec fields if given so the spec and +// annotations are written atomically. func buildSignalPatch(signal, id string, config ...json.RawMessage) ([]byte, error) { patch := map[string]any{ "metadata": map[string]any{ @@ -98,15 +125,119 @@ func buildSignalPatch(signal, id string, config ...json.RawMessage) ([]byte, err if err := json.Unmarshal(config[0], &specPatch); err != nil { return nil, fmt.Errorf("failed to unmarshal config: %w", err) } - // Top-level maps.Copy is safe because the config currently only contains - // "spec" keys and never "metadata". If the config ever includes metadata, - // this will need a deep merge to avoid overwriting the signal annotations. + // Safe as a shallow copy only because config never contains "metadata". maps.Copy(patch, specPatch) } return json.Marshal(patch) } +// extractReplaceSpec returns config's "spec" value. It doesn't type-check the +// spec — the API server does that — but it does reject an empty spec ({}), +// since the API server would accept that and wipe the resource instead. +func extractReplaceSpec(config json.RawMessage) (json.RawMessage, error) { + var raw map[string]json.RawMessage + if err := json.Unmarshal(config, &raw); err != nil { + return nil, fmt.Errorf("config must be a JSON object: %w", err) + } + specRaw, ok := raw["spec"] + if !ok { + return nil, fmt.Errorf(`config for replace operation must contain a "spec" key`) + } + var spec map[string]json.RawMessage + if err := json.Unmarshal(specRaw, &spec); err == nil && len(spec) == 0 { + return nil, fmt.Errorf(`config "spec" must not be empty`) + } + return specRaw, nil +} + +// buildReplaceSignalPatch creates a JSON Patch that replaces the DatadogAgent's +// spec wholesale and writes the experiment signal and ID annotations. +// bootstrapAnnotations should be true if the DatadogAgent has no annotations +// map yet, so the patch creates one first. Pending-task annotations are added +// later, by injectPendingAnnotations. +func buildReplaceSignalPatch(signal, id string, spec json.RawMessage, bootstrapAnnotations bool) []jsonpatch.Operation { + ops := make([]jsonpatch.Operation, 0, 5) + + if bootstrapAnnotations { + // "test" fails the patch if annotations already exist, instead of "add" overwriting them. + ops = append(ops, + jsonpatch.Operation{Operation: "test", Path: "/metadata/annotations", Value: nil}, + jsonpatch.Operation{Operation: "add", Path: "/metadata/annotations", Value: map[string]string{}}, + ) + } + + ops = append(ops, + jsonpatch.Operation{Operation: "add", Path: "/spec", Value: spec}, + jsonpatch.Operation{Operation: "add", Path: kubernetes.AnnotationJSONPatchPath(v2alpha1.AnnotationExperimentSignal), Value: signal}, + jsonpatch.Operation{Operation: "add", Path: kubernetes.AnnotationJSONPatchPath(v2alpha1.AnnotationExperimentID), Value: id}, + ) + + return ops +} + +// injectPendingAnnotations adds pending's task-tracking annotations to sp's +// patch, merging them in for a merge patch or appending JSON Patch ops for a +// JSON Patch. This is the only place pending annotations are written. +func injectPendingAnnotations(sp signalPatch, pending *pendingOperation) ([]byte, error) { + // resultVersion is handled separately below: merge and JSON Patch clear a + // stale value differently. + entries := []struct{ key, value string }{ + {v2alpha1.AnnotationPendingTaskID, pending.taskID}, + {v2alpha1.AnnotationPendingAction, string(pending.intent)}, + {v2alpha1.AnnotationPendingExperimentID, pending.experimentID}, + {v2alpha1.AnnotationPendingPackage, pending.packageName}, + } + + switch sp.Type { + case types.JSONPatchType: + ops := sp.Ops + for _, e := range entries { + ops = append(ops, jsonpatch.Operation{Operation: "add", Path: kubernetes.AnnotationJSONPatchPath(e.key), Value: e.value}) + } + // Always written, even as "", to clear a stale value from an earlier + // promote — pendingOperationFromAnnotations treats "" as absent. + ops = append(ops, jsonpatch.Operation{Operation: "add", Path: kubernetes.AnnotationJSONPatchPath(v2alpha1.AnnotationPendingResultVersion), Value: pending.resultVersion}) + return json.Marshal(ops) + + case types.MergePatchType: + var patchMap map[string]any + if len(sp.Data) != 0 { + if err := json.Unmarshal(sp.Data, &patchMap); err != nil { + return nil, fmt.Errorf("failed to unmarshal base patch: %w", err) + } + } else { + patchMap = make(map[string]any) + } + + metadata, ok := patchMap["metadata"].(map[string]any) + if !ok { + metadata = make(map[string]any) + patchMap["metadata"] = metadata + } + annotations, ok := metadata["annotations"].(map[string]any) + if !ok { + annotations = make(map[string]any) + metadata["annotations"] = annotations + } + for _, e := range entries { + annotations[e.key] = e.value + } + if pending.resultVersion != "" { + annotations[v2alpha1.AnnotationPendingResultVersion] = pending.resultVersion + } else { + // Clear any old promote result version. Merge patch leaves keys alone + // when they are omitted. + annotations[v2alpha1.AnnotationPendingResultVersion] = nil + } + + return json.Marshal(patchMap) + + default: + return nil, fmt.Errorf("unsupported patch type %q", sp.Type) + } +} + // isTerminalPhase returns true for terminal experiment phases. func isTerminalPhase(phase v2alpha1.ExperimentPhase) bool { switch phase { diff --git a/pkg/fleet/remote_config.go b/pkg/fleet/remote_config.go index aa6e126c94..8ad220db13 100644 --- a/pkg/fleet/remote_config.go +++ b/pkg/fleet/remote_config.go @@ -26,13 +26,16 @@ type installerConfig struct { type Operation string const ( - OperationCreate Operation = "create" - OperationUpdate Operation = "update" - OperationDelete Operation = "delete" + OperationCreate Operation = "create" + OperationUpdate Operation = "update" + OperationDelete Operation = "delete" + OperationReplace Operation = "replace" ) // fleetManagementOperation is a single fleet operation for config management of a Kubernetes resource. -// Config is a JSON merge patch (no strategic merge patch). +// For OperationUpdate, Config is a JSON merge patch (no strategic merge patch) applied to the resource. +// For OperationReplace, Config must contain a "spec" key, whose value wholly replaces the +// resource's spec (no merging); other top-level keys are ignored. type fleetManagementOperation struct { Operation Operation `json:"operation"` Config json.RawMessage `json:"config"` diff --git a/pkg/kubernetes/jsonpatch.go b/pkg/kubernetes/jsonpatch.go new file mode 100644 index 0000000000..356a83cba3 --- /dev/null +++ b/pkg/kubernetes/jsonpatch.go @@ -0,0 +1,20 @@ +// Unless explicitly stated otherwise all files in this repository are licensed +// under the Apache License Version 2.0. +// This product includes software developed at Datadog (https://www.datadoghq.com/). +// Copyright 2016-present Datadog, Inc. + +package kubernetes + +import "strings" + +// JSONPointerEscape escapes a string for use as a segment of an RFC 6901 JSON +// Pointer, e.g. an annotation key used in a JSON Patch "path". +func JSONPointerEscape(s string) string { + return strings.NewReplacer("~", "~0", "/", "~1").Replace(s) +} + +// AnnotationJSONPatchPath converts an annotation key to a JSON Patch path +// under /metadata/annotations, escaping "~" and "/" per RFC 6901. +func AnnotationJSONPatchPath(key string) string { + return "/metadata/annotations/" + JSONPointerEscape(key) +}