Report event source failures and node coverage in policy status

Agents.md

@@ -94,7 +94,7 @@ is the one-line-per-package index.
 | `api/v1alpha1` | The `RuntimePolicy` CRD: spec, `mode`, and the node-sharded status + conditions. |
 | `pkg/compiler` | Compiles a `RuntimePolicy` into CEL programs and evaluates it into an `EvaluationResult`. Policy-time, not per event. |
 | `pkg/utils` | `Guard(op, fn)` — the panic barrier used at handler fan-out boundaries so one bad handler cannot take out its siblings. |
-| `pkg/controller` | `RuntimePolicy` and `Pod` informers (typed queue keys, lister-fetch-at-process, deletes keyed by UID) plus `StatusWriter`. |
+| `pkg/controller` | `RuntimePolicy` and `Pod` informers (typed queue keys, lister-fetch-at-process, deletes keyed by UID), `StatusWriter`, and `DaemonPlacement` for expected event-source nodes. |
 | `pkg/containers` | Resolves a pod's container cgroup paths/IDs across containerd/CRI-O/Docker and systemd/cgroupfs layouts. |
 | `pkg/bpf/openexec`, `pkg/bpf/egressfilter` | The enforcing eBPF programs: an open/exec dispatcher plus a tail-called policy executor, and a `cgroup_skb/egress` IPv4 filter. Open/exec attaches as BPF-LSM on `file_open`/`bprm_check_security` where the kernel allows it and as `fmod_ret` on `security_file_open` otherwise. Policies are map entries, not programs. Both map-driven, plus per-cgroup observation counters. |
 | `pkg/bpf/exectrace` | Observation-only `raw_tp/sched_process_exec` program streaming per-exec events with argv over a ring buffer; a `runtimeevent.Source`. |
@@ -136,6 +136,9 @@ The filtering rules that apply to that pipeline:
   it is counted: a silent drop hides an attribution regression.
 - Buffer-full drops are likewise counted, labeled by source and reason. Never add a drop path
   without a counter.
+- Register source availability before initialization, and signal readiness only when collection
+  is usable. Poll and ring-buffer source failures reach metrics and relevant policies' node-sharded
+  `EventSourcesAvailable`; a missing expected daemon report is unknown, never healthy.
 - `open`/`exec` observations are kept even when metadata is sparse, so long as the pod is known.
 - Egress observation is destination-IPv4 only. It does see flows a default-deny drops: the BPF
   program computes its decision, records it, and only then returns, and the decision is part of the

api/v1alpha1/runtimepolicy_types.go

@@ -212,6 +212,13 @@ const (
 	ConditionPodsMatched = "PodsMatched"
 	ReasonNoMatchingPods = "NoMatchingPods"
 	ReasonPodsMatched    = "PodsMatched"
+
+	// ConditionEventSourcesAvailable reports whether every event source a
+	// monitor-mode policy needs is ready on every daemon.
+	ConditionEventSourcesAvailable = "EventSourcesAvailable"
+	ReasonEventSourcesAvailable    = "EventSourcesAvailable"
+	ReasonEventSourcesUnavailable  = "EventSourcesUnavailable"
+	ReasonEventSourcesUnknown      = "EventSourcesUnknown"
 )
 
 // PodsMatchedCondition builds the PodsMatched condition every manager records,
@@ -262,6 +269,24 @@ type NodePolicyStatus struct {
 	// Message explains a false EnforcementAvailable or ObservationAvailable.
 	// +optional
 	Message string `json:"message,omitempty"`
+
+	// EventSources reports the event sources this monitor-mode policy needs on
+	// this node.
+	// +optional
+	// +listType=map
+	// +listMapKey=name
+	EventSources []EventSourceStatus `json:"eventSources,omitempty"`
+}
+
+// EventSourceStatus is one source's lifecycle status on a daemon node.
+type EventSourceStatus struct {
+	Name string `json:"name"`
+
+	Status metav1.ConditionStatus `json:"status"`
+
+	Reason string `json:"reason"`
+
+	Message string `json:"message"`
 }
 
 // +genclient

