pkg/k8score/cache.go
Rama main · réplica de lectura del repositorio original.
package k8score
import (
"context"
"errors"
"fmt"
"log"
"maps"
"runtime"
"slices"
"sort"
"strings"
"sync"
"sync/atomic"
"time"
appsv1 "k8s.io/api/apps/v1"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
apiruntime "k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/watch"
"k8s.io/client-go/informers"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/tools/cache"
"k8s.io/client-go/tools/pager"
)
// ResourceCache provides fast, eventually-consistent access to K8s resources
// using SharedInformers. It is the shared core used by both Radar and
// skyhook-connector.
type ResourceCache struct {
factory informers.SharedInformerFactory // cluster-wide factory (always present)
nsFactories map[string]informers.SharedInformerFactory // per-namespace factories (for mixed-scope mode)
factoryByKind map[string]informers.SharedInformerFactory // resolved factory per enabled kind — listers MUST go through this
indexerByKind map[string]cache.Indexer // union indexer for kinds watched in multiple namespace factories
pagedInformers map[string]cache.SharedIndexInformer // per-kind paginating informers (ListPageSize>0); listers read these instead of the factory's
// kindStores holds each enabled kind's informer stores keyed by the
// namespace each one watches ("" for one informer over the kind's whole
// scope), so a read can trust a namespace whose own informer has synced
// while another namespace of the same kind is still loading.
kindStores map[string]map[string]kindStore
changes chan ResourceChange
stopCh chan struct{}
stopOnce sync.Once
enabledResources map[string]bool
deferredSynced map[string]bool
deferredMu sync.RWMutex
deferredDone chan struct{}
deferredFailed atomic.Bool // true if WaitForCacheSync failed for deferred informers
syncComplete atomic.Bool
config CacheConfig
stdlog *log.Logger
// Per-informer sync tracking for diagnostics
informerStatuses []InformerSyncStatus
// informerHasSynced is the informer's own HasSynced, parallel to
// informerStatuses. The Synced flag beside it is written by a tracking
// goroutine and lags it, so a caller deciding whether it may trust an
// empty list must ask this rather than read the flag. Also backs
// KindReadiness and GetSyncSnapshot for the same reason.
informerHasSynced []func() bool
informerMu sync.RWMutex
promotedKinds []string // set when SyncTimeout fires; empty on normal sync
syncStartTime time.Time
// backgroundKeys marks kinds that sync independently with no deadline
// (Events). They share deferredSynced tracking but must never classify
// as failed off the deferred-timeout flag — a late Events LIST on a big
// cluster is normal, not terminal.
backgroundKeys map[string]bool
}
// InformerSyncStatus tracks the sync state of a single informer.
type InformerSyncStatus struct {
Kind string `json:"kind"`
Key string `json:"key"` // e.g. "pods", "deployments"
Deferred bool `json:"deferred"` // true if deferred (non-critical)
Synced bool `json:"synced"`
SyncedAt string `json:"syncedAt,omitempty"` // RFC3339 timestamp
Items int `json:"items"` // current item count in lister
LastError string `json:"lastError,omitempty"` // most recent watch error (truncated to keep diagnostics small)
LastErrorAt string `json:"lastErrorAt,omitempty"` // RFC3339 timestamp of last watch error
ForbiddenSeen bool `json:"forbiddenSeen,omitempty"` // reflector hit 403/401 — probe and reality disagree
}
// SyncPhase describes the current phase of cache initialization.
type SyncPhase string
const (
SyncPhaseNotStarted SyncPhase = "not_started"
SyncPhaseCritical SyncPhase = "syncing_critical"
SyncPhaseDeferred SyncPhase = "syncing_deferred"
SyncPhaseComplete SyncPhase = "complete"
)
// CacheSyncStatus is the overall sync status exposed for diagnostics.
type CacheSyncStatus struct {
Phase SyncPhase `json:"phase"`
SyncStarted string `json:"syncStarted,omitempty"` // RFC3339
ElapsedSec float64 `json:"elapsedSec"`
CriticalTotal int `json:"criticalTotal"`
CriticalSynced int `json:"criticalSynced"`
DeferredTotal int `json:"deferredTotal"`
DeferredSynced int `json:"deferredSynced"`
Informers []InformerSyncStatus `json:"informers"`
PendingCritical []string `json:"pendingCritical,omitempty"` // kinds not yet synced
PendingDeferred []string `json:"pendingDeferred,omitempty"`
PromotedKinds []string `json:"promotedKinds,omitempty"` // critical informers that timed out
}
type informerSetup struct {
key string
kind string
setup func(factory informers.SharedInformerFactory) cache.SharedIndexInformer
isEvent bool
isClusterScoped bool // true for nodes, namespaces, PV, storageclasses, ingressclasses
// pagedSetup, when non-nil and CacheConfig.ListPageSize > 0, builds a
// paginating informer for this kind instead of the factory one. Set only
// for high-cardinality kinds whose single-shot initial LIST can fail on
// very large clusters.
pagedSetup func(client kubernetes.Interface, namespace string, pageSize int64) cache.SharedIndexInformer
}
// newPagedInformer builds a SharedIndexInformer whose initial LIST is fetched in
// pages via a consistent (resourceVersion="") read rather than one unpaginated
// response. The watch is unchanged. On clusters without WatchList streaming, a
// single giant LIST of a high-cardinality kind can exceed the response-read
// deadline or spike memory; paging bounds each request. When WatchList IS
// available, client-go streams the initial state and never calls this ListFunc.
func newPagedInformer(
example apiruntime.Object,
pageSize int64,
listFn pager.ListPageFunc,
watchFn func(ctx context.Context, opts metav1.ListOptions) (watch.Interface, error),
) cache.SharedIndexInformer {
lw := &cache.ListWatch{
ListWithContextFunc: func(ctx context.Context, _ metav1.ListOptions) (apiruntime.Object, error) {
p := pager.New(listFn)
p.PageSize = pageSize
// resourceVersion="" forces a consistent read the apiserver will
// paginate — RV=0 (watch-cache) reads ignore limit. We deliberately
// override the reflector's default RV here.
obj, _, err := p.List(ctx, metav1.ListOptions{ResourceVersion: ""})
return obj, err
},
WatchFuncWithContext: watchFn,
}
inf := cache.NewSharedIndexInformer(lw, example, 0, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc})
return inf
}
func cloneScopeNamespaces(in map[string][]string) map[string][]string {
if in == nil {
return nil
}
out := make(map[string][]string, len(in))
for k, v := range in {
if len(v) == 0 {
continue
}
out[k] = append([]string(nil), v...)
}
return out
}
func resourceScopeNamespaces(key string, scope ResourceScope, byKind map[string][]string) []string {
if !scope.Enabled || scope.Namespace == "" {
return nil
}
namespaces := byKind[key]
if len(namespaces) == 0 {
return []string{scope.Namespace}
}
seen := make(map[string]struct{}, len(namespaces)+1)
out := make([]string, 0, len(namespaces)+1)
add := func(ns string) {
if ns == "" {
return
}
if _, ok := seen[ns]; ok {
return
}
seen[ns] = struct{}{}
out = append(out, ns)
}
add(scope.Namespace)
for _, ns := range namespaces {
add(ns)
}
return out
}
// unionIndexer is a read-only cache.Indexer that fans reads out across the
// per-namespace informer stores of one kind. Each underlying informer watches
// a distinct namespace, so the union cannot duplicate an object. Reading the
// informers' own stores — rather than mirroring events into a separate
// indexer — keeps HasSynced meaningful (an informer's store is fully
// populated before HasSynced flips, while a mirror fills in asynchronously
// after) and avoids holding every object twice.
type unionIndexer struct {
indexers []cache.Indexer
}
var errUnionIndexerReadOnly = errors.New("union indexer is read-only")
func (u *unionIndexer) Add(any) error { return errUnionIndexerReadOnly }
func (u *unionIndexer) Bookmark(string) {}
// LastStoreSyncResourceVersion is only meaningful on stores a reflector
// writes into; nothing writes into the union, so report "unknown".
func (u *unionIndexer) LastStoreSyncResourceVersion() string { return "" }
func (u *unionIndexer) Update(any) error { return errUnionIndexerReadOnly }
func (u *unionIndexer) Delete(any) error { return errUnionIndexerReadOnly }
func (u *unionIndexer) Replace([]any, string) error { return errUnionIndexerReadOnly }
func (u *unionIndexer) Resync() error { return errUnionIndexerReadOnly }
func (u *unionIndexer) AddIndexers(cache.Indexers) error { return errUnionIndexerReadOnly }
func (u *unionIndexer) List() []any {
var out []any
for _, idx := range u.indexers {
out = append(out, idx.List()...)
}
return out
}
func (u *unionIndexer) ListKeys() []string {
var out []string
for _, idx := range u.indexers {
out = append(out, idx.ListKeys()...)
}
return out
}
func (u *unionIndexer) Get(obj any) (any, bool, error) {
for _, idx := range u.indexers {
if item, exists, err := idx.Get(obj); err != nil || exists {
return item, exists, err
}
}
return nil, false, nil
}
func (u *unionIndexer) GetByKey(key string) (any, bool, error) {
for _, idx := range u.indexers {
if item, exists, err := idx.GetByKey(key); err != nil || exists {
return item, exists, err
}
}
return nil, false, nil
}
func (u *unionIndexer) Index(indexName string, obj any) ([]any, error) {
var out []any
for _, idx := range u.indexers {
items, err := idx.Index(indexName, obj)
if err != nil {
return nil, err
}
out = append(out, items...)
}
return out, nil
}
func (u *unionIndexer) IndexKeys(indexName, indexedValue string) ([]string, error) {
var out []string
for _, idx := range u.indexers {
keys, err := idx.IndexKeys(indexName, indexedValue)
if err != nil {
return nil, err
}
out = append(out, keys...)
}
return out, nil
}
func (u *unionIndexer) ListIndexFuncValues(indexName string) []string {
seen := map[string]struct{}{}
var out []string
for _, idx := range u.indexers {
for _, v := range idx.ListIndexFuncValues(indexName) {
if _, ok := seen[v]; ok {
continue
}
seen[v] = struct{}{}
out = append(out, v)
}
}
return out
}
func (u *unionIndexer) ByIndex(indexName, indexedValue string) ([]any, error) {
var out []any
for _, idx := range u.indexers {
items, err := idx.ByIndex(indexName, indexedValue)
if err != nil {
return nil, err
}
out = append(out, items...)
}
return out, nil
}
func (u *unionIndexer) GetIndexers() cache.Indexers {
if len(u.indexers) == 0 {
return cache.Indexers{}
}
return u.indexers[0].GetIndexers()
}
// NewResourceCache creates and starts a ResourceCache from the given config.
// Startup has three tiers:
// - Critical informers block until synced. With PatienceWindow + MinimalSet,
// the cache returns as soon as the minimal set is ready *after* the
// patience window elapses (rest of critical promoted to deferred). With
// SyncTimeout, the cache returns at most after the timeout.
// - Deferred informers sync in the background after critical completes.
// - Background informers (e.g. Events) sync independently on their own goroutine.
func NewResourceCache(cfg CacheConfig) (*ResourceCache, error) {
if cfg.Client == nil {
return nil, fmt.Errorf("CacheConfig.Client must not be nil")
}
if cfg.NamespaceScoped && cfg.Namespace == "" {
return nil, fmt.Errorf("CacheConfig.Namespace must be set when NamespaceScoped is true")
}
for k, scope := range cfg.ResourceScopes {
if scope.Enabled && scope.Namespace == "" {
continue
}
if scope.Enabled && scope.Namespace != "" && k == "" {
return nil, fmt.Errorf("CacheConfig.ResourceScopes contains entry with empty key")
}
}
for k, namespaces := range cfg.ResourceScopeNamespaces {
if k == "" {
return nil, fmt.Errorf("CacheConfig.ResourceScopeNamespaces contains entry with empty key")
}
if len(namespaces) == 0 {
continue
}
scope, ok := cfg.ResourceScopes[k]
if !ok || !scope.Enabled {
return nil, fmt.Errorf("CacheConfig.ResourceScopeNamespaces contains namespaces for disabled key %q", k)
}
if scope.Namespace == "" {
return nil, fmt.Errorf("CacheConfig.ResourceScopeNamespaces contains namespaces for cluster-wide key %q", k)
}
}
channelSize := cfg.ChannelSize
if channelSize <= 0 {
channelSize = 10000
}
logf := cfg.TimingLogger
if logf == nil {
logf = func(string, ...any) {} // no-op
}
stdlog := cfg.Logger
if stdlog == nil {
stdlog = log.Default()
}
// Clone caller-owned maps to prevent mutation after construction.
cfg.ResourceTypes = maps.Clone(cfg.ResourceTypes)
cfg.ResourceScopes = maps.Clone(cfg.ResourceScopes)
cfg.ResourceScopeNamespaces = cloneScopeNamespaces(cfg.ResourceScopeNamespaces)
cfg.DeferredTypes = maps.Clone(cfg.DeferredTypes)
cfg.MinimalSet = maps.Clone(cfg.MinimalSet)
// Derive a per-kind scope view that captures both modes:
// - ResourceScopes (when set) is authoritative.
// - Otherwise fall back to ResourceTypes + NamespaceScoped + Namespace.
// This single map drives both the enabled set and which factory each
// informer is wired into.
scopes := make(map[string]ResourceScope)
if cfg.ResourceScopes != nil {
for k, s := range cfg.ResourceScopes {
if s.Enabled {
scopes[k] = s
}
}
} else {
ns := ""
if cfg.NamespaceScoped {
ns = cfg.Namespace
}
for k, ok := range cfg.ResourceTypes {
if ok {
scopes[k] = ResourceScope{Enabled: true, Namespace: ns}
}
}
}
setups := buildInformerSetups()
for _, setup := range setups {
if !setup.isClusterScoped {
continue
}
scope, ok := scopes[setup.key]
if !ok {
continue
}
scope.Namespace = ""
scopes[setup.key] = scope
delete(cfg.ResourceScopeNamespaces, setup.key)
}
cfg.ResourceScopes = scopes
stopCh := make(chan struct{})
changes := make(chan ResourceChange, channelSize)
// Build one factory per unique scope. The cluster-wide factory is
// always created so cluster-scoped kinds (nodes, namespaces, PV…)
// have a home; namespace-scoped factories are created on demand.
transform := func(obj any) (any, error) {
if cfg.OnTransform != nil {
cfg.OnTransform(obj)
}
return DropManagedFields(obj)
}
clusterFactory := informers.NewSharedInformerFactoryWithOptions(
cfg.Client,
0, // no resync — updates come via watch
informers.WithTransform(transform),
)
nsFactories := make(map[string]informers.SharedInformerFactory)
for key, s := range scopes {
namespaces := resourceScopeNamespaces(key, s, cfg.ResourceScopeNamespaces)
if len(namespaces) == 0 {
continue
}
for _, namespace := range namespaces {
if _, ok := nsFactories[namespace]; ok {
continue
}
nsFactories[namespace] = informers.NewSharedInformerFactoryWithOptions(
cfg.Client,
0,
informers.WithTransform(transform),
informers.WithNamespace(namespace),
)
}
}
if len(nsFactories) > 0 {
var nsList []string
for ns := range nsFactories {
nsList = append(nsList, ns)
}
sort.Strings(nsList)
// %q on the slice quotes each element, neutering newlines / control
// characters in caller-supplied namespace strings before they hit logs.
stdlog.Printf("Using namespace-scoped informers for namespace(s): %q", nsList)
}
// pickFactory routes each informer to the factory matching its scope.
// Cluster-scoped kinds always use the cluster-wide factory regardless
// of caller intent — namespace-scoping them would 404.
pickFactory := func(s informerSetup) informers.SharedInformerFactory {
scope, ok := scopes[s.key]
if !ok || scope.Namespace == "" || s.isClusterScoped {
return clusterFactory
}
return nsFactories[scope.Namespace]
}
// enabledMap is the boolean projection over `scopes` exposed via
// GetEnabledResources for callers that only need "is this kind on?"
// without the per-kind scope detail.
enabled := make(map[string]bool, len(scopes))
for k := range scopes {
enabled[k] = true
}
deferredTypes := cfg.DeferredTypes
if deferredTypes == nil {
deferredTypes = map[string]bool{}
}
var criticalSyncFuncs []cache.InformerSynced
var deferredSyncFuncs []cache.InformerSynced
var deferredKeys []string
var backgroundSyncFuncs []cache.InformerSynced // Events — sync independently, don't block deferredDone
var backgroundKeys []string
enabledCount := 0
rc := &ResourceCache{
factory: clusterFactory,
nsFactories: nsFactories,
factoryByKind: map[string]informers.SharedInformerFactory{},
indexerByKind: map[string]cache.Indexer{},
pagedInformers: map[string]cache.SharedIndexInformer{},
kindStores: map[string]map[string]kindStore{},
changes: changes,
stopCh: stopCh,
enabledResources: enabled,
config: cfg,
stdlog: stdlog,
}
// Track per-informer sync status, HasSynced funcs, and the informer
// handle (needed for staggered start).
type informerEntry struct {
kind string
key string
deferred bool
synced cache.InformerSynced
run func(stopCh <-chan struct{})
}
var allEntries []informerEntry
buildInformer := func(s informerSetup, factory informers.SharedInformerFactory, namespace string) cache.SharedIndexInformer {
if cfg.ListPageSize > 0 && s.pagedSetup != nil {
inf := s.pagedSetup(cfg.Client, namespace, cfg.ListPageSize)
// Paged informers are built outside the factories, so they need the
// same caller hook and managed-fields stripping transform here.
_ = inf.SetTransform(transform)
return inf
}
return s.setup(factory)
}
wireInformer := func(inf cache.SharedIndexInformer, s informerSetup, idx int) error {
var err error
if s.isEvent {
err = rc.addEventHandlers(inf, changes)
} else {
err = rc.addChangeHandlers(inf, s.kind, changes)
}
if err != nil {
return fmt.Errorf("failed to register %s event handler: %w", s.kind, err)
}
if !s.isEvent {
if err := inf.AddIndexers(ownerUIDIndexers); err != nil {
return fmt.Errorf("failed to index %s by owner: %w", s.kind, err)
}
}
// Wire a WatchErrorHandler so reflector-level failures (most
// notably 403/Unauthorized that the probe didn't predict — e.g.
// the token expired between probe and watch, or RBAC changed
// mid-session) are visible in diagnostics. The reflector keeps
// retrying with exponential backoff regardless; this just makes
// the failure observable instead of silent.
key := s.key
kind := s.kind
if hErr := inf.SetWatchErrorHandler(func(_ *cache.Reflector, err error) {
rc.recordWatchError(idx, key, kind, err)
}); hErr != nil {
// SetWatchErrorHandler only errors after the informer has
// started, which can't happen here — log and continue.
stdlog.Printf("Warning: failed to set watch error handler for %s: %v", s.kind, hErr)
}
return nil
}
for _, s := range setups {
if !enabled[s.key] {
continue
}
enabledCount++
idx := len(allEntries) // status index for this logical kind
scopeNamespaces := resourceScopeNamespaces(s.key, scopes[s.key], cfg.ResourceScopeNamespaces)
var synced cache.InformerSynced
var run func(stopCh <-chan struct{})
if !s.isClusterScoped && len(scopeNamespaces) > 1 {
informersForKind := make([]cache.SharedIndexInformer, 0, len(scopeNamespaces))
stores := make([]cache.Indexer, 0, len(scopeNamespaces))
for _, namespace := range scopeNamespaces {
factory := nsFactories[namespace]
inf := buildInformer(s, factory, namespace)
if err := wireInformer(inf, s, idx); err != nil {
close(stopCh)
return nil, err
}
informersForKind = append(informersForKind, inf)
stores = append(stores, inf.GetIndexer())
rc.addKindStore(s.key, namespace, inf)
}
rc.indexerByKind[s.key] = &unionIndexer{indexers: stores}
synced = func() bool {
for _, inf := range informersForKind {
if !inf.HasSynced() {
return false
}
}
return true
}
run = func(stopCh <-chan struct{}) {
for _, inf := range informersForKind {
go inf.Run(stopCh)
}
}
} else {
factory := pickFactory(s)
rc.factoryByKind[s.key] = factory
namespace := scopes[s.key].Namespace
if s.isClusterScoped {
namespace = ""
}
inf := buildInformer(s, factory, namespace)
if cfg.ListPageSize > 0 && s.pagedSetup != nil {
// Listers must read this informer's store, not the factory's
// (the factory's informer for this kind is never started).
rc.pagedInformers[s.key] = inf
}
if err := wireInformer(inf, s, idx); err != nil {
close(stopCh)
return nil, err
}
rc.addKindStore(s.key, "", inf)
synced = inf.HasSynced
run = inf.Run
}
if delay := cfg.DebugSyncDelays[s.key]; delay > 0 {
inner := run
run = func(stopCh <-chan struct{}) {
stdlog.Printf("DEBUG: delaying %s informer start by %v (DebugSyncDelays)", s.key, delay)
select {
case <-time.After(delay):
case <-stopCh:
return
}
inner(stopCh)
}
}
isDeferred := deferredTypes[s.key]
entry := informerEntry{kind: s.kind, key: s.key, deferred: isDeferred, synced: synced, run: run}
allEntries = append(allEntries, entry)
if isDeferred && s.isEvent {
// Events sync independently — they can take 60s+ on large clusters
// and shouldn't block topology completion or warmup transition.
backgroundSyncFuncs = append(backgroundSyncFuncs, synced)
backgroundKeys = append(backgroundKeys, s.key)
if rc.backgroundKeys == nil {
rc.backgroundKeys = map[string]bool{}
}
rc.backgroundKeys[s.key] = true
} else if isDeferred {
deferredSyncFuncs = append(deferredSyncFuncs, synced)
deferredKeys = append(deferredKeys, s.key)
} else {
criticalSyncFuncs = append(criticalSyncFuncs, synced)
}
}
// Initialize per-informer tracking
statuses := make([]InformerSyncStatus, len(allEntries))
hasSynced := make([]func() bool, len(allEntries))
for i, e := range allEntries {
statuses[i] = InformerSyncStatus{Kind: e.kind, Key: e.key, Deferred: e.deferred}
hasSynced[i] = e.synced
}
rc.informerStatuses = statuses
rc.informerHasSynced = hasSynced
if enabledCount == 0 {
stdlog.Printf("Warning: No resource types are accessible (all RBAC checks failed)")
rc.deferredSynced = make(map[string]bool)
rc.deferredDone = make(chan struct{})
close(rc.deferredDone)
rc.syncComplete.Store(true)
return rc, nil
}
// Start critical informers first. Deferred informers are started after
// Phase 1 completes to reduce concurrent LIST pressure on the API server.
// On large clusters (300+ nodes), ~10 concurrent LISTs is significantly
// lighter than ~19, giving the heaviest resource (Pods) more API server
// bandwidth during the critical path.
for _, e := range allEntries {
if !e.deferred {
go e.run(stopCh)
}
}
if len(backgroundKeys) > 0 {
stdlog.Printf("Starting resource cache: %d critical + %d deferred + %d background informers (%d total)",
len(criticalSyncFuncs), len(deferredSyncFuncs), len(backgroundSyncFuncs), enabledCount)
} else {
stdlog.Printf("Starting resource cache: %d critical + %d deferred informers (%d total, deferred start after critical sync)",
len(criticalSyncFuncs), len(deferredSyncFuncs), enabledCount)
}
syncStart := time.Now()
rc.syncStartTime = syncStart
// Hand out the still-syncing handle for progressive per-kind reads.
// Deliberately after informer start + status registration + sync-start
// stamping and before the Phase-1 wait: the handle is structurally
// complete, only sync state is in flux, and KindReadinessFor answers
// that per kind.
if cfg.OnInformersStarted != nil {
cfg.OnInformersStarted(rc)
}
// Track per-informer sync completion in background goroutines.
// Each goroutine updates the InformerSyncStatus when its informer syncs.
for i, e := range allEntries {
idx := i
entry := e
go func() {
t := time.Now()
for !entry.synced() {
select {
case <-stopCh:
return
default:
}
time.Sleep(10 * time.Millisecond)
}
tag := "critical"
if entry.deferred {
tag = "deferred"
}
logf(" Informer synced: %-28s %v (%s)", entry.kind, time.Since(t), tag)
rc.informerMu.Lock()
rc.informerStatuses[idx].Synced = true
rc.informerStatuses[idx].SyncedAt = time.Now().Format(time.RFC3339)
rc.informerMu.Unlock()
}()
}
// Phase 1: Wait for critical informers.
//
// Two modes (combinable):
// - Patience+MinimalSet: wait for ALL critical until the patience
// window elapses. Once elapsed, return as soon as the minimal set
// is ready; promote the rest of critical to deferred.
// - SyncTimeout: hard upper bound. When it fires, promote everything
// still pending and return. Acts as a backstop on the minimal-set
// path so a permanently-stuck informer doesn't trap the caller
// forever (caller still gets to render with whatever synced).
useMinimalSet := cfg.PatienceWindow > 0 && len(cfg.MinimalSet) > 0
timedOut := false
patienceElapsed := false
if len(criticalSyncFuncs) > 0 {
// Build the index of "must-sync" entries for the minimal-set check.
// Filter to entries that are critical AND in MinimalSet (deferred
// types or unknown keys don't gate first paint).
var minimalEntries []informerEntry
if useMinimalSet {
knownCritical := make(map[string]bool, len(allEntries))
for _, e := range allEntries {
if !e.deferred {
knownCritical[e.key] = true
}
if e.deferred {
continue
}
if cfg.MinimalSet[e.key] {
minimalEntries = append(minimalEntries, e)
}
}
// Validate MinimalSet keys: typos or kinds not enabled produce a
// silently-empty minimalEntries → cache returns ~PatienceWindow
// later with nothing meaningful synced. Surface that loud.
var unknown []string
for k := range cfg.MinimalSet {
if !knownCritical[k] {
unknown = append(unknown, k)
}
}
if len(unknown) > 0 {
sort.Strings(unknown)
stdlog.Printf("WARNING: MinimalSet keys not registered as critical informers (typo or RBAC-denied?): %s",
strings.Join(unknown, ", "))
}
if len(minimalEntries) == 0 {
stdlog.Printf("WARNING: MinimalSet matched no enabled critical informers; first paint will fire as soon as PatienceWindow elapses regardless of sync state")
}
}
progressTicker := time.NewTicker(1 * time.Second)
defer progressTicker.Stop()
var patienceCh <-chan time.Time
if useMinimalSet {
patienceTimer := time.NewTimer(cfg.PatienceWindow)
defer patienceTimer.Stop()
patienceCh = patienceTimer.C
}
var deadlineCh <-chan time.Time
if cfg.SyncTimeout > 0 {
deadline := time.NewTimer(cfg.SyncTimeout)
defer deadline.Stop()
deadlineCh = deadline.C
}
// emitProgress reports current sync progress to the application
// callback (typically wired to the connection's progressMessage).
emitProgress := func(minimalReady bool) {
if cfg.SyncProgress == nil {
return
}
rc.informerMu.RLock()
var synced, total int
for _, s := range rc.informerStatuses {
if s.Deferred {
continue
}
total++
if s.Synced {
synced++
}
}
rc.informerMu.RUnlock()
cfg.SyncProgress(synced, total, minimalReady)
}
minimalReady := func() bool {
for _, e := range minimalEntries {
if !e.synced() {
return false
}
}
return true
}
emitProgress(false)
// Set of minimal-set keys for the post-patience log discrimination.
minimalKindSet := make(map[string]bool, len(minimalEntries))
for _, e := range minimalEntries {
minimalKindSet[e.kind] = true
}
for {
allSynced := true
for _, fn := range criticalSyncFuncs {
if !fn() {
allSynced = false
break
}
}
if allSynced {
break
}
if patienceElapsed && useMinimalSet && minimalReady() {
break
}
select {
case <-stopCh:
return nil, fmt.Errorf("failed to sync critical resource caches")
case <-patienceCh:
patienceElapsed = true
patienceCh = nil // disable further receives on the drained timer channel
case <-deadlineCh:
timedOut = true
case <-progressTicker.C:
counts := rc.GetKindObjectCounts()
rc.informerMu.RLock()
var synced, pendingParts, minimalPending []string
for _, s := range rc.informerStatuses {
if s.Deferred {
continue
}
if s.Synced {
synced = append(synced, s.Kind)
} else {
n := counts[s.Kind]
pendingParts = append(pendingParts, fmt.Sprintf("%s(%d)", s.Kind, n))
if minimalKindSet[s.Kind] {
minimalPending = append(minimalPending, s.Kind)
}
}
}
rc.informerMu.RUnlock()
if len(pendingParts) > 0 {
// After patience elapses, the actual blocker for first paint
// is the minimal-set subset, not all of critical — surface it
// distinctly so operators know which kind is holding render.
if patienceElapsed && len(minimalPending) > 0 {
stdlog.Printf("First-paint blocked: %d/%d minimal-set kinds still syncing (%.0fs elapsed) — pending: %s",
len(minimalPending), len(minimalEntries), time.Since(syncStart).Seconds(), strings.Join(minimalPending, ", "))
} else {
stdlog.Printf("Critical sync progress: %d/%d synced (%.0fs elapsed) — pending: %s",
len(synced), len(synced)+len(pendingParts), time.Since(syncStart).Seconds(), strings.Join(pendingParts, ", "))
}
}
emitProgress(useMinimalSet && patienceElapsed && minimalReady())
default:
time.Sleep(100 * time.Millisecond)
}
if timedOut {
break
}
}
}
// Reclassify any critical informer still pending as deferred so the
// cache can return. No-op on the all-synced path.
var promoted []string
rc.informerMu.Lock()
for i, e := range allEntries {
if e.deferred || e.synced() {
continue
}
promoted = append(promoted, e.kind)
deferredSyncFuncs = append(deferredSyncFuncs, e.synced)
deferredKeys = append(deferredKeys, e.key)
rc.informerStatuses[i].Deferred = true
}
rc.informerMu.Unlock()
switch {
case timedOut:
stdlog.Printf("WARNING: Critical sync timed out after %v — promoting %d informers to deferred: %s",
cfg.SyncTimeout, len(promoted), strings.Join(promoted, ", "))
stdlog.Printf("UI renders now; promoted kinds keep syncing in background and their views stay in a loading state until complete")
logf(" Phase 1 sync TIMED OUT (%d critical, %d promoted to deferred): %v",
len(criticalSyncFuncs), len(promoted), time.Since(syncStart))
rc.informerMu.Lock()
rc.promotedKinds = promoted
rc.informerMu.Unlock()
case patienceElapsed && len(promoted) > 0:
stdlog.Printf("First-paint ready after %v: minimal set synced; %d slower informers continue in background: %s",
time.Since(syncStart), len(promoted), strings.Join(promoted, ", "))
logf(" Phase 1 minimal-set sync (%d/%d critical, %d still loading): %v",
len(criticalSyncFuncs)-len(promoted), len(criticalSyncFuncs), len(promoted), time.Since(syncStart))
rc.informerMu.Lock()
rc.promotedKinds = promoted
rc.informerMu.Unlock()
default:
logf(" Phase 1 sync (%d critical informers): %v", len(criticalSyncFuncs), time.Since(syncStart))
stdlog.Printf("Critical resource caches synced in %v — UI can render", time.Since(syncStart))
}
if cfg.SyncProgress != nil {
// Count via e.synced() (same source as the Phase 1 loop) rather than
// informerStatuses[].Synced, which is set asynchronously by per-informer
// tracking goroutines and can lag by ~10ms after HasSynced() flips. On
// the all-synced happy path the lag would otherwise produce synced<total
// here even though we just exited the loop on allSynced — surfaced to
// callers as a misleading "showing partial" final message.
rc.informerMu.RLock()
var synced, total int
for i, e := range allEntries {
if rc.informerStatuses[i].Deferred {
continue
}
total++
if e.synced() {
synced++
}
}
rc.informerMu.RUnlock()
cfg.SyncProgress(synced, total, true)
}
// Log per-type resource counts for startup diagnostics
if counts := rc.GetKindObjectCounts(); len(counts) > 0 {
total := 0
var parts []string
for kind, count := range counts {
if count > 0 {
parts = append(parts, fmt.Sprintf("%s:%d", kind, count))
total += count
}
}
sort.Strings(parts)
stdlog.Printf("Resource breakdown (%d total): %s", total, strings.Join(parts, ", "))
}
rc.syncComplete.Store(true)
// Build deferred tracking state (includes both deferred and background keys).
// The handle is already published to HTTP handlers (OnInformersStarted), so
// these fields must be installed under deferredMu — /api/connection polls
// GetSyncSnapshot sub-second during the connecting phase and reads them.
allDeferredKeys := append(append([]string{}, deferredKeys...), backgroundKeys...)
deferredSynced := make(map[string]bool, len(allDeferredKeys))
for _, k := range allDeferredKeys {
deferredSynced[k] = false
}
deferredDone := make(chan struct{})
rc.deferredMu.Lock()
rc.deferredSynced = deferredSynced
rc.deferredDone = deferredDone
rc.deferredMu.Unlock()
// Phase 2: Start deferred informers now that critical sync is done,
// then wait for them in background. This staggers the API server load.
for _, e := range allEntries {
if e.deferred {
go e.run(stopCh)
}
}
if len(deferredSyncFuncs) > 0 {
go func() {
deferredStart := time.Now()
progressTicker := time.NewTicker(5 * time.Second)
defer progressTicker.Stop()
var deadlineCh <-chan time.Time
if cfg.DeferredSyncTimeout > 0 {
t := time.NewTimer(cfg.DeferredSyncTimeout)
defer t.Stop()
deadlineCh = t.C
}
timedOut := false
for {
// Mark each informer synced the moment its own HasSynced() is
// true. A permanently-failing informer (e.g. HPA autoscaling/v2
// on K8s <1.23) must not block siblings from becoming ready.
rc.deferredMu.Lock()
allSynced := true
for i, fn := range deferredSyncFuncs {
k := deferredKeys[i]
if rc.deferredSynced[k] {
continue
}
if fn() {
rc.deferredSynced[k] = true
} else {
allSynced = false
}
}
rc.deferredMu.Unlock()
if allSynced {
break
}
select {
case <-stopCh:
rc.deferredFailed.Store(true)
stdlog.Printf("ERROR: Deferred resource cache sync aborted after %v", time.Since(deferredStart))
close(deferredDone)
return
case <-deadlineCh:
timedOut = true
case <-progressTicker.C:
counts := rc.GetKindObjectCounts()
rc.informerMu.RLock()
var synced, pendingParts []string
for _, s := range rc.informerStatuses {
if !s.Deferred {
continue
}
// Skip background informers (Events) in deferred progress
if slices.Contains(backgroundKeys, s.Key) {
continue
}
if s.Synced {
synced = append(synced, s.Kind)
} else {
n := counts[s.Kind]
pendingParts = append(pendingParts, fmt.Sprintf("%s(%d)", s.Kind, n))
}
}
rc.informerMu.RUnlock()
stdlog.Printf("Deferred sync progress: %d/%d synced (%.0fs elapsed) — pending: %s",
len(synced), len(synced)+len(pendingParts), time.Since(deferredStart).Seconds(), strings.Join(pendingParts, ", "))
default:
time.Sleep(100 * time.Millisecond)
}
if timedOut {
break
}
}
if timedOut {
// Stop waiting on stragglers. deferredFailed is the "give up"
// signal read by IsDeferredPending — stragglers start reporting
// not-pending, so HTTP handlers return 403 instead of perpetual
// 503. Informers that already synced keep their own
// deferredSynced[k]=true and continue serving normally.
rc.deferredMu.RLock()
var pending []string
for _, k := range deferredKeys {
if !rc.deferredSynced[k] {
pending = append(pending, k)
}
}
rc.deferredMu.RUnlock()
rc.deferredFailed.Store(true)
stdlog.Printf("WARNING: Deferred sync timed out after %v; %d informer(s) never synced: %s. "+
"These resources will return 403 to API consumers. Common cause: the corresponding API "+
"version isn't served on this cluster (check `kubectl api-resources | grep <resource>`).",
cfg.DeferredSyncTimeout, len(pending), strings.Join(pending, ", "))
} else {
logf(" Phase 2 sync (%d deferred informers): %v", len(deferredSyncFuncs), time.Since(deferredStart))
stdlog.Printf("Deferred resource caches synced in %v (total: %v)", time.Since(deferredStart), time.Since(syncStart))
}
close(deferredDone)
}()
} else {
close(deferredDone)
}
// Background informers (Events) sync independently — they can take 60s+
// on large clusters and shouldn't block topology/warmup completion.
if len(backgroundSyncFuncs) > 0 {
go func() {
bgStart := time.Now()
if cache.WaitForCacheSync(stopCh, backgroundSyncFuncs...) {
rc.deferredMu.Lock()
for _, k := range backgroundKeys {
rc.deferredSynced[k] = true
}
rc.deferredMu.Unlock()
stdlog.Printf("Background Events sync complete in %v", time.Since(bgStart))
} else {
rc.deferredFailed.Store(true)
stdlog.Printf("WARNING: Background Events sync failed after %v", time.Since(bgStart))
}
}()
}
return rc, nil
}
// InformerResourceKeys returns the resource-type keys that have a typed
// informer setup in this package. Read-only introspection for cross-package
// alignment tests (e.g. internal/k8s asserting that every capability field
// has either a typed informer here or a documented dynamic-cache exception).
// Not intended as a stable API for application use.
func InformerResourceKeys() []string {
setups := buildInformerSetups()
keys := make([]string, 0, len(setups))
for _, s := range setups {
keys = append(keys, s.key)
}
return keys
}
// buildInformerSetups returns the table-driven informer setup list. Each
// entry's setup func takes the factory it should be created from — caller
// picks the cluster-wide or a namespace-scoped factory based on the kind's
// resolved access scope.
func buildInformerSetups() []informerSetup {
type entry = informerSetup
mk := func(key, kind string, isEvent, clusterScoped bool, fn func(f informers.SharedInformerFactory) cache.SharedIndexInformer) entry {
return entry{key: key, kind: kind, setup: fn, isEvent: isEvent, isClusterScoped: clusterScoped}
}
// withPaging marks a high-cardinality kind as paginatable: when
// CacheConfig.ListPageSize > 0, its initial LIST is fetched in pages.
withPaging := func(e entry, paged func(client kubernetes.Interface, namespace string, pageSize int64) cache.SharedIndexInformer) entry {
e.pagedSetup = paged
return e
}
return []informerSetup{
mk(Services, "Service", false, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Core().V1().Services().Informer()
}),
withPaging(mk(Pods, "Pod", false, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Core().V1().Pods().Informer()
}), func(client kubernetes.Interface, ns string, pageSize int64) cache.SharedIndexInformer {
return newPagedInformer(&corev1.Pod{}, pageSize,
func(ctx context.Context, o metav1.ListOptions) (apiruntime.Object, error) {
return client.CoreV1().Pods(ns).List(ctx, o)
},
func(ctx context.Context, o metav1.ListOptions) (watch.Interface, error) {
return client.CoreV1().Pods(ns).Watch(ctx, o)
})
}),
mk(Nodes, "Node", false, true, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Core().V1().Nodes().Informer()
}),
mk(Namespaces, "Namespace", false, true, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Core().V1().Namespaces().Informer()
}),
mk(ConfigMaps, "ConfigMap", false, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Core().V1().ConfigMaps().Informer()
}),
mk(Secrets, "Secret", false, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Core().V1().Secrets().Informer()
}),
mk(Events, "Event", true, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Core().V1().Events().Informer()
}),
mk(PersistentVolumeClaims, "PersistentVolumeClaim", false, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Core().V1().PersistentVolumeClaims().Informer()
}),
mk(PersistentVolumes, "PersistentVolume", false, true, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Core().V1().PersistentVolumes().Informer()
}),
mk(Deployments, "Deployment", false, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Apps().V1().Deployments().Informer()
}),
mk(DaemonSets, "DaemonSet", false, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Apps().V1().DaemonSets().Informer()
}),
mk(StatefulSets, "StatefulSet", false, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Apps().V1().StatefulSets().Informer()
}),
withPaging(mk(ReplicaSets, "ReplicaSet", false, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Apps().V1().ReplicaSets().Informer()
}), func(client kubernetes.Interface, ns string, pageSize int64) cache.SharedIndexInformer {
return newPagedInformer(&appsv1.ReplicaSet{}, pageSize,
func(ctx context.Context, o metav1.ListOptions) (apiruntime.Object, error) {
return client.AppsV1().ReplicaSets(ns).List(ctx, o)
},
func(ctx context.Context, o metav1.ListOptions) (watch.Interface, error) {
return client.AppsV1().ReplicaSets(ns).Watch(ctx, o)
})
}),
mk(Ingresses, "Ingress", false, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Networking().V1().Ingresses().Informer()
}),
mk(IngressClasses, "IngressClass", false, true, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Networking().V1().IngressClasses().Informer()
}),
mk(Jobs, "Job", false, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Batch().V1().Jobs().Informer()
}),
mk(CronJobs, "CronJob", false, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Batch().V1().CronJobs().Informer()
}),
mk(HorizontalPodAutoscalers, "HorizontalPodAutoscaler", false, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Autoscaling().V2().HorizontalPodAutoscalers().Informer()
}),
mk(StorageClasses, "StorageClass", false, true, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Storage().V1().StorageClasses().Informer()
}),
mk(PodDisruptionBudgets, "PodDisruptionBudget", false, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Policy().V1().PodDisruptionBudgets().Informer()
}),
mk(NetworkPolicies, "NetworkPolicy", false, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Networking().V1().NetworkPolicies().Informer()
}),
mk(ServiceAccounts, "ServiceAccount", false, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Core().V1().ServiceAccounts().Informer()
}),
mk(Roles, "Role", false, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Rbac().V1().Roles().Informer()
}),
mk(ClusterRoles, "ClusterRole", false, true, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Rbac().V1().ClusterRoles().Informer()
}),
mk(RoleBindings, "RoleBinding", false, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Rbac().V1().RoleBindings().Informer()
}),
mk(ClusterRoleBindings, "ClusterRoleBinding", false, true, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Rbac().V1().ClusterRoleBindings().Informer()
}),
mk(LimitRanges, "LimitRange", false, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Core().V1().LimitRanges().Informer()
}),
mk(ResourceQuotas, "ResourceQuota", false, false, func(f informers.SharedInformerFactory) cache.SharedIndexInformer {
return f.Core().V1().ResourceQuotas().Informer()
}),
}
}
// addChangeHandlers registers event handlers for non-Event resource changes.
func (rc *ResourceCache) addChangeHandlers(inf cache.SharedIndexInformer, kind string, ch chan<- ResourceChange) error {
_, err := inf.AddEventHandler(cache.ResourceEventHandlerFuncs{
AddFunc: func(obj any) {
rc.enqueueChange(ch, kind, obj, nil, OpAdd)
},
UpdateFunc: func(oldObj, newObj any) {
rc.enqueueChange(ch, kind, newObj, oldObj, OpUpdate)
},
DeleteFunc: func(obj any) {
rc.enqueueChange(ch, kind, obj, nil, OpDelete)
},
})
return err
}
// addEventHandlers registers special handlers for K8s Events.
// Events use a separate path: no noisy filtering, no diff computation.
func (rc *ResourceCache) addEventHandlers(inf cache.SharedIndexInformer, ch chan<- ResourceChange) error {
_, err := inf.AddEventHandler(cache.ResourceEventHandlerFuncs{
AddFunc: func(obj any) {
rc.enqueueEvent(ch, obj, OpAdd)
},
UpdateFunc: func(oldObj, newObj any) {
rc.enqueueEvent(ch, newObj, OpUpdate)
},
DeleteFunc: func(obj any) {
rc.enqueueEvent(ch, obj, OpDelete)
},
})
return err
}
// enqueueChange handles non-Event resource change notifications.
func (rc *ResourceCache) enqueueChange(ch chan<- ResourceChange, kind string, obj any, oldObj any, op string) {
meta, ok := obj.(metav1.Object)
if !ok {
if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok {
meta, ok = tombstone.Obj.(metav1.Object)
if !ok {
rc.stdlog.Printf("Warning: tombstone contained non-metav1.Object for %s %s", kind, op)
return
}
obj = tombstone.Obj
} else {
return
}
}
ns := meta.GetNamespace()
name := meta.GetName()
uid := string(meta.GetUID())
// Track event received (before any filtering)
if rc.config.OnReceived != nil {
rc.safeCallback("OnReceived", func() { rc.config.OnReceived(kind) })
}
// Check if noisy — skip OnChange callback and don't send to changes channel.
// Noisy resources (Lease, Endpoints, EndpointSlice updates, etc.) fire constantly
// and don't affect topology or produce meaningful k8s_event broadcasts.
skipCallback := false
isNoisy := false
if rc.config.IsNoisyResource != nil && rc.config.IsNoisyResource(kind, name, op) {
skipCallback = true
isNoisy = true
if rc.config.OnDrop != nil {
rc.config.OnDrop(kind, ns, name, "noisy_filter", op)
}
}
// SuppressInitialAdds: during initial sync, skip OnChange for adds
if op == "add" && rc.config.SuppressInitialAdds && !rc.syncComplete.Load() {
skipCallback = true
}
// Compute diff for updates
var diff *DiffInfo
if op == "update" && oldObj != nil && obj != nil && rc.config.ComputeDiff != nil {
diff = rc.config.ComputeDiff(kind, oldObj, obj)
}
change := ResourceChange{
Kind: kind,
Namespace: ns,
Name: name,
UID: uid,
Operation: op,
Diff: diff,
}
if rc.config.OnObservedChange != nil {
rc.safeCallback("OnObservedChange", func() { rc.config.OnObservedChange(change, obj, oldObj) })
}
// Fire OnChange callback (before channel send, matching existing behavior)
if !skipCallback && rc.config.OnChange != nil {
rc.safeCallback("OnChange", func() { rc.config.OnChange(change, obj, oldObj) })
}
// Non-blocking send to changes channel (skip noisy resources entirely —
// they don't affect topology and would just trigger unnecessary rebuilds)
if !isNoisy {
select {
case ch <- change:
default:
if rc.config.OnDrop != nil {
rc.config.OnDrop(kind, ns, name, "channel_full", op)
} else {
rc.stdlog.Printf("Warning: change channel full, dropped %s %s/%s op=%s", kind, ns, name, op)
}
}
}
}
// enqueueEvent handles K8s Event resource changes (separate path).
func (rc *ResourceCache) enqueueEvent(ch chan<- ResourceChange, obj any, op string) {
meta, ok := obj.(metav1.Object)
if !ok {
if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok {
meta, ok = tombstone.Obj.(metav1.Object)
if !ok {
rc.stdlog.Printf("Warning: tombstone contained non-metav1.Object for Event %s", op)
return
}
obj = tombstone.Obj
} else {
return
}
}
ns := meta.GetNamespace()
name := meta.GetName()
uid := string(meta.GetUID())
// Fire OnEventChange callback
if rc.config.OnEventChange != nil {
rc.safeCallback("OnEventChange", func() { rc.config.OnEventChange(obj, op) })
}
change := ResourceChange{
Kind: "Event",
Namespace: ns,
Name: name,
UID: uid,
Operation: op,
}
select {
case ch <- change:
default:
if rc.config.OnDrop != nil {
rc.config.OnDrop("Event", ns, name, "channel_full", op)
} else {
rc.stdlog.Printf("Warning: change channel full, dropped Event %s/%s op=%s", ns, name, op)
}
}
}
// safeCallback invokes fn with panic recovery to protect informer goroutines.
func (rc *ResourceCache) safeCallback(name string, fn func()) {
defer func() {
if r := recover(); r != nil {
buf := make([]byte, 4096)
n := runtime.Stack(buf, false)
rc.stdlog.Printf("ERROR: k8score %s callback panicked: %v\n%s", name, r, buf[:n])
}
}()
fn()
}
// recordWatchError is invoked from each informer's WatchErrorHandler when its
// reflector drops its connection. It updates the per-informer status so the
// failure surfaces in diagnostics, and logs the first 403/401 — these signal
// that the access probe and the actual list/watch disagree (token rotation,
// RBAC change mid-session, or webhook authorizer flakiness).
//
// The reflector continues retrying with exponential backoff regardless; this
// hook just makes the failure observable. Cap the recorded error string to
// keep the diagnostics payload small.
func (rc *ResourceCache) recordWatchError(idx int, key, kind string, err error) {
if err == nil {
return
}
msg := err.Error()
const maxLen = 256
if len(msg) > maxLen {
msg = msg[:maxLen] + "…"
}
forbidden := isWatchForbiddenErr(err)
rc.informerMu.Lock()
if idx >= 0 && idx < len(rc.informerStatuses) {
s := &rc.informerStatuses[idx]
firstForbidden := forbidden && !s.ForbiddenSeen
s.LastError = msg
s.LastErrorAt = time.Now().Format(time.RFC3339)
if forbidden {
s.ForbiddenSeen = true
}
if firstForbidden {
rc.stdlog.Printf("WARNING: informer %s saw 403/Unauthorized — probe and reality disagree (key=%s): %v",
kind, key, err)
}
}
rc.informerMu.Unlock()
}
// isWatchForbiddenErr returns true when err signals 403/401 from the
// apiserver. The reflector wraps the typed error in an unstructured
// status error, so apierrors helpers (which need an *apierrors.StatusError)
// can miss it — fall back to substring matching when type-detection fails.
//
// Kept here (rather than reusing the apierrors-aware helper in capabilities.go)
// because pkg/k8score must not import internal/.
func isWatchForbiddenErr(err error) bool {
if err == nil {
return false
}
msg := strings.ToLower(err.Error())
return strings.Contains(msg, "forbidden") || strings.Contains(msg, "unauthorized")
}
// Stop initiates a non-blocking shutdown of the cache.
func (rc *ResourceCache) Stop() {
if rc == nil {
return
}
rc.stopOnce.Do(func() {
rc.stdlog.Println("Stopping resource cache")
close(rc.stopCh)
go func() {
done := make(chan struct{})
go func() {
rc.factory.Shutdown()
for _, f := range rc.nsFactories {
f.Shutdown()
}
close(done)
}()
select {
case <-done:
rc.stdlog.Println("Resource cache factory shutdown complete")
case <-time.After(5 * time.Second):
rc.stdlog.Println("Resource cache factory shutdown taking >5s, abandoning")
}
}()
})
}
// Client returns the clientset the cache's informers were built from, so a
// caller holding its own clientset can tell whether the cache describes the
// same cluster connection before reading from it.
func (rc *ResourceCache) Client() kubernetes.Interface {
if rc == nil {
return nil
}
return rc.config.Client
}
// Changes returns a read-only channel for resource change notifications.
// It is lossy and meant for one reader that refreshes views: changes that
// IsNoisyResource marks are never sent, and a send to a full buffer is dropped
// and reported through OnDrop. Code that must see every change uses
// OnObservedChange instead.
func (rc *ResourceCache) Changes() <-chan ResourceChange {
if rc == nil {
return nil
}
return rc.changes
}
// IsKindClusterWide reports whether the informer for the given resource (plural
// lowercase name, e.g. "services") is enabled and watches ALL namespaces. When
// true, the lister's contents are authoritative cluster-wide; when false
// (disabled or namespace-scoped fallback), reads only cover a subset of
// namespaces, so callers must not infer cluster-wide absence from a miss.
func (rc *ResourceCache) IsKindClusterWide(resource string) bool {
if rc == nil {
return false
}
s, ok := rc.config.ResourceScopes[resource]
return ok && s.Enabled && s.Namespace == "" && len(rc.config.ResourceScopeNamespaces[resource]) == 0
}
// KindCoversNamespace reports whether the informer for a resource (plural
// lowercase, e.g. "pods") reliably covers the given namespace - cluster-wide, or
// namespace-scoped to exactly that namespace. When it returns false, an EMPTY
// list for that namespace is NOT authoritative: the informer may simply not watch
// it (namespace-scoped RBAC), so callers must not infer "0 objects" from a miss.
// A nil ResourceScopes is the legacy/cluster-wide default and covers everything.
func (rc *ResourceCache) KindCoversNamespace(resource, ns string) bool {
if rc == nil {
return false
}
if rc.config.ResourceScopes == nil {
return true
}
s, ok := rc.config.ResourceScopes[resource]
if !ok || !s.Enabled {
return false
}
if s.Namespace == "" {
return true
}
for _, watched := range resourceScopeNamespaces(resource, s, rc.config.ResourceScopeNamespaces) {
if watched == ns {
return true
}
}
return false
}
// KindNamespaces returns the namespaces covered by a namespaced informer.
// A nil result means the kind is cluster-wide or disabled.
func (rc *ResourceCache) KindNamespaces(resource string) []string {
if rc == nil {
return nil
}
s, ok := rc.config.ResourceScopes[resource]
if !ok || !s.Enabled || s.Namespace == "" {
return nil
}
return append([]string(nil), resourceScopeNamespaces(resource, s, rc.config.ResourceScopeNamespaces)...)
}
func (rc *ResourceCache) IsKindReady(resource string) bool {
// Answer from the informer's own HasSynced, like every other readiness
// probe — the tracked Synced flag lags it, so capacity checks and
// missing-reference detectors would briefly disagree with what the
// resource handlers already serve.
synced, known := rc.InformerSynced(resource)
return known && synced
}
// ChangesRaw returns the bidirectional channel for internal use.
func (rc *ResourceCache) ChangesRaw() chan ResourceChange {
if rc == nil {
return nil
}
return rc.changes
}
// PromotedKinds returns the list of resource kinds that were promoted from
// critical to deferred at first paint. Empty if sync completed normally.
// This is a snapshot from cache construction and does NOT shrink as
// promoted informers eventually sync — use PendingPromotedKinds for the
// live "still loading" view (e.g. for UI banners).
func (rc *ResourceCache) PromotedKinds() []string {
if rc == nil {
return nil
}
rc.informerMu.RLock()
defer rc.informerMu.RUnlock()
return rc.promotedKinds
}
// PendingPromotedKinds returns the subset of PromotedKinds whose informers
// have not yet completed their initial sync. Drains to empty as the
// background informers finish, so a UI bound to this method shows a
// truthful "still loading" indicator.
func (rc *ResourceCache) PendingPromotedKinds() []string {
if rc == nil {
return nil
}
rc.informerMu.RLock()
defer rc.informerMu.RUnlock()
if len(rc.promotedKinds) == 0 {
return nil
}
syncedByKind := make(map[string]bool, len(rc.informerStatuses))
for _, s := range rc.informerStatuses {
if s.Synced {
syncedByKind[s.Kind] = true
}
}
var pending []string
for _, k := range rc.promotedKinds {
if !syncedByKind[k] {
pending = append(pending, k)
}
}
return pending
}
// IsSyncComplete returns true after the initial critical informer sync.
func (rc *ResourceCache) IsSyncComplete() bool {
if rc == nil {
return false
}
return rc.syncComplete.Load()
}
// IsDeferredSynced returns true when all deferred (non-background) informers have
// completed sync. Background informers (e.g. Events) sync independently and are
// not included. Returns false if still syncing or if sync failed.
func (rc *ResourceCache) IsDeferredSynced() bool {
if rc == nil {
return false
}
rc.deferredMu.RLock()
done := rc.deferredDone
rc.deferredMu.RUnlock()
if done == nil {
// Phase 1 still running: tracking state not installed yet.
return false
}
select {
case <-done:
return !rc.deferredFailed.Load()
default:
return false
}
}
// DeferredDone returns a channel that is closed when all deferred (non-background)
// informers have completed their initial sync. Background informers sync independently.
func (rc *ResourceCache) DeferredDone() <-chan struct{} {
if rc == nil {
return nil
}
rc.deferredMu.RLock()
defer rc.deferredMu.RUnlock()
return rc.deferredDone
}
// InformerSynced reports one informer's initial-sync state. `known` is false
// when this cache never started that kind, so a caller can tell "not ready"
// from "not watched at all" instead of failing closed on both. Callers that
// need one kind must use this rather than IsDeferredSynced, which stays false
// while any unrelated deferred informer is warming or has permanently failed.
//
// It answers from the informer's own HasSynced, not the Synced flag the
// tracking goroutine writes ~10ms later: a caller gating a read on this would
// otherwise refuse to serve a cache that is in fact ready.
func (rc *ResourceCache) InformerSynced(key string) (synced, known bool) {
if rc == nil {
return false, false
}
rc.informerMu.RLock()
defer rc.informerMu.RUnlock()
for i, status := range rc.informerStatuses {
if status.Key == key {
if i < len(rc.informerHasSynced) && rc.informerHasSynced[i] != nil {
return rc.informerHasSynced[i](), true
}
return status.Synced, true
}
}
return false, false
}
// GetSyncStatus returns the current sync status of all informers for diagnostics.
// Safe to call at any time, including during sync.
func (rc *ResourceCache) GetSyncStatus() CacheSyncStatus {
if rc == nil {
return CacheSyncStatus{Phase: SyncPhaseNotStarted}
}
rc.informerMu.RLock()
statuses := make([]InformerSyncStatus, len(rc.informerStatuses))
copy(statuses, rc.informerStatuses)
rc.informerMu.RUnlock()
// Enrich with live item counts from listers
counts := rc.GetKindObjectCounts()
for i := range statuses {
if n, ok := counts[statuses[i].Kind]; ok {
statuses[i].Items = n
}
}
var critTotal, critSynced, defTotal, defSynced int
var pendingCritical, pendingDeferred []string
for _, s := range statuses {
if s.Deferred {
defTotal++
if s.Synced {
defSynced++
} else {
pendingDeferred = append(pendingDeferred, s.Kind)
}
} else {
critTotal++
if s.Synced {
critSynced++
} else {
pendingCritical = append(pendingCritical, s.Kind)
}
}
}
var phase SyncPhase
switch {
case rc.syncStartTime.IsZero():
phase = SyncPhaseNotStarted
case !rc.syncComplete.Load():
phase = SyncPhaseCritical
case !rc.IsDeferredSynced():
phase = SyncPhaseDeferred
default:
phase = SyncPhaseComplete
}
result := CacheSyncStatus{
Phase: phase,
ElapsedSec: time.Since(rc.syncStartTime).Seconds(),
CriticalTotal: critTotal,
CriticalSynced: critSynced,
DeferredTotal: defTotal,
DeferredSynced: defSynced,
Informers: statuses,
PendingCritical: pendingCritical,
PendingDeferred: pendingDeferred,
PromotedKinds: rc.PromotedKinds(),
}
if !rc.syncStartTime.IsZero() {
result.SyncStarted = rc.syncStartTime.Format(time.RFC3339)
}
return result
}
// GetEnabledResources returns a copy of the enabled resources map.
func (rc *ResourceCache) GetEnabledResources() map[string]bool {
if rc == nil {
return nil
}
result := make(map[string]bool, len(rc.enabledResources))
maps.Copy(result, rc.enabledResources)
return result
}
// GetResourceCount returns total cached resources across all enabled non-Event listers.
func (rc *ResourceCache) GetResourceCount() int {
if rc == nil {
return 0
}
counts := rc.GetKindObjectCounts()
total := 0
for kind, n := range counts {
if kind == "Event" {
continue // Events are not counted as "resources"
}
total += n
}
return total
}
// kindLister maps a Kind name to a lister accessor for table-driven counting.
type kindLister struct {
kind string
group string // API group (empty for core, e.g. "apps", "batch", "networking.k8s.io")
lister func(rc *ResourceCache) any
}
// allKindListers is the table of all resource kinds and their lister accessors.
var allKindListers = []kindLister{
{"Pod", "", func(rc *ResourceCache) any { return rc.Pods() }},
{"Service", "", func(rc *ResourceCache) any { return rc.Services() }},
{"Node", "", func(rc *ResourceCache) any { return rc.Nodes() }},
{"Namespace", "", func(rc *ResourceCache) any { return rc.Namespaces() }},
{"ConfigMap", "", func(rc *ResourceCache) any { return rc.ConfigMaps() }},
{"Secret", "", func(rc *ResourceCache) any { return rc.Secrets() }},
{"Event", "", func(rc *ResourceCache) any { return rc.Events() }},
{"PersistentVolumeClaim", "", func(rc *ResourceCache) any { return rc.PersistentVolumeClaims() }},
{"PersistentVolume", "", func(rc *ResourceCache) any { return rc.PersistentVolumes() }},
{"Deployment", "apps", func(rc *ResourceCache) any { return rc.Deployments() }},
{"DaemonSet", "apps", func(rc *ResourceCache) any { return rc.DaemonSets() }},
{"StatefulSet", "apps", func(rc *ResourceCache) any { return rc.StatefulSets() }},
{"ReplicaSet", "apps", func(rc *ResourceCache) any { return rc.ReplicaSets() }},
{"Ingress", "networking.k8s.io", func(rc *ResourceCache) any { return rc.Ingresses() }},
{"IngressClass", "networking.k8s.io", func(rc *ResourceCache) any { return rc.IngressClasses() }},
{"Job", "batch", func(rc *ResourceCache) any { return rc.Jobs() }},
{"CronJob", "batch", func(rc *ResourceCache) any { return rc.CronJobs() }},
{"HorizontalPodAutoscaler", "autoscaling", func(rc *ResourceCache) any { return rc.HorizontalPodAutoscalers() }},
{"StorageClass", "storage.k8s.io", func(rc *ResourceCache) any { return rc.StorageClasses() }},
{"PodDisruptionBudget", "policy", func(rc *ResourceCache) any { return rc.PodDisruptionBudgets() }},
{"NetworkPolicy", "networking.k8s.io", func(rc *ResourceCache) any { return rc.NetworkPolicies() }},
{"ServiceAccount", "", func(rc *ResourceCache) any { return rc.ServiceAccounts() }},
{"Role", "rbac.authorization.k8s.io", func(rc *ResourceCache) any { return rc.Roles() }},
{"ClusterRole", "rbac.authorization.k8s.io", func(rc *ResourceCache) any { return rc.ClusterRoles() }},
{"RoleBinding", "rbac.authorization.k8s.io", func(rc *ResourceCache) any { return rc.RoleBindings() }},
{"ClusterRoleBinding", "rbac.authorization.k8s.io", func(rc *ResourceCache) any { return rc.ClusterRoleBindings() }},
{"LimitRange", "", func(rc *ResourceCache) any { return rc.LimitRanges() }},
{"ResourceQuota", "", func(rc *ResourceCache) any { return rc.ResourceQuotas() }},
}
// AllKindListers returns the table of all resource kinds with their group and lister.
// Used by the resource-counts endpoint to enumerate typed resources.
func AllKindListers() []kindLister {
return allKindListers
}
// KindReadinessForKindName is KindReadinessFor keyed by the Kind name
// ("Pod") instead of the informer key ("pods") — for callers that enumerate
// the kindLister table, which carries Kinds only.
func (rc *ResourceCache) KindReadinessForKindName(kind string) KindReadiness {
if rc == nil {
return KindUnavailable
}
key := ""
rc.informerMu.RLock()
for _, status := range rc.informerStatuses {
if status.Kind == kind {
key = status.Key
break
}
}
rc.informerMu.RUnlock()
if key == "" {
return KindUnavailable
}
return rc.KindReadinessFor(key)
}
// Kind returns the resource kind name.
func (kl kindLister) Kind() string { return kl.kind }
// Group returns the API group (empty for core resources).
func (kl kindLister) Group() string { return kl.group }
// Lister returns the lister accessor function.
func (kl kindLister) Lister() func(rc *ResourceCache) any { return kl.lister }
// CountKey returns the key used in resource-counts responses: "group/Kind" or just "Kind" for core.
func (kl kindLister) CountKey() string {
if kl.group != "" {
return kl.group + "/" + kl.kind
}
return kl.kind
}
// GetKindObjectCounts returns the number of cached objects per resource kind.
// Only includes kinds that are enabled. Returns nil if cache is nil.
func (rc *ResourceCache) GetKindObjectCounts() map[string]int {
if rc == nil {
return nil
}
counts := make(map[string]int)
for _, kl := range allKindListers {
l := kl.lister(rc)
if l == nil {
continue
}
n := listCount(l)
if n > 0 {
counts[kl.kind] = n
}
}
return counts
}
// isEnabled returns true if the resource type has an informer running.
func (rc *ResourceCache) isEnabled(key string) bool {
if rc == nil || rc.enabledResources == nil {
return false
}
return rc.enabledResources[key]
}
// isReady returns true if the resource is enabled and, if deferred, synced.
func (rc *ResourceCache) isReady(key string) bool {
if !rc.isEnabled(key) {
return false
}
if rc.config.DeferredTypes == nil || !rc.config.DeferredTypes[key] {
return true
}
rc.deferredMu.RLock()
done := rc.deferredSynced[key]
rc.deferredMu.RUnlock()
if done {
return true
}
// The deferred tracker freezes its map when the deferred deadline fires;
// a kind whose LIST completes after that still has a fully-synced store.
// Ask the informer itself, so the lister and KindReadinessFor (which also
// reads live HasSynced) agree — otherwise a late-synced kind reports
// KindReady at the gate and then 403s off a nil lister forever.
synced, known := rc.InformerSynced(key)
return known && synced
}
// KindReadiness classifies a typed kind's serveability. It is meaningful at
// any point in the cache lifecycle, including during the Phase-1 sync wait
// (via the OnInformersStarted handle).
type KindReadiness int
const (
// KindUnavailable: no informer for this kind (RBAC-disabled, unknown key,
// or not part of the typed set). Callers fall through to their existing
// forbidden/not-found/dynamic semantics.
KindUnavailable KindReadiness = iota
// KindPending: the informer exists but its initial LIST hasn't completed.
// Serving now would render a partial store as truth — callers must
// respond "still loading", never an incomplete list.
KindPending
// KindReady: the informer's store is complete; serve normally.
KindReady
// KindFailed: the deferred-sync deadline fired with this kind still
// unsynced. The informer keeps retrying and the kind reads KindReady once
// it syncs, so callers say the load is overdue, not that it has failed for
// good, and suggest a later retry rather than "retry shortly".
KindFailed
)
// KindReadinessFor reports whether key can be served from this cache right
// now. Keys are informer keys (lowercase plural, e.g. "pods") — the same
// vocabulary as DeferredTypes and InformerSyncStatus.Key.
func (rc *ResourceCache) KindReadinessFor(key string) KindReadiness {
if rc == nil || !rc.isEnabled(key) {
return KindUnavailable
}
synced, known := rc.InformerSynced(key)
if !known {
return KindUnavailable
}
if synced {
return KindReady
}
// The give-up deadline only covers kinds in the deferred tracking set
// (statically deferred + promoted criticals). A kind mid-Phase-1 is still
// pending even if a previous generation's flag lingers — the tracking map
// tells them apart. Background kinds (Events) share the tracking map but
// have no deadline: they stay pending, not failed.
if rc.pastDeadline(key) {
return KindFailed
}
return KindPending
}
// KindSyncState is one kind's live sync state in a SyncSnapshot.
type KindSyncState struct {
Kind string `json:"kind"`
Key string `json:"key"`
Synced bool `json:"synced"`
Deferred bool `json:"deferred"`
// Failed marks a kind whose sync deadline fired before it finished
// loading. Its informer keeps retrying and the kind clears once it
// syncs; the shell renders these as errors instead of promising they
// will finish soon.
Failed bool `json:"failed,omitempty"`
}
// SyncSnapshot is a lightweight progress report for connection-status
// payloads: per-kind live sync state without the lister walks GetSyncStatus
// does for item counts. Cheap enough for sub-second polling during sync.
type SyncSnapshot struct {
Phase SyncPhase `json:"phase"`
CriticalTotal int `json:"criticalTotal"`
CriticalSynced int `json:"criticalSynced"`
DeferredTotal int `json:"deferredTotal"`
DeferredSynced int `json:"deferredSynced"`
Kinds []KindSyncState `json:"kinds"`
}
// GetSyncSnapshot returns the live per-kind sync state. Unlike GetSyncStatus
// it reads each informer's HasSynced directly (no status-goroutine lag) and
// never touches listers.
func (rc *ResourceCache) GetSyncSnapshot() SyncSnapshot {
if rc == nil {
return SyncSnapshot{Phase: SyncPhaseNotStarted}
}
rc.informerMu.RLock()
statuses := make([]InformerSyncStatus, len(rc.informerStatuses))
copy(statuses, rc.informerStatuses)
rc.informerMu.RUnlock()
snap := SyncSnapshot{Kinds: make([]KindSyncState, 0, len(statuses))}
for _, s := range statuses {
readiness := rc.KindReadinessFor(s.Key)
synced := readiness == KindReady
snap.Kinds = append(snap.Kinds, KindSyncState{
Kind: s.Kind, Key: s.Key, Synced: synced, Deferred: s.Deferred,
Failed: readiness == KindFailed,
})
if s.Deferred {
snap.DeferredTotal++
if synced {
snap.DeferredSynced++
}
} else {
snap.CriticalTotal++
if synced {
snap.CriticalSynced++
}
}
}
switch {
case rc.syncStartTime.IsZero():
snap.Phase = SyncPhaseNotStarted
case !rc.syncComplete.Load():
snap.Phase = SyncPhaseCritical
case !rc.IsDeferredSynced():
snap.Phase = SyncPhaseDeferred
default:
snap.Phase = SyncPhaseComplete
}
return snap
}
// IsDeferredPending returns true when the resource type passed RBAC checks
// (informer is enabled) but deferred sync has not completed yet. Callers
// can use this to distinguish "no permission" (return 403) from "not ready
// yet" (return 503) when a lister returns nil.
// Returns false once deferred sync has permanently failed (avoids infinite spinner).
func (rc *ResourceCache) IsDeferredPending(key string) bool {
if rc == nil {
return false
}
if !rc.isEnabled(key) {
return false
}
if rc.config.DeferredTypes == nil || !rc.config.DeferredTypes[key] {
return false
}
if rc.deferredFailed.Load() {
return false
}
rc.deferredMu.RLock()
done := rc.deferredSynced[key]
rc.deferredMu.RUnlock()
if done {
return false
}
// Same live fallback as isReady: the tracking goroutine marks the map a
// beat after HasSynced flips, and the two answers must agree — otherwise
// a just-synced kind serves from its lister while this still says 503.
synced, known := rc.InformerSynced(key)
return !(known && synced)
}