Categories 1-3: Sensor Pipeline, Background Loops, and Enrichment/Detection paths
+
+
+
+
Sensor (per cluster)
+
Bidirectional gRPC stream
MsgFromSensor / MsgToSensor
mTLS authenticated
+
partial: connect counter
+
+
→
+
+
Deduper
+
Hash-based dedup
per connection
Prevents reprocessing identical msgs
+
partial: passed/deduped counter
+
+
→
+
+
Worker Queue
+
FNV hash → poolSize+1 shards
Each shard = DedupingQueue + goroutine
Retry: 5 attempts, 5min backoff
+
partial: add/remove/dedupe counters only
+
+
→
+
+
Pipeline Router
+
Match() selects 1 of 25 fragments
safe.Run() wraps execution
Panic recovery per message
+
partial: panic counter, event duration
+
+
→
+
+
Fragment.Run()
+
Type-specific processing
Validates, enriches context
May trigger detection/risk
+
partial: resource_processed_count
+
+
→
+
+
PostgreSQL
+
250+ datastores
KeyFence for dedup
Connection pool
+
has metrics: op duration, table stats
+
+
Each processes a specific MsgFromSensor type. Colored tags show what side-effects they trigger.
+
+
+
+
deploymentevents
+
Deployment CRUD
+
→ Risk reprocess (async)
+
+
+
alerts
+
AlertResults
+
→ Detection lifecycle mgr
+
+
+
processindicators
+
ProcessIndicator
+
→ Detection lifecycle (queued)
+
+
+
networkflowupdate
+
NetworkFlowUpdate
+
→ Flow store + baselines (LOCAL)
+
+
+
nodes
+
Node CRUD
+
→ Vuln enrichment + Risk calc
+
+
+
nodeindex
+
IndexReport (v4)
+
→ Vuln enrichment + Risk calc
+
+
+
nodeinventory
+
NodeInventory (v2)
+
→ Vuln enrichment + Risk calc
+
+
+
podevents
+
Pod CRUD
+
Store only
+
+
+
namespaces
+
Namespace CRUD
+
Store + network graph epoch
+
+
+
networkpolicies
+
NetworkPolicy CRUD
+
Store + network graph epoch
+
+
+
secrets
+
Secret CRUD
+
Store only
+
+
+
roles
+
K8s Role CRUD
+
Store only
+
+
+
rolebindings
+
K8s RoleBinding CRUD
+
Store only
+
+
+
serviceaccounts
+
ServiceAccount CRUD
+
→ Risk reprocess (conditional)
+
+
+
imageintegrations
+
ImageIntegration
+
→ Enrichment + Detection loop
+
+
+
reprocessing
+
ReprocessDeployment
+
→ Risk manager
+
+
+
enhancements
+
DeploymentEnhancementResp
+
→ Enhancement broker
+
+
+
processlisteningonport
+
ProcessListeningOnPort
+
Store only
+
+
+
clustermetrics
+
ClusterMetrics
+
Prometheus gauges + telemetry
+
+
+
clusterstatusupdate
+
ClusterStatusUpdate
+
→ CVE fetcher trigger
+
+
+
clusterhealthupdate
+
ClusterHealthInfo
+
Health status computation
+
+
+
auditlogstateupdate
+
AuditLogStatusInfo
+
Store only
+
+
+
virtualmachines
+
VirtualMachine CRUD
+
Store only (feature-gated)
+
+
+
virtualmachineindex
+
VMIndexReport
+
→ Vuln enrichment (feature-gated)
+
+
Two paths to enrichment: inline (real-time from pipeline) and background (reprocessor loop)
+
+
+
+
Inline Enrichment (Pipeline-triggered)
+
+ Image Enrichment:
+ Pipeline → EnrichDeployment() → imageEnricher.EnrichImage()
+ Semaphore: internalScanSemaphore (configurable)
+ External: Scanner + Registry calls
+ scanWaiterManager deduplicates concurrent scans
+ Node Enrichment:
+ nodeindex/nodeinventory fragments → nodeEnricher.EnrichNode()
+ Vulnerability data merged into node
+ Context Enrichment:
+ Most fragments add cluster name/ID to resources
+
+
partial: scan semaphore gauges only
+
+
+
+
Background Enrichment (Reprocessor)
+
+ Image Reprocessing (every 4h):
+ Query all images → semaphore(5) parallel enrichment
+ ForceRefetchCachedValuesOnly on tick
+ Send updated images back to Sensor clusters
+ Node Reprocessing:
+ All nodes → semaphore(5) parallel enrichment
+ Skip host-scanned nodes
+ Watched Images:
+ Separately reprocessed (may delegate to cluster)
+ Signature Verification:
+ Triggered by signal, re-verifies all image sigs
+
+
partial: total duration gauge only
+
+
+
+
Build-time Detection
+
API-triggered only
Evaluates policies against image
Returns alerts synchronously
+
no pipeline metrics
+
+
+
Deploy-time Detection
+
Sensor sends AlertResults
Lifecycle mgr merges alerts
Alert create/update/resolve
+
partial: resource count
+
+
+
Runtime Detection
+
Process indicators queued
Batched flush every 1 min
Baseline generation/locking
+
partial: queue length gauge
+
+
+
+
Alert Manager & Risk Calculator
+
+ Alert Manager: Merges new alerts with existing → classifies as new/updated/resolved → dispatches notifications → triggers risk reprocess
+ Risk Manager: Scores deployments (image risk + violations + port exposure), images (vulns + age + components), nodes (vulns). Updates namespace/cluster rankings.
+
+
partial: risk_processing_duration histogram
+
+
Only showing loops in Categories 1-3 scope. Loops marked with * are candidates for standard metrics.
+
+
+
+
* Reprocessor (enrichLoop)
+
Every 4h (ROX_REPROCESS_INTERVAL)
+
1 goroutine + semaphore(5) workers per batch
+
Reprocess all images, nodes, watched images. Resync deployments to sensors. Can be short-circuited.
+
partial: duration gauge
+
+
+
* Reprocessor (riskLoop)
+
Every ROX_RISK_REPROCESS_INTERVAL
+
1 goroutine
+
Drains deploymentRiskSet, sends risk reprocess messages to sensor connections.
+
no metrics
+
+
+
* Reprocessor (activeComponentLoop)
+
Every ROX_ACTIVE_VULN_REFRESH_INTERVAL
+
1 goroutine
+
Updates active (exploitable) component tracking.
+
no metrics
+
+
+
* CVE Suppress
+
Every 1h (hardcoded)
+
1 goroutine
+
Unsuppress CVEs with expired suppress state.
+
no metrics
+
+
+
* Pruning (GC)
+
Every ROX_PRUNE_INTERVAL
+
1 goroutine + semaphore(5) for network flows
+
Prune old alerts, orphaned pods/nodes, expired requests, stale baselines, log imbues, network flows.
+
partial: prune_duration histogram
+
+
+
* CVE Fetcher
+
ROX_ORCHESTRATOR_VULN_SCAN_INTERVAL
+
1 goroutine
+
Reconcile orchestrator + Istio CVEs. Throttled on cluster connection.
+
no metrics
+
+
+
* Lifecycle Mgr (indicator flush)
+
Every 1 min
+
1 goroutine
+
Flush queued process indicators to datastore. Batch indexing. Baseline evaluation.
+
partial: process_queue_length gauge
+
+
+
* Network Baseline Flush
+
Every 5s (hardcoded)
+
1 goroutine
+
Flush accumulated network baseline changes to storage.
+
no metrics
+
+
+
* Hash Manager Flush
+
ROX_HASH_FLUSH_INTERVAL
+
1 goroutine (optional)
+
Flush deduplication hashes to database.
+
partial: hash_size gauge
+
+
+
Conn Manager Health
+
Every 30s (hardcoded)
+
1 goroutine
+
Update inactive/active cluster health status.
+
no metrics
+
+
+
Vuln Request Manager
+
ROX_VULN_REQUEST_REVERT_TIMER_DURATION
+
3 goroutines
+
Expire timed deferrals, expire fixable CVE deferrals, unsuppress CVEs.
+
no metrics
+
+
+
Network Gatherer
+
ROX_EXT_NETWORK_SRCS_GATHER_INTERVAL
+
2 goroutines
+
Fetch default external networks, reconcile with storage.
+
no metrics
+
+
+
Observability Gaps Summary
+
+ Worker Queue: No depth gauge, no worker utilization, no retry metrics, no in-flight count, no end-to-end latency
+ Background Loops: 7 of 12 loops have ZERO metrics. No "is running" flag, no batch size, no items processed/failed per run, no time-since-last-run
+ Enrichment: No call count by trigger source, no success/failure/skip rates, no cache hit rates
+ Detection: No alerts-generated-per-evaluation, no violations-found vs clean ratio, no baseline generation timing
+ Cross-cutting: No end-to-end trace from Sensor message arrival to final storage commit
+
+
Each archetype has different concurrency patterns, different metric needs, and different levels of existing instrumentation
+
+
+
Archetype 1: Sensor Message Pipeline (Worker Queue)
+
+
+
+// Hash-sharded deduping queues (poolSize+1 workers)
+type workerQueue struct {
+ poolSize int
+ queues []*dedupingqueue.DedupingQueue[string] // each queue = 1 goroutine
+}
+
+// Push: FNV hash routes msg to a specific queue shard
+func (w *workerQueue) push(msg *central.MsgFromSensor) {
+ idx := w.indexFromKey(msg.GetHashKey())
+ w.queues[idx].Push(msg) // NO depth gauge, NO push timing
+}
+
+// Workers: pull blocking, handle, retry on transient errors
+func (w *workerQueue) runWorker(...) {
+ for msg := queue.PullBlocking(stopSig); msg != nil; ... {
+ err := handler(ctx, msgFromSensor)
+ if pgutils.IsTransientError(err) {
+ // Retry with backoff - NO retry count metric
+ concurrency.AfterFunc(reprocessingDuration, ...)
+ }
+ }
+}
+
+
+
+
Existing Metrics
+
+sensor_event_queue - CounterVec (Add/Remove/Dedupe ops)sensor_event_duration - Histogram (processing time per type)sensor_event_deduper - Counter (passed/deduped)pipeline_panics - Counter per resource type
+
+
+
Missing Metrics
+
+- Queue depth gauge (how deep is each shard?)
+- Worker utilization (busy vs idle time)
+- Retry count / permanent failure count
+- Per-shard processing rate
+- Items in-flight (being processed right now)
+- End-to-end latency (enqueue to completion)
+
+
+
+
+
Archetype 2: Background Ticker Loop (Reprocessor)
+
+
+
+// Three independent ticker loops, each in its own goroutine
+func (l *loopImpl) Start() {
+ l.enrichAndDetectTicker = time.NewTicker(4 * time.Hour)
+ l.deploymentRiskTicker = time.NewTicker(riskInterval)
+ l.activeComponentTicker = time.NewTicker(activeInterval)
+ go l.riskLoop()
+ go l.enrichLoop()
+ go l.activeComponentLoop()
+}
+
+// enrichLoop: select on stop, short-circuit, signature, ticker
+func (l *loopImpl) enrichLoop() {
+ for !l.stopSig.IsDone() {
+ select {
+ case <-l.enrichAndDetectTicker.C:
+ l.runReprocessing(ForceRefetchCachedValuesOnly)
+ // Only tracks total duration, not per-entity
+ }
+ }
+}
+
+// Semaphore-limited concurrent processing (5 goroutines max)
+func (l *loopImpl) runReprocessingForObjects(...) {
+ sema := semaphore.NewWeighted(5)
+ for _, id := range ids {
+ go func(id string) {
+ defer sema.Release(1)
+ individualReprocessFunc(id) // NO per-item timing
+ }(id)
+ }
+}
+
+
+
+
Existing Metrics
+
+reprocessor_duration_seconds - Gauge (total loop time)signature_verification_reprocessor_duration_seconds - Gaugerisk_processing_duration - Histogram
+
+
+
Missing Metrics
+
+- Is the loop currently running? (boolean gauge)
+- How many items in current batch?
+- Items processed / failed per run
+- Semaphore utilization (how many of 5 slots used?)
+- Time since last successful run
+- Individual entity processing duration histogram
+- Number of runs completed (counter)
+
+
+
+
+
Archetype 3: Enrichment/Detection (Request-Driven Processing)
+
+
+
+// Enrichment: called inline from pipeline AND from reprocessor
+// Two different call paths, same underlying function
+
+// Path 1: Pipeline fragment triggers enrichment synchronously
+func (p *pipelineImpl) Run(ctx, clusterID, msg) {
+ deployment := msg.GetEvent().GetDeployment()
+ images, _, _ := e.enricher.EnrichDeployment(ctx, deployment)
+ // NO metrics on enrichment call count or success rate
+ detector.ProcessDeployment(ctx, deployment, images)
+ // NO metrics on detection outcome counts
+}
+
+// Path 2: Reprocessor triggers enrichment in background
+func (l *loopImpl) reprocessImage(id string, ...) {
+ result, err := reprocessingFunc(ctx, enrichCtx, image)
+ if result.ImageUpdated {
+ l.risk.CalculateRiskAndUpsertImage(image)
+ }
+ // Only counts nReprocessed at batch level
+}
+
+// Image scan semaphore limits concurrent scanner calls
+internalScanSemaphore = semaphore.NewWeighted(maxParallelScans)
+// This one HAS metrics: rox_image_scan_semaphore_*
+
+
+
+
Existing Metrics
+
+image_scan_semaphore_holding_size - active scansimage_scan_semaphore_queue_size - queued scansdeployment_enhancement_duration_ms - round trip timeresource_processed_count - per resource type
+
+
+
Missing Metrics
+
+- Enrichment call count (by trigger: pipeline vs reprocessor)
+- Enrichment success/failure/skip rate
+- Detection: alerts generated per policy evaluation
+- Detection: violations found vs clean deployments
+- Risk calculation duration per entity type
+- Cache hit rate for enrichment data
+
+
+
+
+
Key Insight: The DedupingQueue Already Has Metric Hooks
+
Both pkg/queue/Queue[T] and pkg/dedupingqueue/DedupingQueue[K] already accept optional Prometheus metrics via functional options:
+
+// Queue[T] accepts:
+WithCounterVec[T](vec *prometheus.CounterVec) // tracks Add/Remove
+WithDroppedMetric[T](metric prometheus.Counter) // tracks drops when full
+
+// DedupingQueue[K] accepts:
+WithSizeMetrics[K](metric prometheus.Gauge) // tracks queue depth
+WithOperationMetricsFunc[K](fn func(ops.Op, string)) // tracks Add/Remove/Dedupe
+
+// But NO built-in support for:
+// - Processing duration histograms
+// - In-flight item counts
+// - Error/retry counters
+// - Worker utilization gauges
+