queue

package
v0.41.0 Latest Latest
Warning

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

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

Documentation

Index

Constants

View Source 
const (
	// DefaultFactIndexMaxFactsPerType caps one (audit route, group/resource)'s entries across all
	// four match structures. It is the primary cap because it is the fair one: a burst on one noisy
	// type — a deletecollection over ten thousand objects, a large rollout — must not evict every
	// other type's facts. One entry is a fact plus its bookkeeping, a few hundred bytes, so a type
	// at its cap costs low single-digit megabytes.
	DefaultFactIndexMaxFactsPerType = 4096

	// DefaultFactIndexMaxFactsTotal caps the whole index, so the pod's memory is bounded by a number
	// that does not scale with how many types happen to be watched. It sits well above the per-type
	// cap: reaching it takes many types simultaneously busy, and eviction then falls on the type
	// holding the most.
	DefaultFactIndexMaxFactsTotal = 65536

	// DefaultFactCollectionWindow is how long after a deletecollection's stageTimestamp a removal in
	// its scope may still be credited to it. It is far shorter than the fact TTL and can afford to
	// be: under the deletion-as-intent rule the removal being attributed happens at delete-REQUEST
	// time, so finalizers do not stretch it, and the window only has to cover audit batching plus
	// clock skew. Ten times the default grace window leaves room for a slow batch without letting an
	// unrelated delete a minute later be claimed.
	DefaultFactCollectionWindow = 30 * time.Second

	// DefaultFactIndexSweepInterval is how often aged-out entries are reclaimed. It only bounds
	// MEMORY, never correctness: a lookup checks the TTL itself, so an entry past its horizon is
	// never joined merely because the sweep has not run yet.
	DefaultFactIndexSweepInterval = 30 * time.Second
)

Defaults for the in-memory fact index. Redis used to enforce the TTL and the memory ceiling for free; holding the facts in process moves both jobs here, which is why every one of these numbers is a bound rather than a hint.

View Source 
const (
	// DefaultFactStreamMaxLen bounds one stream so a hot type cannot grow without bound between
	// retention trims. It is a count of ENTRIES, and one entry carries a whole audit batch's facts
	// for one (route, group/resource), so it is far more history than the TTL horizon usually holds.
	DefaultFactStreamMaxLen int64 = 10000

	// DefaultFactStreamReadCount bounds how many entries one Next drains per stream. It also decides
	// when a follower is considered BEHIND, which is the precondition for trim-gap detection.
	DefaultFactStreamReadCount int64 = 512

	// DefaultFactStreamBlock is how long one Next waits for a new entry before returning empty. It
	// sets how quickly a change to the followed set takes effect: a follower re-reads its set on
	// every Next, so a subscribe or unsubscribe lands within one block period.
	DefaultFactStreamBlock = time.Second

	// DefaultFactStreamTrimInterval is how often a stream is trimmed to the retention horizon.
	// Trimming is amortized onto the publish path, so this bounds the extra command rate rather
	// than the accuracy of the horizon.
	DefaultFactStreamTrimInterval = time.Minute
)

Defaults for the attribution fact transport. They are shared by both implementations so a conformance test, and an operator reading one implementation's flags, sees one set of numbers.

View Source 
const DefaultAttributionFactTTL = 10 * time.Minute

DefaultAttributionFactTTL is how long an attribution fact stays joinable while it waits for the matching watch event. It bounds the stream's retention horizon and the in-memory index together, and it doubles as the follower's replay horizon, so a restart warms the index with exactly the window that is still usable. After it elapses a miss is simply "absent": there is no tombstone, so an aged-out fact is indistinguishable from one that never arrived. Configurable via --author-attribution-ttl.

View Source 
const DefaultCollectionUIDCap = 10000

DefaultCollectionUIDCap is how many uids a collection fact may carry before the set is dropped and the join falls back to scope matching.

It bounds one entry's size, the broadcast to every subscriber of the type, and the replay on restart. A uid is 36 bytes, so this cap is a few hundred kilobytes at worst — against a response body for the same request that runs to tens of megabytes. It is a tuning number rather than a correctness one: the fallback is already correct, so the cap only decides how often the precise path is taken, and a collection delete large enough to exceed it is exactly the one whose body a production cluster with audit truncation enabled would not have sent in the first place.

View Source 
const DefaultKeyPrefix = "gitops-reverser"

DefaultKeyPrefix is the root namespace every Redis key (cursors, fact streams, and command author records alike) carries when --redis-key-prefix is not set. It is also the value every release before the flag existed used, so the default is a no-op upgrade.

Variables

This section is empty.

Functions

func RecordFactsWritten added in v0.41.0 

func RecordFactsWritten(ctx context.Context, count int)

RecordFactsWritten counts facts appended to the fact log. It is called by the publish side, once per append rather than once per entry, so the counter measures facts rather than audit batches and stays comparable with the matched op on the other side of the join.

func ValidateKeyPrefix 

func ValidateKeyPrefix(prefix string) (string, error)

ValidateKeyPrefix checks a --redis-key-prefix value and returns its normalized form.

