Get(k any) (any, bool)

Set(k any, v any, ttl time.Duration)

Delete(k any)

PostOne(ctx context.Context, record *deploymentrecord.Record) error

CreateClusterJob(ctx context.Context, records []*deploymentrecord.Record, cluster string) (*deploymentrecord.JobResponse, error)

WaitForClusterJob(ctx context.Context, cluster string, jobID int64) (*deploymentrecord.JobStatus, error)

Key string

EventType string

DeletedPod *corev1.Pod // Only populated for delete events

informerFactory informers.SharedInformerFactory

podInformer cache.SharedIndexInformer

workqueue workqueue.TypedRateLimitingInterface[PodEvent]

workloadResolver workloadResolver

metadataAggregator podMetadataAggregator

apiClient deploymentRecordPoster

cfg *Config

// best effort cache to avoid redundant posts

// post requests are idempotent, so if this cache fails due to

// restarts or other events, nothing will break.

observedDeployments ttlCache

// best effort cache to suppress API calls for artifacts that

// returned a 404 (no artifact found). Keyed by image digest.

unknownArtifacts ttlCache

// informerSyncTimeout is the maximum time allowed for all informers to sync

// and prevents sync from hanging indefinitely.

informerSyncTimeout time.Duration

// syncing gates informer event handlers during startup. When true,

// pod add events are suppressed so they can be reported via the bulk

// cluster job instead of individual PostOne calls. Only set when

// BulkClusterSync is enabled.

syncing atomic.Bool

// Create informer factory

informerFactory := createInformerFactory(clientset, namespace, excludeNamespaces)

podInformer := informerFactory.Core().V1().Pods().Informer()

// Create the workload resolver with listers from the factory

resolver := workload.NewResolver(

informerFactory.Apps().V1().Deployments().Lister(),

informerFactory.Apps().V1().DaemonSets().Lister(),

informerFactory.Apps().V1().StatefulSets().Lister(),

informerFactory.Batch().V1().Jobs().Lister(),

informerFactory.Batch().V1().CronJobs().Lister(),

)

// Create work queue with rate limiting

queue := workqueue.NewTypedRateLimitingQueue(

workqueue.DefaultTypedControllerRateLimiter[PodEvent](),

)

// Create API client with optional token

clientOpts := []deploymentrecord.ClientOption{}

if cfg.APIToken != "" {

clientOpts = append(clientOpts, deploymentrecord.WithAPIToken(cfg.APIToken))

}

if cfg.GHAppID != "" &&

cfg.GHInstallID != "" &&

(len(cfg.GHAppPrivateKey) > 0 || cfg.GHAppPrivateKeyPath != "") {

clientOpts = append(clientOpts, deploymentrecord.WithGHApp(

cfg.GHAppID,

cfg.GHInstallID,

cfg.GHAppPrivateKey,

cfg.GHAppPrivateKeyPath,

))

}

clientOpts = append(clientOpts, deploymentrecord.WithTimeout(deploymentRecordClientTimeoutSeconds))

apiClient, err := deploymentrecord.NewClient(

cfg.BaseURL,

cfg.Organization,

clientOpts...,

)

if err != nil {

return nil, fmt.Errorf("failed to create API client: %w", err)

}

cntrl := &Controller{

informerFactory: informerFactory,

podInformer: podInformer,

workqueue: queue,

workloadResolver: resolver,

metadataAggregator: metadataAggregator,

apiClient: apiClient,

cfg: cfg,

observedDeployments: amcache.NewExpiring(),

unknownArtifacts: amcache.NewExpiring(),

informerSyncTimeout: informerSyncTimeoutDuration,

}

// Only gate informer events when bulk cluster sync is enabled.

// When disabled, all pods discovered during informer sync will be

// enqueued as individual events.

if cfg.BulkClusterSync {

cntrl.syncing.Store(true)

}

// Add event handlers to the informer

