diff --git a/go.mod b/go.mod index 4dc7374cc..3c292b92f 100644 --- a/go.mod +++ b/go.mod @@ -12,6 +12,7 @@ require ( github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc github.com/evanphx/json-patch v5.9.11+incompatible github.com/ghodss/yaml v1.0.0 + github.com/go-logr/logr v1.4.3 github.com/google/cel-go v0.29.2 github.com/google/go-cmp v0.7.0 github.com/itchyny/gojq v0.12.19 @@ -26,6 +27,7 @@ require ( github.com/spf13/pflag v1.0.10 github.com/stretchr/testify v1.11.1 github.com/valyala/fasttemplate v1.2.2 + go.uber.org/zap v1.27.0 google.golang.org/grpc v1.83.0 gopkg.in/yaml.v2 v2.4.0 helm.sh/helm/v3 v3.21.1 @@ -40,7 +42,7 @@ require ( k8s.io/kubectl v0.35.4 k8s.io/utils v0.0.0-20260210185600-b8788abfbbc2 open-cluster-management.io/addon-framework v1.3.0 - open-cluster-management.io/api v1.3.1-0.20260709055002-403378b57558 + open-cluster-management.io/api v1.3.1-0.20260824155636-3ff9ad2f868d open-cluster-management.io/sdk-go v1.3.1-0.20260713072928-c812bc0c3b06 sigs.k8s.io/about-api v0.0.0-20250131010323-518069c31c03 sigs.k8s.io/cluster-inventory-api v0.1.3 @@ -93,7 +95,6 @@ require ( github.com/felixge/httpsnoop v1.0.4 // indirect github.com/fsnotify/fsnotify v1.9.0 // indirect github.com/fxamacker/cbor/v2 v2.9.0 // indirect - github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/go-logr/zapr v1.3.0 // indirect github.com/go-openapi/jsonpointer v0.22.5 // indirect @@ -166,7 +167,6 @@ require ( go.opentelemetry.io/otel/trace v1.44.0 // indirect go.opentelemetry.io/proto/otlp v1.10.0 // indirect go.uber.org/multierr v1.11.0 // indirect - go.uber.org/zap v1.27.0 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/crypto v0.54.0 // indirect diff --git a/go.sum b/go.sum index 02eaa8f78..a2b73fe14 100644 --- a/go.sum +++ b/go.sum @@ -612,8 +612,8 @@ k8s.io/utils v0.0.0-20260210185600-b8788abfbbc2 h1:AZYQSJemyQB5eRxqcPky+/7EdBj0x k8s.io/utils v0.0.0-20260210185600-b8788abfbbc2/go.mod h1:xDxuJ0whA3d0I4mf/C4ppKHxXynQ+fxnkmQH0vTHnuk= open-cluster-management.io/addon-framework v1.3.0 h1:rqW8Dl2Gcbac+8C4mQhOtbbmXb+JbIp3jPwP6AA5DBw= open-cluster-management.io/addon-framework v1.3.0/go.mod h1:UP3lXnEKduMF+6CTPJKzsIahrO6Hd55MSPhDYo1OwaM= -open-cluster-management.io/api v1.3.1-0.20260709055002-403378b57558 h1:Ny12QGibGK2KAZ9KM0E89+jeqmYtQQi3fpi1dOLG5SA= -open-cluster-management.io/api v1.3.1-0.20260709055002-403378b57558/go.mod h1:/qLKuMbMS1+MpirTjaw4BoS3INKG8jV3s5J7iju48tg= +open-cluster-management.io/api v1.3.1-0.20260824155636-3ff9ad2f868d h1:ggO+UQBCE/hNxjTVdw4CUu2YWN/llXRJ7v+lHBrOXeo= +open-cluster-management.io/api v1.3.1-0.20260824155636-3ff9ad2f868d/go.mod h1:/qLKuMbMS1+MpirTjaw4BoS3INKG8jV3s5J7iju48tg= open-cluster-management.io/sdk-go v1.3.1-0.20260713072928-c812bc0c3b06 h1:gpk3LkPj0zMI8lt0A619JQ5KzB5RjxQob5eiF7zeZ28= open-cluster-management.io/sdk-go v1.3.1-0.20260713072928-c812bc0c3b06/go.mod h1:j2M5g9Shhgvh9Y79/T7ZI1iyokZKaZ7ricRa9yHrzd8= sigs.k8s.io/about-api v0.0.0-20250131010323-518069c31c03 h1:1ShFiMjGQOR/8jTBkmZrk1gORxnvMwm1nOy2/DbHg4U= diff --git a/pkg/work/spoke/controllers/manifestcontroller/apply_log.go b/pkg/work/spoke/controllers/manifestcontroller/apply_log.go new file mode 100644 index 000000000..2b46f710e --- /dev/null +++ b/pkg/work/spoke/controllers/manifestcontroller/apply_log.go @@ -0,0 +1,280 @@ +package manifestcontroller + +import ( + "context" + "os" + "sync" + "time" + + "github.com/go-logr/logr" + "go.uber.org/zap/zapcore" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/component-base/logs/json" + + workapiv1 "open-cluster-management.io/api/work/v1" +) + +// applyLogSink is the logger used for the apply-log lines below. +// +// These lines are a machine-consumed telemetry stream, not operator-facing prose: a log +// pipeline joins them to the hub's write-time line on {mw_namespace, mw_name, generation} to +// derive hub->spoke propagation latency, which requires every field to arrive as a first-class +// attribute. A collector can only do that if the whole line is JSON, and klog's format is +// process-global — so the agent's default text output would force every consumer to re-parse +// these fields out of a formatted string. +// +// This sink therefore emits the apply-log lines as JSON regardless of the process-wide +// logging format, while every other line in the agent keeps that format. It is built from +// component-base's own JSON logger, so the encoding is identical to what +// --logging-format=json produces (ts / caller / msg plus the structured key-values) rather +// than a format private to this package. Writes are serialised by zapcore.Lock, so a JSON +// line can never interleave with concurrent klog output on the same stream. +// +// Emission stays gated behind the ManifestWorkApplyLogs feature gate; when the gate is off +// nothing is written here at all. Overridden in tests to capture the emitted key-values. +var applyLogSink = newApplyLogSink() + +func newApplyLogSink() logr.Logger { + // verbosity 0: these lines are unconditional Info once the feature gate admits them, + // so they must not be filtered by -v. nil errorStream keeps everything on stdout. + logger, _ := json.NewJSONLogger(0, zapcore.Lock(zapcore.AddSync(os.Stdout)), nil, nil) + return logger +} + +// Flow discriminators for the spoke apply-log lines. Gated behind the +// ManifestWorkApplyLogs feature gate; paired with the hub webhook line (mw_hub_apply) for +// hub->spoke propagation-latency measurement in Datadog. +const ( + flowSpokeApply = "mw_spoke_apply" // rollup start, once per generation (latency join key) + flowResourceSpokeApply = "mw_resource_spoke_apply" // per resource apply attempt, first apply of a generation (drill-down) + flowResourceSpokeSync = "mw_resource_spoke_sync" // per resource, later reconcile where the outcome changed + flowSpokeApplyResult = "mw_spoke_apply_result" // rollup end, once per generation (outcome) +) + +// applyLogTimeFormat is RFC3339 with millisecond precision, UTC. +const applyLogTimeFormat = "2006-01-02T15:04:05.000Z07:00" + +// Per-resource outcome values. read_only marks a resource tracked by a read-only manifest, which +// the agent observes but never writes, so it is neither applied nor failed. +const ( + outcomeApplied = "applied" + outcomeFailed = "failed" + outcomeReadOnly = "read_only" +) + +// workMeta carries the ManifestWork identity down to the per-resource apply emit, which runs +// deep in applyOneManifest where the ManifestWork object itself is no longer in scope. +type workMeta struct { + name string + namespace string + generation int64 + labels map[string]string +} + +// applyCounts is the per-generation apply tally carried by the rollup end line. +type applyCounts struct { + applied int + failed int + readOnly int +} + +// emittedRollups records the newest generation whose rollup pair this agent process has already +// emitted, keyed by ManifestWork name (the work informer is scoped to the agent's own cluster +// namespace, so the name is unique within a process). +// +// It closes a retry window the persisted guard cannot see. priorAppliedGeneration reads +// WorkApplied.ObservedGeneration, but that value is written by sync *after* reconcile returns. +// When the write fails — a concurrent spec update conflicting on resourceVersion is the common +// case — the controller requeues and reconcile runs again against a status that still names the +// previous generation, so the persisted guard admits the same generation a second time and emits +// a duplicate mw_spoke_apply. That line is the key the propagation-latency join groups on, so a +// duplicate leaves the join with two candidate timestamps for one propagation event. +// +// This layers on top of the persisted guard rather than replacing it: a restarted agent starts +// with an empty ledger and falls back to the status it reads from the API server, which is the +// restart-safe behaviour D4 chose the persisted signal for. +var emittedRollups = newRollupGenerations() + +// rollupGenerations is the process-local half of the once-per-generation guard. +type rollupGenerations struct { + mu sync.Mutex + seen map[string]int64 +} + +func newRollupGenerations() *rollupGenerations { + return &rollupGenerations{seen: map[string]int64{}} +} + +// admit reports whether this process has yet to emit the rollup pair for this generation, +// recording it when so. Because it records, callers must place it last in a short-circuiting +// condition, so a work rejected by an earlier check is not marked as emitted. +func (r *rollupGenerations) admit(name string, generation int64) bool { + r.mu.Lock() + defer r.mu.Unlock() + if seen, ok := r.seen[name]; ok && seen >= generation { + return false + } + r.seen[name] = generation + return true +} + +// forget drops a ManifestWork's entry once the work is gone, so the ledger tracks live works +// rather than every work the process has ever reconciled. +func (r *rollupGenerations) forget(name string) { + r.mu.Lock() + defer r.mu.Unlock() + delete(r.seen, name) +} + +// priorAppliedGeneration returns the ObservedGeneration of the WorkApplied condition, or -1 +// when the condition is absent. This persisted value is the restart-safe signal used to emit +// the rollup lines exactly once per generation (a new agent pod reads the same status). +func priorAppliedGeneration(mw *workapiv1.ManifestWork) int64 { + if c := meta.FindStatusCondition(mw.Status.Conditions, workapiv1.WorkApplied); c != nil { + return c.ObservedGeneration + } + return -1 +} + +// applyOutcome maps an apply error to the per-resource outcome value. +func applyOutcome(err error) string { + if err != nil { + return outcomeFailed + } + return outcomeApplied +} + +// priorManifestOutcome returns the outcome recorded by the resource's last-persisted +// ManifestApplied condition ("applied"/"failed"), or "" when unknown. It is the restart-safe +// signal used to detect a per-resource outcome change on a later reconcile. +func priorManifestOutcome(mc *workapiv1.ManifestCondition) string { + if mc == nil { + return "" + } + c := meta.FindStatusCondition(mc.Conditions, workapiv1.ManifestApplied) + if c == nil { + return "" + } + if c.Status == metav1.ConditionTrue { + return outcomeApplied + } + return outcomeFailed +} + +// countApplyResults tallies results by strategy: read-only manifests apply nothing (counted +// separately), otherwise a non-nil error is a failure and success is an apply. +func countApplyResults(results []applyResult) (applied, failed, readOnly int) { + for _, r := range results { + switch { + case r.strategy == workapiv1.UpdateStrategyTypeReadOnly: + readOnly++ + case r.Error != nil: + failed++ + default: + applied++ + } + } + return applied, failed, readOnly +} + +// rollupOutcome maps the applying-manifest counts to the work-level outcome value. A work with no +// applying manifests (all read-only) reports read_only. +func rollupOutcome(applied, failed int) string { + switch { + case applied == 0 && failed == 0: + return outcomeReadOnly + case failed == 0: + return outcomeApplied + case applied == 0: + return outcomeFailed + default: + return "partial" + } +} + +// emitResourceApply logs one per-resource apply-log line. It is a noop when the feature is +// disabled. Otherwise it emits: +// - flowResourceSpokeApply, on the first apply of a generation (firstApply), for every resource — +// read-only manifests get outcome=read_only (observed, not applied), others applied/failed; or +// - flowResourceSpokeSync, on a later reconcile of the same generation, only when a non-read-only +// resource's outcome changed vs prevOutcome (its last-persisted ManifestApplied result). +// +// Read-only manifests never emit a sync line (their outcome cannot change), and a steady-state resync +// with no outcome change emits nothing — this bounds the per-resource stream to the propagation event +// plus genuine recoveries/regressions. +func emitResourceApply(ctx context.Context, logApply, firstApply bool, wm workMeta, om orderedManifest, + strategy workapiv1.UpdateStrategyType, result applyResult, prevOutcome string) { + if !logApply { + return + } + + readOnly := strategy == workapiv1.UpdateStrategyTypeReadOnly + outcome := applyOutcome(result.Error) + if readOnly { + outcome = outcomeReadOnly + } + + flow := "" + switch { + case firstApply: + flow = flowResourceSpokeApply + case !readOnly && prevOutcome != "" && prevOutcome != outcome: + flow = flowResourceSpokeSync + default: + return + } + + now := time.Now().UTC() + resourceVersion := "" + if result.Error == nil && result.Result != nil { + if accessor, err := meta.Accessor(result.Result); err == nil { + resourceVersion = accessor.GetResourceVersion() + } + } + + kv := []any{ + "flow", flow, + "mw_name", wm.name, + "mw_namespace", wm.namespace, + "generation", wm.generation, + "applied_kind", om.resourceMeta.Kind, + "applied_name", om.resourceMeta.Name, + "applied_namespace", om.resourceMeta.Namespace, + "applied_resource_version", resourceVersion, + "ts_utc", now.Format(applyLogTimeFormat), + "ts_epoch_ms", now.UnixMilli(), + "outcome", outcome, + "labels", wm.labels, + } + if flow == flowResourceSpokeSync { + kv = append(kv, "prev_outcome", prevOutcome) + } + applyLogSink.Info("manifestwork resource applied", kv...) +} + +// emitApplyRollup logs one work-level rollup line. With counts == nil it is the start line +// (mw_spoke_apply, no outcome, stamped before the apply loop); with counts != nil it is the end +// line (mw_spoke_apply_result, carrying applied/failed/outcome, stamped after the apply loop). +func emitApplyRollup(ctx context.Context, wm workMeta, flow string, resourceCount int, counts *applyCounts) { + now := time.Now().UTC() + kv := []any{ + "flow", flow, + "mw_name", wm.name, + "mw_namespace", wm.namespace, + "generation", wm.generation, + "resource_count", resourceCount, + "ts_utc", now.Format(applyLogTimeFormat), + "ts_epoch_ms", now.UnixMilli(), + "labels", wm.labels, + } + if counts != nil { + kv = append(kv, + "applied_count", counts.applied, + "failed_count", counts.failed, + "read_only_count", counts.readOnly, + "outcome", rollupOutcome(counts.applied, counts.failed), + ) + } + applyLogSink.Info("manifestwork apply", kv...) +} diff --git a/pkg/work/spoke/controllers/manifestcontroller/apply_log_test.go b/pkg/work/spoke/controllers/manifestcontroller/apply_log_test.go new file mode 100644 index 000000000..d3fd60d1d --- /dev/null +++ b/pkg/work/spoke/controllers/manifestcontroller/apply_log_test.go @@ -0,0 +1,466 @@ +package manifestcontroller + +import ( + "context" + "errors" + "fmt" + "os" + "testing" + + "github.com/go-logr/logr" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + utilruntime "k8s.io/apimachinery/pkg/util/runtime" + + ocmfeature "open-cluster-management.io/api/feature" + workapiv1 "open-cluster-management.io/api/work/v1" + + testingcommon "open-cluster-management.io/ocm/pkg/common/testing" + "open-cluster-management.io/ocm/pkg/features" + "open-cluster-management.io/ocm/pkg/work/spoke/spoketesting" +) + +// TestMain registers the spoke-work feature gates so reconcile's +// SpokeMutableFeatureGate.Enabled(ManifestWorkApplyLogs) check does not panic in tests. +func TestMain(m *testing.M) { + utilruntime.Must(features.SpokeMutableFeatureGate.Add(ocmfeature.DefaultSpokeWorkFeatureGates)) + os.Exit(m.Run()) +} + +// logEntry is a single captured structured log line. +type logEntry struct { + msg string + kv map[string]any +} + +// captureSink is a logr.LogSink that records Info lines and their key/value pairs +// so tests can assert on the structured attributes the emitters produce. +type captureSink struct { + entries *[]logEntry +} + +func (s *captureSink) Init(logr.RuntimeInfo) {} +func (s *captureSink) Enabled(int) bool { return true } +func (s *captureSink) Error(error, string, ...any) {} +func (s *captureSink) WithValues(...any) logr.LogSink { return s } +func (s *captureSink) WithName(string) logr.LogSink { return s } + +func (s *captureSink) Info(_ int, msg string, kv ...any) { + m := map[string]any{} + for i := 0; i+1 < len(kv); i += 2 { + key, ok := kv[i].(string) + if !ok { + continue + } + m[key] = kv[i+1] + } + *s.entries = append(*s.entries, logEntry{msg: msg, kv: m}) +} + +// captureContext swaps the package-level apply-log sink for a capturing one and returns a +// context plus the slice the emitted entries land in. The sink is package-level rather than +// context-carried because these lines are emitted as JSON independently of the process-wide +// klog format, so tests assert against the sink instead of a context logger. +func captureContext() (context.Context, *[]logEntry) { + entries := &[]logEntry{} + applyLogSink = logr.New(&captureSink{entries: entries}) + return context.Background(), entries +} + +func TestApplyOutcome(t *testing.T) { + if got := applyOutcome(nil); got != "applied" { + t.Errorf("applyOutcome(nil) = %q, want applied", got) + } + if got := applyOutcome(errors.New("boom")); got != "failed" { + t.Errorf("applyOutcome(err) = %q, want failed", got) + } +} + +func TestRollupOutcome(t *testing.T) { + cases := []struct { + applied, failed int + want string + }{ + {3, 0, "applied"}, + {0, 2, "failed"}, + {2, 1, "partial"}, + {0, 0, "read_only"}, + } + for _, c := range cases { + if got := rollupOutcome(c.applied, c.failed); got != c.want { + t.Errorf("rollupOutcome(%d,%d) = %q, want %q", c.applied, c.failed, got, c.want) + } + } +} + +func TestCountApplyResults(t *testing.T) { + results := []applyResult{ + {Error: nil}, // applied + {Error: errors.New("x")}, // failed + {Error: nil}, // applied + {strategy: workapiv1.UpdateStrategyTypeReadOnly}, // read-only, counted separately + } + applied, failed, readOnly := countApplyResults(results) + if applied != 2 || failed != 1 || readOnly != 1 { + t.Errorf("countApplyResults = (%d,%d,%d), want (2,1,1)", applied, failed, readOnly) + } +} + +func TestPriorAppliedGeneration(t *testing.T) { + empty := &workapiv1.ManifestWork{} + if got := priorAppliedGeneration(empty); got != -1 { + t.Errorf("priorAppliedGeneration(no condition) = %d, want -1", got) + } + + mw := &workapiv1.ManifestWork{} + mw.Status.Conditions = []metav1.Condition{{ + Type: workapiv1.WorkApplied, + Status: metav1.ConditionTrue, + ObservedGeneration: 5, + }} + if got := priorAppliedGeneration(mw); got != 5 { + t.Errorf("priorAppliedGeneration = %d, want 5", got) + } +} + +func testWorkMeta() workMeta { + return workMeta{ + name: "demo-mw", + namespace: "cluster1", + generation: 7, + labels: map[string]string{"example.com/team": "platform"}, + } +} + +func testManifest() orderedManifest { + return orderedManifest{ + resourceMeta: workapiv1.ManifestResourceMeta{ + Kind: "ConfigMap", + Name: "cm1", + Namespace: "default", + }, + } +} + +func TestEmitResourceApply_GateOff(t *testing.T) { + ctx, entries := captureContext() + // logApply=false: never emits, even on a first apply. + emitResourceApply(ctx, false, true, testWorkMeta(), testManifest(), + workapiv1.UpdateStrategyTypeUpdate, applyResult{}, "") + if len(*entries) != 0 { + t.Errorf("gate off: expected no log, got %d", len(*entries)) + } +} + +func TestEmitResourceApply_ReadOnlyFirstApply(t *testing.T) { + ctx, entries := captureContext() + emitResourceApply(ctx, true, true, testWorkMeta(), testManifest(), + workapiv1.UpdateStrategyTypeReadOnly, applyResult{Result: &unstructured.Unstructured{}}, "") + if len(*entries) != 1 { + t.Fatalf("read-only first apply: expected 1 log, got %d", len(*entries)) + } + kv := (*entries)[0].kv + assertKV(t, kv, "flow", flowResourceSpokeApply) + assertKV(t, kv, "outcome", outcomeReadOnly) + if _, ok := kv["prev_outcome"]; ok { + t.Error("read-only apply line must not carry prev_outcome") + } +} + +func TestEmitResourceApply_ReadOnlyNoSyncOnResync(t *testing.T) { + ctx, entries := captureContext() + // firstApply=false with a differing prevOutcome: a read-only manifest must NOT emit a sync line + // (its outcome cannot change). + emitResourceApply(ctx, true, false, testWorkMeta(), testManifest(), + workapiv1.UpdateStrategyTypeReadOnly, applyResult{Result: &unstructured.Unstructured{}}, "applied") + if len(*entries) != 0 { + t.Errorf("read-only resync: expected no log, got %d", len(*entries)) + } +} + +func TestEmitResourceApply_FirstApplySuccess(t *testing.T) { + ctx, entries := captureContext() + obj := &unstructured.Unstructured{} + obj.SetResourceVersion("123") + emitResourceApply(ctx, true, true, testWorkMeta(), testManifest(), + workapiv1.UpdateStrategyTypeUpdate, applyResult{Result: obj}, "") + + if len(*entries) != 1 { + t.Fatalf("expected 1 log, got %d", len(*entries)) + } + kv := (*entries)[0].kv + assertKV(t, kv, "flow", flowResourceSpokeApply) + assertKV(t, kv, "mw_name", "demo-mw") + assertKV(t, kv, "mw_namespace", "cluster1") + assertKV(t, kv, "generation", int64(7)) + assertKV(t, kv, "applied_kind", "ConfigMap") + assertKV(t, kv, "applied_name", "cm1") + assertKV(t, kv, "applied_namespace", "default") + assertKV(t, kv, "applied_resource_version", "123") + assertKV(t, kv, "outcome", "applied") + if _, ok := kv["prev_outcome"]; ok { + t.Error("first-apply line must not carry prev_outcome") + } + if _, ok := kv["ts_epoch_ms"]; !ok { + t.Error("missing ts_epoch_ms") + } +} + +func TestEmitResourceApply_FirstApplyFailure(t *testing.T) { + ctx, entries := captureContext() + emitResourceApply(ctx, true, true, testWorkMeta(), testManifest(), + workapiv1.UpdateStrategyTypeServerSideApply, applyResult{Error: errors.New("denied")}, "") + + if len(*entries) != 1 { + t.Fatalf("expected 1 log, got %d", len(*entries)) + } + kv := (*entries)[0].kv + assertKV(t, kv, "flow", flowResourceSpokeApply) + assertKV(t, kv, "outcome", "failed") + assertKV(t, kv, "applied_resource_version", "") +} + +// Resync (firstApply=false) where the resource's outcome changed emits the sync flow. +func TestEmitResourceApply_SyncOutcomeChanged(t *testing.T) { + ctx, entries := captureContext() + obj := &unstructured.Unstructured{} + obj.SetResourceVersion("456") + emitResourceApply(ctx, true, false, testWorkMeta(), testManifest(), + workapiv1.UpdateStrategyTypeUpdate, applyResult{Result: obj}, "failed") + + if len(*entries) != 1 { + t.Fatalf("expected 1 log, got %d", len(*entries)) + } + kv := (*entries)[0].kv + assertKV(t, kv, "flow", flowResourceSpokeSync) + assertKV(t, kv, "outcome", "applied") + assertKV(t, kv, "prev_outcome", "failed") +} + +// Resync where the outcome is unchanged emits nothing. +func TestEmitResourceApply_SyncNoChange(t *testing.T) { + ctx, entries := captureContext() + obj := &unstructured.Unstructured{} + emitResourceApply(ctx, true, false, testWorkMeta(), testManifest(), + workapiv1.UpdateStrategyTypeUpdate, applyResult{Result: obj}, "applied") + if len(*entries) != 0 { + t.Errorf("unchanged resync: expected no log, got %d", len(*entries)) + } +} + +// Resync with no known prior outcome emits nothing (cannot determine a change). +func TestEmitResourceApply_SyncUnknownPrev(t *testing.T) { + ctx, entries := captureContext() + emitResourceApply(ctx, true, false, testWorkMeta(), testManifest(), + workapiv1.UpdateStrategyTypeUpdate, applyResult{}, "") + if len(*entries) != 0 { + t.Errorf("unknown prev: expected no log, got %d", len(*entries)) + } +} + +func TestPriorManifestOutcome(t *testing.T) { + if got := priorManifestOutcome(nil); got != "" { + t.Errorf("nil condition = %q, want empty", got) + } + applied := &workapiv1.ManifestCondition{Conditions: []metav1.Condition{{ + Type: workapiv1.ManifestApplied, Status: metav1.ConditionTrue, + }}} + if got := priorManifestOutcome(applied); got != "applied" { + t.Errorf("applied condition = %q, want applied", got) + } + failed := &workapiv1.ManifestCondition{Conditions: []metav1.Condition{{ + Type: workapiv1.ManifestApplied, Status: metav1.ConditionFalse, + }}} + if got := priorManifestOutcome(failed); got != "failed" { + t.Errorf("failed condition = %q, want failed", got) + } +} + +func TestEmitApplyRollup_Start(t *testing.T) { + ctx, entries := captureContext() + emitApplyRollup(ctx, testWorkMeta(), flowSpokeApply, 3, nil) + + if len(*entries) != 1 { + t.Fatalf("expected 1 log, got %d", len(*entries)) + } + kv := (*entries)[0].kv + assertKV(t, kv, "flow", flowSpokeApply) + assertKV(t, kv, "generation", int64(7)) + assertKV(t, kv, "resource_count", 3) + if _, ok := kv["outcome"]; ok { + t.Error("start line must not carry outcome") + } + if _, ok := kv["applied_count"]; ok { + t.Error("start line must not carry applied_count") + } +} + +func TestEmitApplyRollup_End(t *testing.T) { + ctx, entries := captureContext() + emitApplyRollup(ctx, testWorkMeta(), flowSpokeApplyResult, 6, + &applyCounts{applied: 2, failed: 1, readOnly: 3}) + + if len(*entries) != 1 { + t.Fatalf("expected 1 log, got %d", len(*entries)) + } + kv := (*entries)[0].kv + assertKV(t, kv, "flow", flowSpokeApplyResult) + assertKV(t, kv, "resource_count", 6) + assertKV(t, kv, "applied_count", 2) + assertKV(t, kv, "failed_count", 1) + assertKV(t, kv, "read_only_count", 3) + assertKV(t, kv, "outcome", "partial") +} + +func assertKV(t *testing.T, kv map[string]any, key string, want any) { + t.Helper() + got, ok := kv[key] + if !ok { + t.Errorf("missing key %q", key) + return + } + if got != want { + t.Errorf("key %q = %v (%T), want %v (%T)", key, got, got, want, want) + } +} + +func flowCounts(entries *[]logEntry) map[string]int { + counts := map[string]int{} + for _, e := range *entries { + if f, ok := e.kv["flow"].(string); ok { + counts[f]++ + } + } + return counts +} + +// resetRollupState gives a test a clean process-local emit ledger, so generation state does not +// leak between reconcile-level tests in either direction. +func resetRollupState(t *testing.T) { + t.Helper() + emittedRollups = newRollupGenerations() + t.Cleanup(func() { emittedRollups = newRollupGenerations() }) +} + +func setApplyLogsGate(t *testing.T, enabled bool) { + t.Helper() + if err := features.SpokeMutableFeatureGate.Set( + fmt.Sprintf("%s=%t", ocmfeature.ManifestWorkApplyLogs, enabled)); err != nil { + t.Fatal(err) + } +} + +// TestReconcileEmitsApplyLogLines drives a full reconcile with the gate on for a new +// generation and asserts all three lines fire exactly once. +func TestReconcileEmitsApplyLogLines(t *testing.T) { + setApplyLogsGate(t, true) + defer setApplyLogsGate(t, false) + resetRollupState(t) + + work, workKey := newTestCase("emit"). + withWorkManifest(testingcommon.NewUnstructured("v1", "Secret", "ns1", "test")). + newManifestWork() + controller := newController(t, work, nil, spoketesting.NewFakeRestMapper()). + withKubeObject().withUnstructuredObject() + syncContext := testingcommon.NewFakeSyncContext(t, workKey) + + ctx, entries := captureContext() + if err := controller.toController().sync(ctx, syncContext, work.Name); err != nil { + t.Fatalf("sync: %v", err) + } + + got := flowCounts(entries) + if got[flowSpokeApply] != 1 || got[flowResourceSpokeApply] != 1 || got[flowSpokeApplyResult] != 1 { + t.Errorf("flow counts = %v, want each of mw_spoke_apply/mw_resource_spoke_apply/mw_spoke_apply_result == 1", got) + } +} + +// TestReconcileDedupsRollupBySameGeneration verifies that when the generation has already been +// applied (WorkApplied.ObservedGeneration == generation), no apply-log lines are emitted — +// the once-per-generation guard suppresses a resync. +func TestReconcileDedupsRollupBySameGeneration(t *testing.T) { + setApplyLogsGate(t, true) + defer setApplyLogsGate(t, false) + resetRollupState(t) + + work, workKey := newTestCase("dedup"). + withWorkManifest(testingcommon.NewUnstructured("v1", "Secret", "ns1", "test")). + withExistingWorkCondition(newCondition(workapiv1.WorkApplied, "True", "", "", 0, nil)). + newManifestWork() + controller := newController(t, work, nil, spoketesting.NewFakeRestMapper()). + withKubeObject().withUnstructuredObject() + syncContext := testingcommon.NewFakeSyncContext(t, workKey) + + ctx, entries := captureContext() + if err := controller.toController().sync(ctx, syncContext, work.Name); err != nil { + t.Fatalf("sync: %v", err) + } + + if got := flowCounts(entries); len(got) != 0 { + t.Errorf("flow counts = %v, want none (generation already applied)", got) + } +} + +// TestReconcileRollupSurvivesUnpersistedStatus covers the retry window the persisted guard cannot +// see. sync emits the rollup pair and then writes WorkApplied.ObservedGeneration; when that write +// fails the controller requeues and reconciles the same generation again. Driving sync twice +// against a lister that never observes the status update is that exact sequence, and the latency +// join key must still be emitted once. +func TestReconcileRollupSurvivesUnpersistedStatus(t *testing.T) { + setApplyLogsGate(t, true) + defer setApplyLogsGate(t, false) + resetRollupState(t) + + work, workKey := newTestCase("retry"). + withWorkManifest(testingcommon.NewUnstructured("v1", "Secret", "ns1", "test")). + newManifestWork() + // Seed the AppliedManifestWork the controller would otherwise create on the first sync. The + // fake lister is not fed by that create, so without it the second sync fails on AlreadyExists + // before ever reaching reconcile. The name mirrors applyAppliedManifestWork's + // "-", and the test controller's hubHash is empty. + appliedWork := &workapiv1.AppliedManifestWork{ + ObjectMeta: metav1.ObjectMeta{ + Name: "-" + work.Name, + Finalizers: []string{workapiv1.AppliedManifestWorkFinalizer}, + }, + Spec: workapiv1.AppliedManifestWorkSpec{ManifestWorkName: work.Name}, + } + controller := newController(t, work, appliedWork, spoketesting.NewFakeRestMapper()). + withKubeObject().withUnstructuredObject() + syncContext := testingcommon.NewFakeSyncContext(t, workKey) + + ctx, entries := captureContext() + for i := range 2 { + if err := controller.toController().sync(ctx, syncContext, work.Name); err != nil { + t.Fatalf("sync %d: %v", i+1, err) + } + } + + got := flowCounts(entries) + if got[flowSpokeApply] != 1 || got[flowSpokeApplyResult] != 1 { + t.Errorf("flow counts = %v, want mw_spoke_apply and mw_spoke_apply_result == 1 across both syncs", got) + } +} + +// TestReconcileNoEmitWhenGateOff confirms a full reconcile emits no apply-log lines with the +// gate disabled. +func TestReconcileNoEmitWhenGateOff(t *testing.T) { + setApplyLogsGate(t, false) + resetRollupState(t) + + work, workKey := newTestCase("gate-off"). + withWorkManifest(testingcommon.NewUnstructured("v1", "Secret", "ns1", "test")). + newManifestWork() + controller := newController(t, work, nil, spoketesting.NewFakeRestMapper()). + withKubeObject().withUnstructuredObject() + syncContext := testingcommon.NewFakeSyncContext(t, workKey) + + ctx, entries := captureContext() + if err := controller.toController().sync(ctx, syncContext, work.Name); err != nil { + t.Fatalf("sync: %v", err) + } + + if got := flowCounts(entries); len(got) != 0 { + t.Errorf("flow counts = %v, want none (gate off)", got) + } +} diff --git a/pkg/work/spoke/controllers/manifestcontroller/manifestwork_controller.go b/pkg/work/spoke/controllers/manifestcontroller/manifestwork_controller.go index 9e08248d1..c5c29e2f3 100644 --- a/pkg/work/spoke/controllers/manifestcontroller/manifestwork_controller.go +++ b/pkg/work/spoke/controllers/manifestcontroller/manifestwork_controller.go @@ -132,6 +132,8 @@ func (m *ManifestWorkController) sync(ctx context.Context, controllerContext fac oldManifestWork, err := m.manifestWorkLister.Get(manifestWorkName) if apierrors.IsNotFound(err) { // work not found, could have been deleted, do nothing. + // Drop any apply-log emit state so the ledger tracks live works only. + emittedRollups.forget(manifestWorkName) return nil } if err != nil { diff --git a/pkg/work/spoke/controllers/manifestcontroller/manifestwork_reconciler.go b/pkg/work/spoke/controllers/manifestcontroller/manifestwork_reconciler.go index 75b78b5f7..9c2e52f19 100644 --- a/pkg/work/spoke/controllers/manifestcontroller/manifestwork_reconciler.go +++ b/pkg/work/spoke/controllers/manifestcontroller/manifestwork_reconciler.go @@ -17,11 +17,13 @@ import ( "k8s.io/client-go/util/retry" "k8s.io/klog/v2" + ocmfeature "open-cluster-management.io/api/feature" workapiv1 "open-cluster-management.io/api/work/v1" "open-cluster-management.io/sdk-go/pkg/basecontroller/events" "open-cluster-management.io/sdk-go/pkg/basecontroller/factory" commonhelper "open-cluster-management.io/ocm/pkg/common/helpers" + "open-cluster-management.io/ocm/pkg/features" "open-cluster-management.io/ocm/pkg/work/helper" "open-cluster-management.io/ocm/pkg/work/spoke/apply" "open-cluster-management.io/ocm/pkg/work/spoke/auth" @@ -33,6 +35,9 @@ type applyResult struct { Error error resourceMeta workapiv1.ManifestResourceMeta + // strategy is the resolved update strategy for this manifest, used to classify the apply-log + // rollup counts (read-only manifests apply nothing and are counted separately). + strategy workapiv1.UpdateStrategyType } // resourceApplyOrder defines the priority rank for applying resources by kind. @@ -112,12 +117,36 @@ func (m *manifestworkReconciler) reconcile( // We creat a ownerref instead of controller ref since multiple controller can declare the ownership of a manifests owner := helper.NewAppliedManifestWorkOwner(appliedManifestWork) + // Emit apply logs behind the ManifestWorkApplyLogs gate. The per-resource + // line is emitted deep in applyOneManifest; the two work-level rollup lines are emitted here, once + // per generation, gated on the persisted WorkApplied.ObservedGeneration (restart-safe). + wm := workMeta{ + name: manifestWork.Name, + namespace: manifestWork.Namespace, + generation: manifestWork.Generation, + labels: manifestWork.Labels, + } + logApply := features.SpokeMutableFeatureGate.Enabled(ocmfeature.ManifestWorkApplyLogs) + // emittedRollups.admit records, so it stays last: the checks before it decide whether this + // generation is a candidate at all, and a work they reject must not be marked as emitted. + emitRollup := logApply && + len(manifestWork.Spec.Workload.Manifests) > 0 && + priorAppliedGeneration(manifestWork) < manifestWork.Generation && + emittedRollups.admit(manifestWork.Name, manifestWork.Generation) + if emitRollup { + emitApplyRollup(ctx, wm, flowSpokeApply, len(manifestWork.Spec.Workload.Manifests), nil) + } + var errs []error // Apply resources on spoke cluster. resourceResults := make([]applyResult, len(manifestWork.Spec.Workload.Manifests)) + // A line is emitted per resource apply attempt from inside this retry loop, not once after the + // loop settles, so the timestamp is when the hub change first reached this cluster — that is what + // the hub->spoke latency is calculated from. One line per manifestwork resource apply attempt, + // plus the work-level line at the end of a successful apply. err := retry.RetryOnConflict(retry.DefaultBackoff, func() error { resourceResults = m.applyManifests( - ctx, manifestWork.Spec.Workload.Manifests, manifestWork.Spec, manifestWork.Status, controllerContext.Recorder(), *owner, resourceResults) + ctx, manifestWork.Spec.Workload.Manifests, manifestWork.Spec, manifestWork.Status, controllerContext.Recorder(), *owner, wm, logApply, emitRollup, resourceResults) for _, result := range resourceResults { if apierrors.IsConflict(result.Error) { @@ -191,6 +220,13 @@ func (m *manifestworkReconciler) reconcile( ) } + // Work-level rollup end line, paired with the start line above (same guard). + if emitRollup { + applied, failed, readOnly := countApplyResults(resourceResults) + emitApplyRollup(ctx, wm, flowSpokeApplyResult, len(resourceResults), + &applyCounts{applied: applied, failed: failed, readOnly: readOnly}) + } + return manifestWork, appliedManifestWork, resourceResults, err } @@ -244,6 +280,9 @@ func (m *manifestworkReconciler) applyManifests( workStatus workapiv1.ManifestWorkStatus, recorder events.Recorder, owner metav1.OwnerReference, + wm workMeta, + logApply bool, + firstApply bool, existingResults []applyResult) []applyResult { ordered := m.parseAndSortManifests(manifests) @@ -256,7 +295,7 @@ func (m *manifestworkReconciler) applyManifests( if om.err != nil { existingResults[om.specIndex] = applyResult{Error: om.err, resourceMeta: om.resourceMeta} } else { - existingResults[om.specIndex] = m.applyOneManifest(ctx, om, workSpec, workStatus, recorder, owner) + existingResults[om.specIndex] = m.applyOneManifest(ctx, om, workSpec, workStatus, recorder, owner, wm, logApply, firstApply) } } @@ -269,7 +308,10 @@ func (m *manifestworkReconciler) applyOneManifest( workSpec workapiv1.ManifestWorkSpec, workStatus workapiv1.ManifestWorkStatus, recorder events.Recorder, - owner metav1.OwnerReference) applyResult { + owner metav1.OwnerReference, + wm workMeta, + logApply bool, + firstApply bool) applyResult { logger := klog.FromContext(ctx) result := applyResult{resourceMeta: om.resourceMeta} @@ -309,9 +351,15 @@ func (m *manifestworkReconciler) applyOneManifest( strategy = *option.UpdateStrategy } + result.strategy = strategy.Type applier := m.appliers.GetApplier(strategy.Type) result.Result, result.Error = applier.Apply(ctx, om.gvr, om.obj, requiredOwner, option, recorder) + // Per-resource apply-log line. On the first apply of a generation it emits the + // apply flow; on a later reconcile it emits the sync flow only when the outcome changed vs the + // last-persisted ManifestApplied condition. Noop when gated off or read-only. + emitResourceApply(ctx, logApply, firstApply, wm, om, strategy.Type, result, priorManifestOutcome(manifestCondition)) + return result } diff --git a/vendor/k8s.io/component-base/logs/json/json.go b/vendor/k8s.io/component-base/logs/json/json.go new file mode 100644 index 000000000..20723687e --- /dev/null +++ b/vendor/k8s.io/component-base/logs/json/json.go @@ -0,0 +1,159 @@ +/* +Copyright 2020 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package json + +import ( + "io" + "sync/atomic" + "time" + + "github.com/go-logr/logr" + "github.com/go-logr/zapr" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + + "k8s.io/component-base/featuregate" + logsapi "k8s.io/component-base/logs/api/v1" +) + +var ( + // timeNow stubbed out for testing + timeNow = time.Now +) + +type runtime struct { + v uint32 +} + +func (r *runtime) ZapV() zapcore.Level { + // zap levels are inverted: everything with a verbosity >= threshold gets logged. + return -zapcore.Level(atomic.LoadUint32(&r.v)) +} + +// Enabled implements the zapcore.LevelEnabler interface. +func (r *runtime) Enabled(level zapcore.Level) bool { + return level >= r.ZapV() +} + +func (r *runtime) SetVerbosityLevel(v uint32) error { + atomic.StoreUint32(&r.v, v) + return nil +} + +var _ zapcore.LevelEnabler = &runtime{} + +// NewJSONLogger creates a new json logr.Logger and its associated +// control interface. The separate error stream is optional and may be nil. +// The encoder config is also optional. +func NewJSONLogger(v logsapi.VerbosityLevel, infoStream, errorStream zapcore.WriteSyncer, encoderConfig *zapcore.EncoderConfig) (logr.Logger, logsapi.RuntimeControl) { + r := &runtime{v: uint32(v)} + + if encoderConfig == nil { + encoderConfig = &zapcore.EncoderConfig{ + MessageKey: "msg", + CallerKey: "caller", + NameKey: "logger", + TimeKey: "ts", + EncodeTime: epochMillisTimeEncoder, + EncodeDuration: zapcore.StringDurationEncoder, + EncodeCaller: zapcore.ShortCallerEncoder, + } + } + + encoder := zapcore.NewJSONEncoder(*encoderConfig) + var core zapcore.Core + if errorStream == nil { + core = zapcore.NewCore(encoder, infoStream, r) + } else { + highPriority := zap.LevelEnablerFunc(func(lvl zapcore.Level) bool { + return lvl >= zapcore.ErrorLevel && r.Enabled(lvl) + }) + lowPriority := zap.LevelEnablerFunc(func(lvl zapcore.Level) bool { + return lvl < zapcore.ErrorLevel && r.Enabled(lvl) + }) + core = zapcore.NewTee( + zapcore.NewCore(encoder, errorStream, highPriority), + zapcore.NewCore(encoder, infoStream, lowPriority), + ) + } + l := zap.New(core, zap.WithCaller(true)) + return zapr.NewLoggerWithOptions(l, zapr.LogInfoLevel("v"), zapr.ErrorKey("err")), + logsapi.RuntimeControl{ + SetVerbosityLevel: r.SetVerbosityLevel, + Flush: func() { + _ = l.Sync() + }, + } +} + +func epochMillisTimeEncoder(_ time.Time, enc zapcore.PrimitiveArrayEncoder) { + nanos := timeNow().UnixNano() + millis := float64(nanos) / float64(time.Millisecond) + enc.AppendFloat64(millis) +} + +// Factory produces JSON logger instances. +type Factory struct{} + +var _ logsapi.LogFormatFactory = Factory{} + +func (f Factory) Feature() featuregate.Feature { + return logsapi.LoggingBetaOptions +} + +func (f Factory) Create(c logsapi.LoggingConfiguration, o logsapi.LoggingOptions) (logr.Logger, logsapi.RuntimeControl) { + // We intentionally avoid all os.File.Sync calls. Output is unbuffered, + // therefore we don't need to flush, and calling the underlying fsync + // would just slow down writing. + // + // The assumption is that logging only needs to ensure that data gets + // written to the output stream before the process terminates, but + // doesn't need to worry about data not being written because of a + // system crash or powerloss. + stderr := zapcore.Lock(AddNopSync(o.ErrorStream)) + if c.Options.JSON.SplitStream { + stdout := zapcore.Lock(AddNopSync(o.InfoStream)) + size := c.Options.JSON.InfoBufferSize.Value() + if size > 0 { + // Prevent integer overflow. + if size > 2*1024*1024*1024 { + size = 2 * 1024 * 1024 * 1024 + } + stdout = &zapcore.BufferedWriteSyncer{ + WS: stdout, + Size: int(size), + } + } + // stdout for info messages, stderr for errors. + return NewJSONLogger(c.Verbosity, stdout, stderr, nil) + } + // Write info messages and errors to stderr to prevent mixing with normal program output. + return NewJSONLogger(c.Verbosity, stderr, nil, nil) +} + +// AddNoSync adds a NOP Sync implementation. +func AddNopSync(writer io.Writer) zapcore.WriteSyncer { + return nopSync{Writer: writer} +} + +type nopSync struct { + io.Writer +} + +func (f nopSync) Sync() error { + return nil +} diff --git a/vendor/modules.txt b/vendor/modules.txt index 4398cd02b..d64d2bfee 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -1841,6 +1841,7 @@ k8s.io/component-base/featuregate k8s.io/component-base/logs k8s.io/component-base/logs/api/v1 k8s.io/component-base/logs/internal/setverbositylevel +k8s.io/component-base/logs/json k8s.io/component-base/logs/klogflags k8s.io/component-base/metrics k8s.io/component-base/metrics/features @@ -1951,7 +1952,7 @@ open-cluster-management.io/addon-framework/pkg/agent/v1alpha1 open-cluster-management.io/addon-framework/pkg/assets open-cluster-management.io/addon-framework/pkg/index open-cluster-management.io/addon-framework/pkg/utils -# open-cluster-management.io/api v1.3.1-0.20260709055002-403378b57558 +# open-cluster-management.io/api v1.3.1-0.20260824155636-3ff9ad2f868d ## explicit; go 1.26.0 open-cluster-management.io/api/addon/v1alpha1 open-cluster-management.io/api/addon/v1beta1 diff --git a/vendor/open-cluster-management.io/api/feature/feature.go b/vendor/open-cluster-management.io/api/feature/feature.go index 72f507528..7464af7b2 100644 --- a/vendor/open-cluster-management.io/api/feature/feature.go +++ b/vendor/open-cluster-management.io/api/feature/feature.go @@ -77,6 +77,13 @@ const ( // is not a scalar value. RawFeedbackJsonString featuregate.Feature = "RawFeedbackJsonString" + // ManifestWorkApplyLogs makes the work agent in the spoke cluster emit structured logs on the apply of the manifestwork and its resources + // Log lines are written with three types of logs - mw_spoke_apply at reconcile start, mw_resource_spoke_apply per applied resource, and + // mw_spoke_apply_result at reconcile. + // The lines carry the manifestwork name/namespace/generation, the applied resource identity, an apply + // timestamp, and the manifestwork labels. Read-only applies emit an observation log instead. + ManifestWorkApplyLogs featuregate.Feature = "ManifestWorkApplyLogs" + // ResourceCleanup will start gc controller to clean up resources in cluster ns after cluster is deleted. ResourceCleanup featuregate.Feature = "ResourceCleanup" @@ -154,4 +161,5 @@ var DefaultHubWorkFeatureGates = map[featuregate.Feature]featuregate.FeatureSpec var DefaultSpokeWorkFeatureGates = map[featuregate.Feature]featuregate.FeatureSpec{ ExecutorValidatingCaches: {Default: false, PreRelease: featuregate.Alpha}, RawFeedbackJsonString: {Default: false, PreRelease: featuregate.Alpha}, + ManifestWorkApplyLogs: {Default: false, PreRelease: featuregate.Alpha}, }