watch

package
v0.47.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 16, 2026 License: Apache-2.0 Imports: 45 Imported by: 0

Documentation

Overview

Package watch drives the api-source-of-truth reconcile: it keeps the followability registry and the per-GitTarget watched-type tables fresh, and mirrors each watched type into Git from a long-lived Kubernetes watch per claimed (GitTarget, GVR, scope). The live watch is the only source of ongoing object state, and its events are the writes. The INITIAL desired set the mark-and-sweep folds over the Git folder comes from the same watch's sendInitialEvents replay, or, on an apiserver that does not serve it, from a LIST with the watch's events buffered behind it (targetWatchListAndStream). See docs/architecture.md.

Index

Constants

View Source 
const (
	// TriggerReasonDeclare is the GitTarget controller submitting a declaration.
	TriggerReasonDeclare = "declare"
	// TriggerReasonRuleChange is a WatchRule or ClusterWatchRule edit, refusal, or deletion.
	TriggerReasonRuleChange = "rule_change"
	// TriggerReasonSharedRefresh is a shared snapshot moving under a target: a CRD installed or
	// removed, an APIService appearing, a source-namespace grant or revocation. It is derived
	// rather than posted — the refresh compares each target's rendered plan across the
	// re-projection and marks only the ones it actually changed.
	TriggerReasonSharedRefresh = "shared_refresh"
	// TriggerReasonPeriodic is the 30s floor, the one path that still marks everything.
	TriggerReasonPeriodic = "periodic"
)

Trigger reasons, as they appear on the triggers-by-reason metric and in logs. There are three because there are three ways intent reaches the owner, and telling them apart is what shows which source is noisy.

View Source 
const (
	StreamReasonInitialReplay          = "InitialReplay"
	StreamReasonResumeReplay           = "ResumeReplay"
	StreamReasonExpiredResourceVersion = "ExpiredResourceVersion"
	StreamReasonWatchError             = "WatchError"
	StreamReasonWatchNotPermitted      = "WatchNotPermitted"
	StreamReasonAllStreamsReady        = "AllStreamsReady"
	StreamReasonReplaying              = "Replaying"
	StreamReasonNoResolvedTypes        = "NoResolvedTypes"
)
View Source 
const (
	// ClusterWatchRuleReasonGitTargetNamespaceNotAuthorized is the terminal reason when the
	// referenced GitTarget's namespace is not admitted by that target's ClusterProvider — either
	// because spec.accessFrom excludes it or because the provider does not exist at all.
	//
	// One rule-side reason covers both provider-side causes on purpose: from the ClusterWatchRule's
	// point of view the single fact that matters is that this rule may not compile against this
	// target. The Message carries which of the two it was.
	ClusterWatchRuleReasonGitTargetNamespaceNotAuthorized = "GitTargetNamespaceNotAuthorized"
)

ClusterWatchRule compile refusal reasons. They live here rather than in the controller because bootstrap refuses on the same grounds without a controller in sight, and one vocabulary is what keeps the two from drifting.

View Source 
const DefaultAttributionGraceWindow = 3 * time.Second

DefaultAttributionGraceWindow is the bounded wait a watch event spends for a matching audit fact to arrive in the index before it ships as committer. It is the "slack" that makes "a late audit arrival must not rewrite a shipped commit" enforceable: we wait briefly BEFORE shipping rather than rewrite afterwards.

Variables

This section is empty.

Functions

func CompileWatchRule added in v0.39.0 

func CompileWatchRule(
	ctx context.Context,
	reader client.Reader,
	store *rulestore.RuleStore,
	rule configv1alpha3.WatchRule,
	target configv1alpha3.GitTarget,
	provider configv1alpha3.GitProvider,
) (authz.ResolvedSourceScope, error)

CompileWatchRule is THE ONLY PATH from a WatchRule to a compiled rule. It resolves the whole per-item source-namespace scope first and compiles only on an admitted verdict.

It is one function, called by both the WatchRule reconciler and the watch manager's startup bootstrap, because two call sites that each remember to check is an arrangement this codebase has already got wrong once. Bootstrap lists every WatchRule and seeds the store BEFORE the first reconcile, then marks the store ready — so a gate the reconciler alone enforced would be bypassed for the whole startup window, on EVERY restart, which is exactly when nobody is watching. Routing compilation through here closes that by construction rather than by discipline: there is no second place that can call AddOrUpdateWatchRule for a WatchRule.

Its two outcomes map onto the two things the caller must do:

  • ADMITTED — the rule is compiled with every item resolved to its source-namespace set. The caller publishes SourceNamespaceAuthorized=True.
  • DENIED — any previously compiled rule is REMOVED here, before the caller publishes anything. A gate that only writes a condition is not a gate; the caller must still replan the watch manager and then publish the Failed trio, in that order.

There is no third "cannot say yet" outcome, and there used to be. It existed because the gate's selector half read Namespace labels in ANOTHER cluster, which could be syncing, unreachable or Forbidden, and a rule with an already-resolved scope had to RETAIN it through such a gap rather than narrow to the empty set (a narrowed set is the input to a sweep, so failing closed there would have deleted a tenant's Git content over an outage). Every input is now a control-plane object this reconcile already read, so there is nothing left to retain a scope through, and the per-rule grant memory that existed only to serve that retention is gone with it.

Bootstrap cannot publish status (it runs before controllers start), so a rule denied there is simply not compiled and the first reconcile writes the terminal condition. That ordering — fail closed first, explain second — is correct, not a limitation.

Types

type APIResourceCatalog 

type APIResourceCatalog struct {
	// contains filtered or unexported fields
}