Two independent constraints shape the allowed character set:

  • A prefix names a keyspace an operator inspects and, on a bad day, deletes by glob. Redis glob metacharacters (*, ?, [, ], \) in it would make "<prefix>:*" match more than this install's keys, so they are rejected rather than escaped — a prefix is an operator-chosen identifier, not user data.
  • Key fields (uid, resourceVersion, namespace) are ':'-delimited and %-escaped by escapeKeyField. '%' in the prefix would make an escaped key ambiguous with an unescaped one, so it is rejected too.

':' is allowed, because a prefix like "cell-a:tenant-7" is a natural nesting and every suffix constant already begins with ':'. A trailing ':' is normalized away so "tenant-7:" and "tenant-7" name the same keyspace rather than two that differ by an empty segment.

An empty prefix is rejected: an unprefixed keyspace collides with Redis's own key namespace conventions and, more importantly, silently un-namespaces an install that meant to set the flag and passed the empty string. Use DefaultKeyPrefix to opt out.

Types

type ActorKind added in v0.41.0 

type ActorKind string

ActorKind is the bounded kind of actor a resolution named. It is the same vocabulary commits_total{author_kind} uses, so the two metrics stop disagreeing about the shape of one distinction, and it is orthogonal to the tier: every tier can name either kind of actor, or none. 

const (
	// ActorKindUser is a human (or any non-service-account subject) named by the matched fact.
	ActorKindUser ActorKind = "user"
	// ActorKindServiceAccount is a named service account.
	ActorKindServiceAccount ActorKind = "serviceaccount"
	// ActorKindNone is no actor at all: nothing matched, or the fact that matched carried no author.
	ActorKindNone ActorKind = "none"
)

type AttributionResult 

type AttributionResult string

AttributionResult is the bounded resolver outcome recorded for each watch event. The set is the join's tier table, so a reading of attribution_resolutions_total{tier} says which evidence named the author, not merely that one was named. It is the source of truth for that label: a lookup that could not say which tier answered is the thing the tier label exists to fix, so the split lives in the enum rather than at the metric boundary.

WHO was named is a separate question and a separate label — see ActorKind. The two used to be crammed into one value (exact_user against exact_serviceaccount), which made counting exact resolutions a sum of two series and made the actor kind unaskable of every other tier. 

const (
	// AttributionDeleteSticky is the sticky removal pointer: a fact whose own verb is a delete, filed by
	// uid into a slot no later WRITE fact may overwrite. It is the strongest evidence a removal can
	// have about itself, so it is consulted before the exact tier — and only by a removal, because an
	// exact-capable event asks who produced a version rather than who deleted an object.
	//
	// "sticky" is the half of the name that is not the verb, and it is there because the stickiness is
	// the whole reason this tier can answer at all: every other structure would have been overwritten
	// by the finalizer patch that followed the delete.
	//
	// It is also the only tier the TTL does not bound. A uid is unique across space and time, so the
	// statement can never be superseded; its horizon is the index's caps instead. See
	// docs/spec/attribution.md.
	AttributionDeleteSticky AttributionResult = "delete_sticky"
	// AttributionExact is an exact UID+resourceVersion match: this actor produced this exact version.
	AttributionExact AttributionResult = "exact"
	// AttributionLatest is the uid-latest tier — the object's own last write or its own delete fact,
	// keyed by uid alone. It is the tier the removal path turns on: a match here that describes a
	// WRITE is held as a fallback while the wait continues for evidence about the deletion.
	AttributionLatest AttributionResult = "latest"
	// AttributionResourceVersion is the rv-only escape hatch: a fact that carried a resourceVersion
	// and no uid, matched on that version alone. It and AttributionLatest were one value ("weak")
	// and are different evidence, which is why they are now two.
	AttributionResourceVersion AttributionResult = "resource_version"
	// AttributionDeleteCollectionBodyUID is a removal matched to a deletecollection fact whose uid set
	// contains this object. There is no over-attribution risk in it: either the API server said it
	// deleted this object, or it did not.
	//
	// "body" is in the name because the uid set comes from the RESPONSE BODY, which is the part the
	// API server may not send — a proxied collection delete sends none, and a set past
	// DefaultCollectionUIDCap is dropped. Both land on the scope tier instead, counted on
	// attribution_collection_without_uidset_total, so the name says which half of the pair answered
	// and why the other one exists.
	AttributionDeleteCollectionBodyUID AttributionResult = "deletecollection_body_uid"
	// AttributionName is a match on (namespace, name) for a fact that carries neither a uid nor a
	// resourceVersion. It is the tier of last resort for a type whose audit event cannot express
	// object identity: the kube-apiserver proxies an aggregated-API request and never decodes the
	// response, so the objectRef carries the name from the URL path and nothing else, and there is no
	// body to backfill from. Measured in corpus flunder/aggregated-api-delete.
	//
	// It ranks below every other per-object tier because a name is REUSED after a delete and
	// recreate where a uid is not, so it can name the author of a previous object that held this
	// name. The TTL is what bounds that: the wrong answer requires the recreate to happen inside it.
	AttributionName AttributionResult = "name"
	// AttributionDeleteCollectionScope is a removal matched to a deletecollection fact by scope alone —
	// same type and namespace, selector accepting the object's labels, within the collection window.
	// It is the weakest evidence the join has, which is why it is reached only when every more
	// specific tier missed. It is also the tier that resolves what the deleted expander gave up on:
	// a collection delete the API server sent no response body for.
	AttributionDeleteCollectionScope AttributionResult = "deletecollection_scope"
	// AttributionAbsent means no usable author fact matched before the grace elapsed.
	AttributionAbsent AttributionResult = "absent"
)