_, err = podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{

AddFunc: func(obj any) {

// Skip adding sync events

if cntrl.syncing.Load() {

return

}

pod, ok := obj.(*corev1.Pod)

if !ok {

slog.Error("Invalid object returned",

"object", obj,

)

return

}

// Only process pods that are running and belong

// to a supported workload (Deployment, DaemonSet, StatefulSet, Job, or CronJob)

if pod.Status.Phase == corev1.PodRunning && workload.HasSupportedOwner(pod) {

key, err := cache.MetaNamespaceKeyFunc(obj)

// For our purposes, there are in practice

// no error event we care about, so don't

// bother with handling it.

if err == nil {

queue.Add(PodEvent{

Key: key,

EventType: EventCreated,

})

}

}

// Also process Job-owned pods that completed before

// we observed them in Running phase (e.g. sub-second Jobs).

if workload.IsTerminalPhase(pod) && workload.GetJobOwnerName(pod) != "" {

key, err := cache.MetaNamespaceKeyFunc(obj)

if err == nil {

queue.Add(PodEvent{

Key: key,

EventType: EventCreated,

})

}

}

},

UpdateFunc: func(oldObj, newObj any) {

oldPod, ok := oldObj.(*corev1.Pod)

if !ok {

slog.Error("Invalid old object returned",

"object", oldObj,

)

return

}

newPod, ok := newObj.(*corev1.Pod)

if !ok {

slog.Error("Invalid new object returned",

"object", newObj,

)

return

}

// Skip if pod is being deleted or doesn't belong

// to a supported workload.

// Exception: Job-owned pods transitioning to a terminal phase

// (Succeeded/Failed) from a non-Running state should still be

// processed — this catches short-lived Jobs that skip Running.

// We exclude Running→terminal transitions since those pods

// were already enqueued when they entered Running.

isJobTerminal := !workload.IsTerminalPhase(oldPod) && workload.IsTerminalPhase(newPod) &&

oldPod.Status.Phase != corev1.PodRunning && workload.GetJobOwnerName(newPod) != ""

if !isJobTerminal {

if newPod.DeletionTimestamp != nil || !workload.HasSupportedOwner(newPod) {

return

}

}

// Only process if pod just became running.

// We need to process this as often when a container

// is created, the spec does not contain the digest

// so we need to wait for the status field to be

// populated from where we can get the digest.

if oldPod.Status.Phase != corev1.PodRunning &&

newPod.Status.Phase == corev1.PodRunning {

key, err := cache.MetaNamespaceKeyFunc(newObj)

// For our purposes, there are in practice

// no error event we care about, so don't

// bother with handling it.

if err == nil {

queue.Add(PodEvent{

Key: key,

EventType: EventCreated,

})

}

}

// Also catch Job-owned pods that transitioned directly

// to a terminal phase without us seeing them as Running.

if isJobTerminal {

key, err := cache.MetaNamespaceKeyFunc(newObj)

if err == nil {

queue.Add(PodEvent{

Key: key,

EventType: EventCreated,

})

}

}

},

DeleteFunc: func(obj any) {

pod, ok := obj.(*corev1.Pod)

if !ok {

// Handle deleted final state unknown

tombstone, ok := obj.(cache.DeletedFinalStateUnknown)

if !ok {

return

}

pod, ok = tombstone.Obj.(*corev1.Pod)

if !ok {

return

}

}

// Only process pods that belong to a supported workload

if !workload.HasSupportedOwner(pod) {

return

}

key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)

// For our purposes, there are in practice

// no error event we care about, so don't

// bother with handling it.

if err == nil {

queue.Add(PodEvent{

Key: key,

EventType: EventDeleted,

DeletedPod: pod,

})

}

},

})

if err != nil {

return nil, fmt.Errorf("failed to add event handlers: %w", err)

}

return cntrl, nil

defer runtime.HandleCrash()

defer c.workqueue.ShutDown()

slog.Info("Starting informers")

// Start all informers via the factory

c.informerFactory.Start(ctx.Done())

// Wait for the caches to be synced

slog.Info("Waiting for informer caches to sync")

informerSyncCtx, cancel := context.WithTimeout(ctx, c.informerSyncTimeout)

syncResults := c.informerFactory.WaitForCacheSync(informerSyncCtx.Done())

cancel()

for _, synced := range syncResults {

if !synced {

if ctx.Err() != nil {

return fmt.Errorf("cache sync interrupted: %w", ctx.Err())

}

return errors.New("timed out waiting for caches to sync - please ensure deployment tracker has the correct kubernetes permissions")

}

}

c.syncing.Store(false)

syncClusterPods := c.podInformer.GetIndexer().List()

err := c.processSyncEvents(ctx, syncClusterPods)

if err != nil {

return fmt.Errorf("sync events failed: %w", err)

}

slog.Info("Starting workers",

"count", workers,

)

// Start workers

for i := 0; i < workers; i++ {

go wait.UntilWithContext(ctx, c.runWorker, time.Second)

}

slog.Info("Controller started")

<-ctx.Done()

slog.Info("Shutting down workers")

return nil

event, shutdown := c.workqueue.Get()

if shutdown {

return false

}

defer c.workqueue.Done(event)

start := time.Now()

err := c.processEvent(ctx, event)

dur := time.Since(start)

if err == nil {

dtmetrics.EventsProcessedOk.WithLabelValues(event.EventType).Inc()

dtmetrics.EventsProcessedTimer.WithLabelValues("ok").Observe(dur.Seconds())

c.workqueue.Forget(event)

return true

}

dtmetrics.EventsProcessedTimer.WithLabelValues("failed").Observe(dur.Seconds())

dtmetrics.EventsProcessedFailed.WithLabelValues(event.EventType).Inc()

// Requeue on error with rate limiting

slog.Error("Failed to process event, requeuing",

"event_key", event.Key,

"error", err,

)

c.workqueue.AddRateLimited(event)

return true

var factory informers.SharedInformerFactory

switch {

case namespace != "":

slog.Info("Namespace to watch",

"namespace",

namespace,

)

factory = informers.NewSharedInformerFactoryWithOptions(

clientset,

30*time.Second,

informers.WithNamespace(namespace),

)

case excludeNamespaces != "":

seenNamespaces := make(map[string]bool)

fieldSelectorParts := make([]string, 0)

for _, ns := range strings.Split(excludeNamespaces, ",") {

ns = strings.TrimSpace(ns)

if ns != "" && !seenNamespaces[ns] {

seenNamespaces[ns] = true

fieldSelectorParts = append(fieldSelectorParts, fmt.Sprintf("metadata.namespace!=%s", ns))

}

}

slog.Info("Excluding namespaces from watch",

"field_selector",

strings.Join(fieldSelectorParts, ","),

)

tweakListOptions := func(options *metav1.ListOptions) {

options.FieldSelector = strings.Join(fieldSelectorParts, ",")

}

factory = informers.NewSharedInformerFactoryWithOptions(

clientset,

30*time.Second,

informers.WithTweakListOptions(tweakListOptions),

)

default:

factory = informers.NewSharedInformerFactory(clientset,

30*time.Second,

)

}

return factory