APIResourceCatalog is the per-scan normalizer between Kubernetes discovery and the typeset registry: it turns one ServerGroupsAndResources() result into a policy-annotated typeset.Scan. It holds NO judgement and no time-sensitive state — retain-on-error and the removal grace for omissions both live in typeset.Registry.UpdateFromScan (see docs/spec/typeset-owns-discovery-grace.md). The only state kept is mechanical bookkeeping: the last normalized scan (the change fingerprint, and the registry's re-derive source for refreshes without a discovery round-trip), the scan generation, and readiness.

func NewAPIResourceCatalog 

func NewAPIResourceCatalog() *APIResourceCatalog

NewAPIResourceCatalog constructs an empty API resource catalog.

func (*APIResourceCatalog) DegradedGroupVersions 

func (c *APIResourceCatalog) DegradedGroupVersions() []schema.GroupVersion

DegradedGroupVersions returns the group/versions the latest scan reported as failed, sorted for stable logging.

func (*APIResourceCatalog) Generation 

func (c *APIResourceCatalog) Generation() uint64

Generation reports the current scan generation. It bumps only when the normalized scan facts change, so downstream consumers gated on it do not churn on steady rescans.

func (*APIResourceCatalog) Ready 

func (c *APIResourceCatalog) Ready() bool

Ready reports whether the catalog has accepted any trusted discovery data.

func (*APIResourceCatalog) Refresh 

func (c *APIResourceCatalog) Refresh(disco apiResourceDiscovery) (bool, error)

Refresh normalizes one discovery scan and stores it as the latest. It returns whether the normalized facts changed from the previous scan. It makes no retention decision of any kind: a group/version this scan failed on or no longer lists is simply reported as such in the scan; the registry judges what that means.

func (*APIResourceCatalog) Scan 

func (c *APIResourceCatalog) Scan(sensitive types.SensitiveResourcePolicy) (typeset.Scan, bool)

Scan returns the last normalized scan, policy-annotated and stamped with the current generation — the registry's one input (UpdateFromScan). ok is false before the first trusted scan. sensitive is the operator-configured SensitiveResourcePolicy, applied at projection exactly like the allow/deny resource policy is applied at normalization: a startup-known fact typeset never infers.

func (*APIResourceCatalog) ServesWatchable 

func (c *APIResourceCatalog) ServesWatchable(gvr schema.GroupVersionResource) bool

ServesWatchable reports whether the latest scan saw this exact group/version/resource served with both list and watch verbs. It is the precondition for opening an informer on a resource the operator does not require: an API server that does not aggregate does not serve apiregistration.k8s.io at all, and a blind informer on it retries forever. It deliberately ignores the allow/deny watch policy, which governs what the operator mirrors, not what it may inspect. Reports false before the first trusted scan.

func (*APIResourceCatalog) Stats 

func (c *APIResourceCatalog) Stats() CatalogStats

Stats returns a point-in-time summary of the latest scan for metrics.

type AttributionLookup 

type AttributionLookup interface {
	// Await resolves the strongest author fact for a watch event, waiting up to grace for one that
	// has not been delivered yet. It returns an AttributionAbsent resolution when nothing matched in
	// time; it never blocks longer than grace and never returns an error path.
	Await(ctx context.Context, query queue.FactQuery, grace time.Duration) queue.AuthorResolution
}

AttributionLookup is the read side of the attribution fact index. The in-process queue.FactIndex satisfies it; nil means configured-author.

It waits rather than polls. The index registers the waiter BEFORE it reads itself, so a fact delivered in the gap between the two wakes a waiter that is already listening — the race the old 150ms poll loop papered over by looking again. There is no Redis call anywhere on this path: the fast case is a map read and the waiting case is a channel receive.

type AuthorQuery added in v0.41.0 

type AuthorQuery struct {
	// AuditRoute partitions the facts. It is the route the API server posts audit under, NOT the
	// ClusterProvider's name: several providers may name one cluster and all read its facts.
	AuditRoute string
	// GVR is the watched type. The version serves the metric labels only; the join is keyed on the
	// group/resource, which is what a fact carries.
	GVR             schema.GroupVersionResource
	UID             k8stypes.UID
	ResourceVersion string
	// Namespace and Labels serve the collection tier only: they are how a removal finds the
	// deletecollection whose scope covered it.
	Namespace string
	Labels    map[string]string
	// Name serves the name tier only, the floor for a fact carrying neither a uid nor a
	// resourceVersion — an aggregated-API write, whose audit objectRef holds the name and nothing
	// else. The watch event always carries it, so supplying it costs nothing when no fact needs it.
	Name string
	// ExactCapable is true for ADDED and MODIFIED, whose resourceVersion is the one the write
	// produced. A removal's is not, so it consults the weaker tiers the exact-capable events skip.
	ExactCapable bool
}

AuthorQuery is one watch event's identity, as the author resolver reads it.

It is a struct rather than the parameter list this used to be because the collection tier needs the object's namespace and labels: a body-less deletecollection is joined by scope — type, namespace, selector, window — and neither the namespace nor the labels could be expressed in the old six arguments, so that tier would have been unreachable from the resolver.

type AuthorResolver 

type AuthorResolver interface {
	// ResolveAuthor returns the author UserInfo for a watch event together with the
	// attribution OUTCOME. It may wait up to the grace window for a matching fact; it never
	// blocks indefinitely and never returns an error path. exactCapable distinguishes
	// ADDED/MODIFIED events (true) from known RV-mismatch removals (false).
	//
	// The outcome is returned explicitly rather than as an ok bool because the two possible
	// "no author" cases are NOT the same and callers must be able to tell them apart:
	// AttributionNotAttempted (configured-author mode — the committer legitimately is the
	// author) versus AttributionUnresolved (attribution ran and found nothing — a gap worth
	// surfacing). An empty UserInfo cannot distinguish them, which is exactly how the loss
	// stayed invisible. A resolved outcome always carries a non-empty UserInfo.
	//
	// In production this method only ever returns the latter two: configured-author mode is
	// expressed by leaving Manager.AuthorResolver nil (attachAuthor returns early, leaving the
	// event's zero AttributionNotAttempted), never by constructing a resolver over a nil
	// lookup. cmd/main.go only builds one with a non-nil index.
	ResolveAuthor(ctx context.Context, query AuthorQuery) (git.UserInfo, git.AttributionOutcome)
}

AuthorResolver names the commit author for a live watch event from audit facts.

func NewAuthorResolver 

func NewAuthorResolver(
	lookup AttributionLookup,
	grace time.Duration,
	log logr.Logger,
	health *attribution.RouteHealth,
) AuthorResolver

NewAuthorResolver builds the conservative author resolver over the attribution index. grace bounds the per-event wait for a late fact; a zero grace disables waiting (single lookup). A matched actor — human or service account — is always named by its own username.

health is the process-wide route registry the audit ingress publishes into; nil is accepted (every method is nil-safe) and only costs the never-resolves warning.

type CatalogStats 

type CatalogStats struct {
	// AllowedResources is the count of served top-level resources the default
	// watch policy permits.
	AllowedResources int
	// ExcludedResources is the count of served top-level resources the default
	// watch policy excludes (pods, events, leases, …).
	ExcludedResources int
	// TrustedGroupVersions is the count of group/versions the latest scan served cleanly.
	TrustedGroupVersions int
	// DegradedGroupVersions is the count of group/versions the latest scan reported as failed.
	DegradedGroupVersions int
	// Generation is the current scan generation.
	Generation uint64
}

CatalogStats is a point-in-time summary of the latest scan used to set the api_catalog_* gauges. All counts exclude subresources.

type ClusterWatchRuleDecision added in v0.39.0 

type ClusterWatchRuleDecision struct {
	// Admitted reports whether the rule compiled.
	Admitted bool
	// Reason is the terminal condition reason when it did not.
	Reason string
	// Message explains the refusal to an operator.
	Message string
}

ClusterWatchRuleDecision is the outcome of the shared ClusterWatchRule compile path.

func CompileClusterWatchRule added in v0.39.0 

func CompileClusterWatchRule(
	ctx context.Context,
	reader client.Reader,
	store *rulestore.RuleStore,
	rule configv1alpha3.ClusterWatchRule,
	target configv1alpha3.GitTarget,
	provider configv1alpha3.GitProvider,
) (ClusterWatchRuleDecision, error)

CompileClusterWatchRule is THE ONLY PATH from a ClusterWatchRule to a compiled cluster rule, and it is the compile-time half of the cluster-scope-only narrowing.

Two refusals, both terminal, in this order:

  1. the referenced GitTarget's namespace must be admitted by that target's ClusterProvider — a ClusterWatchRule's gitTargetRef carries a namespace, so it can name a target in ANY namespace and widen that target's mirror scope cluster-wide;
  2. the rule must not carry a stored scope other than "Cluster". Admission rejects the value on write, but a pre-release object keeps it in etcd, and resolving it as if it had asked for cluster scope would silently change what a running rule mirrors.

Like CompileWatchRule it is shared by the reconciler and the startup bootstrap, so a restart cannot open an unauthorized or namespaced watch before the first reconcile can publish status.

type CursorStore 

type CursorStore interface {
	LookupWatchCursor(
		ctx context.Context,
		gitTargetUID string,
		gvr schema.GroupVersionResource,
		namespace string,
	) (string, bool)
	RecordWatchCursor(
		ctx context.Context,
		gitTargetUID string,
		gvr schema.GroupVersionResource,
		namespace, rv string,
	) error
}

CursorStore persists the last processed resourceVersion for each (GitTarget UID, GVR, scope) watch shard, bounded by a TTL. The GitTarget is identified by its UID alone — globally unique, so namespace/name would be redundant. Cursors are refreshed on write and never deleted: a live watch keeps its cursor fresh, a dead one's cursor expires. Nil means every new watch session rebuilds from a fresh replay.

type DeclareStatus added in v0.42.0 

type DeclareStatus struct {
	// Declared reports whether the GitTarget has ever reached DeclareForGitTarget. It is the
	// observable form of the capture-on-Declare contract: a GitTarget the controller's Validated
	// gate refused never reaches Declare, so it never appears here, which makes "an unauthorized
	// namespace starts no watch" assertable from outside this package.
	Declared bool
	// ClusterID is the source cluster the declaration named — a ClusterProvider name, empty for
	// the cluster the operator runs in. Meaningful only when Declared.
	ClusterID string
	// Pending reports whether a plan pass is owed — the target is dirty, or has never
	// successfully applied a plan.
	Pending bool
	// Failures counts consecutive failed passes; LastError carries the most recent message.
	Failures  int
	LastError string
}

DeclareStatus is the observable state of a GitTarget's declaration: whether the owner still owes it a plan pass, and how the last one ended. It is what makes a target whose passes keep timing out read as failing rather than as idle.

func (DeclareStatus) Settled added in v0.42.0 

func (s DeclareStatus) Settled() bool

Settled reports whether the target's declaration has landed: it declared, nothing is owed, and the last pass succeeded.

type EventRouter 

type EventRouter struct {
	WorkerManager *git.WorkerManager
	WatchManager  *Manager
	Client        client.Client
	Log           logr.Logger
	// contains filtered or unexported fields
}

EventRouter orchestrates control flow between components. It dispatches live events to BranchWorkers, routes them through per-GitTarget event streams for buffering and deduplication, and drives the synchronous streaming-snapshot resync (M8).

func NewEventRouter 

func NewEventRouter(
	workerManager *git.WorkerManager,
	watchManager *Manager,
	client client.Client,
	log logr.Logger,
) *EventRouter

NewEventRouter creates a new event router.

func (*EventRouter) GetGitTargetEventStream 

func (r *EventRouter) GetGitTargetEventStream(gitDest types.ResourceReference) *reconcile.GitTargetEventStream

GetGitTargetEventStream returns the registered GitTargetEventStream for a GitTarget.

func (*EventRouter) RegisterGitTargetEventStream 

func (r *EventRouter) RegisterGitTargetEventStream(
	gitDest types.ResourceReference,
	stream *reconcile.GitTargetEventStream,
)

RegisterGitTargetEventStream registers a GitTargetEventStream with the router. This allows routing events to specific GitTargetEventStreams for buffering and deduplication.

func (*EventRouter) RouteEvent 

func (r *EventRouter) RouteEvent(
	providerName, providerNamespace string,
	branch string,
	event git.Event,
) error

RouteEvent sends an event to the worker for (provider, branch). The target info is used to lookup the worker, then the event is queued. Returns an error if no worker exists for the given (provider, branch) combination.

func (*EventRouter) RouteToGitTargetEventStream 

func (r *EventRouter) RouteToGitTargetEventStream(
	event git.Event,
	gitDest types.ResourceReference,
) error

RouteToGitTargetEventStream routes an event to a specific GitTargetEventStream. This replaces direct routing to BranchWorkers, enabling event buffering and deduplication.

func (*EventRouter) ServiceCommitRequest 

func (r *EventRouter) ServiceCommitRequest(
	ctx context.Context,
	attach git.AttachCommitRequest,
) (git.FinalizeResult, bool, error)

ServiceCommitRequest is the controller's attach-then-poll seam (§6.4.3): it resolves the GitTarget's branch worker, registers the CommitRequest attach idempotently on that worker's FIFO event queue (bind the message to the author's open window, finalize after the grace), and returns the request's current outcome. resolved=false means the worker has not finished — the controller requeues and polls again.