Three values are named for the VERB that produced the fact and how it matched — delete_sticky, deletecollection_body_uid, deletecollection_scope — because those three are the tiers reachable only by a removal, and naming the source is the honest thing to do: it says where the evidence came from rather than asserting what it proves. Two of them are statements about this exact object; the third, deletecollection_scope, is a statement about a request whose scope covered it, and can name the wrong actor. A name that claimed otherwise would be read as a guarantee it cannot give.

`latest` and `name` can hold a delete fact too, so they are NOT named for one: either can equally hold a write, and a value that could mean either must not claim a verb.

type AuthorFact 

type AuthorFact struct {
	Namespace string `json:"namespace,omitempty"`
	UID       string `json:"uid,omitempty"`
	// Name is the object's name, and it feeds one tier only: the (namespace, name) join a fact with
	// no uid and no resourceVersion is otherwise unreachable through. A collection fact clears it,
	// because a collection request names no object.
	Name string `json:"name,omitempty"`
	// Author is the actor's username, and it is the ONE required field on the wire: a fact exists to
	// name somebody, so a fact that names nobody is not a weak fact, it is not a fact. It is never
	// empty and never null — see UnmarshalJSON, which refuses an entry carrying one.
	Author string `json:"author"`
	// DisplayName and Email are the actor's, when the API server supplied them. They are the only
	// fields here that are not identity or evidence: they exist because a commit author is a name
	// and an email, and re-deriving them at commit time would need a second lookup.
	DisplayName string `json:"displayName,omitempty"`
	Email       string `json:"email,omitempty"`
	Verb        string `json:"verb,omitempty"`
	// AuditID is the one field the join never reads and is kept anyway. It is what ties a commit
	// authored by the wrong person back to the audit event that named them, which is the single
	// question a mis-attribution investigation asks and the one thing nothing else in the system
	// can answer.
	AuditID         string `json:"auditID,omitempty"`
	ResourceVersion string `json:"resourceVersion,omitempty"`
	StageTimestamp  string `json:"stageTimestamp,omitempty"`

	// LabelSelector is the selector the request URI expressed, carried on a COLLECTION fact only.
	// It is the intent the actor stated, and evaluating it against the object a watch event carries
	// is a better test of membership than reading back a list the API server may not have sent.
	// Empty means the collection covered everything of its type in its namespace, which is what
	// --all means.
	LabelSelector string `json:"labelSelector,omitempty"`
	// UIDs is the set of objects a collection delete covered, reduced from the response body at the
	// receiver, on a COLLECTION fact only. It is absent when the API server sent no body — a
	// truncated, aggregated, or metadata-only response — and when the set was larger than the cap,
	// in which case the join falls back to scope matching, which is already correct.
	UIDs []string `json:"uids,omitempty"`
}

AuthorFact is the minimal attribution fact published per accepted, mutating audit event and read back by the watch-event resolver. It names an author candidate and carries the evidence the join needs to decide confidence; it is never object state.

Every field here is either read by the join or printed when a fact is investigated. That is a deliberate bar, because a fact is not stored once: it is broadcast to every process following its type, held for the whole TTL, and replayed into memory on every restart, so a field nothing reads is paid for on all three. Three fields were removed for failing it:

  • the group/resource, which is the STREAM'S OWN NAME — the index takes the scope from the entry's key, never from the fact, so carrying it duplicated the routing on every entry;
  • the subresource, which no tier joins on and nothing logs.

A stored is-service-account bool went the same way, for a different reason: it is not evidence, it is a prefix check on Author that the reader can do for itself (see ActorKind).

Name was removed with the subresource, on the same observation — no tier read it — and is back, because that observation was true of the code and false of the domain. An aggregated-API write is audited with no uid and no resourceVersion, and the name from the URL path is the ONLY identity it carries, so a fact without it could not be joined at all for that whole population. "No code reads it" and "nothing could ever read it" are different claims, and only the second justifies dropping a field.

func AuthorFactFromEvent added in v0.41.0 

func AuthorFactFromEvent(
	ctx context.Context,
	event auditv1.Event,
	uidCap int,
) (AuthorFact, schema.GroupResource, bool)

AuthorFactFromEvent reduces one accepted, mutating audit event to the fact the stream carries, reporting false when the event can never name an author. Only facts that WOULD have been stored may be published: an event with no objectRef or no user produces nothing, or waiters are woken by facts that can name nobody.

The one rule that changes from the per-key write path is the name check. A deletecollection is name-less by nature and is now exactly the case that produces a fact — one fact describing the COLLECTION, which every removal in its scope joins — so "no resolvable name" becomes "no name and not a collection verb".

The caller has already applied the intrinsic accept gate: reads, failures, dry runs, and non-ResponseComplete stages never reach here.

