watch

package
v0.39.2 Latest Latest
Warning

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

 Go to latest Published: Jul 27, 2026 License: Apache-2.0 Imports: 46 Imported by: 0

Documentation

Overview

Package watch drives the api-source-of-truth reconcile: it keeps the followability registry and the demand-driven materialization axis fresh, fills per-type checkpoints, and reconciles each watched type into Git by SPLICING the per-type Redis materialization (checkpoint + audit log) into a desired set — no long-lived object watch is held (R3).

Index

Constants

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.allowedNamespaces 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"

	// ClusterWatchRuleReasonScopeNotSupported is the terminal reason for a STORED ClusterWatchRule
	// that still selects namespaced resources through the removed scope choice.
	ClusterWatchRuleReasonScopeNotSupported = "ClusterScopeOnly"
)

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 ClusterWatchRuleNamespacedScopeMessage = "ClusterWatchRule is cluster-scoped only; watch " +
	"namespaced resources with a WatchRule and `rules[].sourceNamespace`."

ClusterWatchRuleNamespacedScopeMessage is the operator-facing refusal for a stored scope: Namespaced. It names the replacement, because the migration is cross-kind and cannot be performed automatically.

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,
	scope SourceScopeService,
	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 three outcomes map onto the three things the caller must do:

  • ADMITTED — the rule is compiled with every item expanded to concrete namespaces, and the resolved scope is recorded. The caller publishes SourceNamespaceAuthorized=True.
  • TERMINAL (any item denied, or a permanently unevaluatable policy with no scope ever resolved for this spec) — 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.
  • CANNOT SAY YET (retryable), or a rule MAINTAINING an already-resolved scope through an unevaluatable policy — nothing is compiled and nothing is removed. The caller leaves status InProgress and retries. Never narrow to the empty set here: a narrowed set is the input to a sweep, so failing closed while maintaining would delete a tenant's Git content over a transient outage.

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.

func SourceScopeSpecHash added in v0.39.0 

func SourceScopeSpecHash(rule *configv1alpha3.WatchRule) string

SourceScopeSpecHash fingerprints the part of a WatchRule that decides its resolved scope: every item's requested source namespace, in order, plus the rule's own namespace (the value an omitted item resolves to). A change to any of them means the rule is ESTABLISHING a new scope rather than maintaining its old one, so the retained grant must not be reused.

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 {
	// LookupAuthorResolution resolves the strongest author fact for a watch event.
	// exactCapable is true for ADDED/MODIFIED events (try only the immutable exact key
	// and the rv-only hatch) and false for known RV-mismatch events such as DELETED
	// (also consult the last-writer-wins /last pointer).
	LookupAuthorResolution(
		ctx context.Context,
		auditRoute string,
		gvr schema.GroupVersionResource,
		uid k8stypes.UID,
		rv string,
		exactCapable bool,
	) queue.AuthorResolution
}

AttributionLookup is the read side of the optional audit attribution index. The Redis-backed queue.AttributionIndex satisfies it; nil means configured-author.

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:258 only builds one with a non-nil index.
	ResolveAuthor(
		ctx context.Context,
		auditRoute string,
		gvr schema.GroupVersionResource,
		uid k8stypes.UID,
		rv string,
		exactCapable bool,
	) (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,
) 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.

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 ClusterSnapshot 

type ClusterSnapshot struct {
	Desired      []manifestanalyzer.DesiredResource
	Revision     string
	CoverageHead string
}

ClusterSnapshot is one type's revision-pinned desired set for a GitTarget: Desired is the scoped object set the worker folds over the git folder; Revision is the checkpoint resourceVersion the set is anchored at (it stays the commit-message {{.Revision}} and the resync request revision); CoverageHead is the splice coverage head Hc — a FULL Redis stream position "<rv>-<seq>" (the last folded entry's ID, or "<Revision>-<maxseq>" when nothing was folded), the value the per-(GitTarget, GVR) freshness watermark gates the audit tail on. It is a stream position, not a bare rv, because distinct audit entries can share an rv; comparing positions is what keeps the gate from dropping a legitimate same-rv live entry. See signing-snapshot-tail-replay-failure-investigation.md §5/§7.

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 targetRef 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 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
}

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
	// 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 demand-driven materialization axis fresh and drives the per-type splice reconcile. It holds NO long-lived object informers: the only always-on resource intake is the audit-webhook push (mirrored into the per-type :audit:stream); the only API touch on a schedule is the brief checkpoint fill (mirrorTypeObjects) the materialization driver runs for claimed types. 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(
	ctx context.Context,
	gitDest types.ResourceReference,
	clusterID string,
	auditRoute string,
	pruneMode v1alpha3.PruneMode,
	forceRecheck ...bool,
) error