attach.GitTargetName/GitTargetNamespace name the GitTarget; the worker is keyed by its provider+branch. When no worker exists there is, by definition, no window to collect into, so the request resolves NoOpenWindow (as before). A GitTarget that cannot be read is a transient error the controller surfaces and retries.

func (*EventRouter) UnregisterGitTargetEventStream 

func (r *EventRouter) UnregisterGitTargetEventStream(gitDest types.ResourceReference)

UnregisterGitTargetEventStream removes a GitTargetEventStream from the router. This is called during GitTarget deletion cleanup.

type GVR 

type GVR struct {
	Group    string
	Version  string
	Resource string
	Scope    configv1alpha3.ResourceScope
}

GVR represents a concrete Group/Version/Resource target with a scope. This is used to plan dynamic informer creation from active rules.

type GitPathAcceptanceStatus 

type GitPathAcceptanceStatus struct {
	Accepted       bool
	Reason         string
	Message        string
	At             metav1.Time
	RefusedCell    types.CellKey
	RefusedCellSet bool
}

GitPathAcceptanceStatus is the whole-target write-safety status for a GitTarget path.

type Manager 

type Manager struct {
	// Client provides cluster access.
	Client client.Client
	// Log is the logger to use.
	Log logr.Logger
	// RuleStore gives access to compiled WatchRule/ClusterWatchRule.
	RuleStore *rulestore.RuleStore
	// EventRouter dispatches per-type reconciles/sweeps and field-patch events to branch workers.
	EventRouter *EventRouter
	// AuthorResolver optionally names the commit author for a live watch event by
	// joining the audit attribution index (RV/UID match, bounded grace window). Nil
	// is configured-author mode (no audit/Redis): every event commits as the committer.
	AuthorResolver AuthorResolver
	// FactStreams is the process-wide subscription set the attribution fact follower reads. Each
	// running target watch holds one reference on the (audit route, group/resource) it covers, so
	// the process follows a type while at least one watch needs it and stops following it when the
	// last one goes away — which is the whole point of the per-type fan-out: facts for a type nobody
	// watches are appended and never received. Nil is configured-author mode: no follower runs, so
	// no subscription is taken.
	FactStreams *queue.FactStreamSet
	// WatchCursorStore optionally persists per-watch resourceVersion cursors so
	// reconnects can resume without replaying the full type snapshot.
	WatchCursorStore CursorStore
	// SensitiveResources is the startup-configured policy classifying which types must
	// use the encrypted Git write path. It is applied when the followability registry
	// builds its observations, so each TypeRecord carries the right Sensitive fact. The
	// zero value still treats core Secrets as sensitive.
	SensitiveResources types.SensitiveResourcePolicy

	// SourceClusters resolves a GitTarget's source cluster — a ClusterProvider NAME — into a
	// rest.Config, reading the kubeconfig Secret the provider names from the config plane. It is
	// required for any GitTarget to mirror, single-cluster installs included: a source cluster is
	// always a ClusterProvider, and only this resolver can say whether that provider is in-cluster
	// (kubeConfig omitted) or remote. Nil leaves every source cluster unresolvable; only the config
	// plane, which needs no provider, still works.
	SourceClusters SourceClusterResolver
	// contains filtered or unexported fields
}