func (AuthorFact) ActorKind added in v0.41.0 

func (f AuthorFact) ActorKind() ActorKind

ActorKind classifies the actor a fact names. It is derived rather than carried: the API server spells a service account one way and only one way, so a stored kind would be the same check, denormalized onto every fact and able to disagree with the name beside it.

func (*AuthorFact) UnmarshalJSON added in v0.41.0 

func (f *AuthorFact) UnmarshalJSON(raw []byte) error

UnmarshalJSON decodes a fact and refuses one that names nobody, which is the whole wire contract: `author` must be present, a string, and non-empty. Missing, `null`, and `""` are the same violation and are all refused.

Go cannot express "a string of at least one character" as a type — every type has a zero value that is constructible without going through any constructor, and `encoding/json` writes exported fields straight past one anyway — so the constraint lives at the only boundary that can hold it: the point where a fact written by somebody else enters this process.

The refusal is deliberately at ENTRY granularity, not per fact. This operator's publish gate cannot produce an authorless fact (AuthorFactFromEvent refuses an event whose user is unresolvable, and counts it as no_attribution_fact), so an entry carrying one was written by something else: a different version, a different producer, or a hand-written entry. That is a protocol violation rather than a low-quality fact, and it is better counted and logged loudly — it lands on attribution_fact_stream_decode_errors_total with the stream and entry id — than half-absorbed by silently dropping one fact out of a batch.

type AuthorResolution 

type AuthorResolution struct {
	Fact   AuthorFact
	Result AttributionResult
}

AuthorResolution is the structured result of an attribution lookup.

func (AuthorResolution) ActorKind added in v0.41.0 

func (r AuthorResolution) ActorKind() ActorKind

ActorKind classifies the actor this resolution named, which is none when no fact matched.

type CommandAuthor 

type CommandAuthor struct {
	Author      string `json:"author"`
	DisplayName string `json:"displayName,omitempty"`
	Email       string `json:"email,omitempty"`
	RequestedAt string `json:"requestedAt,omitempty"` // RFC3339Nano, for lag metrics/debug
}

CommandAuthor is the minimal authorship captured at admission for one command object. It carries only what a git commit author needs — no RV, no auditID, no conflict bit: this is a 1:1 command capture, not a post-persist join.

type CommandAuthorStore 

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

CommandAuthorStore records and reads command authorship. It shares RedisStore's connection but is wired whenever the validate-operator-types webhook is enabled — independent of --author-attribution, which only governs mirrored-resource attribution.

func (*CommandAuthorStore) LookupCommandAuthor 

func (s *CommandAuthorStore) LookupCommandAuthor(
	ctx context.Context, uid types.UID,
) (CommandAuthor, bool)

LookupCommandAuthor is the controller-side read, keyed by the persisted object's UID. ok=false means no record was captured — the validate-operator-types webhook is not configured (or a best-effort write missed) — and the controller finalizes as the committer, immediately.

func (*CommandAuthorStore) RecordCommandAuthor 

func (s *CommandAuthorStore) RecordCommandAuthor(
	ctx context.Context, uid types.UID, author CommandAuthor,
) error

RecordCommandAuthor is the admission-side write: capture the authenticated submitter the instant a command CREATE is admitted, before it persists. Last-write-wins (a CREATE fires admission once; a retried admission re-asserts the same user).

type FactDelivery added in v0.41.0 

type FactDelivery struct {
	Entries []FactEntry
	Gaps    []FactStreamGap
}

FactDelivery is one Next's result: the entries read this round, in append order per stream, plus any trim gaps noticed. Both may be empty — that is an idle block period, not an error.

type FactEntry added in v0.41.0 

type FactEntry struct {
	Key   FactStreamKey
	ID    string
	Facts []AuthorFact
}

FactEntry is one appended batch of facts, as it comes back to a follower. ID is the transport-assigned position, "<unix-millis>-<sequence>", which increases strictly within a stream and is what a follower resumes from.

type FactFollower added in v0.41.0 

type FactFollower interface {
	// FollowFacts starts following keys, reading each from horizon before now. A horizon of the
	// fact TTL is what makes a restart cost nothing: the follower replays the whole retention
	// window before the first watch event needs it. Entries older than the horizon are skipped,
	// to within the millisecond granularity of a stream position.
	FollowFacts(keys []FactStreamKey, horizon time.Duration) FactSubscription

	// TransportKind names this transport for the metric labels. It is on the follower half rather
	// than beside the wiring because the index is what records follower health, and a counter that
	// cannot say which transport erred says half of what an operator needs.
	TransportKind() FactTransportKind
}

FactFollower follows a set of streams from a horizon. The fact index is its only caller: it follows the union of the types any watch covers and applies what it reads into memory.

type FactIndex added in v0.41.0 

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

FactIndex is the transport-agnostic half of attribution: the four match structures the join reads, the waiter registry a blocked resolver parks on, and the loop that fills both from whatever transport it was handed.

There is exactly ONE index per process, not one per GitTarget. A fact names a write that happened in Kubernetes, not a consumer interested in it, so one fact already serves every GitTarget that needs it; five GitTargets mirroring one Deployment would otherwise hold five copies of every fact and bill memory against a number that has nothing to do with how much is happening in the cluster. The fan-out that does do useful work is the SUBSCRIPTION set, which is per type.