DeclareForGitTarget ensures the GitTarget's watch-first data plane is running against the source cluster it mirrors from. clusterID is (api/v1alpha3).GitTarget.SourceCluster() — the referenced ClusterProvider's name, "default" for the cluster the operator runs in. It is captured here, the same capture-on-Declare pattern as the UID: because spec.clusterProviderRef is immutable it is learned once and never changes, so there is no per-rule propagation and no cross-rule disagreement window.

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.

func (*Manager) DeclaredSourceCluster added in v0.38.0 

func (m *Manager) DeclaredSourceCluster(gitDest types.ResourceReference) (string, bool)

DeclaredSourceCluster reports the source cluster captured for a GitTarget at Declare time and whether that GitTarget has declared at all. It is the observable form of the capture-on-Declare contract: a GitTarget the controller's Validated gate refused never reaches DeclareForGitTarget, so it never appears here. That makes "an unauthorized namespace starts no watch" assertable from outside this package — unlike clusterIDForGitTarget, which deliberately hides the not-yet-declared case behind the local-cluster default.

func (*Manager) EnsureGitTargetWatches 

func (m *Manager) EnsureGitTargetWatches(
	ctx context.Context,
	gitDest types.ResourceReference,
	forceRecheck ...bool,
) error

EnsureGitTargetWatches makes the GitTarget's raw watch set match its current claimed, followable (GVR, scope) table. Each watch resumes from its stored cursor when possible; otherwise it initializes with sendInitialEvents and a scoped mark-and-sweep before streaming live object events.

func (*Manager) EnumerateSourceNamespaces added in v0.39.0 

func (m *Manager) EnumerateSourceNamespaces(
	_ context.Context,
	target *configv1alpha3.GitTarget,
) ([]string, authz.SourceScopeResult)

EnumerateSourceNamespaces expands a GitTarget's allowedSourceNamespaces SELECTOR into the concrete set of source-cluster namespaces it currently admits. It implements the wildcard half of authz.SourceNamespaceResolver.

It answers only the SELECTOR half; authz unions the policy's exact names itself, without a cache and without any source-cluster access, which is what keeps a `sourceNamespace: "*"` item against a names-only policy resolving on a cluster whose Namespace list is Forbidden.

An empty slice with an Admitted verdict is a real answer — the selector currently matches nothing — while Unknown and Unavailable mean the set could not be computed. The caller must never read the latter as the empty set: an empty resolved scope is the input to a resync sweep.

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.

func (*Manager) ForgetSourceScopeGrant added in v0.39.0 

func (m *Manager) ForgetSourceScopeGrant(rule k8stypes.NamespacedName)

ForgetSourceScopeGrant drops a rule's resolved scope. It is called on a REFUSAL or a deletion — never on an unevaluatable policy, which must retain the scope.

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) MarkTargetGitPathAccepted 

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

MarkTargetGitPathAccepted clears any prior refusal for the GitTarget path.

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.

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. A fresh watch epoch is the only recovery route.

func (*Manager) MarkTargetRenderFidelityScopeClean added in v0.37.0 

func (m *Manager) MarkTargetRenderFidelityScopeClean(
	target types.ResourceReference,
	epoch uint64,
	key targetWatchKey,
)

MarkTargetRenderFidelityScopeClean records one complete clean replay result from the current epoch. A stale cancellation tail is ignored by the gate and cannot reopen a failed target.

func (*Manager) MarkTargetRenderFidelityScopeDiverged added in v0.37.0 

func (m *Manager) MarkTargetRenderFidelityScopeDiverged(
	target types.ResourceReference,
	epoch uint64,
	key targetWatchKey,
	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,
	key targetWatchKey,
	epoch uint64,
	mode v1alpha3.PruneMode,
	retained int,
)

MarkTargetRetention records what one scope's resync retained.

Scope lifecycle is handled by the EPOCH rather than by eviction, reusing the watch epoch RenderFidelityGate already defines: records carry the epoch they were produced under, a new epoch replaces the whole per-scope map, and a record from an older epoch is dropped. A scope that leaves the watch plan therefore takes its count with it at the next declaration, with no per-key deletion logic to get wrong — and a stale in-flight reply from a cancelled watch cannot resurrect a count for a scope this target no longer has.

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) ReconcileForRuleChange 

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

ReconcileForRuleChange refreshes the trusted API catalog and the resident watched-type tables when rules change or a CRD is installed/removed. It no longer starts object informers or gathers a whole-GitTarget snapshot (R3): the catalog refresh drives the followability registry, whose transitions gate the materialization axis (which types get a checkpoint) and fan per-type reconciles; the splice off that checkpoint is the only resource-mirror path. Called by the WatchRule/ClusterWatchRule controllers after rule modifications, by the periodic ticker, and by the API-surface trigger.

func (*Manager) RecordSourceScopeGrant added in v0.39.0 

func (m *Manager) RecordSourceScopeGrant(
	rule k8stypes.NamespacedName,
	specHash string,
	namespaces [][]string,
)