Manager is a controller-runtime Runnable that keeps the followability registry and the per-GitTarget watched-type tables fresh, and owns every target watch. It holds no object informers and no object cache: each stream is a raw watch, opened per claimed (GVR, scope) against the GitTarget's own source cluster. The only API touch on a schedule is the discovery refresh that keeps the catalogs current. See docs/architecture.md.

func (*Manager) ClusterTypeLookup added in v0.38.0 

func (m *Manager) ClusterTypeLookup(clusterID string) typeset.Lookup

ClusterTypeLookup returns the GVK->GVR resolver the git writer scans a folder's manifests with, scoped to ONE source cluster — its own registry. A folder is owned by exactly one GitTarget (one materialization), so the writer resolves each document against that GitTarget's cluster, never a union: two clusters can validly serve one GVK under different GVRs/scopes, and a first-wins union would mis-file or delete manifests. In a single-cluster install every target resolves against the one local registry, unchanged. An unknown cluster id yields the (possibly unready) context registry, which fails closed via the acceptance gate.

func (*Manager) ComputeRequestedGVRs 

func (m *Manager) ComputeRequestedGVRs() []GVR

ComputeRequestedGVRs aggregates the watched GVRs from the active RuleStore: the union of every GitTarget's watched types, read from the resident tables.