func NewFactIndex added in v0.41.0 

func NewFactIndex(cfg FactIndexConfig) *FactIndex

NewFactIndex builds an empty index.

func (*FactIndex) Apply added in v0.41.0 

func (i *FactIndex) Apply(ctx context.Context, entry FactEntry)

Apply stores one delivered entry's facts and wakes whoever was waiting for them. Facts are applied in the order they were delivered, which is what makes the latest tier last-writer-wins mean the last fact APPENDED rather than whichever goroutine reached the map first. A fact ages from when it was APPENDED, not from when this process happened to read it. The two differ by more than a hair in the case that matters most: the follower replays the whole retention window on start, so stamping those entries with the read time would hand every one of them a second full TTL and let a restart resurrect facts the horizon had already retired. A follower that falls behind, or a transport that hands back an entry its own retention should have dropped, lands in the same place. Reading the append time off the entry's position makes the TTL mean the same thing on both transports and on every delivery path, which is what SweepInterval bounding memory rather than correctness depends on.

func (*FactIndex) Await added in v0.41.0 

func (i *FactIndex) Await(ctx context.Context, query FactQuery, grace time.Duration) AuthorResolution

Await resolves a watch event, waiting up to grace for a fact that has not been delivered yet. It returns an AttributionAbsent resolution when nothing matched in time; it never blocks longer than the grace and never returns an error path.

The order of the first two statements is the design, not a detail. The waiter is registered BEFORE the index is read, so a fact applied in the gap between the two signals a waiter that is already listening. Checking first and registering after loses exactly that fact — the race the poll loop used to paper over by looking again.

A match does not always end the wait. For a REMOVAL, the strongest fact present early is often the object's last WRITE, which says who edited it and nothing about who deleted it — and the watch event reliably beats the audit batch that carries the delete, which is the entire reason the grace window exists. Returning on that first match answered "who deleted this" with "who last edited it", every time an object was touched by someone else before being removed. Such a match is held as a FALLBACK instead: the wait continues for evidence about the deletion itself, and the fallback is returned only when the grace expires without any arriving. Attribution is never lost by waiting — the worst case returns exactly what returning early would have.

func (*FactIndex) Len added in v0.41.0 

func (i *FactIndex) Len() int

Len reports how many entries the index holds across every scope and structure.

func (*FactIndex) Lookup added in v0.41.0 

func (i *FactIndex) Lookup(query FactQuery) AuthorResolution

Lookup reads the index once, trying the tiers strongest-first:

  1. the sticky removal pointer for that uid, for a removal only;
  2. the exact (uid, rv) fact, the only exact-capable join;
  3. a collection fact whose uid set contains this object;
  4. the last-writer-wins fact for that uid, for a removal whose rv never matches;
  5. a collection fact whose scope, selector, and window cover it;
  6. the rv-only escape hatch;
  7. the (namespace, name) floor.

The name tier is last because it is the weakest per-object evidence here: a name is reused after a delete and recreate, so it can name the author of a previous object that held it, where a uid cannot and an rv identifies one specific write. Nothing that carries a uid or an rv ever reaches it, so ranking it last costs the stronger tiers nothing and only picks up what they cannot express.

Precedence is the correctness argument for the collection tiers, and the two of them sit on OPPOSITE sides of the latest tier on purpose.

Uid membership outranks it because the two tiers answer different questions. The latest tier says who last WROTE an object; a removal asks who DELETED it. For a single-object delete those coincide, because the delete files its own fact under that uid — but a collection delete files one fact about the collection, so the uid's latest entry is left holding whoever happened to write the object last. Ranking it above the collection's uid set credited a removal to the previous editor and never reached the actor who actually ran the delete, which is the one thing the deleted expander did get right: it overwrote that entry per object. Uid membership is the API server stating that THIS request deleted THIS object, so nothing weaker may answer ahead of it.

Scope matching stays below, because it is the weakest evidence here and can name the wrong human: an unrelated delete by another actor during the same window is claimed by its own fact at tier 3 and never reaches tier 4.

func (*FactIndex) Run added in v0.41.0 

func (i *FactIndex) Run(ctx context.Context, follower FactFollower) error

Run follows the subscription set until the context ends, applying what it reads and reporting what it lost. It returns only when the context ends: a transport failure is retried, because a follower that gave up would leave attribution silently dead for the life of the process.

func (*FactIndex) Streams added in v0.41.0 

func (i *FactIndex) Streams() *FactStreamSet

Streams is the reference-counted set of (route, group/resource) pairs this process follows. The watch side acquires a reference when it starts covering a type and releases it when the last watch on that type goes away; Run makes the follower track it.

func (*FactIndex) Sweep added in v0.41.0 

func (i *FactIndex) Sweep(now time.Time) int

Sweep reclaims every entry past the TTL horizon. Lookups already ignore aged-out entries, so this bounds memory rather than deciding what may be joined.

type FactIndexConfig added in v0.41.0 

