kubernetes

package
v0.9.0 Latest Latest
Warning

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

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

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrCacheWarming = fmt.Errorf("kubernetes read cache is still warming up")

ErrCacheWarming is returned by GetReadClient while InitReadCache's initial sync is still in flight. Handler code (main_helper.go's getK8sReadClient) turns this into a 503 instead of blocking the request on the sync — warm-up for a fresh pod is bounded (InitReadCache's syncTimeout) and ought to be visible as a clean, fast outage response, not request latency.

View Source 
var ErrDeploymentNotFound = errors.New("deployment not found")

ErrDeploymentNotFound wraps a failed Deployment lookup so callers (the HTTP handler) can tell "the deployment itself is missing" (404) apart from every other failure mode here (500) via errors.Is, without parsing an error string.

View Source 
var Hub = NewEventHub(250 * time.Millisecond)

Hub is the process-lifetime event hub. cache.go's InitReadCache registers an informer event handler per cached type that calls Hub.Publish; main.go's /api/events/stream handler calls Hub.Register/Unregister per SSE connection.

Functions

func AllowedNamespaces added in v0.9.0 

func AllowedNamespaces(c *gin.Context, namespaces []string) (map[string]bool, error)

AllowedNamespaces evaluates CanListRolloutsInNamespace for each of the given namespaces (deduplicated) and returns the subset that are allowed, as a set. Used by cluster-wide read routes (e.g. /api/rollouts) to filter an already-fetched, shared-client-read result down to what the caller may see. Checks run concurrently (bounded by errgroup's default unlimited — namespace counts here are small, and repeat calls hit the cache) so N distinct namespaces cost one round trip each, not N sequential ones.

func CanListRolloutsInNamespace added in v0.9.0 

func CanListRolloutsInNamespace(c *gin.Context, ns string) (bool, error)

CanListRolloutsInNamespace answers whether the caller behind c may list Rollouts in namespace ns, per SelfSubjectAccessReview under their own OIDC token, cached per (token, namespace) for 5 minutes. Always true when no token is present on the request (service-account mode).

func FilterByNamespace added in v0.9.0 

func FilterByNamespace[T any](items []T, namespaceOf func(T) string, allowed map[string]bool) []T

FilterByNamespace returns the subset of items whose namespace (per namespaceOf) is present and true in allowed. Used to trim an already-fetched typed list down to the namespaces a viewer may see.

func FilterKustomizationsByRolloutAnnotation added in v0.9.0 

func FilterKustomizationsByRolloutAnnotation(kustomizations *kustomizev1.KustomizationList, ociRepositories *sourcev1.OCIRepositoryList, rolloutName string) *kustomizev1.KustomizationList

FilterKustomizationsByRolloutAnnotation filters an already-fetched KustomizationList down to the kustomizations that reference rolloutName, either directly via a rollout.kuberik.com/substitute.<var>.from annotation or indirectly via an OCIRepository sourceRef that itself carries the rollout annotation (ociRepositories must already be filtered to that set, e.g. by GetOCIRepositoriesByRolloutAnnotation).

Split out from GetKustomizationsByRolloutAnnotation so callers that need both the Kustomization and OCIRepository lists (main.go's rollout-detail handler) can share one OCIRepositories LIST instead of issuing it twice.

func InitReadCache added in v0.9.0 

func InitReadCache(ctx context.Context, syncTimeout time.Duration)

InitReadCache builds the informer cache and its cache-backed read client, starts the cache, eagerly registers an informer for every type in cachedByObject (GetInformer — without this, informers are created lazily on first Get/List, which would defeat "wait for sync at startup"), and blocks this goroutine (not the caller — call this via `go InitReadCache(...)`) until WaitForCacheSync returns or syncTimeout elapses, whichever is first.

A single informer that never syncs (e.g. a CRD not installed on this cluster — RolloutDependency is optional today, see client.go's existing non-fatal handling for it) does not block the others or leave the read path permanently 503: after the timeout, readCacheReady is set regardless, and any type that never synced falls through to the cache library's own per-call error for that one type, exactly like an uncached client would return for a missing CRD.

func RunMultiStream added in v0.9.0 

func RunMultiStream(ctx context.Context, opts MultiStreamOptions, handlers MultiStreamHandlers)

RunMultiStream merges this process's own hub-local ChangeEvent stream (opts.LocalHub) with a live subscription to every opts.Spokes dashboard's own GET /api/events/stream, until ctx is done or the local hub drops this subscriber for backpressure. It is the whole engine behind the multi- cluster GET /api/events/stream contract:

  • Local events are tagged with opts.LocalName (if not already tagged) and passed through opts.Filter before being handed to handlers.OnChanges.
  • Each spoke is subscribed to under opts.Token — the caller's own bearer token, exactly as main_fanout.go's fetchSpoke forwards it — so the spoke applies the caller's own RBAC visibility and this process never needs to re-filter what the spoke already filtered and tagged. Reconnects with jittered exponential backoff (1s→30s cap) on any connect/read error; a spoke's own local coalescing (its own EventHub) means each batch that does arrive is forwarded to handlers.OnChanges as-is, no further coalescing added here.
  • handlers.OnClusters is called once immediately with the starting snapshot (LocalName always true; every spoke false until its first successful connect) and again on every connectivity flip.
  • A spoke that is down, slow, or erroring never blocks local events — each spoke runs its own goroutine writing into a shared, buffered, drop-on-full channel that Run's single select loop reads alongside the local hub channel.

On return, every goroutine RunMultiStream spawned has already exited (it waits for them) — safe to call repeatedly (e.g. once per test iteration, or once per SSE request) without leaking.

func SetReadClientForTest added in v0.9.0 

func SetReadClientForTest(c *Client) func()

SetReadClientForTest overrides GetReadClient's result for the duration of a test, bypassing InitReadCache (and therefore any real apiserver/kubeconfig) entirely. Returns a restore func that puts back whatever GetReadClient would have returned before the override — call it via `defer`.

Exists for handler-level tests (main_list_order_test.go) that need getK8sReadClient (main_helper.go) to hand back a client backed by a fake controller-runtime client loaded with fixture objects, so the test exercises the real production handler in main.go rather than a reimplementation of it.

func SortByNamespaceName added in v0.9.0 

func SortByNamespaceName[T any, PT nsNamed[T]](items []T)

SortByNamespaceName sorts items in place by (namespace, name), ascending, using a stable sort so equal keys (there are none in practice — namespace scoping and the apiserver both guarantee unique names) never move relative to each other between calls.

T is inferred from the slice argument at the call site (e.g. SortByNamespaceName(rollouts.Items) infers T = rolloutv1alpha1.Rollout); PT is then inferred as *T. Safe for cluster-scoped types too (ClusterRolloutSchedule): GetNamespace() always returns "" for those, which just collapses the sort to name-only order.

Types

type ChangeEvent added in v0.9.0 

type ChangeEvent struct {
	Type            string `json:"type"` // "add" | "update" | "delete"
	Kind            string `json:"kind"`
	Namespace       string `json:"namespace"`
	Name            string `json:"name"`
	Cluster         string `json:"cluster"`
	ResourceVersion string `json:"resourceVersion"`
	Ts              int64  `json:"ts"` // unix millis

	// Object is the full object as the API would serve it (managedFields
	// already stripped by the informer cache's own DefaultTransform —
	// cache.go's cache.TransformStripManagedFields() — and the
	// kubectl.kubernetes.io/last-applied-configuration annotation stripped
	// by AttachObjects below), for the 10 kinds AttachObjects knows how to
	// hydrate. Omitted (nil) on delete events, for any other Kind, when the
	// object could not be re-fetched, and when its marshaled JSON exceeds
	// maxEventObjectBytes — see AttachObjects's doc comment for the exact
	// contract (EVENTS-2026-09-04 Part 2). A spoke's own handler attaches
	// this the same way to its own local events before the hub ever sees
	// them; the hub forwards an already-hydrated spoke batch unchanged, the
	// same way it already forwards a spoke batch's Cluster tag unchanged.
	Object json.RawMessage `json:"object,omitempty"`

	// Derived carries the server-computed response bodies the frontend would
	// otherwise fetch separately on seeing this event — DERIVED-2026-09-04.
	// See attachDerived's doc comment for the exact per-Kind contract, the
	// once-per-batch computation point, and the debounce/size-guard rules.
	// A spoke computes this for its own local events the same way it
	// computes Object; the hub forwards an already-derived spoke batch
	// unchanged.
	Derived json.RawMessage `json:"derived,omitempty"`
	// contains filtered or unexported fields
}

ChangeEvent is one informer add/update/delete on a cached type (cache.go's cachedByObject), coalesced by EventHub and broadcast over SSE (main.go's GET /api/events/stream) so the frontend can invalidate its TanStack queries the instant something changes instead of polling on a timer. See PERF-2026-09-04 §C.6/C.7.

Cluster is the display name of the cluster this event happened on — the same name the frontend already sees from /api/clusters and the [cluster] route segment. cache.go's publishChange (the only in-process producer) never sets it; it is always empty when an event first lands in this EventHub, because a bare EventHub has no notion of "which cluster am I." It is filled in one layer up, by whichever code turns a Hub batch into an outbound SSE message: main.go's /api/events/stream handler (via RunMultiStream, multistream.go) stamps every local batch with this process's own cluster name before it ever reaches a client. A spoke's own events therefore arrive at the hub already carrying the spoke's name (the spoke's own handler stamped them the same way), so the hub forwards spoke batches verbatim rather than re-tagging them.

func AttachObjects added in v0.9.0 

func AttachObjects(ctx context.Context, k8sClient *Client, events []ChangeEvent) []ChangeEvent

AttachObjects fills in ChangeEvent.Object for each event in events whose Kind is one of objectCarryingKinds, by re-Getting that object through k8sClient. For the hub-local read client (kubernetes.GetReadClient) this is an in-memory informer-cache lookup for every kind this function knows about (see objectCarryingKinds's doc comment) — not a live apiserver round trip, so calling this once per coalesced batch is cheap.

Returns a NEW slice; events itself is never mutated in place. That matters because RunMultiStream's local batch is the same slice shared with every other client currently registered on the hub for this coalescing window (see its own tagging step's doc comment) — main.go's handler runs AttachObjects downstream of that copy today, but keeping the same non-mutating discipline here means a future caller that reorders Filter/AttachObjects can't reintroduce that bug by accident.

Object is left nil (never an error) for:

  • delete events — the object is gone, there is nothing to embed;
  • any Kind not in objectCarryingKinds;
  • a Get that errors — most commonly the object was deleted again between the informer callback that produced this event and this call, the same "stale by the time you look" race every read in this codebase already tolerates;
  • anything whose marshaled JSON exceeds maxEventObjectBytes.

Every one of these falls back the same way: the frontend invalidates and refetches, exactly as it did for every event before this field existed.

k8sClient may be nil (e.g. the caller's own GetReadClient call failed) — AttachObjects then returns events unchanged rather than panicking, so a caller can pass its read-client result straight through without an extra nil check.

func FilterEventsByVisibility added in v0.9.0 

func FilterEventsByVisibility(c *gin.Context, events []ChangeEvent) []ChangeEvent

FilterEventsByVisibility trims a coalesced batch of change events (from EventHub, see eventhub.go) down to the ones the caller behind c may see, using the same "list rollouts in this namespace" SelfSubjectAccessReview as every other namespaced read (CanListRolloutsInNamespace) — the change stream must not leak the existence of an object in a namespace the caller couldn't otherwise list. No-op when the request carries no OIDC token (service-account mode streams every event, matching every other read path's no-token behavior). A single denied/erroring namespace only drops that namespace's events, never the whole batch.

type Client 

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

func GetClientFromGoContext added in v0.6.0 

func GetClientFromGoContext(ctx context.Context) (*Client, error)

GetClientFromGoContext is a helper for non-Gin contexts (e.g., background operations) It always uses the default client

func GetDefaultClient added in v0.6.0 

func GetDefaultClient() (*Client, error)

GetDefaultClient returns the default Kubernetes client (using service account credentials) This is lazily initialized on first use

func GetReadClient added in v0.9.0 

func GetReadClient(c *gin.Context) (*Client, error)

GetReadClient returns the shared, process-lifetime client used for every read (LIST/GET) behind the JSON routes. Backed by the informer cache built by InitReadCache (cache.go) once it has completed its startup sync; ErrCacheWarming is returned (never a block) if that sync is still in flight, and this falls back to the plain (uncached) default client if InitReadCache was never called at all (e.g. in tests, or a binary that doesn't wire it up) — same behavior as before slice 2. The per-request token, if any, is intentionally ignored — see the visibility-filter requirement above for how per-user RBAC is still enforced on data read through this client.

func GetReadClientForGoContext added in v0.9.0 

func GetReadClientForGoContext(ctx context.Context) (*Client, error)

GetReadClientForGoContext is GetReadClient for non-Gin callers (background operations that have no per-request user identity to consider anyway).

func GetWriteClient added in v0.9.0 

func GetWriteClient(c *gin.Context) (*Client, error)

GetWriteClient returns a client authorized as the actual signed-in operator for a mutating call or a SelfSubjectAccessReview. When an OIDC token is present on the request it is used verbatim (never silently downgraded to the service account); otherwise this falls back to the shared default client, matching today's service-account-only behavior. Construction is cheap — see NewClientWithToken — it reuses the shared Scheme/RESTMapper instead of redoing API discovery per request.

func NewClient 

func NewClient() (*Client, error)

NewClient creates a Kubernetes client using service account credentials (in-cluster) or kubeconfig

func NewClientWithToken added in v0.6.0 

func NewClientWithToken(token string) (*Client, error)

NewClientWithToken creates a Kubernetes client using the provided OIDC token. If token is empty, falls back to service account credentials (in-cluster) or kubeconfig.

Cheap to call per request: the Scheme and RESTMapper are built once (sharedSchemeAndMapper) and reused here, so this only builds a fresh rest.Config (cheap — no network call) plus a client.New/clientset construction that does no discovery of its own. Callers needing a long-lived, cache-backed reader should prefer GetReadClient/GetDefaultClient instead of calling this directly — see context.go for the read/write split.

func NewTestClient added in v0.9.0 

func NewTestClient(objs ...client.Object) (*Client, error)

NewTestClient builds a *Client backed by a controller-runtime fake client preloaded with objs, sharing the same Scheme every real client in this package builds from (buildScheme, in client.go). For tests only — production code always goes through NewClient/NewClientWithToken or the informer-cache-backed client InitReadCache builds; this has no clientset, so anything reading through c.clientset (pod logs, GetEventsForRollout's typed Events LIST) is out of scope for tests built on top of this.

The fake client's own List order is not guaranteed stable across calls — its backing store is itself a Go map (client-go's testing.ObjectTracker) — which is exactly why this exists: it reproduces the same class of nondeterminism InitReadCache's real informer cache has (PERF-2026-09-04 §D), so a test built on it actually exercises whether the code sorts its own output rather than happening to pass because the fixture data was already in order.

func NewTestClientWithPods added in v0.9.0 

func NewTestClientWithPods(objs []client.Object, pods ...corev1.Pod) (*Client, *k8sfake.Clientset, error)

NewTestClientWithPods is NewTestClient plus a client-go fake clientset (k8s.io/client-go/kubernetes/fake) preloaded with pods, wired in as c.clientset — DERIVED-2026-09-04. BuildDeploymentChildren's Pods LIST goes through GetClientset() rather than the controller-runtime client (see that function's doc comment for why: Pods are deliberately not part of InitReadCache's informer cache), which is exactly what NewTestClient's own doc comment says is out of scope for it. Client.clientset is typed as the kubernetes.Interface a *kubernetes.Clientset implements rather than that concrete type specifically so a fake clientset can stand in here.

Returns the fake Clientset itself (not just *Client) so a test can attach its own Fake.PrependReactor to count LIST calls — used to verify derived-data computation happens once per coalesced batch rather than once per subscriber (eventhub_derived_test.go).

func (*Client) AddBypassGatesAnnotation 

func (c *Client) AddBypassGatesAnnotation(ctx context.Context, namespace, name string, version string) (*rolloutv1alpha1.Rollout, error)

AddBypassGatesAnnotation adds the rollout.kuberik.com/bypass-gates annotation to a rollout This allows the rollout to bypass gate checks for a specific version

func (*Client) AddForceDeployAnnotation added in v0.4.0 

func (c *Client) AddForceDeployAnnotation(ctx context.Context, namespace, name string, version string, message string) (*rolloutv1alpha1.Rollout, error)

AddForceDeployAnnotation adds the rollout.kuberik.com/force-deploy annotation to a rollout This allows the rollout to force deploy a specific version Optionally includes a message explaining why the force deploy was triggered

func (*Client) AddUnblockFailedAnnotation 

func (c *Client) AddUnblockFailedAnnotation(ctx context.Context, namespace, name string) (*rolloutv1alpha1.Rollout, error)

AddUnblockFailedAnnotation adds the rollout.kuberik.com/unblock-failed annotation to a rollout This allows the rollout to resume after a failed bake

func (*Client) ChangeVersion added in v0.5.0 

func (c *Client) ChangeVersion(ctx context.Context, namespace, name string, version string, pin bool, message string) (*rolloutv1alpha1.Rollout, error)

ChangeVersion updates the rollout version with an option to pin or unpin atomically. When pin is true, it sets spec.wantedVersion to the version and optionally sets a deploy message. When pin is false, it adds the force-deploy annotation for the version and clears spec.wantedVersion in the same server-side apply operation, optionally setting a deploy message.

func (*Client) CheckPermission added in v0.6.0 

func (c *Client) CheckPermission(ctx context.Context, apiGroup, resource, verb, namespace, name string) (bool, error)

CheckPermission checks if the current user has permission to perform an action using SelfSubjectAccessReview API Uses the stored REST config which includes the user's OIDC token

func (*Client) CheckRolloutPermission added in v0.6.0 

func (c *Client) CheckRolloutPermission(ctx context.Context, verb, namespace, name string) (bool, error)

CheckRolloutPermission checks if the current user has permission to perform an action on a Rollout

func (*Client) ClearKruiseRolloutStalledCondition added in v0.8.0 

func (c *Client) ClearKruiseRolloutStalledCondition(ctx context.Context, namespace, name string) error

ClearKruiseRolloutStalledCondition sets the Stalled condition on a kruise rollout to False and resets the step started-at annotation so the step timeout window is refreshed. This allows the rollouttest controller to create new jobs after a retry.

func (*Client) ContinueKruiseRollout added in v0.4.0 

func (c *Client) ContinueKruiseRollout(ctx context.Context, namespace, name string) (*kruiserolloutv1beta1.Rollout, error)

ContinueKruiseRollout updates the currentStepState of an OpenKruise rollout to continue the rollout

func (*Client) FormatUserInfo added in v0.6.0 

func (c *Client) FormatUserInfo(ctx context.Context) (string, error)

FormatUserInfo formats user information for appending to deploy messages Returns empty string if user is a service account

func (*Client) GetAllPods added in v0.6.0 

func (c *Client) GetAllPods(ctx context.Context, namespace string) (*corev1.PodList, error)

GetAllPods lists all pods in a namespace

func (*Client) GetAllRolloutTests added in v0.6.0 

func (c *Client) GetAllRolloutTests(ctx context.Context, namespace string) (*openkruisev1alpha1.RolloutTestList, error)

GetAllRolloutTests fetches all RolloutTests in a namespace

func (*Client) GetClientset added in v0.6.0 

func (c *Client) GetClientset() kubernetes.Interface

GetClientset returns the Kubernetes clientset for direct API access.

Typed as the kubernetes.Interface the real *kubernetes.Clientset implements, not the concrete type — DERIVED-2026-09-04: this is what lets a test build a *Client around client-go's fake.NewSimpleClientset instead of the real thing, so BuildDeploymentChildren's live Pods LIST (the one call this package's read cache deliberately does not serve — see cache.go's cachedByObject doc comment) can be exercised, and counted, without a real apiserver.

func (*Client) GetClusterRolloutSchedule added in v0.7.0 

func (c *Client) GetClusterRolloutSchedule(ctx context.Context, name string) (*rolloutv1alpha1.ClusterRolloutSchedule, error)

GetClusterRolloutSchedule gets a single ClusterRolloutSchedule

func (*Client) GetClusterRolloutSchedules added in v0.7.0 

func (c *Client) GetClusterRolloutSchedules(ctx context.Context) (*rolloutv1alpha1.ClusterRolloutScheduleList, error)

GetClusterRolloutSchedules gets all ClusterRolloutSchedules

func (*Client) GetClusterRolloutSchedulesByRollout added in v0.7.0 

func (c *Client) GetClusterRolloutSchedulesByRollout(ctx context.Context, namespace, rolloutName string, rolloutLabels, namespaceLabels map[string]string) (*rolloutv1alpha1.ClusterRolloutScheduleList, error)

GetClusterRolloutSchedulesByRollout gets ClusterRolloutSchedules that match a specific rollout.

Same inverted-selector situation as GetRolloutSchedulesByRollout — the match is evaluated against this rollout's (and its namespace's) labels using each schedule's own embedded selector, which has no server-side query equivalent. ClusterRolloutSchedule is cluster-scoped by definition, so there's no namespace to narrow the LIST by either; this is already the minimum round trip for this resource.

func (*Client) GetCurrentUserIdentity added in v0.6.0 

func (c *Client) GetCurrentUserIdentity(ctx context.Context) (string, bool, error)

GetCurrentUserIdentity gets the current user's identity using SelfSubjectReview API This is the same API that kubectl auth whoami uses Returns the username and a boolean indicating if it's a service account Returns empty string and false if unable to determine identity

func (*Client) GetDeployment added in v0.9.0 

func (c *Client) GetDeployment(ctx context.Context, namespace, name string) (*appsv1.Deployment, error)

GetDeployment fetches a single Deployment by namespace/name, for the GET .../deployments/:name/children handler (main.go). Goes through c's controller-runtime client — when c is the shared read client (GetReadClient), that's an in-memory informer-cache hit (cache.go's cachedByObject now includes apps/v1 Deployment), not a live apiserver round trip, which is what makes the refetch the frontend does on a Deployment/ReplicaSet stream event cheap.

func (*Client) GetEnvironmentByRolloutReference added in v0.6.0 

func (c *Client) GetEnvironmentByRolloutReference(ctx context.Context, namespace, rolloutName string) (*envv1alpha1.Environment, error)

GetEnvironmentByRolloutReference fetches Environment that references a specific rollout.

Left as list-then-filter: Environment carries spec.rolloutRef.Name as a plain object reference, not a label — environment-controller never stamps a corresponding label on the object (checked against the vendored environment-controller source), so there's no selector to push down.

func (*Client) GetEnvironments added in v0.6.0 

func (c *Client) GetEnvironments(ctx context.Context, namespace string) (*envv1alpha1.EnvironmentList, error)

GetEnvironments fetches all Environments in a namespace

func (*Client) GetEnvironmentsAllNamespaces added in v0.8.0 

func (c *Client) GetEnvironmentsAllNamespaces(ctx context.Context) (*envv1alpha1.EnvironmentList, error)

GetEnvironmentsAllNamespaces fetches all Environments across all namespaces

func (*Client) GetEventsForRollout added in v0.8.0 

func (c *Client) GetEventsForRollout(ctx context.Context, namespace, rolloutName string) ([]corev1.Event, error)

GetEventsForRollout collects events relevant to a rollout: 1. Events for the Rollout object itself 2. Events for all Deployments found via kustomizations linked to the rollout 3. Events for ReplicaSets owned by those Deployments

func (*Client) GetHealthChecksBySelector 

func (c *Client) GetHealthChecksBySelector(ctx context.Context, namespace string, selector *rolloutv1alpha1.HealthCheckSelectorConfig) ([]rolloutv1alpha1.HealthCheck, error)

GetHealthChecksBySelector returns health checks that match the given selector

func (*Client) GetImagePolicies 

func (c *Client) GetImagePolicies(ctx context.Context, namespace string) (*imagereflectorv1beta2.ImagePolicyList, error)

func (*Client) GetImagePoliciesAllNamespaces 

func (c *Client) GetImagePoliciesAllNamespaces(ctx context.Context) (*imagereflectorv1beta2.ImagePolicyList, error)

New: list image policies across all namespaces

func (*Client) GetImagePolicy 

func (c *Client) GetImagePolicy(ctx context.Context, namespace, name string) (*imagereflectorv1beta2.ImagePolicy, error)

func (*Client) GetImageRepositories 

func (c *Client) GetImageRepositories(ctx context.Context, namespace string) (*imagereflectorv1beta2.ImageRepositoryList, error)

func (*Client) GetImageRepositoriesAllNamespaces 

func (c *Client) GetImageRepositoriesAllNamespaces(ctx context.Context) (*imagereflectorv1beta2.ImageRepositoryList, error)

New: list image repositories across all namespaces

func (*Client) GetImageRepository 

func (c *Client) GetImageRepository(ctx context.Context, namespace, name string) (*imagereflectorv1beta2.ImageRepository, error)

func (*Client) GetKruiseRollout added in v0.6.0 

func (c *Client) GetKruiseRollout(ctx context.Context, namespace, name string) (*kruiserolloutv1beta1.Rollout, error)

GetKruiseRollout fetches a KruiseRollout by name and namespace

func (*Client) GetKruiseRollouts added in v0.8.0 

func (c *Client) GetKruiseRollouts(ctx context.Context, namespace string) (*kruiserolloutv1beta1.RolloutList, error)

GetKruiseRollouts lists KruiseRollouts in a single namespace.

func (*Client) GetKruiseRolloutsAllNamespaces added in v0.8.0 

func (c *Client) GetKruiseRolloutsAllNamespaces(ctx context.Context) (*kruiserolloutv1beta1.RolloutList, error)

GetKruiseRolloutsAllNamespaces lists KruiseRollouts across all namespaces. Used by the rollouts list endpoint so the frontend can correlate each kuberik Rollout to its underlying KruiseRollouts (via the linked Kustomization's inventory entries) and render a real pipeline glyph.

func (*Client) GetKustomization 

func (c *Client) GetKustomization(ctx context.Context, namespace, name string) (*kustomizev1.Kustomization, error)

func (*Client) GetKustomizationManagedResources 

func (c *Client) GetKustomizationManagedResources(ctx context.Context, namespace, name string) ([]ManagedResourceStatus, error)

func (*Client) GetKustomizations 

func (c *Client) GetKustomizations(ctx context.Context, namespace string) (*kustomizev1.KustomizationList, error)

func (*Client) GetKustomizationsAllNamespaces 

func (c *Client) GetKustomizationsAllNamespaces(ctx context.Context) (*kustomizev1.KustomizationList, error)

New: list kustomizations across all namespaces

func (*Client) GetKustomizationsByRolloutAnnotation 

func (c *Client) GetKustomizationsByRolloutAnnotation(ctx context.Context, namespace, rolloutName string) (*kustomizev1.KustomizationList, error)

func (*Client) GetOCIRepositories 

func (c *Client) GetOCIRepositories(ctx context.Context, namespace string) (*sourcev1.OCIRepositoryList, error)

func (*Client) GetOCIRepositoriesAllNamespaces 

func (c *Client) GetOCIRepositoriesAllNamespaces(ctx context.Context) (*sourcev1.OCIRepositoryList, error)

New: list OCI repositories across all namespaces

func (*Client) GetOCIRepositoriesByRolloutAnnotation 

func (c *Client) GetOCIRepositoriesByRolloutAnnotation(ctx context.Context, namespace, rolloutName string) (*sourcev1.OCIRepositoryList, error)

func (*Client) GetPodLogs added in v0.6.0 

func (c *Client) GetPodLogs(ctx context.Context, namespace, podName, containerName string, tailLines *int64, follow bool) (string, error)

GetPodLogs retrieves logs from a pod

func (*Client) GetPodsByJobName added in v0.6.0 

func (c *Client) GetPodsByJobName(ctx context.Context, namespace, jobName string) (*corev1.PodList, error)

GetPodsByJobName lists pods owned by a job

func (*Client) GetPodsByOwnerReference added in v0.6.0 

func (c *Client) GetPodsByOwnerReference(ctx context.Context, namespace string, ownerUID string) (*corev1.PodList, error)

GetPodsByOwnerReference lists pods owned by a specific resource (by UID)

func (*Client) GetPodsBySelector added in v0.6.0 

func (c *Client) GetPodsBySelector(ctx context.Context, namespace string, selector labels.Selector) (*corev1.PodList, error)

GetPodsBySelector lists pods matching the given label selector

func (*Client) GetReplicaSets added in v0.6.0 

func (c *Client) GetReplicaSets(ctx context.Context, namespace string) (*appsv1.ReplicaSetList, error)

GetReplicaSets lists replica sets in a namespace

func (*Client) GetReplicaSetsBySelector added in v0.9.0 

func (c *Client) GetReplicaSetsBySelector(ctx context.Context, namespace string, selector labels.Selector) (*appsv1.ReplicaSetList, error)

GetReplicaSetsBySelector lists ReplicaSets in namespace matching selector, for the same children handler as GetDeployment — cache-backed the same way when c is the shared read client. Pods stay a separate, uncached, live LIST in the handler (cache.go's cachedByObject deliberately excludes Pod; see its doc comment).

func (*Client) GetRollout 

func (c *Client) GetRollout(ctx context.Context, namespace, name string) (*rolloutv1alpha1.Rollout, error)

func (*Client) GetRolloutDependencies added in v0.9.0 

func (c *Client) GetRolloutDependencies(ctx context.Context, namespace string) (*rolloutv1alpha1.RolloutDependencyList, error)

GetRolloutDependencies lists RolloutDependencies in a namespace, with spec defaults resolved.

func (*Client) GetRolloutDependenciesAllNamespaces added in v0.9.0 

func (c *Client) GetRolloutDependenciesAllNamespaces(ctx context.Context) (*rolloutv1alpha1.RolloutDependencyList, error)

GetRolloutDependenciesAllNamespaces lists RolloutDependencies across all namespaces, with spec defaults resolved.

func (*Client) GetRolloutGatesByRolloutReference 

func (c *Client) GetRolloutGatesByRolloutReference(ctx context.Context, namespace, rolloutName string) (*rolloutv1alpha1.RolloutGateList, error)

GetRolloutGatesByRolloutReference fetches RolloutGates that reference a specific rollout.

Left as list-then-filter-in-Go rather than a label selector: the rolloutschedule and rolloutdependency controllers both stamp gate.kuberik.com/rollout-name on the gates they create, but that's an implementation detail of those two controllers, not a guarantee on the RolloutGate CRD — a hand-authored or third-party-controller-created gate can set spec.rolloutRef.Name without the label. Selecting on the label would silently drop any such gate from this list even though it correctly targets the rollout. Namespace-scoped LIST + filter keeps that correct.

func (*Client) GetRolloutSchedule added in v0.7.0 

func (c *Client) GetRolloutSchedule(ctx context.Context, namespace, name string) (*rolloutv1alpha1.RolloutSchedule, error)

GetRolloutSchedule gets a single RolloutSchedule

func (*Client) GetRolloutSchedules added in v0.7.0 

func (c *Client) GetRolloutSchedules(ctx context.Context, namespace string) (*rolloutv1alpha1.RolloutScheduleList, error)

GetRolloutSchedules gets all RolloutSchedules in a namespace

func (*Client) GetRolloutSchedulesAllNamespaces added in v0.7.0 

func (c *Client) GetRolloutSchedulesAllNamespaces(ctx context.Context) (*rolloutv1alpha1.RolloutScheduleList, error)

GetRolloutSchedulesAllNamespaces gets all RolloutSchedules across all namespaces

func (*Client) GetRolloutSchedulesByRollout added in v0.7.0 

func (c *Client) GetRolloutSchedulesByRollout(ctx context.Context, namespace, rolloutName string, rolloutLabels map[string]string) (*rolloutv1alpha1.RolloutScheduleList, error)

GetRolloutSchedulesByRollout gets RolloutSchedules that match a specific rollout.

Left as list-then-filter, and can't be pushed to a label selector: the direction is inverted from the usual "object has a label, filter on it" case. Here each RolloutSchedule carries its own spec.rolloutSelector, and whether it matches is a function of the *rollout's* labels, not any label on the RolloutSchedule itself — there's no server-side query for "list objects whose embedded selector matches this label set." Already namespace-scoped, which is the LIST-narrowing that is available here.

func (*Client) GetRolloutTests added in v0.6.0 

func (c *Client) GetRolloutTests(ctx context.Context, namespace string) (*openkruisev1alpha1.RolloutTestList, error)

func (*Client) GetRolloutTestsByRolloutName added in v0.6.0 

func (c *Client) GetRolloutTestsByRolloutName(ctx context.Context, namespace, rolloutName string) (*openkruisev1alpha1.RolloutTestList, error)

GetRolloutTestsByRolloutName fetches RolloutTests that reference a specific KruiseRollout by name.

Left as list-then-filter: RolloutTest carries spec.rolloutName as a plain string field, not a label — openkruise-controller never stamps a corresponding label on the object (checked against the vendored openkruise-controller source), so there's nothing to select on server-side without the controller changing first.

func (*Client) GetRollouts 

func (c *Client) GetRollouts(ctx context.Context, namespace string) (*rolloutv1alpha1.RolloutList, error)

func (*Client) GetRolloutsAllNamespaces 

func (c *Client) GetRolloutsAllNamespaces(ctx context.Context) (*rolloutv1alpha1.RolloutList, error)

New: list rollouts across all namespaces

func (*Client) GetSecret 

func (c *Client) GetSecret(ctx context.Context, namespace, name string) (*corev1.Secret, error)

func (*Client) ManagedResourceStatusForDeployment added in v0.9.0 

func (c *Client) ManagedResourceStatusForDeployment(ctx context.Context, namespace, name string) (*ManagedResourceStatus, error)

ManagedResourceStatusForDeployment computes the exact ManagedResourceStatus entry GetKustomizationManagedResources would return for one Deployment — same GVK string ("apps/v1/Deployment"), same conversion (toUnstructuredWithGVK), same status.Compute call via managedResourceStatusFromObject — without needing to know which Kustomization owns it. Used by eventhub.go's derived-data computation (DERIVED-2026-09-04) so a Deployment ChangeEvent can carry the same status word the managed-resources endpoint would compute, from the same cached object, without the frontend making a second round trip.

dep is fetched via c.GetDeployment — an informer-cache hit for the InitReadCache-backed client this is always called against in production — so this costs one cache lookup, not a live apiserver GET.

func (*Client) MarkDeploymentSuccessful added in v0.4.0 

func (c *Client) MarkDeploymentSuccessful(ctx context.Context, namespace, name string, message string) (*rolloutv1alpha1.Rollout, error)

MarkDeploymentSuccessful marks the latest deployment as successful by updating the rollout status

func (*Client) ReconcileAllFluxResources 

func (c *Client) ReconcileAllFluxResources(ctx context.Context, namespace, rolloutName string) (previousScanTime string, err error)

ReconcileAllFluxResources reconciles all associated Flux resources for a rollout Returns the previous scanTime of the ImageRepository (if found) so the caller can detect completion

func (*Client) ReconcileImageRepository added in v0.7.0 

func (c *Client) ReconcileImageRepository(ctx context.Context, namespace, name string) error

ReconcileImageRepository adds the reconcile annotation to trigger a reconciliation

func (*Client) ReconcileKustomization 

func (c *Client) ReconcileKustomization(ctx context.Context, namespace, name string) error

ReconcileKustomization adds the reconcile annotation to trigger a reconciliation

func (*Client) ReconcileOCIRepository 

func (c *Client) ReconcileOCIRepository(ctx context.Context, namespace, name string) error

ReconcileOCIRepository adds the reconcile annotation to trigger a reconciliation

func (*Client) ResetBakeStatusToDeploying added in v0.7.4 

func (c *Client) ResetBakeStatusToDeploying(ctx context.Context, namespace, name string) (*rolloutv1alpha1.Rollout, error)

ResetBakeStatusToDeploying resets the rollout's bake status to "Deploying" This should be called when continuing a rollout to indicate a new deployment phase

func (*Client) ResetHealthChecksToPending added in v0.7.4 

func (c *Client) ResetHealthChecksToPending(ctx context.Context, namespace, name string) error

ResetHealthChecksToPending resets all health checks matching the rollout's selector to "Pending" This should be called when continuing a rollout to reset health monitoring

func (*Client) SetRetryAnnotation added in v0.8.0 

func (c *Client) SetRetryAnnotation(ctx context.Context, namespace, name, mode string) error

SetRetryAnnotation patches the Rollout with the rollout.kuberik.com/retry annotation (presence-only trigger consumed by rollout-controller) and, when mode is "skip", also sets rollouttest.kuberik.com/retry-mode so the openkruise stepgate marks failed RolloutTests as Skipped instead of re-running them.

func (*Client) UpdateRolloutVersion 

func (c *Client) UpdateRolloutVersion(ctx context.Context, namespace, name string, version *string, explanation string) (*rolloutv1alpha1.Rollout, error)

type ClusterSpec added in v0.9.0 

type ClusterSpec struct {
	Name string
	URL  string
}

ClusterSpec names one spoke dashboard for RunMultiStream to subscribe to — the same {name, url} shape main.go's ClusterInfo carries, duplicated here (rather than imported) so this package stays free of a dependency on the main package.

type DeploymentChildren added in v0.9.0 

type DeploymentChildren struct {
	ReplicaSets []RSInfo       `json:"replicaSets"`
	Deployment  DeploymentInfo `json:"deployment"`
}

DeploymentChildren is the exact JSON body GET /namespaces/:namespace/deployments/:name/children returns.

func BuildDeploymentChildren added in v0.9.0 

func BuildDeploymentChildren(ctx context.Context, k8sClient *Client, namespace, name string) (*DeploymentChildren, error)

BuildDeploymentChildren fetches the Deployment, its owned ReplicaSets, and their Pods, and assembles them into the same response shape the /namespaces/:namespace/deployments/:name/children handler has always returned. Both that handler (main.go) and eventhub.go's derived-data computation call this — the only difference between them is who pays for it and how often (see eventhub.go's attachDerived doc comment for the once-per-batch, debounced version of that story).

The Deployment and its ReplicaSets are read via k8sClient's controller-runtime client (GetDeployment/GetReplicaSetsBySelector) — an informer-cache hit for the InitReadCache-backed client both call sites use in production. Pods stay a live, namespaced LIST through k8sClient's clientset (GetClientset) — cache.go's cachedByObject deliberately excludes cluster-wide Pods, so this is unavoidably a real apiserver round trip, same as before this function existed.

Returns an error wrapping ErrDeploymentNotFound (errors.Is) if the Deployment itself can't be fetched; any other error is a ReplicaSet/Pod LIST or selector-parse failure with its own detail in err.Error().

type DeploymentInfo added in v0.9.0 

type DeploymentInfo struct {
	Name              string `json:"name"`
	Namespace         string `json:"namespace"`
	Replicas          int32  `json:"replicas"`
	ReadyReplicas     int32  `json:"readyReplicas"`
	UpdatedReplicas   int32  `json:"updatedReplicas"`
	AvailableReplicas int32  `json:"availableReplicas"`
}

DeploymentInfo is the small subset of Deployment status the children response echoes back alongside its ReplicaSets, matching the shape the handler has always returned (a map[string]interface{} before this file existed).

type DerivedData added in v0.9.0 

type DerivedData struct {
	ManagedResource *ManagedResourceStatus `json:"managedResource,omitempty"`
	OwnerDeployment *OwnerDeploymentRef    `json:"ownerDeployment,omitempty"`
	Children        *DeploymentChildren    `json:"children,omitempty"`
}

DerivedData is the JSON shape of ChangeEvent.Derived, decoded from the json.RawMessage that field actually carries on the wire (marshalDerived below stores the marshaled bytes directly, the same way AttachObjects stores Object's marshaled bytes directly, rather than a struct that gets remarshaled a second time when the whole batch is encoded for SSE).

  • On a Deployment add/update event: ManagedResource is the exact ManagedResourceStatus entry GET .../managed-resources would return for that Deployment (client.go's managedResourceStatusFromObject, via ManagedResourceStatusForDeployment), and Children is the exact GET .../children response for that same Deployment — a Deployment status write usually precedes the RS/Pod changes it caused, so shipping Children here too saves the frontend a second refetch a beat later. OwnerDeployment is unset.
  • On a ReplicaSet add/update/delete event: OwnerDeployment names the Deployment that owns it (from the ReplicaSet's own OwnerReferences, captured at publish time — cache.go's publishChange), and Children is that owner Deployment's exact GET .../children response. ManagedResource is unset.

Every field is a pointer so an unset one is simply absent from the JSON (`omitempty`) rather than present-but-zero — the frontend's existing invalidate-and-refetch fallback is what fills in anything left out here, exactly as it already does for a nil Object.

type EventHub added in v0.9.0 

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

EventHub coalesces informer events into small batches (one flush per window) and fans them out to connected SSE clients, each with its own bounded buffer.

Deliberately does NOT do per-user namespace visibility filtering — that requires a SelfSubjectAccessReview round trip (kubernetes. CanListRolloutsInNamespace / FilterEventsByVisibility), and running that inside flush() while holding h.mu would let one slow/uncached check stall delivery to every other connected client. The HTTP handler (main.go's /api/events/stream) filters the raw batch it reads from its own channel, in its own goroutine, after Register returns it — same visibility rule as every other read, just applied one layer up from here.

func NewEventHub added in v0.9.0 

func NewEventHub(window time.Duration) *EventHub

NewEventHub builds a hub and starts its coalescing loop in a background goroutine. window is how often pending events are flushed to clients (the spec: 250ms in production; tests use a much shorter window so they don't sleep for real time).

func (*EventHub) ClientCount added in v0.9.0 

func (h *EventHub) ClientCount() int

ClientCount reports the number of currently registered clients.

func (*EventHub) Publish added in v0.9.0 

func (h *EventHub) Publish(ev ChangeEvent)

Publish records one informer event, keyed by (kind, namespace, name) so repeated updates to the same object inside one coalescing window collapse into that object's latest state instead of one SSE message per watch event — e.g. a Rollout that gets three status patches in 250ms produces one "update" event, not three.

func (*EventHub) Register added in v0.9.0 

func (h *EventHub) Register(bufSize int) (uint64, <-chan []ChangeEvent)

Register adds a new client with a buffered channel of size bufSize and returns its id (for Unregister) and the channel to read coalesced batches from. The channel is closed by the hub itself when the client is dropped for backpressure (see flush) — callers must treat a closed channel as "reconnect", not as an error.

func (*EventHub) RegisterWithCap added in v0.9.0 

func (h *EventHub) RegisterWithCap(identity string, bufSize int, limits IdentityLimits) (id uint64, ch <-chan []ChangeEvent, total int, ok bool)

RegisterWithCap is Register plus the concurrent-subscriber caps that defend the hub against a leaked proxy holding open thousands of abandoned SSE requests (2026-09-04 incident: a dev proxy in front of the hub never closed upstream requests when browsers went away; ~1000 orphaned /api/events/stream subscribers, each with its own spoke subscriptions and heartbeat ticker, saturated the k8s client's rate limiter and crashed the pod).

identity is caller-defined — main.go's handler uses a hash of the bearer token when present, else the client IP — this package attaches no meaning to it beyond "same string means same caller for capping purposes".

If registering this client would put the hub's total at or beyond limits.MaxTotal, registration is refused entirely (ok=false, id/ch zero) — the caller must respond (503) without this client ever touching the hub, exactly as if RegisterWithCap were never called. Otherwise the client is registered exactly as Register would, and if that pushes this SAME identity's own concurrent count over limits.MaxPerClient, the OLDEST still-registered subscription for this identity is evicted — closing its channel via the same dropClientLocked path flush() uses for backpressure, which is what makes eviction tear down everything the evicted request owned: the losing RunMultiStream call reads a closed local channel, fires OnLocalDropped, and returns — cancelling its spoke subscriptions and stopping its heartbeat ticker via its own defers (see multistream.go). A real user opening a 9th tab therefore evicts their own oldest tab, never another identity's stream; a leaked-proxy flood loses its oldest copy of itself every time a new one opens.

Returns the new client's id and channel (nil/0 if refused) plus the hub's total client count immediately after this call (or, when refused, the count that caused the refusal) — so the caller can set X-Kuberik-Stream-Clients from this single locked call without a second ClientCount round trip.

func (*EventHub) Stop added in v0.9.0 

func (h *EventHub) Stop()

Stop ends the coalescing goroutine. Production never calls this (the hub runs for the process lifetime, same as the informer cache it sits behind); it exists so tests can assert on goroutine counts deterministically instead of leaking one background goroutine per test.

func (*EventHub) Unregister added in v0.9.0 

func (h *EventHub) Unregister(id uint64)

Unregister removes a client. Safe to call after the hub has already dropped the client for backpressure, or evicted it via RegisterWithCap — the map entry is already gone by then, so this is a no-op, not a double- close (dropClientLocked's h.clients lookup guards the close).

type IdentityLimits added in v0.9.0 

type IdentityLimits struct {
	// MaxPerClient is how many concurrently-registered clients one identity
	// may hold before RegisterWithCap starts evicting that identity's own
	// oldest subscription to make room for a new one.
	MaxPerClient int
	// MaxTotal is the hard ceiling on clients registered across every
	// identity — reaching it refuses new registrations outright rather than
	// evicting anyone, so a flood can't grow the process's memory even
	// transiently.
	MaxTotal int
}

IdentityLimits configures the concurrent-subscriber caps RegisterWithCap enforces. Either field <= 0 disables that particular cap.

type ManagedResourceStatus 

type ManagedResourceStatus struct {
	GroupVersionKind string                     `json:"groupVersionKind"`
	Name             string                     `json:"name"`
	Namespace        string                     `json:"namespace"`
	Status           string                     `json:"status"`
	Message          string                     `json:"message"`
	LastModified     time.Time                  `json:"lastModified"`
	Object           *unstructured.Unstructured `json:"object"`
}

type MultiStreamHandlers added in v0.9.0 

type MultiStreamHandlers struct {
	// OnChanges is called once per source batch ready to forward — one call
	// per local flush, and one call per batch a spoke sent, kept separate
	// rather than merged into a bigger batch (the 250ms coalescing already
	// happened at its source; merging further here would just add latency
	// with no coalescing benefit).
	OnChanges func([]ChangeEvent)

	// OnClusters is called once immediately (before Run's select loop
	// starts, i.e. before returning control to the caller) with the initial
	// {name: connected} snapshot, and again every time any cluster's
	// connected state changes.
	OnClusters func(map[string]bool)

	// OnHeartbeat fires on Options.HeartbeatInterval, if set.
	OnHeartbeat func()

	// OnLocalDropped is called if LocalHub drops this subscriber for
	// backpressure (its channel closed — see EventHub.flush). Run returns
	// immediately afterward; the caller should end the SSE response so the
	// browser's EventSource reconnects with a clean buffer.
	OnLocalDropped func()
}

MultiStreamHandlers are Run's callbacks. Every call happens synchronously from Run's own goroutine, one at a time, never concurrently with another — so a caller (main.go's SSE handler) can write straight to an http.ResponseWriter from inside them with no locking of its own.

type MultiStreamOptions added in v0.9.0 

type MultiStreamOptions struct {
	// LocalHub is the EventHub to read this process's own informer events
	// from. Defaults to the package-level Hub when nil; tests pass their own
	// so they don't share state with other tests or the real cache.
	LocalHub *EventHub

	// LocalName is this process's own cluster display name — stamped onto
	// every local ChangeEvent whose Cluster field is still empty (cache.go's
	// publishChange never sets it), and reported as always-connected in the
	// "clusters" snapshot.
	LocalName string

	// Spokes are the other dashboards to subscribe to. Empty means
	// single-cluster behavior: local events only, plus one "clusters" event
	// naming just LocalName.
	Spokes []ClusterSpec

	// Token is forwarded as every spoke request's bearer token — the
	// caller's own OIDC token, never a shared/cached credential — so each
	// spoke applies the caller's own RBAC visibility filter to what it
	// sends back. This process does not re-filter events a spoke already
	// filtered and tagged.
	Token string

	// FanoutHeader, if non-empty, is set on every spoke request so the
	// spoke's own handler knows this is already a fan-out leg and must not
	// discover/subscribe to spokes of its own — avoids hub↔spoke
	// subscription cycles. Matches main_fanout.go's fanoutHeader.
	FanoutHeader string

	// HTTPClient issues the spoke requests. Defaults to http.DefaultClient.
	HTTPClient *http.Client

	// Filter, if set, is applied to each LOCAL batch (after cluster
	// tagging) before it reaches Handlers.OnChanges — the per-user
	// visibility check every other read in this codebase applies. Spoke
	// batches are never passed through Filter; the spoke already applied
	// the caller's own visibility to them.
	Filter func([]ChangeEvent) []ChangeEvent

	// HeartbeatInterval, if > 0, fires Handlers.OnHeartbeat on this cadence.
	HeartbeatInterval time.Duration

	// LocalBufSize is the buffer size passed to LocalHub.Register. Defaults
	// to 32. Ignored when LocalClientCh is set (see below) — the caller
	// already chose a buffer size when it registered.
	LocalBufSize int

	// LocalClientID/LocalClientCh, when LocalClientCh is non-nil, are an
	// already-registered local hub subscription — e.g. from
	// EventHub.RegisterWithCap — that Run uses instead of calling
	// hub.Register itself. This is what lets a caller (main.go's stream
	// handler) apply the concurrent-subscriber cap and decide whether to
	// respond at all (503 on refusal) BEFORE writing any SSE bytes, while
	// still letting Run own the client's lifecycle exactly as it would for
	// a self-registered one: LocalClientID must be the id that came back
	// alongside LocalClientCh, since Run's own deferred hub.Unregister
	// uses it symmetrically, and a closed LocalClientCh (whether from
	// backpressure or a RegisterWithCap eviction) is handled identically
	// to a self-registered client hitting OnLocalDropped. When
	// LocalClientCh is nil, Run registers its own client exactly as
	// before.
	LocalClientID uint64
	LocalClientCh <-chan []ChangeEvent

	// SpokeOutBufSize sizes the channel every spoke subscription's parsed
	// batches land on before Run forwards them. Defaults to 64. A full
	// buffer drops the batch (same backpressure rule EventHub itself
	// applies to a slow client) rather than blocking a spoke's reader.
	SpokeOutBufSize int
}

MultiStreamOptions configures one hub-side aggregated change-event stream — the engine behind GET /api/events/stream. See RunMultiStream's doc comment for the merge model.

type OwnerDeploymentRef added in v0.9.0 

type OwnerDeploymentRef struct {
	Namespace string `json:"namespace"`
	Name      string `json:"name"`
}

OwnerDeploymentRef identifies the Deployment that owns a ReplicaSet event.

type PodInfo added in v0.9.0 

type PodInfo struct {
	Name        string   `json:"name"`
	Namespace   string   `json:"namespace"`
	Phase       string   `json:"phase"`
	Ready       bool     `json:"ready"`
	Terminating bool     `json:"terminating"`
	Restarts    int32    `json:"restarts"`
	Node        string   `json:"node"`
	Age         string   `json:"age"`
	Images      []string `json:"images"`
	Message     string   `json:"message,omitempty"`
}

PodInfo is one Pod's display-ready summary within a ReplicaSet's child list — restart count, readiness, first meaningful waiting/terminated message, image list, and a human-rendered age string, computed once here instead of separately in every place a caller might otherwise recompute them from a raw corev1.Pod.

type RSInfo added in v0.9.0 

type RSInfo struct {
	Name            string    `json:"name"`
	Namespace       string    `json:"namespace"`
	Replicas        int32     `json:"replicas"`
	ReadyReplicas   int32     `json:"readyReplicas"`
	DesiredReplicas int32     `json:"desiredReplicas"`
	IsCurrentRS     bool      `json:"isCurrentRS"`
	Pods            []PodInfo `json:"pods"`
}

RSInfo is one ReplicaSet owned by the Deployment, plus the Pods owned by that ReplicaSet.

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.