func (*Manager) DeclareForGitTarget 

func (m *Manager) DeclareForGitTarget(
	gitDest types.ResourceReference,
	clusterID string,
	auditRoute string,
	pruneMode v1alpha3.PruneMode,
	forceRecheck ...bool,
)

DeclareForGitTarget submits the GitTarget's declaration to the watch-plane owner and returns. It does no work on the caller's goroutine.

clusterID is (api/v1alpha3).GitTarget.SourceCluster() — the referenced ClusterProvider's name, "default" for the cluster the operator runs in. It is captured with the UID and the audit route, because none of the three can be derived from the rule-derived watch tables the owner plans from.

pruneMode is (api/v1alpha3).GitTarget.EffectivePruneMode(). Unlike the other two it is mutable, and widening it to a sweeping mode forces a fresh replay — see prune_declaration.go for why the edge, and only that edge, has to be the trigger.

It returns no error, and that is the contract: the pass runs after the target has been quiet for the settle window, so there is no result to hand back here. The controller reads DeclareStatusForGitTarget for whether the declaration has landed and how the last pass ended. A GitTarget applied together with its WatchRules is ONE piece of configuration, and waiting for the silence window is what makes the first plan it declares the one the user wrote rather than an empty one superseded four times as each rule arrives.

func (*Manager) DeclareStatusForGitTarget added in v0.42.0 

func (m *Manager) DeclareStatusForGitTarget(gitDest types.ResourceReference) DeclareStatus

DeclareStatusForGitTarget reports whether the owner still owes this GitTarget a pass and how the last one ended. A target whose passes keep timing out must not read as idle.

func (*Manager) FollowableTypeRecords 

func (m *Manager) FollowableTypeRecords() []typeset.TypeRecord

FollowableTypeRecords returns the LOCAL cluster's currently-followable type records (verdict followable or retained), sorted by identity. It is the inventory the status and visibility surfaces read; it never recomputes followability.

func (*Manager) ForgetGitTargetDeclaration 

func (m *Manager) ForgetGitTargetDeclaration(gitDest types.ResourceReference)

ForgetGitTargetDeclaration drops in-memory watch state for a deleted GitTarget, and tears down its source cluster's context when it was the last GitTarget mirroring from it.

The pending trigger is dropped synchronously, in the same step that queues the teardown: a trigger that outlived the delete would have the next pass re-create what the delete tore down. The teardown itself carries the UID it was issued for, so a GitTarget deleted and recreated under the same namespace and name cannot have its successor's watches torn down by its predecessor's delete.

func (*Manager) GitPathAcceptanceForGitTarget 

func (m *Manager) GitPathAcceptanceForGitTarget(gitDest types.ResourceReference) GitPathAcceptanceStatus

GitPathAcceptanceForGitTarget returns the latest acceptance status for the GitTarget. Missing state means no refusal has been observed, so the path is accepted.

func (*Manager) GitPathEvents 

func (m *Manager) GitPathEvents() <-chan event.GenericEvent

GitPathEvents returns the channel the GitTarget controller wires via source.Channel so a GitPath acceptance transition enqueues the owning GitTarget. It is lazily created so a zero-value Manager (tests) and the cmd-wired Manager share one channel.

func (*Manager) LayoutForGitTarget added in v0.43.0 

func (m *Manager) LayoutForGitTarget(gitDest types.ResourceReference) (git.LayoutReport, bool)

LayoutForGitTarget returns the most recently resolved layout for a GitTarget, and whether one has been resolved at all. Absent means no scan has reported yet — which is different from a folder that resolved to nothing, and the controller has to distinguish the two: the first leaves LayoutResolved Unknown, the second sets it with reason None.

func (*Manager) MarkTargetGitPathAccepted 

func (m *Manager) MarkTargetGitPathAccepted(gitDest types.ResourceReference)

MarkTargetGitPathAccepted clears any prior unscoped refusal for the GitTarget path. Scoped resyncs should call MarkTargetGitPathScopeAccepted so they only clear the refusal they proved.

func (*Manager) MarkTargetGitPathRefused 

func (m *Manager) MarkTargetGitPathRefused(gitDest types.ResourceReference, reason, message string)

MarkTargetGitPathRefused records that the GitTarget path failed the structure-only acceptance gate. The refusal is target-wide, not stream-specific.

It is a report from a branch-worker drain goroutine. Only a real TRANSITION republishes the snapshot and enqueues a reconcile, so the happy-path resync stream does not enqueue one per event.

func (*Manager) MarkTargetGitPathScopeAccepted added in v0.45.1 

func (m *Manager) MarkTargetGitPathScopeAccepted(gitDest types.ResourceReference, cell types.CellKey)

MarkTargetGitPathScopeAccepted clears a scoped refusal only when the successful resync covered the same cell that produced the refusal. A successful replay of secrets must not hide a still impossible configmaps path, because the GitTarget status is the operator's only durable clue.

func (*Manager) MarkTargetGitPathScopeRefused added in v0.45.1 

func (m *Manager) MarkTargetGitPathScopeRefused(
	gitDest types.ResourceReference,
	cell types.CellKey,
	reason string,
	message string,
)