type FactIndexConfig struct {
	// TTL is how long a fact stays joinable, and doubles as the follower's replay horizon so a
	// restart warms the index with exactly the window that is still usable.
	TTL time.Duration
	// MaxFactsPerType caps one (route, group/resource); MaxFactsTotal caps the whole index.
	MaxFactsPerType int
	MaxFactsTotal   int
	// CollectionWindow bounds the scope-matching tier.
	CollectionWindow time.Duration
	// SweepInterval is how often aged-out entries are reclaimed.
	SweepInterval time.Duration
	Log           logr.Logger
}

FactIndexConfig configures the index. Every zero field falls back to its Default… constant, so the zero value is the supported configuration.

type FactPublisher added in v0.41.0 

type FactPublisher interface {
	// PublishFacts appends facts as ONE entry on the key's stream. An empty batch is a no-op.
	// Entries appear to followers in the order they were published per stream; there is no
	// ordering promise across streams, and none is needed — an object belongs to exactly one
	// group/resource and therefore to exactly one stream.
	PublishFacts(ctx context.Context, key FactStreamKey, facts []AuthorFact) error
}

FactPublisher appends a batch of facts for one (route, group/resource). The audit receiver is its only caller: it decodes one EventList per request, groups the facts it accepted by stream, and appends once per group.

type FactQuery added in v0.41.0 

type FactQuery struct {
	// AuditRoute partitions the index. It leads every key for the same reason the streams are named
	// per route: a fact from cluster A must never name the author of an object watched on cluster B.
	AuditRoute      string
	GroupResource   schema.GroupResource
	UID             string
	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 reached when a fact carries neither a uid nor a
	// resourceVersion. The watch side always knows it; only the audit side can be missing 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
}

FactQuery is one watch event's identity, as the join reads it. It is everything the index needs to try all five tiers, so a caller assembles it once rather than threading five arguments.

type FactStreamGap added in v0.41.0 

type FactStreamGap struct {
	Key FactStreamKey
	// Cursor is the position the follower had reached; FirstSurviving is the oldest entry the
	// stream still holds. FirstSurviving is newer than Cursor, and everything between them is gone.
	Cursor         string
	FirstSurviving string
}

FactStreamGap reports that a follower was trimmed past on one stream: entries it had not read were dropped by retention before it got to them, so the facts they carried are lost for good. It is the one loss this transport can see, and reporting it is why the transport is a log with positions rather than fire-and-forget publish/subscribe.

type FactStreamKey added in v0.41.0 

type FactStreamKey struct {
	AuditRoute    string
	GroupResource schema.GroupResource
}

FactStreamKey identifies one attribution fact stream: the audit route the facts arrived under, and the group/resource they are about. The route is part of the identity for the same reason it is part of the v1 fact keys — a fact from cluster A must never name the author of an object watched on cluster B.

GroupResource is the typed identity rather than a rendered string. The rendering matters: the stream name embeds the API-path form groupResourceKey produces ("configmaps", "apps/deployments"), and schema.GroupResource.String() produces the reversed dotted form ("deployments.apps"), so a caller that rendered its own key would publish to a stream nobody follows and get no compile error for it. Holding the type and rendering at the transport boundary makes that mistake unrepresentable.

func FactStreamKeyFor added in v0.41.0 

func FactStreamKeyFor(auditRoute string, gr schema.GroupResource) FactStreamKey

FactStreamKeyFor builds the key for one audit route and group/resource.

func (FactStreamKey) String added in v0.41.0 

func (k FactStreamKey) String() string

String renders the key for logs and metrics as "<route>/<group-resource>".

type FactStreamSet added in v0.41.0 

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

FactStreamSet is the reference-counted union of the (audit route, group/resource) pairs the watches running in this process cover. It is what makes the per-type fan-out mean anything: the process follows a type while at least one watch needs it and stops following it when the last one goes away, so facts for a type nobody watches are written and never received.

Reference counting rather than a plain set is the point. Several WatchRules, and several GitTargets, routinely cover one type; the type must stay followed while ANY of them does, and a set that had forgotten how many watches added it would unfollow on the first one to stop.

It is deliberately independent of the index and of the transport: the watch side acquires and releases, and whoever is following the streams observes the union.

func NewFactStreamSet added in v0.41.0 

func NewFactStreamSet() *FactStreamSet

NewFactStreamSet builds an empty subscription set.

func (*FactStreamSet) Acquire added in v0.41.0 

func (s *FactStreamSet) Acquire(key FactStreamKey) func()

Acquire takes one reference on a stream and returns the release for it. The returned release is idempotent, so a caller that releases twice — a watch torn down on both an error path and its deferred cleanup — cannot unfollow a type another watch still needs.

func (*FactStreamSet) Keys added in v0.41.0 

func (s *FactStreamSet) Keys() []FactStreamKey

Keys returns the followed set in a stable order, which is what a follower is given.

func (*FactStreamSet) Len added in v0.41.0 

func (s *FactStreamSet) Len() int

Len reports how many distinct streams are followed.

func (*FactStreamSet) Observe added in v0.41.0 

func (s *FactStreamSet) Observe(observe func([]FactStreamKey))