charts/kyverno-runtime/crds/runtime.nirmata.io_runtimepolicies.yaml

@@ -523,6 +523,32 @@ spec:
                         EnforcementAvailable reports whether the kernel programs and maps
                         enforcing this policy are attached and programmed on this node.
                       type: boolean
+                    eventSources:
+                      description: |-
+                        EventSources reports the event sources this monitor-mode policy needs on
+                        this node.
+                      items:
+                        description: EventSourceStatus is one source's lifecycle status
+                          on a daemon node.
+                        properties:
+                          message:
+                            type: string
+                          name:
+                            type: string
+                          reason:
+                            type: string
+                          status:
+                            type: string
+                        required:
+                        - message
+                        - name
+                        - reason
+                        - status
+                        type: object
+                      type: array
+                      x-kubernetes-list-map-keys:
+                      - name
+                      x-kubernetes-list-type: map
                     lastEvaluatedTime:
                       format: date-time
                       type: string

charts/kyverno-runtime/templates/clusterrole.yaml

@@ -18,6 +18,10 @@ rules:
 - apiGroups: [""]
   resources: ["nodes"]
   verbs: ["get", "list", "watch"]
+# daemon: identifies expected event-source nodes from its own deployment
+- apiGroups: ["apps"]
+  resources: ["daemonsets"]
+  verbs: ["get", "list", "watch"]
 - apiGroups: ["runtime.nirmata.io"]
   resources: ["runtimepolicies"]
   verbs: ["get", "list", "watch"]

charts/kyverno-runtime/templates/daemonset.yaml

@@ -78,6 +78,12 @@ spec:
           valueFrom:
             fieldRef:
               fieldPath: spec.nodeName
+        - name: POD_NAMESPACE
+          valueFrom:
+            fieldRef:
+              fieldPath: metadata.namespace
+        - name: DAEMONSET_NAME
+          value: {{ include "kyverno-runtime.fullname" . | quote }}
         ports:
         - name: metrics
           containerPort: {{ .Values.daemon.metrics.port }}

cmd/kyverno-runtime/daemon.go

@@ -22,6 +22,7 @@ import (
 	"github.com/nirmata/runtime/pkg/pushsink"
 	"github.com/nirmata/runtime/pkg/reporter"
 	"github.com/nirmata/runtime/pkg/reportevents"
+	"github.com/nirmata/runtime/pkg/runtimeevent"
 	"github.com/nirmata/runtime/pkg/services"
 	"github.com/nirmata/runtime/pkg/utils"
 
@@ -224,7 +225,6 @@ func runDaemon(cmd *cobra.Command, args []string) error {
 		}
 		return obj, nil
 	})