MarkTargetGitPathScopeRefused records that a scoped writer pass failed the structure-only acceptance gate. The status remains target-level, but the remembered scope prevents an unrelated successful resync from clearing it.

func (*Manager) MarkTargetRenderFidelityDiverged added in v0.37.0 

func (m *Manager) MarkTargetRenderFidelityDiverged(
	target types.ResourceReference,
	divergence manifestanalyzer.RenderDivergence,
)

MarkTargetRenderFidelityDiverged closes normal writes immediately when a live window hits the same boundary outside a scoped replay. Restarting every scope is the only recovery route.

func (*Manager) MarkTargetRenderFidelityScopeClean added in v0.37.0 

func (m *Manager) MarkTargetRenderFidelityScopeClean(
	target types.ResourceReference,
	revision uint64,
	cell types.CellKey,
)

MarkTargetRenderFidelityScopeClean records one complete clean replay result from the cell's current stream. A stale cancellation tail carries the retired stream's revision, so the gate ignores it and it cannot reopen a failed target.

func (*Manager) MarkTargetRenderFidelityScopeDiverged added in v0.37.0 

func (m *Manager) MarkTargetRenderFidelityScopeDiverged(
	target types.ResourceReference,
	revision uint64,
	cell types.CellKey,
	divergence manifestanalyzer.RenderDivergence,
)

MarkTargetRenderFidelityScopeDiverged records a replay refusal caused by a rendered token.

func (*Manager) MarkTargetRetention added in v0.39.0 

func (m *Manager) MarkTargetRetention(
	gitDest types.ResourceReference,
	cell types.CellKey,
	revision uint64,
	mode v1alpha3.PruneMode,
	retained int,
)

MarkTargetRetention records what one scope's resync retained.

Scope lifecycle is handled by the watch plan rather than by eviction here: retainTargetRetentionScopes installs the selected cells and the revision each one's stream reports under, so a cell that leaves the plan takes its count with it, and a stale in-flight reply from a cancelled watch — which carries the retired stream's revision — cannot resurrect or overwrite a count. A report for a cell the plan does not hold is dropped for the same reason.

Zero is recorded as actively as any other number: it is the converged signal.

func (*Manager) NeedLeaderElection 

func (m *Manager) NeedLeaderElection() bool

NeedLeaderElection ensures only the elected leader runs the watch manager.

func (*Manager) RenderFidelityForGitTarget added in v0.37.0 

func (m *Manager) RenderFidelityForGitTarget(target types.ResourceReference) RenderFidelityStatus

RenderFidelityForGitTarget returns the latest condition projection. Missing state means the target has not installed watches yet and remains writable for compatibility.

func (*Manager) ReportGitPathRefusal 

func (m *Manager) ReportGitPathRefusal(
	gitDest types.ResourceReference,
	cell types.CellKey,
	refused *manifestanalyzer.AcceptanceRefusedError,
)

ReportGitPathRefusal records a write plan the branch worker refused on a live-event path, where no result channel carries the error back to the router. It is installed on the WorkerManager (git.GitPathRefusalReporter) at startup, and applies the same reason mapping the resync path uses, so a refusal reaches the user as GitPathAccepted=False / Stalled=True whether it was a live write or a background resync that hit it.

func (*Manager) ReportLayoutResolved added in v0.43.0 

func (m *Manager) ReportLayoutResolved(gitDest types.ResourceReference, report git.LayoutReport)

ReportLayoutResolved records what a scan resolved about a GitTarget's folder. It is installed on the WorkerManager (git.LayoutReporter) at startup, and it is the only writer of the layouts projection.

Every scan calls it, including the ones that resolve exactly what the last one did — the branch worker has no memory across scans and should not grow one. The transition test lives here, so a steady-state target republishes nothing and enqueues no reconcile, and a folder whose shape actually changed does both.

func (*Manager) ResolveClusterWatchRuleResources 

func (m *Manager) ResolveClusterWatchRuleResources(
	_ context.Context,
	rule configv1alpha3.ClusterWatchRule,
) (bool, string)

ResolveClusterWatchRuleResources reports one ClusterWatchRule's resource-resolution status for controller feedback. See resolveRuleResourceStatus. A ClusterWatchRule is cluster-scoped, so its gitTargetRef names both the GitTarget and its namespace; the status resolves against that GitTarget's source cluster.

func (*Manager) ResolveWatchRuleResources 

func (m *Manager) ResolveWatchRuleResources(
	_ context.Context,
	rule configv1alpha3.WatchRule,
) (bool, string)

ResolveWatchRuleResources reports one WatchRule's resource-resolution status for controller feedback. See resolveRuleResourceStatus. A WatchRule's GitTarget lives in the WatchRule's own namespace, and the status resolves against THAT GitTarget's source cluster — so a remote-only CRD is reported as watched, and a local-only CRD selected by a remote target is not, instead of both being answered from the local registry.

func (*Manager) RetentionForGitTarget added in v0.39.0 

func (m *Manager) RetentionForGitTarget(gitDest types.ResourceReference) RetentionSummary

RetentionForGitTarget returns the roll-up across the target's currently tracked scopes.

func (*Manager) SetupWithManager 

func (m *Manager) SetupWithManager(mgr ctrl.Manager) error

SetupWithManager is a placeholder to enable kubebuilder RBAC marker scanning. The Manager is manually added to the controller-runtime manager in main.go as a Runnable, but this method allows kubebuilder's controller-gen to discover and process the RBAC markers.

func (*Manager) SourceClusterReachable added in v0.38.0 

func (m *Manager) SourceClusterReachable(clusterID string) SourceClusterReachableStatus

SourceClusterReachable projects a source cluster's runtime reachability for a GitTarget's SourceClusterReachable condition. The local cluster is always reachable; a remote is Unknown until the data plane's first discovery attempt, then True or False with a classified reason.

func (*Manager) Start 