Observe installs the callback that receives the followed set whenever it changes, and hands it the current set immediately so the follower and the set never start out disagreeing. A nil callback detaches. Only one observer is supported: one process follows one subscription.

type FactSubscription added in v0.41.0 

type FactSubscription interface {
	// SetStreams replaces the followed set. A newly followed stream starts from the horizon, so
	// it replays its retention window; an unfollowed stream's cursor is forgotten, so following it
	// again replays too. It takes effect on the next Next, hence within one block period.
	SetStreams(keys []FactStreamKey)

	// Next returns the entries appended since the last call, waiting up to the block period for
	// the first of them. An empty delivery with a nil error means the block period elapsed with
	// nothing new, which is the ordinary idle case. It returns an error only when the context ends
	// or the transport fails.
	Next(ctx context.Context) (FactDelivery, error)
}

FactSubscription is one follower's live position across its followed streams. It is not safe to call Next concurrently with itself; SetStreams may be called from another goroutine at any time. A subscription owns no resources beyond its cursors, so it is dropped rather than closed.

type FactTransport added in v0.41.0 

type FactTransport interface {
	FactPublisher
	FactFollower
}

FactTransport is the whole seam: publish a batch, follow a set. Everything above it — the in-memory index, the TTL sweep, the waiter registry, the resolver — has one implementation and never learns which transport it has. Consumers should depend on FactPublisher or FactFollower, the half they use; this composition exists for wiring and for the conformance suite.

type FactTransportKind added in v0.41.0 

type FactTransportKind string

FactTransportKind is the bounded name of the transport carrying the facts. It is metric metadata rather than behavior: nothing above the seam branches on it, but every reading of the attribution metrics depends on it, because the two transports fail differently — a burst of unresolved commits after a restart is expected under memory, which drops every fact with the process, and a bug under redis. 

const (
	// FactTransportRedis is the Redis Streams transport, the default and the only multi-replica one.
	FactTransportRedis FactTransportKind = "redis"
	// FactTransportMemory is the in-process ring, which requires a single replica and loses every
	// fact on restart by design.
	FactTransportMemory FactTransportKind = "memory"
)

type MemoryFactStream added in v0.41.0 

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

MemoryFactStream is the in-process implementation of the attribution fact transport: one ring buffer per (audit route, group/resource), trimmed by TTL and entry count, with one cursor per follower.

It is the same data structure as the Redis one rather than an approximation of it. A Redis stream is a capped, ordered log with per-reader cursors, which is a ring buffer: replay from the horizon is reading the ring from its tail, and trim-gap detection is comparing a follower's cursor against the ring's oldest surviving entry. Both fall out; neither is simulated.

It only works when the audit receiver and the resolver are the same process. Selecting it alongside more than one replica is a configuration error, and belongs at startup validation rather than here — a transport cannot see how many replicas it has.

func NewMemoryFactStream added in v0.41.0 

func NewMemoryFactStream(cfg MemoryFactStreamConfig) *MemoryFactStream

NewMemoryFactStream builds the in-process fact transport.

func (*MemoryFactStream) FollowFacts added in v0.41.0 

func (m *MemoryFactStream) FollowFacts(keys []FactStreamKey, horizon time.Duration) FactSubscription

FollowFacts starts following keys from horizon before now.

func (*MemoryFactStream) PublishFacts added in v0.41.0 

func (m *MemoryFactStream) PublishFacts(ctx context.Context, key FactStreamKey, facts []AuthorFact) error

PublishFacts appends the batch as one entry on the key's ring and wakes every parked follower.

func (*MemoryFactStream) TransportKind added in v0.41.0 

func (m *MemoryFactStream) TransportKind() FactTransportKind

TransportKind names this transport for the metric labels.

type MemoryFactStreamConfig added in v0.41.0 

type MemoryFactStreamConfig struct {
	// TTL is the retention horizon: entries older than it are dropped from the ring.
	TTL time.Duration
	// MaxLen caps one ring's entry count, oldest first. Zero means DefaultFactStreamMaxLen; a
	// negative value means no cap, leaving the TTL as the only bound.
	MaxLen int64
	// Block is how long one Next waits for a new entry, which bounds how long a change to the
	// followed set waits to take effect.
	Block time.Duration
	// ReadCount bounds how many entries one Next drains per stream.
	ReadCount int64
}

MemoryFactStreamConfig configures the in-memory transport. Every zero field falls back to its Default… constant, so the zero value is the supported configuration.

type RedisFactStream added in v0.41.0 

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

RedisFactStream is the Redis Streams implementation of the attribution fact transport: one stream per (audit route, group/resource), appended to with XADD and followed with a single blocking XREAD across the followed set.

It uses no consumer groups, deliberately. A consumer group distributes entries BETWEEN consumers, and this is a fan-out: every process watching a type needs every fact for that type. Each follower reads independently from its own in-memory cursor.

func (*RedisFactStream) FollowFacts added in v0.41.0 

func (s *RedisFactStream) FollowFacts(keys []FactStreamKey, horizon time.Duration) FactSubscription

FollowFacts starts following keys from horizon before now.

func (*RedisFactStream) PublishFacts added in v0.41.0 