RecordSourceScopeGrant remembers that a rule resolved a whole scope under this spec, establishing what RetainedSourceScope will later report. The grant replaces any previous one atomically.

func (*Manager) RefreshAPIResourceCatalog 

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

RefreshAPIResourceCatalog refreshes trusted catalog data from Kubernetes discovery, for the local cluster and every source cluster a GitTarget currently mirrors from. It returns the LOCAL cluster's error only: a remote that cannot be reached fails its OWN GitTargets (through their unready registries and a SourceClusterReachable=False projection), never the local cluster's reconcile. A remote rotation is also picked up here, on the refresh cadence.

func (*Manager) RenderFidelityEpochForGitTarget added in v0.37.0 

func (m *Manager) RenderFidelityEpochForGitTarget(target types.ResourceReference) uint64

RenderFidelityEpochForGitTarget returns the epoch a replay result must carry. A zero epoch means no shared gate is wired, so callers preserve the legacy data path.

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,
	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) 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 targetRef names both the GitTarget and its namespace; the status resolves against that GitTarget's source cluster.

func (*Manager) ResolveSourceNamespace added in v0.39.0 

func (m *Manager) ResolveSourceNamespace(
	_ context.Context,
	target *configv1alpha3.GitTarget,
	namespace string,
) authz.SourceScopeResult

ResolveSourceNamespace answers whether a GitTarget's declared allowedSourceNamespaces admits a namespace in that target's source cluster. It implements authz.SourceNamespaceResolver.

It only ever sees SELECTOR questions: authz answers the exact-name half itself, without a cache and without any source-cluster access at all, which is what keeps name-based policies working against a cluster whose Namespace reads are denied.

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) RetainedSourceScope added in v0.39.0 

func (m *Manager) RetainedSourceScope(rule k8stypes.NamespacedName, specHash string) ([][]string, bool)

RetainedSourceScope reports the resolved scope last GRANTED to a rule FOR A GIVEN SPEC, and whether any grant was ever established for that spec. It is what separates ESTABLISHING a scope from MAINTAINING one: an unevaluatable policy must never produce a resolved namespace set, so while establishing the rule simply does not compile, and while maintaining the last known-good scope is retained instead of being narrowed to nothing — because a narrowed set is the input to a sweep, and failing closed there would delete a tenant's Git content on a transient outage.

A grant recorded under a DIFFERENT spec hash is not reported: a rule whose items changed is establishing a new scope, so it must not inherit the old one.

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) SourceNamespaceEvents added in v0.39.0 

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

SourceNamespaceEvents returns the channel the WatchRule controller wires via source.Channel so a source-cluster Namespace label change re-reconciles the rules it grants or revokes. It carries GitTargets — the object the rules are mapped from — and is lazily created so a zero-value Manager (tests) and the cmd-wired Manager share one channel.

func (*Manager) SourceScope added in v0.39.0 

func (m *Manager) SourceScope() SourceScopeService

SourceScope exposes the manager itself as the source-scope service the WatchRule gate resolves through. It is a method rather than a bare interface assertion so the controller's WatchManagerInterface can carry it and tests can supply a stand-in.

func (*Manager) Start 

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

Start begins the watch ingestion manager and blocks until context cancellation. Performs initial reconciliation then runs periodic discovery refresh.

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) 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
	// ObservedTime is when the most recent contributing resync reported.
	ObservedTime 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 SourceScopeService added in v0.39.0 

type SourceScopeService interface {
	authz.SourceNamespaceResolver

	// RetainedSourceScope reports the resolved scope last granted to a rule FOR A GIVEN SPEC, and
	// whether any grant was ever established for that spec.
	//
	// It is keyed by the rule's spec hash rather than by item index on purpose: retention applies
	// only while the spec is unchanged, so an edit discards the memory and re-establishes from
	// scratch, and a reorder can never let one item inherit another item's grant.
	RetainedSourceScope(rule k8stypes.NamespacedName, specHash string) ([][]string, bool)
	// RecordSourceScopeGrant remembers a successful whole-rule resolution.
	RecordSourceScopeGrant(rule k8stypes.NamespacedName, specHash string, namespaces [][]string)
	// ForgetSourceScopeGrant drops a rule's resolved scope on a refusal or a deletion.
	ForgetSourceScopeGrant(rule k8stypes.NamespacedName)
}

SourceScopeService is the source-scope service as its consumers need it: the policy resolution and enumeration authz calls, plus the per-rule resolved-scope memory that separates ESTABLISHING a grant from MAINTAINING one. *Manager implements it; a nil value is legitimate and means "not wired yet" (a zero-value manager in tests, or a controller running before the data plane is up), which degrades to name-only policy evaluation rather than to a denial.

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. See docs/design/watchrule-source-namespace/pr2-stream-scope-collapse.md.

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.