func (m *Manager) Start(ctx context.Context) error

Start begins the watch ingestion manager and blocks until context cancellation. Everything the watch plane does happens on the loop it enters: controllers submit intent, this owns the state.

func (*Manager) StreamStateEvents added in v0.42.0 

func (m *Manager) StreamStateEvents() <-chan event.GenericEvent

StreamStateEvents returns a channel carrying a GenericEvent for the GitTarget whose per-cell stream readiness just changed, so a controller projecting that readiness re-reconciles on the transition instead of on its next periodic requeue.

Unlike GitPathEvents this is a FAN-OUT: every call registers a new subscriber and returns its own channel, because a Go channel has one consumer and three controllers project this state (the GitTarget and both rule kinds). Sending is best-effort into each subscriber's buffer.

Why it exists: a stream reaching Streaming is the last thing that has to happen before a rule or a target can honestly report StreamsRunning=True, and it was the one report on the watch plane that notified nobody. The data plane would converge in about two seconds and the status would follow up to ten seconds later, on RequeueStreamSettleInterval, because nothing said so.

func (*Manager) StreamSummaryForClusterWatchRule 

func (m *Manager) StreamSummaryForClusterWatchRule(rule configv1alpha3.ClusterWatchRule) StreamSummary

StreamSummaryForClusterWatchRule reports stream readiness for one ClusterWatchRule, resolved against the source cluster its GitTarget mirrors from. It always matches cluster-scoped records, because a ClusterWatchRule is cluster-scope-only.

func (*Manager) StreamSummaryForGitTarget 

func (m *Manager) StreamSummaryForGitTarget(gitDest types.ResourceReference) StreamSummary

StreamSummaryForGitTarget reports the GitTarget stream-readiness roll-up.

func (*Manager) StreamSummaryForWatchRule 

func (m *Manager) StreamSummaryForWatchRule(rule configv1alpha3.WatchRule) StreamSummary

StreamSummaryForWatchRule reports stream readiness for one namespaced WatchRule, resolved against the source cluster its GitTarget mirrors from.

It reads the COMPILED rule, not the spec. A rule's watched namespaces can no longer be derived from its spec at all: a `sourceNamespace: "*"` item's set exists only after resolution against the GitTarget's policy and the source-cluster snapshot. Rebuilding the keys from the spec would look for streams under keys that were never opened, so a perfectly healthy wildcard rule would report permanently not-ready while its streams run — the same class of bug the singular field already hit once, one level up.

A rule that is not compiled expects no streams, which is correct: the gate refused it, or the store has not been seeded yet.

func (*Manager) TriggerAllRuleChange added in v0.42.0 

func (m *Manager) TriggerAllRuleChange()

TriggerAllRuleChange marks every declared GitTarget dirty. It is the deletion path only: a WatchRule that is already gone cannot be read for the target it named, so the owner cannot narrow the invalidation and does not pretend to.

func (*Manager) TriggerRuleChange added in v0.42.0 

func (m *Manager) TriggerRuleChange(gitDest types.ResourceReference)

TriggerRuleChange marks the GitTarget a rule names as needing a new plan pass.

It posts intent and returns, where the inline call sites it replaced did a discovery call, a namespace list, a re-projection and a replan of EVERY running GitTarget synchronously.

It deliberately does NOT also enqueue the GitTarget: the pass itself enqueues on a render-fidelity change, the one moment its answer differs. Enqueueing here would fire before anything it would report has changed.

func (*Manager) TypeRecords 

func (m *Manager) TypeRecords() []typeset.TypeRecord

TypeRecords returns every known type record — followable, retained, and refused — for the LOCAL cluster, for inventory and "why is this type not picked up?" views.

func (*Manager) TypeRegistry 

func (m *Manager) TypeRegistry() *typeset.Registry

TypeRegistry returns the LOCAL cluster's followability registry. Retained for the git writer's manager-wide mapper wiring (cmd/main.go); the cluster-scoped writer lookup uses ClusterTypeLookup instead (see gvr.go / Step 4).

type OperationSet 

type OperationSet map[string]struct{}

OperationSet is the set of operation filters recorded for a watched type in one namespace. The sentinel "*" means all operations and subsumes the rest, exactly as the effective-plan hash encodes operations today.

func (OperationSet) Match 

func (s OperationSet) Match(op string) bool

Match reports whether the operation is included in the operation set. A nil or empty set means all operations, matching WatchRule semantics.

func (OperationSet) Sorted 

func (s OperationSet) Sorted() []string

Sorted returns the operations in a stable order, collapsing to ["*"] when the all-operations sentinel is present.

type RenderFidelityStatus added in v0.37.0 

type RenderFidelityStatus = git.RenderFidelityStatus

RenderFidelityStatus is the GitTarget render-vs-live condition state shared with the writer.

type RetentionSummary added in v0.39.0 

type RetentionSummary struct {
	// Reported distinguishes "no resync has reported yet" from "a resync reported zero". Both are
	// legitimate states and they mean opposite things: the first is unknown, the second is the
	// converged signal, which is half the value of publishing this at all.
	Reported bool
	// Mode is the effective spec.prune.mode the most recent contributing resync ran under. It
	// travels WITH the count rather than being read from the spec at projection time, so the two
	// always describe the same observation — a target switched to `always` does not briefly
	// publish `always` beside a count that a retaining policy produced.
	Mode v1alpha3.PruneMode
	// RetainedDocuments is the sum over the target's currently tracked scopes.
	RetainedDocuments int
	// LastChangedTime is when this roll-up last changed. It advances only in the mutation that
	// moves the published count or mode, never on a re-report of the same numbers, so it dates the
	// RESULT rather than the last scan. Non-zero whenever Reported is true.
	LastChangedTime time.Time
}

RetentionSummary is the per-GitTarget roll-up the controller projects onto status.

