watch

package
v0.36.0 Latest Latest
Warning

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

 Go to latest Published: Jul 14, 2026 License: Apache-2.0 Imports: 40 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 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

This section is empty.

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,
		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, or ok=false to
	// commit as the configured committer. It may wait up to the grace window for a
	// matching fact; it never blocks indefinitely and never returns an error path —
	// an absent fact is a committer commit, not a failure. exactCapable distinguishes
	// ADDED/MODIFIED events (true) from known RV-mismatch removals (false).
	ResolveAuthor(
		ctx context.Context,
		gvr schema.GroupVersionResource,
		uid k8stypes.UID,
		rv string,
		exactCapable bool,
	) (git.UserInfo, bool)
}

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 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
	// 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) 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,
	forceRecheck ...bool,
) error

DeclareForGitTarget ensures the GitTarget's watch-first data plane is running.

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

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

FollowableTypeRecords returns every currently-followable type record (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.

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

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

RefreshAPIResourceCatalog refreshes trusted catalog data from Kubernetes discovery.

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.

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.

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

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.

func (*Manager) TypeRecords 

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

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

func (*Manager) TypeRegistry 

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

TypeRegistry returns the live followability registry, the single decision surface (a typeset.Lookup). The git worker reads it to resolve manifest GVKs; the manager refreshes it in place, so the returned pointer tracks discovery updates.

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 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
	ObservedTime  metav1.Time
}

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 with a single cluster-wide stream, true for a cluster-scoped resource and for a namespaced resource a ClusterWatchRule follows across all namespaces.

func (WatchedType) SnapshotNamespaces 

func (t WatchedType) SnapshotNamespaces() []string

SnapshotNamespaces returns the namespaces this type is gathered under for the streaming snapshot and informers: an empty slice means cluster-wide. A cluster-wide selection overrides any named namespaces, matching the historic gvrSnapshotEntry collapse.

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.