mirror of
https://github.com/open-cluster-management-io/ocm.git
synced 2026-08-29 08:37:21 +00:00
✨ Update work-controller to emit logs on ManifestWork apply (#1658)
* chore(work): vendor the ManifestWorkApplyLatency spoke-work feature gate Pull in the new ManifestWorkApplyLatency spoke-work feature gate from open-cluster-management.io/api and re-vendor. Uses a temporary replace directive so the gate resolves before the upstream api change is tagged. This replace MUST be dropped before merge: bump the api require to the merged revision and re-run go mod vendor. Signed-off-by: Ramesh Krishna <ramekris3163@gmail.com> Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * feat(work): emit apply-latency logs from the manifestwork reconciler Add structured apply-latency logging to the spoke work agent, behind the ManifestWorkApplyLatency feature gate (alpha, off by default), to measure hub->spoke ManifestWork propagation latency. On the first apply of a generation the reconciler emits, once per generation, a rollup line at reconcile start (mw_spoke_apply, the propagation instant) and a rollup line at reconcile end (mw_spoke_apply_result, with applied/failed/ read-only counts and outcome), plus a per-resource line (mw_resource_spoke_apply) carrying each resource's outcome (read-only manifests report outcome=read_only, observed but not applied). On a later reconcile of the same generation, a resource re-emits (mw_resource_spoke_sync, with prev_outcome) only when its outcome changed, so steady-state resyncs stay quiet. The once-per-generation guard reads the persisted WorkApplied ObservedGeneration and the per-resource change check reads the persisted ManifestApplied condition, so both are restart-safe with no in-memory state. Timestamps mark the attempt, not eventual success, so the delta reflects one-way propagation. New apply_log.go holds the emitters and helpers; the reconciler threads the ManifestWork identity, the gate flag, and the first-apply flag down to the per-resource emit, and records each result's resolved strategy for the counts. Covered by unit tests for the emitters and helpers and reconcile-level tests for the gate, the once-per-generation dedup, and the gate-off path. Signed-off-by: Ramesh Krishna <ramekris3163@gmail.com> Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * feat: adding json structured logs Signed-off-by: suvaanshkumar <suvaanshkumar@gmail.com> * feat: update ocm to emit json logs instead Signed-off-by: Jeffrey Wong <jeffreywong0417@gmail.com> * feat: revert temp changes used for testing Signed-off-by: Jeffrey Wong <jeffreywong0417@gmail.com> * feat: go mod tidy + vendor Signed-off-by: Jeffrey Wong <jeffreywong0417@gmail.com> * feat: updating flag and addressing review comments Signed-off-by: suvaanshkumar <suvaanshkumar@gmail.com> --------- Signed-off-by: Ramesh Krishna <ramekris3163@gmail.com> Signed-off-by: suvaanshkumar <suvaanshkumar@gmail.com> Signed-off-by: Jeffrey Wong <jeffreywong0417@gmail.com> Co-authored-by: Ramesh Krishna <ramekris3163@gmail.com> Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Co-authored-by: suvaanshkumar <suvaanshkumar@gmail.com>
This commit is contained in:
Jeffrey
co-authored by
Claude Opus 4.8
Ramesh Krishna
suvaanshkumar
Claude Opus 4.8
Ramesh Krishna
suvaanshkumar
parent
b8dd057783
commit
d4b1c465cc
@@ -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
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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...)
|
||||
}
|
||||
@@ -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
|
||||
// "<hubHash>-<workName>", 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)
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
+159
@@ -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
|
||||
}
|
||||
Vendored
+2
-1
@@ -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
|
||||
|
||||
+8
@@ -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},
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user