type SourceClusterReachableStatus added in v0.38.0 

type SourceClusterReachableStatus struct {
	State   string
	Reason  string
	Message string
}

SourceClusterReachableStatus is the kstatus-shaped projection of a source cluster's reachability, for the GitTarget controller to set as the SourceClusterReachable condition. State is "True" | "False" | "Unknown"; the controller maps it to metav1.ConditionStatus.

type SourceClusterResolver added in v0.38.0 

type SourceClusterResolver interface {
	// ResolveSourceCluster returns the rest.Config for a ClusterProvider name, and an opaque
	// version token that changes when the resolved config changes (the provider generation and
	// the kubeconfig Secret's resourceVersion). An unknown or unreadable name is an error:
	// mirroring the wrong cluster into a folder is worse than mirroring none.
	//
	// A NIL config with a nil error means the provider omits spec.kubeConfig and therefore names
	// the operator's OWN cluster — the in-cluster answer, available to every provider name. This
	// is the ONLY authority on whether a source cluster is in-cluster; nothing keys that off the
	// provider's name.
	ResolveSourceCluster(ctx context.Context, providerName string) (cfg *rest.Config, version string, err error)
}

SourceClusterResolver turns a source-cluster NAME (a ClusterProvider's name) into a rest.Config by looking up the ClusterProvider and reading the kubeconfig Secret it names from the operator namespace. It is an interface so the watch manager grows no Kubernetes client of its own for this, and so tests can stand up a remote cluster without a Secret. The concrete implementation lives in source_cluster_resolver.go.

func NewSecretSourceClusterResolver added in v0.38.0 

func NewSecretSourceClusterResolver(
	c client.Client,
	operatorNamespace string,
	safety kubeconfig.SafetyPolicy,
	qps float32,
	burst int,
) SourceClusterResolver

NewSecretSourceClusterResolver builds the production source-cluster resolver.

type StreamState 

type StreamState string

StreamState names the per-type watch readiness state. 

const (
	// StreamStateReplaying means the initial-events replay is still being folded.
	StreamStateReplaying StreamState = "Replaying"
	// StreamStateStreaming means the watch is routing live, attributable events.
	StreamStateStreaming StreamState = "Streaming"
	// StreamStateBlocked means the watch cannot currently run.
	StreamStateBlocked StreamState = "Blocked"
)

type StreamSummary 

type StreamSummary struct {
	Total         int
	Ready         int
	Replaying     int
	Blocked       int
	Reason        string
	Message       string
	PendingSample []string
}

StreamSummary is a bounded status roll-up for a target or rule.

func (StreamSummary) StreamsRunning 

func (s StreamSummary) StreamsRunning() bool

StreamsRunning reports whether all resolved streams are Streaming.

func (StreamSummary) Summary 

func (s StreamSummary) Summary() string

Summary returns the display ratio stored in status.streams.summary.

type WatchedType 

type WatchedType struct {
	GVK           schema.GroupVersionKind
	GVR           schema.GroupVersionResource
	Namespaced    bool
	Scope         configv1alpha3.ResourceScope
	ServedVersion string
	Preferred     bool

	// NamespaceOps maps each watched namespace to the union of operation filters
	// for this type in that namespace. The empty-string key is a cluster-wide
	// stream: a cluster-scoped resource, or a namespaced resource a ClusterWatchRule
	// follows across every namespace.
	NamespaceOps map[string]OperationSet
}

WatchedType is one followable type a GitTarget watches: a (GVK, GVR, scope) triple plus the namespace scope and served-version metadata, projected straight from the type registry's followable set. The registry owns identity (GVK<->GVR is 1:1 there), followability, and the removal grace, so a WatchedType is a copy of a registry fact, never a re-decision.

func (WatchedType) ClusterWide 

func (t WatchedType) ClusterWide() bool

ClusterWide reports whether this type is gathered under a cluster-wide scope: true for a cluster-scoped resource and for a namespaced resource a ClusterWatchRule follows across all namespaces. It reports the presence of that scope, NOT that it is the only one — a namespaced type may carry the cluster-wide scope alongside named ones, and each is streamed in its own right. Use WatchScopes to enumerate them.

func (WatchedType) WatchScopes added in v0.39.0 

func (t WatchedType) WatchScopes() []string

WatchScopes returns the distinct namespace scopes this type is gathered under — one per stream — in a stable order. The empty string is the cluster-wide scope (a cluster-scoped resource, or a namespaced resource a ClusterWatchRule follows across every namespace) and sorts first.

A cluster-wide selection does NOT suppress co-resident named namespaces. A WatchRule scoped to one namespace and a ClusterWatchRule scoped cluster-wide, on the same GVR and the same GitTarget, stay two scopes here, each keeping its own operation filters. This previously collapsed to a single cluster-wide scope, which silently widened the named rule's stream to every namespace the credential could read and discarded its operation set — a gate bypass once a WatchRule declares the source namespaces it is authorized for.

Every read site must project the same scope set, because a gather's scope becomes the mark-and-sweep's scope: a gather wider than the stream that triggered it deletes managed documents that were never in scope (see git.ResyncScope).

type WatchedTypeTable 

type WatchedTypeTable struct {
	GitDest types.ResourceReference
	// Dest is the GitTarget's write destination fingerprint (provider/branch/path),
	// carried so the effective-plan hash can be derived from the table alone.
	Dest       string
	Types      []WatchedType
	ResolvedAt uint64
}

WatchedTypeTable is a GitTarget's resident, resolved-once set of watched types: the subset of the type registry's followable set its WatchRules and ClusterWatchRules select. It is re-resolved only on a deliberate trigger (a rule-set change or a catalog/registry generation bump) and read by the snapshot, informer, and plan-hash paths instead of each re-resolving inline.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.