-	nodeFactory.Start(ctx.Done())
 	nodeGone := func(name string) bool {
 		// before the first sync the watch cannot distinguish a deleted node
 		// from one it has not listed yet
@@ -242,6 +242,30 @@ func runDaemon(cmd *cobra.Command, args []string) error {
 
 	// sw owns this node's shard of every RuntimePolicy status.
 	sw := controller.NewStatusWriter(c, nodeName, controller.DefaultStatusFlushInterval, logger.WithName("statuswriter"), nodeGone, onConditionChanged)
+	sw.SetExpectedSourceNodes(func() controller.ExpectedSourceNodes { return controller.ExpectedSourceNodes{} })
+	if namespace, daemonSetName := os.Getenv("POD_NAMESPACE"), os.Getenv("DAEMONSET_NAME"); namespace != "" && daemonSetName != "" {
+		placement, err := controller.NewDaemonPlacement(k8sClient, namespace, daemonSetName, nodeInformer, sw.MarkAllDirty)
+		if err != nil {
+			return err
+		}
+		sw.SetExpectedSourceNodes(placement.Snapshot)
+		g.Go(func() error { return placement.Run(ctx) })
+	} else {
+		logger.Info("daemon placement unavailable; POD_NAMESPACE and DAEMONSET_NAME are required for event source coverage status")
+		if _, err := nodeInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+			AddFunc: func(any) { sw.MarkAllDirty() }, DeleteFunc: func(any) { sw.MarkAllDirty() },
+		}); err != nil {
+			return err
+		}
+	}
+	nodeFactory.Start(ctx.Done())
+	recordSourceStatus := func(source string, state runtimeevent.SourceState, reason string) {
+		m.RecordSourceStatus(source, state, reason)
+		sw.RecordSourceStatus(source, state, reason)
+	}
+	for _, source := range []string{egressObserveSource, openExecSource, exectrace.SourceName, dnsquery.SourceName} {
+		recordSourceStatus(source, runtimeevent.SourceStateStarting, runtimeevent.SourceReasonStarting)
+	}
 
 	em := egressmgr.NewEgressManager(logger, sw, func(reason string, delta uint64) {
 		m.EventsDropped.WithLabelValues(egressObserveSource, reason).Add(float64(delta))
@@ -254,6 +278,7 @@ func runDaemon(cmd *cobra.Command, args []string) error {
 	execSrc, err := exectrace.New(logger.WithName("exectrace"), observeInterval)
 	if err != nil {
 		logger.Error(err, "exec tracing unavailable; argv will not be observed")
+		recordSourceStatus(exectrace.SourceName, runtimeevent.SourceStateUnavailable, runtimeevent.SourceReasonInitializationFailed)
 		execSrc = nil
 	} else {
 		defer func() { _ = execSrc.Close() }()
@@ -296,7 +321,8 @@ func runDaemon(cmd *cobra.Command, args []string) error {
 	policyHandlers := []events.RuntimePolicyEventHandler{em, sw, mon}
 
 	// Poll the managers' observation maps, attribute, then hand to the monitor.
-	col := collector.New(logger.WithName("collector"), eventBufferSize, sourceRestartBackoff, m)
+	col := collector.New(logger.WithName("collector"), eventBufferSize, sourceRestartBackoff, m,
+		collector.WithSourceStatusFunc(recordSourceStatus))
 	col.AddSource(collector.NewPollSource(egressObserveSource, observeInterval, em.CollectObservations))
 
 	lsmEnabled, err := utils.BpfLSMEnabled()
@@ -310,6 +336,10 @@ func runDaemon(cmd *cobra.Command, args []string) error {
 	}, lsmEnabled, execSinks...)
 	if err != nil {
 		logger.Error(err, "failed to create openexec manager, exec and open enforcement won't work")
+		recordSourceStatus(openExecSource, runtimeevent.SourceStateUnavailable, runtimeevent.SourceReasonInitializationFailed)
+		if execSrc != nil {
+			recordSourceStatus(exectrace.SourceName, runtimeevent.SourceStateUnavailable, runtimeevent.SourceReasonDependencyUnavailable)
+		}
 	} else {
 		podHandlers = append(podHandlers, execMgr)
 		policyHandlers = append(policyHandlers, execMgr)
@@ -318,7 +348,7 @@ func runDaemon(cmd *cobra.Command, args []string) error {
 
 	// A typed nil in the Source interface is not nil, so the check is here
 	// rather than left to AddSource.
-	if execSrc != nil {
+	if execSrc != nil && execMgr != nil {
 		col.AddSource(execSrc)
 	}
 	col.AddStage(attrIdx)
@@ -328,6 +358,7 @@ func runDaemon(cmd *cobra.Command, args []string) error {
 	// cgroup_skb program leaves every other behavior working.
 	if dnsObs, err := dnsquery.New(); err != nil {
 		logger.Error(err, "dns question observation disabled")
+		recordSourceStatus(dnsquery.SourceName, runtimeevent.SourceStateUnavailable, runtimeevent.SourceReasonInitializationFailed)
 	} else {
 		defer func() { _ = dnsObs.Close() }()
 		dm := dnsmgr.New(logger.WithName("dnsmgr"), dnsObs)

docs/dev/DESIGN.md

@@ -837,7 +837,8 @@ an unzeroed tail is a cross-pod argv leak, not untidiness.
 ## Status reporting
 
 `pkg/controller.StatusWriter` is the single writer of `RuntimePolicyStatus` and the single
-implementation of `runtimeevent.PolicyStatusRecorder`. It consumes the policy event stream only;
+implementation of `runtimeevent.PolicyStatusRecorder`. It consumes policy events, source lifecycle
+transitions, and daemon placement changes;
 pod-level detail belongs to the Reports and the Prometheus counters, not to the status.
 
 Because every node runs a daemon and `RuntimePolicy` is cluster-scoped, status is **sharded**:
@@ -873,11 +874,57 @@ shard, and never prunes before its node watch has synced. A node that still exis
 runs a daemon (a taint, an unscheduled DaemonSet) keeps its shard; the watch only answers
 whether the node object is there.
 
+Event sources have their own lifecycle. The daemon registers all four sources before
+attempting to initialize their dependencies. The collector records each start
+and failure, and a source signals readiness only after its reader is usable. A poll source
+waits for its first successful poll, including an empty result, before announcing readiness;
+restarting a failing poller does not establish recovery. Starting and
+unavailable sources expose a zero `source_available` gauge; constructor and reader failures
+increment `source_failures_total` with bounded reasons. Quiet sources remain available without
+needing an event. Constructor failures require a daemon restart; reader failures use the
+collector's restart backoff. The exec tracer also requires a functioning open/exec manager to
+populate its cgroup gate.
+
+Each node shard's `eventSources` list includes every producer relevant to the monitor policy:
+
+| Behavior | Required sources |
+| --- | --- |
+| `open` | `openexec-observe` |
+| `exec` | `openexec-observe`, `exec-trace` |
+| `network`, `protocol` | `egress-observe` |
+| `dns` | `dnsquery` |
+
+The exec tracer represents argv coverage. If that reader fails independently, filename
+observations can still arrive through the open/exec counter source. If its manager dependency
+fails to initialize, neither argv nor filename coverage is available. `EventSourcesAvailable`
+is `False` if any relevant source fails, `Unknown` while a required source or node has not
+reported, and `True` when every expected daemon node reports readiness. Monitor `Applied`
+inherits a false or unknown source condition. Enforce policies do not depend on observation
+source availability, and a monitor policy only depends on the producers of its active behaviors.
+A behavior is active when an allow or deny rule contains literal values or a nonempty CEL
+expression. Empty behaviors and empty rules add no dependency. Expressions retain their source
+dependencies even if one evaluation returns an empty list, because reevaluation can produce
+targets. The API server defaults an omitted mode to `monitor`; an internal spec that bypasses
+defaulting and has a nil mode is not classified as observe mode.
+
+`pkg/controller.DaemonPlacement` watches the daemon's DaemonSet and its owned pods. The chart
+injects `POD_NAMESPACE` and `DAEMONSET_NAME` to identify that deployment. Pod node assignments
+(including the DaemonSet controller's target affinity on pending pods) identify expected nodes;
+`desiredNumberScheduled` accounts for nodes whose pods have not appeared yet. The daemon does
+not duplicate Kubernetes scheduling rules for selectors, affinity, or tolerations. Unobserved
+DaemonSet generations and incomplete inventories produce `Unknown` unless a known source
+failure already requires `False`. Node, DaemonSet, and pod changes dirty policy status, so
+membership changes are reconciled without a policy edit. Completed placement changes exclude
+departed daemon nodes from source aggregation; only Node deletion removes their shards.
+Unavailable placement discovery also produces `Unknown`. These are last-reported source states,
+not heartbeats: a daemon restart replaces its shard, but temporary node unreachability keeps
+the last report. Source readiness does not establish lossless delivery or detect kernel stalls.
+
 `Applied` is derived rather than recorded: `StatusWriter` computes it at flush time from
-`spec.mode` plus the aggregated `EnforcementAvailable` / `ObservationAvailable` for that mode and
-the aggregated `PodsMatched` — a mode that promises enforcement or observation does not read as
+`spec.mode` plus the aggregated `EnforcementAvailable` / `ObservationAvailable` for that mode,
+`EventSourcesAvailable` for relevant monitor policies, and `PodsMatched` — a mode that promises enforcement or observation does not read as
 applied while any node's attachment behind it never took, or while no node has a matching pod.
-The two are checked in that order, so an attachment failure (the more actionable case) is
+The gates are checked in that order, so an attachment failure (the more actionable case) is
 reported ahead of, and is never masked by, a selector that also happens to match nothing at the
 same time. The one direct exception to the derivation is `reportCompileFailure`, which records
 `Applied=False/CompileFailed` itself for a policy the compiler rejected outright — there is no
@@ -913,11 +960,13 @@ has not ticked recently, and is otherwise honest about having nothing else to ch
 `--metrics-addr=:{{ .Values.daemon.metrics.port }}` and declares the matching `containerPort`.
 An empty value disables the endpoint without disabling the counters.
 
-`pkg/metrics/metrics.go` registers exactly six collectors, all under the `nirmata_runtime`
+`pkg/metrics/metrics.go` registers collectors under the `nirmata_runtime`
 namespace: `events_ingested_total{source,kind}`, `events_dropped_total{source,reason}`,
 `attribution_misses_total`, `findings_emitted_total{policy,behavior}`,
 `monitor_filter_eval_errors_total{policy,expression}`, and
-`report_writes_total{result}`. The `reason` values something produces are `buffer_full`
+`report_writes_total{result}`, `source_available{source}`, and
+`source_failures_total{source,reason}`. Source failure reasons are `InitializationFailed`,
+`ReaderFailed`, `UnexpectedExit`, and `DependencyUnavailable`. Drop reasons are `buffer_full`
 (`pkg/collector`), `unattributed` (`pkg/monitor`, `pkg/reporter`),
 `unattributed_kernel_deny` (`pkg/monitor`), `ringbuf_full` / `name_unreadable` /
 `undecodable` (`pkg/bpf/dnsquery`, all under `source="dnsquery"`), and `queue_full` /

docs/dev/DEVELOPMENT.md

@@ -86,6 +86,23 @@ make test-e2e-protocol # protocol enforcement behavior
 `make kind-install` rebuilds and reloads the image every time. Running only the Chainsaw suites
 validates whatever image was loaded last.
 
+### Validating event source availability
+
+The chart supplies `POD_NAMESPACE` and `DAEMONSET_NAME` and grants read access to DaemonSets.
+Custom daemon manifests must supply the same identity and permissions for expected-node
+discovery; otherwise relevant monitor policies report `EventSourcesAvailable=Unknown`.
+
+After `make kind-install`, create a monitor policy with exec and DNS behaviors, then inspect
+`status.nodes[*].eventSources`, `EventSourcesAvailable`, and the daemon's
+`nirmata_runtime_source_available` metrics. In an isolated kind cluster, temporarily removing
+the daemon's BPF privileges exercises constructor failures without a production fault flag;
+restore its exact security context and wait for rollout before checking recovery. Constructor
+failures require restarting the daemon, while reader failures are covered by the collector's
+deterministic retry tests. Dependency tests cover every behavior's poll or ring-buffer producer,
+empty rules, expression-backed rules, and internal specs without a defaulted mode. The placement
+tests cover pending target affinity, DaemonSet ownership,
+rollout deduplication, deletion, and unobserved placement generations.
+
 ### Validating the push sink
 
 `hack/pushsink-testcollector` is a dev-only test double for the findings push sink