func (s *RedisFactStream) PublishFacts(ctx context.Context, key FactStreamKey, facts []AuthorFact) error

PublishFacts appends the batch as one XADD on the key's stream, capped with MAXLEN ~, and trims the stream to the retention horizon when one is due.

func (*RedisFactStream) TransportKind added in v0.41.0 

func (s *RedisFactStream) TransportKind() FactTransportKind

TransportKind names this transport for the metric labels.

type RedisFactStreamConfig added in v0.41.0 

type RedisFactStreamConfig struct {
	// TTL is the retention horizon: entries older than it are trimmed away. It is the same number
	// as --author-attribution-ttl, which now bounds stream retention and the in-memory index
	// together.
	TTL time.Duration
	// MaxLen caps one stream's entry count via XADD MAXLEN ~, so a hot type cannot grow without
	// bound between retention trims. Zero means DefaultFactStreamMaxLen; a negative value means no
	// cap, leaving the TTL as the only bound.
	MaxLen int64
	// TrimInterval is how often one stream is XTRIMmed to the retention horizon, zero meaning
	// DefaultFactStreamTrimInterval. The trim is amortized onto the publish path rather than run
	// from a goroutine: a stream nobody writes to needs no trimming, since nothing is arriving to
	// age. It bounds a command round trip per publish, which is why the in-memory transport, whose
	// trim is a slice re-slice, has no equivalent knob and trims on every append.
	TrimInterval time.Duration
	// Block is the XREAD BLOCK period, which bounds how long a change to the followed set waits to
	// take effect.
	Block time.Duration
	// ReadCount is the XREAD COUNT per stream per read.
	ReadCount int64
}

RedisFactStreamConfig configures the Redis Streams transport. Every zero field falls back to its Default… constant, so the zero value is the supported configuration.

type RedisStore 

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

RedisStore is the optional Redis/Valkey-backed store. It owns the connection and persists each GitTarget watch shard's resume cursor (state continuity / work re-pickup), and the readiness gate pings it. It is NOT a hard dependency: --redis-addr may be empty, in which case watches cold-replay on restart instead of resuming. What does require it is the admission webhook's command-author capture, and attribution when it runs over the Redis fact-stream transport; attribution over the in-memory transport needs no Redis at all.

It knows nothing about attribution facts. Those live in their own streams, built on the same connection by RedisFactStream.

func NewRedisStore 

func NewRedisStore(cfg RedisStoreConfig) (*RedisStore, error)

NewRedisStore opens the Redis/Valkey connection that backs the resume cursors.

func (*RedisStore) CommandAuthorStore 

func (s *RedisStore) CommandAuthorStore() *CommandAuthorStore

CommandAuthorStore builds the command-authorship store on this connection. Wire it when the validate-operator-types webhook is enabled; it does not depend on attribution. The record lives in the same top-level author domain as audit facts but in the separate command subfamily, with its own fixed cleanup TTL.

func (*RedisStore) FactStream added in v0.41.0 

func (s *RedisStore) FactStream(cfg RedisFactStreamConfig) *RedisFactStream

FactStream builds the Redis Streams attribution fact transport on this store's connection, in the same keyspace as its other keys. The blocking follower parks on one pooled connection for the duration of each block period, so a process runs one follower rather than one per type.

func (*RedisStore) KeyPrefix 

func (s *RedisStore) KeyPrefix() string

KeyPrefix returns the root namespace every key this store's family owns is written under. Reported at startup so an operator can confirm which keyspace a pod claims.

func (*RedisStore) LookupWatchCursor 

func (s *RedisStore) LookupWatchCursor(
	ctx context.Context,
	gitTargetUID string,
	gvr schema.GroupVersionResource,
	namespace string,
) (string, bool)

LookupWatchCursor returns the last resourceVersion durably processed for one GitTarget watch shard. A miss means the watch must rebuild from a fresh replay.

func (*RedisStore) Ping 

func (s *RedisStore) Ping(ctx context.Context) error

Ping checks liveness of the underlying Redis/Valkey connection. The readiness gate uses it so the pod does not report ready before its resume-cursor store is reachable.

func (*RedisStore) RecordWatchCursor 

func (s *RedisStore) RecordWatchCursor(
	ctx context.Context,
	gitTargetUID string,
	gvr schema.GroupVersionResource,
	namespace, rv string,
) error

RecordWatchCursor stores the last resourceVersion durably processed for one GitTarget watch shard, refreshing watchCursorTTL on each write. The cursor is keyed by GitTarget UID and bounded by the TTL, so it never needs explicit deletion: a live watch keeps it fresh, and a dead one's cursor simply expires.

type RedisStoreConfig 

type RedisStoreConfig struct {
	Addr       string
	Username   string
	AuthValue  string
	DB         int
	TLSEnabled bool
	// KeyPrefix is the root namespace for every key this connection's stores own. Empty
	// means DefaultKeyPrefix. Several reversers can then share one Redis/Valkey without
	// sharing a logical database — Redis offers only 16 of those, and --redis-db was the
	// only separator this operator provided.
	KeyPrefix string
}

RedisStoreConfig configures the Redis/Valkey connection that backs the required watch-resume cursor store.

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.