diff --git a/Agents.md b/Agents.md index 59e1a3d0..f63945f6 100644 --- a/Agents.md +++ b/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 diff --git a/api/v1alpha1/runtimepolicy_types.go b/api/v1alpha1/runtimepolicy_types.go index 6babf865..1f5470cf 100644 --- a/api/v1alpha1/runtimepolicy_types.go +++ b/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 diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index 5e6971a9..fb91a991 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -71,6 +71,21 @@ func (in *BehaviorRule) DeepCopy() *BehaviorRule { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *EventSourceStatus) DeepCopyInto(out *EventSourceStatus) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new EventSourceStatus. +func (in *EventSourceStatus) DeepCopy() *EventSourceStatus { + if in == nil { + return nil + } + out := new(EventSourceStatus) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *MonitorFilter) DeepCopyInto(out *MonitorFilter) { *out = *in @@ -128,6 +143,11 @@ func (in *NodePolicyStatus) DeepCopyInto(out *NodePolicyStatus) { *out = new(bool) **out = **in } + if in.EventSources != nil { + in, out := &in.EventSources, &out.EventSources + *out = make([]EventSourceStatus, len(*in)) + copy(*out, *in) + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new NodePolicyStatus. diff --git a/charts/kyverno-runtime/crds/runtime.nirmata.io_runtimepolicies.yaml b/charts/kyverno-runtime/crds/runtime.nirmata.io_runtimepolicies.yaml index 712a62be..d97a113e 100644 --- a/charts/kyverno-runtime/crds/runtime.nirmata.io_runtimepolicies.yaml +++ b/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 diff --git a/charts/kyverno-runtime/templates/clusterrole.yaml b/charts/kyverno-runtime/templates/clusterrole.yaml index c2db082f..8d5198ab 100644 --- a/charts/kyverno-runtime/templates/clusterrole.yaml +++ b/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"] diff --git a/charts/kyverno-runtime/templates/daemonset.yaml b/charts/kyverno-runtime/templates/daemonset.yaml index 1541124d..d1468b7b 100644 --- a/charts/kyverno-runtime/templates/daemonset.yaml +++ b/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 }} diff --git a/cmd/kyverno-runtime/daemon.go b/cmd/kyverno-runtime/daemon.go index 3658b157..f850d9b1 100644 --- a/cmd/kyverno-runtime/daemon.go +++ b/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) diff --git a/docs/dev/DESIGN.md b/docs/dev/DESIGN.md index 6ddd0cea..f07a5d55 100644 --- a/docs/dev/DESIGN.md +++ b/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` / diff --git a/docs/dev/DEVELOPMENT.md b/docs/dev/DEVELOPMENT.md index f1b40e81..83a9a2ad 100644 --- a/docs/dev/DEVELOPMENT.md +++ b/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 diff --git a/docs/users/reference/metrics.md b/docs/users/reference/metrics.md index 6df955ed..0fe89a00 100644 --- a/docs/users/reference/metrics.md +++ b/docs/users/reference/metrics.md @@ -4,7 +4,7 @@ The daemon serves Prometheus metrics on `--metrics-addr` (default `:9090`, set b `daemon.metrics.port`). Setting the flag to the empty string disables the endpoint; the counters themselves keep working. -## Counters +## Available metrics Every metric is prefixed `nirmata_runtime_`. @@ -12,6 +12,8 @@ Every metric is prefixed `nirmata_runtime_`. | --- | --- | --- | | `nirmata_runtime_events_ingested_total` | `source`, `kind` | Observations ingested by the collector. | | `nirmata_runtime_events_dropped_total` | `source`, `reason` | Dropped observations. | +| `nirmata_runtime_source_available` | `source` | Gauge: `1` after the source is ready to collect, `0` while starting or unavailable. Series exist even when initialization fails. | +| `nirmata_runtime_source_failures_total` | `source`, `reason` | Source initialization failures and collector restart failures. | | `nirmata_runtime_attribution_misses_total` | — | Observations that could not be tied to a pod. | | `nirmata_runtime_findings_emitted_total` | `policy`, `behavior` | Findings handed to the reporter. | | `nirmata_runtime_monitor_filter_eval_errors_total` | `policy`, `expression` | `spec.monitorFilter` expressions that failed to evaluate. The finding is reported anyway. | @@ -26,7 +28,7 @@ Label values: | Label | Values | | --- | --- | -| `source` | `egress-observe`, `lsm-observe` (the two poll sources), `dnsquery` (the DNS question source), `exec-trace` (the streamed exec source), `monitor`, `reporter` | +| `source` | `egress-observe`, `openexec-observe` (the two poll sources), `dnsquery` (the DNS question source), `exec-trace` (the streamed exec source), `monitor`, `reporter` | | `kind` | `net`, `protocol`, `exec`, `open`, `dns` | | `reason` | `buffer_full`, `unattributed`, `unattributed_kernel_deny`, `count_map_full`, `ringbuf_full`, `name_unreadable`, `undecodable`, `queue_full`, `send_failed` | | `behavior` | `network`, `protocol`, `exec`, `open`, `dns` | @@ -48,6 +50,35 @@ The pipeline-wide drop reasons: poll interval than the map holds (2048), so narrow the `podSelector` of the policies selecting it, or accept the gap knowingly. +## Source availability + +All four collector sources expose `nirmata_runtime_source_available`. A quiet source can be +healthy: readiness means its collection machinery is usable, independent of event volume. +A poll source announces readiness after its first successful poll, even if it returns no events. +The collector's `/healthz` endpoint checks its dispatch loop and policy cache; individual +source failures do not fail that endpoint. + +The `reason` labels on `nirmata_runtime_source_failures_total` are: + +| Reason | Meaning | +| --- | --- | +| `InitializationFailed` | Kernel resources could not be loaded. Fix the cause in the daemon logs and restart the daemon. | +| `ReaderFailed` | A source run failed. The collector retries after `--source-restart-backoff`. | +| `UnexpectedExit` | A source returned while the daemon was still running. The collector retries it. | +| `DependencyUnavailable` | The exec tracer's open/exec manager could not load, so the tracer's cgroup gate cannot be populated. | + +Normal shutdown does not increment failures. A retry restores the availability gauge only +after the source confirms readiness. Failure series appear when that failure occurs; gauge +series are initialized before source construction. Availability does not prove lossless +delivery or detect every kernel-side stall; continue checking the drop counters separately. + +Monitor policies expose `EventSourcesAvailable` and per-node `eventSources` for their active +behaviors in [policy status](runtimepolicy.md#status). Open and exec depend on +`openexec-observe`, network and protocol depend on `egress-observe`, DNS depends on `dnsquery`, +and exec additionally depends on `exec-trace`. Losing only `exec-trace` removes argv coverage; +filename observations can remain available through `openexec-observe`. A failure to initialize +the open/exec manager removes both forms of exec coverage. + ## DNS question loss The DNS question source counts three ways a question can be lost, all under @@ -116,7 +147,8 @@ What to look at: - `nirmata_runtime_findings_emitted_total` staying at zero while a `monitor` policy is applied means nothing matched, or the observation path is not producing — check the - `ObservationAvailable` condition on the policy. + `ObservationAvailable` and `EventSourcesAvailable` conditions on the policy and the + source availability gauges. - `nirmata_runtime_attribution_misses_total` rising steadily is expected on a busy node: node and host-process activity is never attributed to a pod. A step change alongside missing findings for a specific workload is not. diff --git a/docs/users/reference/runtimepolicy.md b/docs/users/reference/runtimepolicy.md index 1834405a..458f8794 100644 --- a/docs/users/reference/runtimepolicy.md +++ b/docs/users/reference/runtimepolicy.md @@ -734,9 +734,9 @@ program must never inherit deny entries and an enforcing one must not start from `status` is written per node. Each daemon owns exactly one entry in `status.nodes` (keyed by `nodeName`) and never touches another node's entry; each shard carries that node's compact -answers — `enforcementAvailable`, `observationAvailable`, `podsMatched`, and a `message` naming +answers — `enforcementAvailable`, `observationAvailable`, `eventSources`, `podsMatched`, and a `message` naming what is unavailable there. `status.lastEvaluatedTime` is the newest shard timestamp, and the -cluster-scoped `Applied`, `EnforcementAvailable`, `ObservationAvailable` and `PodsMatched` +cluster-scoped `Applied`, `EnforcementAvailable`, `ObservationAvailable`, `EventSourcesAvailable` and `PodsMatched` conditions are derived from the shards, so on a mixed cluster the top-level value states something true of the cluster rather than of whichever node wrote last. Updates are flushed every 30 seconds with conflict retry. @@ -769,11 +769,12 @@ Conditions: | Type | Reasons | Meaning | | --- | --- | --- | -| `Applied` | `Enforcing`, `Monitoring`, `NoMode`, `CompileFailed`, `EnforcementUnavailable`, `ObservationUnavailable`, `NoMatchingPods` | Whether the daemon has the policy loaded, and in which mode. `NoMode` reports `False` for a policy that omits `spec.mode`, which is neither enforced nor reported. `CompileFailed` reports `False` when the spec could not be compiled, with the offending field path and value in the message; nothing in such a policy is applied, including the rules either side of the bad one. `EnforcementUnavailable` / `ObservationUnavailable` report `False` when the `EnforcementAvailable` / `ObservationAvailable` condition (below) is `False` for the policy's mode: a mode that promises enforcement or observation does not read as applied while any node's attachment behind it never took. `NoMatchingPods` reports `False` when the `PodsMatched` condition (below) is `False` — no node has a matching pod — and the mode's own `EnforcementAvailable` / `ObservationAvailable` is not — that is, is either `True` or was never recorded at all, which is the normal case for a policy whose behaviors never hit a programming failure. When both conditions are `False` at once, `EnforcementUnavailable` / `ObservationUnavailable` takes priority and `NoMatchingPods` is not reported. | +| `Applied` | `Enforcing`, `Monitoring`, `NoMode`, `CompileFailed`, `EnforcementUnavailable`, `ObservationUnavailable`, `EventSourcesUnavailable`, `EventSourcesUnknown`, `NoMatchingPods` | Whether the daemon has the policy loaded, and in which mode. `NoMode` reports `False` for an internal policy constructed without a mode; the API server defaults an omitted `spec.mode` to `monitor`. `CompileFailed` reports `False` when the spec could not be compiled, with the offending field path and value in the message; nothing in such a policy is applied, including the rules either side of the bad one. `EnforcementUnavailable` / `ObservationUnavailable` report `False` when the `EnforcementAvailable` / `ObservationAvailable` condition (below) is `False` for the policy's mode: a mode that promises enforcement or observation does not read as applied while any node's attachment behind it never took. `NoMatchingPods` reports `False` when the `PodsMatched` condition (below) is `False` — no node has a matching pod — and the mode's own `EnforcementAvailable` / `ObservationAvailable` is not — that is, is either `True` or was never recorded at all, which is the normal case for a policy whose behaviors never hit a programming failure. When both conditions are `False` at once, `EnforcementUnavailable` / `ObservationUnavailable` takes priority and `NoMatchingPods` is not reported. | | `TargetsValid` | `AllTargetsSupported`, `NoTargets`, `UnsupportedTargets`, `UnresolvedServices` | Whether every `network` and `protocol` target could be programmed. `UnsupportedTargets` lists the rejected values and why; `UnresolvedServices` lists the Service and endpoint names that are not in cache. | | `ExecRulesValid` | `AllPathsSupported`, `NoPaths`, `UnsupportedPaths` | Whether every `exec` path could be programmed. `UnsupportedPaths` lists the rejected values and why. | | `OpenRulesValid` | `AllPathsSupported`, `NoPaths`, `UnsupportedPaths` | Whether every `open` path could be programmed. | | `ObservationAvailable` | `ObservationAvailable`, `ObservationUnavailable` | Set to `False` when observation could not be attached on at least one node — a node not booted with `lsm=bpf`, or a loaded LSM program with no observation maps — with the failing nodes and their causes named in the message; a monitor-mode policy on such a node silently produces no findings until this clears. Set to `True` when every node reporting it has observation attached. Each node's own answer is `status.nodes[*].observationAvailable`. | +| `EventSourcesAvailable` | `EventSourcesAvailable`, `EventSourcesUnavailable`, `EventSourcesUnknown` | Availability of poll and ring-buffer sources required by the monitor policy's active behaviors. `False` identifies source failures, `Unknown` means an expected daemon or source has not reported, and `True` means all expected nodes report ready sources. A false or unknown value also gates monitor `Applied`. Each node records source name, status, reason, and message in `status.nodes[*].eventSources`. | | `EnforcementAvailable` | `EnforcementAvailable`, `EnforcementUnavailable` | Set to `False` when a kernel map could not be programmed or attached on at least one node — a node not booted with `lsm=bpf`, a full map, a failed update — with the failing nodes and their causes named in the message, so part of the policy is not enforced there. Set to `True` when every node reporting it has enforcement programmed. Each node's own answer is `status.nodes[*].enforcementAvailable`. | | `PodsMatched` | `PodsMatched`, `NoMatchingPods` | Whether any node currently has a pod selected by `spec.podSelector` / `spec.namespaceSelector`. `NoMatchingPods` catches a selector that is well-formed but matches nothing anywhere — otherwise indistinguishable from a policy that is enforcing on pods that simply never triggered it. Nodes where none of the policy's pods are scheduled do not make it `False`. Each node's own answer is `status.nodes[*].podsMatched`. | @@ -783,6 +784,35 @@ and a bad one reports `Applied=False` with `CompileFailed`, while a value an `ex produced is checked when it is programmed and reports the per-behavior condition. Either way it also reaches an operator-visible log line. +Source coverage follows the daemon's DaemonSet placement. The controller's desired pod count +and its pods' target nodes identify expected reporters, including pending daemons. A newly +expected node without a source report makes coverage `Unknown`; a known failure still takes +priority and keeps it `False`. Node additions, deletions, and daemon placement changes trigger +status reconciliation without a policy edit. Deleted nodes' shards are removed, and nodes +excluded by a completed placement change stop contributing to source coverage. During incomplete +placement discovery, retained failures remain visible. If discovery is unavailable or the +DaemonSet has no expected nodes, coverage is `Unknown` rather than healthy. +Membership changes converge after the DaemonSet controller and informer caches update, followed +by the status flush interval; a stale desired count can temporarily keep coverage `Unknown` +after a node has been deleted. + +Required producers are `openexec-observe` for open and exec, `egress-observe` for network and +protocol, `dnsquery` for DNS, and additionally `exec-trace` for exec argv. +A behavior contributes dependencies only when an allow or deny rule contains literal values or +a nonempty CEL expression. Empty behaviors and empty rules contribute none. Expressions retain +their dependencies when an evaluation returns an empty list, since reevaluation can produce +targets. This is a conservative coverage check, independent of the current finding count. + +These are last-reported states, not a daemon heartbeat. A temporarily unreachable node retains +its last report; a restarted daemon replaces its node's source states. An available source does +not guarantee lossless observations or detect every kernel stall. Losing `exec-trace` removes +argv coverage while filename observations may continue; losing `openexec-observe` removes +file-open and exec filename counter observations, and losing `egress-observe` removes network and protocol +observations. A failure to initialize the open/exec manager also prevents argv coverage because +the tracer's cgroup gate cannot be populated. Losing `dnsquery` removes DNS question findings. +Observation source availability does not gate enforcement. Constructor failures require a daemon +restart after correcting the logged cause; reader failures retry automatically. + ## Findings and Reports Findings — monitor-mode matches and enforce-mode denials alike — are written as @@ -978,6 +1008,11 @@ exception in shape — a program of its own, streamed rather than counted — an preserved — not the ordering or timing of individual occurrences within a window. A `dns` question is a single record delivered as it happens, so only the reporter's flush interval applies to it. +- **Observation sources can be unavailable.** Open and exec filename counters use + `openexec-observe`; network and protocol use `egress-observe`. Exec argv uses `exec-trace`, + a streamed source separate from filename counters; DNS questions use `dnsquery`. + `EventSourcesAvailable` and source availability metrics report initialization and reader failures. Their last-known + readiness is not a heartbeat and does not guarantee lossless observation. - **Open/exec path counters cap per cgroup.** The per-cgroup path map holds 2048 distinct `(path, decision)` keys; a workload touching more than that within one poll interval loses the excess. The read-and-reset drain mitigates this but does not eliminate it. diff --git a/pkg/bpf/dnsquery/source.go b/pkg/bpf/dnsquery/source.go index 49962c5d..c6a6bc9a 100644 --- a/pkg/bpf/dnsquery/source.go +++ b/pkg/bpf/dnsquery/source.go @@ -75,6 +75,7 @@ func (s *Source) Run(ctx context.Context, out chan<- runtimeevent.Event) error { if err != nil { return fmt.Errorf("%s: open ring buffer: %w", SourceName, err) } + runtimeevent.SourceReady(ctx) // Read blocks in the kernel and does not observe ctx, so closing the reader // is what unblocks it. diff --git a/pkg/bpf/exectrace/source.go b/pkg/bpf/exectrace/source.go index fa9fc02f..ad549daf 100644 --- a/pkg/bpf/exectrace/source.go +++ b/pkg/bpf/exectrace/source.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "os" + "sync" "time" "github.com/nirmata/runtime/pkg/runtimeevent" @@ -47,10 +48,18 @@ type Source struct { statInterval time.Duration objs execTraceObjects link link.Link - rd *ringbuf.Reader + mu sync.Mutex + rd ringReader + closed bool + newReader func() (ringReader, error) clock func() time.Time } +type ringReader interface { + Read() (ringbuf.Record, error) + Close() error +} + // New loads and attaches the kernel program. The caller owns Close. func New(log logr.Logger, statInterval time.Duration) (*Source, error) { if statInterval <= 0 { @@ -72,13 +81,9 @@ func New(log logr.Logger, statInterval time.Duration) (*Source, error) { } s.link = l - rd, err := ringbuf.NewReader(s.objs.Events) - if err != nil { - _ = l.Close() - _ = s.objs.Close() - return nil, fmt.Errorf("%s: opening ring buffer: %w", SourceName, err) + s.newReader = func() (ringReader, error) { + return ringbuf.NewReader(s.objs.Events) } - s.rd = rd return s, nil } @@ -110,21 +115,38 @@ func (s *Source) DeleteCgids(cgids []uint64) error { // Run drains the ring buffer until ctx is done. func (s *Source) Run(ctx context.Context, out chan<- runtimeevent.Event) error { + // Reader creation and publication share Close's lock so teardown cannot + // miss a reader or close its map while it is being opened. + s.mu.Lock() + if s.closed { + s.mu.Unlock() + return fmt.Errorf("%s: source is closed: %w", SourceName, os.ErrClosed) + } + rd, err := s.newReader() + if err != nil { + s.mu.Unlock() + return fmt.Errorf("%s: opening ring buffer: %w", SourceName, err) + } + s.rd = rd + s.mu.Unlock() + defer func() { _ = rd.Close() }() + // ringbuf.Read has no deadline; closing the reader is what unblocks it. done := make(chan struct{}) defer close(done) go func() { select { case <-ctx.Done(): - _ = s.rd.Close() + _ = rd.Close() case <-done: } }() go s.pollStats(ctx, done) + runtimeevent.SourceReady(ctx) for { - rec, err := s.rd.Read() + rec, err := rd.Read() if err != nil { if errors.Is(err, ringbuf.ErrClosed) || ctx.Err() != nil { return nil @@ -204,6 +226,13 @@ func (s *Source) readStats() ([statCount]uint64, error) { } func (s *Source) Close() error { + s.mu.Lock() + defer s.mu.Unlock() + if s.closed { + return nil + } + s.closed = true + var errs []error if s.rd != nil { if err := s.rd.Close(); err != nil && !errors.Is(err, os.ErrClosed) { diff --git a/pkg/bpf/exectrace/source_test.go b/pkg/bpf/exectrace/source_test.go new file mode 100644 index 00000000..ef7af8ce --- /dev/null +++ b/pkg/bpf/exectrace/source_test.go @@ -0,0 +1,178 @@ +package exectrace + +import ( + "context" + "errors" + "os" + "sync" + "testing" + "time" + + "github.com/nirmata/runtime/pkg/runtimeevent" + + "github.com/cilium/ebpf/ringbuf" + "github.com/go-logr/logr" +) + +type fakeRingReader struct { + err error + closed chan struct{} + once sync.Once +} + +func newFakeRingReader(err error) *fakeRingReader { + return &fakeRingReader{err: err, closed: make(chan struct{})} +} + +func (r *fakeRingReader) Read() (ringbuf.Record, error) { + if r.err != nil { + return ringbuf.Record{}, r.err + } + <-r.closed + return ringbuf.Record{}, ringbuf.ErrClosed +} + +func (r *fakeRingReader) Close() error { + r.once.Do(func() { close(r.closed) }) + return nil +} + +// TestSourceRunReopensReaderBeforeAnnouncingReady prevents a failed reader from +// being advertised as recovered without a usable replacement. +func TestSourceRunReopensReaderBeforeAnnouncingReady(t *testing.T) { + readers := []ringReader{ + newFakeRingReader(errors.New("reader failed")), + newFakeRingReader(nil), + } + opened := 0 + s := &Source{ + log: logr.Discard(), + statInterval: time.Hour, + clock: time.Now, + newReader: func() (ringReader, error) { + r := readers[opened] + opened++ + return r, nil + }, + } + ready := make(chan struct{}, 2) + ctx := runtimeevent.WithSourceReady(context.Background(), func() { ready <- struct{}{} }) + if err := s.Run(ctx, make(chan runtimeevent.Event)); err == nil { + t.Fatal("first Run returned nil, want reader error") + } + select { + case <-ready: + case <-time.After(time.Second): + t.Fatal("first reader did not announce ready") + } + + secondCtx, cancel := context.WithCancel(runtimeevent.WithSourceReady(context.Background(), func() { ready <- struct{}{} })) + done := make(chan error, 1) + go func() { done <- s.Run(secondCtx, make(chan runtimeevent.Event)) }() + select { + case <-ready: + case <-time.After(time.Second): + t.Fatal("replacement reader did not announce ready") + } + cancel() + select { + case err := <-done: + if err != nil { + t.Errorf("second Run = %v, want nil after cancellation", err) + } + case <-time.After(time.Second): + t.Fatal("second Run did not return after cancellation") + } + if opened != 2 { + t.Errorf("opened readers = %d, want 2", opened) + } +} + +// TestSourceRunDoesNotAnnounceReadyWhenReaderCannotOpen keeps an unsuccessful +// retry from clearing a source failure. +func TestSourceRunDoesNotAnnounceReadyWhenReaderCannotOpen(t *testing.T) { + s := &Source{ + newReader: func() (ringReader, error) { + return nil, errors.New("cannot open reader") + }, + } + ready := make(chan struct{}, 1) + ctx := runtimeevent.WithSourceReady(context.Background(), func() { ready <- struct{}{} }) + if err := s.Run(ctx, make(chan runtimeevent.Event)); err == nil { + t.Fatal("Run returned nil, want reader-open error") + } + select { + case <-ready: + t.Error("reader-open failure announced readiness") + default: + } +} + +// TestSourceCloseSerializesReaderPublication keeps concurrent teardown from +// missing a reader opened by a source run or allowing one to open afterward. +func TestSourceCloseSerializesReaderPublication(t *testing.T) { + for _, timing := range []string{"before run", "during reader creation", "while reading"} { + t.Run(timing, func(t *testing.T) { + rd := newFakeRingReader(nil) + creating, release := make(chan struct{}), make(chan struct{}) + ready := make(chan struct{}, 1) + s := &Source{ + log: logr.Discard(), statInterval: time.Hour, clock: time.Now, + newReader: func() (ringReader, error) { + close(creating) + <-release + return rd, nil + }, + } + if timing == "before run" { + if err := s.Close(); err != nil { + t.Fatal(err) + } + close(release) + } + ctx, cancel := context.WithCancel(runtimeevent.WithSourceReady(context.Background(), func() { ready <- struct{}{} })) + defer cancel() + done := make(chan error, 1) + go func() { done <- s.Run(ctx, make(chan runtimeevent.Event)) }() + + if timing != "before run" { + <-creating + if timing == "while reading" { + close(release) + <-ready + } + started, closed := make(chan struct{}), make(chan error, 2) + go func() { close(started); closed <- s.Close() }() + <-started + go func() { closed <- s.Close() }() + if timing == "during reader creation" { + close(release) + } + for range 2 { + if err := <-closed; err != nil { + t.Errorf("Close = %v", err) + } + } + } + select { + case err := <-done: + if timing == "before run" { + if !errors.Is(err, os.ErrClosed) { + t.Errorf("Run = %v, want a closed source error", err) + } + select { + case <-creating: + t.Error("a closed source opened a reader") + default: + } + } else if err != nil { + t.Errorf("Run = %v, want nil after Close", err) + } + case <-time.After(time.Second): + cancel() + <-done + t.Error("Run did not return after Close") + } + }) + } +} diff --git a/pkg/collector/collector.go b/pkg/collector/collector.go index 74f62600..38653d29 100644 --- a/pkg/collector/collector.go +++ b/pkg/collector/collector.go @@ -59,18 +59,28 @@ type Collector struct { // after is the sleep seam used for restart backoff; tests replace it to // keep restart behavior deterministic. Must be set before Run. after func(time.Duration) <-chan time.Time + + sourceStatus runtimeevent.SourceStatusFunc +} + +// Option configures a Collector. +type Option func(*Collector) + +// WithSourceStatusFunc observes the lifecycle of every registered source. +func WithSourceStatusFunc(f runtimeevent.SourceStatusFunc) Option { + return func(c *Collector) { c.sourceStatus = f } } // New builds a Collector. Sources, stages, and sinks are registered separately // so that daemon wiring can be conditional. -func New(log logr.Logger, bufferSize int, backoff time.Duration, m *metrics.Metrics) *Collector { +func New(log logr.Logger, bufferSize int, backoff time.Duration, m *metrics.Metrics, opts ...Option) *Collector { if bufferSize <= 0 { bufferSize = DefaultBufferSize } if backoff <= 0 { backoff = DefaultRestartBackoff } - return &Collector{ + c := &Collector{ log: log, metrics: m, bufferSize: bufferSize, @@ -78,6 +88,12 @@ func New(log logr.Logger, bufferSize int, backoff time.Duration, m *metrics.Metr events: make(chan taggedEvent, bufferSize), after: time.After, } + for _, opt := range opts { + if opt != nil { + opt(c) + } + } + return c } // AddSource registers an event producer. Nil sources are ignored. @@ -177,20 +193,28 @@ func (c *Collector) runSource(ctx context.Context, src runtimeevent.Source, out return } - // Typically a poll source wrapping a single manager: it ticks on its - // own interval, calls that manager's collect function and writes the - // events it returns to out. - err := utils.Guard("collector: source "+name, func() error { - return src.Run(ctx, out) + c.recordSourceStatus(name, runtimeevent.SourceStateStarting, runtimeevent.SourceReasonStarting) + var ready sync.Once + runCtx := runtimeevent.WithSourceReady(ctx, func() { + if ctx.Err() != nil { + return + } + ready.Do(func() { + c.recordSourceStatus(name, runtimeevent.SourceStateAvailable, runtimeevent.SourceReasonReady) + }) }) + err := src.Run(runCtx, out) + switch { case ctx.Err() != nil: return case err == nil: - c.log.V(2).Info("source finished", "source", name) - return + c.recordSourceStatus(name, runtimeevent.SourceStateUnavailable, runtimeevent.SourceReasonUnexpectedExit) + c.log.Error(errors.New("source exited without cancellation"), "source failed; restarting after backoff", + "source", name, "backoff", c.backoff) default: + c.recordSourceStatus(name, runtimeevent.SourceStateUnavailable, runtimeevent.SourceReasonReaderFailed) c.log.Error(err, "source failed; restarting after backoff", "source", name, "backoff", c.backoff) } @@ -203,6 +227,12 @@ func (c *Collector) runSource(ctx context.Context, src runtimeevent.Source, out } } +func (c *Collector) recordSourceStatus(source string, state runtimeevent.SourceState, reason string) { + if c.sourceStatus != nil { + c.sourceStatus(source, state, reason) + } +} + // forward takes the events runSource produced on in and hands them to the // shared fan-in buffer, tagged with the source they came from. func (c *Collector) forward(ctx context.Context, source string, in <-chan runtimeevent.Event) { diff --git a/pkg/collector/collector_test.go b/pkg/collector/collector_test.go index 163b38e5..64585a01 100644 --- a/pkg/collector/collector_test.go +++ b/pkg/collector/collector_test.go @@ -473,13 +473,14 @@ func TestSourceRestartsWithBackoffAfterError(t *testing.T) { } } -func TestSourceReturningNilIsNotRestarted(t *testing.T) { +func TestSourceReturningNilIsRestarted(t *testing.T) { runs := make(chan int, 8) + backoffs := make(chan time.Duration, 8) + fire := make(chan time.Time) c := New(logr.Discard(), 4, time.Millisecond, nil) - c.after = func(time.Duration) <-chan time.Time { - t.Error("backoff scheduled for a source that finished cleanly") - ch := make(chan time.Time) - return ch + c.after = func(d time.Duration) <-chan time.Time { + backoffs <- d + return fire } attempt := 0 c.AddSource(&funcSource{name: "oneshot", run: func(context.Context, chan<- runtimeevent.Event) error { @@ -498,14 +499,112 @@ func TestSourceReturningNilIsNotRestarted(t *testing.T) { if got := recvInt(t, runs); got != 1 { t.Fatalf("run count = %d, want 1", got) } + if got := recvDuration(t, backoffs); got != time.Millisecond { + t.Errorf("restart backoff = %v, want 1ms", got) + } + fire <- time.Now() + if got := recvInt(t, runs); got != 2 { + t.Fatalf("run count = %d, want 2 after unexpected exit", got) + } recvInt(t, blocked) stop() +} + +func TestSourceStatusTracksFailureRecoveryAndNormalShutdown(t *testing.T) { + type status struct { + State runtimeevent.SourceState + Reason string + } + statuses := make(chan status, 8) + runs := make(chan int, 2) + backoff := make(chan time.Duration, 1) + fire := make(chan time.Time, 1) + c := New(logr.Discard(), 4, 42*time.Millisecond, nil, + WithSourceStatusFunc(func(_ string, state runtimeevent.SourceState, reason string) { + statuses <- status{State: state, Reason: reason} + })) + c.after = func(d time.Duration) <-chan time.Time { + backoff <- d + return fire + } + attempt := 0 + c.AddSource(&funcSource{name: "flaky", run: func(ctx context.Context, _ chan<- runtimeevent.Event) error { + attempt++ + runs <- attempt + if attempt == 1 { + return errors.New("reader failed") + } + runtimeevent.SourceReady(ctx) + <-ctx.Done() + return nil + }}) + + _, stop := runCollector(t, c) + if got := recvInt(t, runs); got != 1 { + t.Fatalf("first run = %d, want 1", got) + } + if got := recvDuration(t, backoff); got != 42*time.Millisecond { + t.Errorf("backoff = %v, want 42ms", got) + } + fire <- time.Now() + if got := recvInt(t, runs); got != 2 { + t.Fatalf("second run = %d, want 2", got) + } + stop() + + var got []status + for i := 0; i < 4; i++ { + select { + case s := <-statuses: + got = append(got, s) + default: + t.Fatalf("status %d missing; got %v", i, got) + } + } + want := []status{ + {runtimeevent.SourceStateStarting, runtimeevent.SourceReasonStarting}, + {runtimeevent.SourceStateUnavailable, runtimeevent.SourceReasonReaderFailed}, + {runtimeevent.SourceStateStarting, runtimeevent.SourceReasonStarting}, + {runtimeevent.SourceStateAvailable, runtimeevent.SourceReasonReady}, + } + if diff := cmp.Diff(want, got); diff != "" { + t.Errorf("source statuses (-want +got):\n%s", diff) + } + select { + case s := <-statuses: + t.Errorf("normal shutdown reported unexpected status %+v", s) + default: + } +} + +func TestQuietSourceRemainsStartingUntilItSignalsReady(t *testing.T) { + statuses := make(chan runtimeevent.SourceState, 2) + started := make(chan struct{}, 1) + c := New(logr.Discard(), 4, DefaultRestartBackoff, nil, + WithSourceStatusFunc(func(_ string, state runtimeevent.SourceState, _ string) { statuses <- state })) + c.AddSource(&funcSource{name: "quiet", run: func(ctx context.Context, _ chan<- runtimeevent.Event) error { + started <- struct{}{} + <-ctx.Done() + runtimeevent.SourceReady(ctx) + return nil + }}) + + _, stop := runCollector(t, c) + select { + case <-started: + case <-time.After(testTimeout): + t.Fatal("source did not start") + } + if got := <-statuses; got != runtimeevent.SourceStateStarting { + t.Errorf("first status = %q, want Starting", got) + } select { - case n := <-runs: - t.Fatalf("finished source was restarted (run #%d)", n) + case got := <-statuses: + t.Errorf("quiet source reported %q before readiness", got) default: } + stop() } func TestRunReturnsCleanlyOnContextCancel(t *testing.T) { diff --git a/pkg/collector/pollsource.go b/pkg/collector/pollsource.go index d975e3a4..14729f04 100644 --- a/pkg/collector/pollsource.go +++ b/pkg/collector/pollsource.go @@ -53,6 +53,7 @@ func (p *pollSource) Name() string { return p.name } func (p *pollSource) Run(ctx context.Context, out chan<- runtimeevent.Event) error { tick, stop := p.ticks(p.interval) defer stop() + ready := false for { select { @@ -63,6 +64,10 @@ func (p *pollSource) Run(ctx context.Context, out chan<- runtimeevent.Event) err if err != nil { return fmt.Errorf("polling %s: %w", p.name, err) } + if !ready { + runtimeevent.SourceReady(ctx) + ready = true + } for _, ev := range evs { select { case <-ctx.Done(): diff --git a/pkg/collector/pollsource_test.go b/pkg/collector/pollsource_test.go index 7d29e394..a6d2fa77 100644 --- a/pkg/collector/pollsource_test.go +++ b/pkg/collector/pollsource_test.go @@ -141,6 +141,165 @@ func TestPollSourceReturnsPollError(t *testing.T) { } } +// TestPollSourceReadinessFollowsFirstSuccessfulPoll prevents a failed reader +// retry from claiming recovery before it has read its counter map. +func TestPollSourceReadinessFollowsFirstSuccessfulPoll(t *testing.T) { + cases := []struct { + name string + events []runtimeevent.Event + fail bool + wantReady bool + }{ + {name: "first poll error never announces ready", fail: true}, + {name: "empty first poll announces ready", wantReady: true}, + {name: "event first poll announces ready", events: []runtimeevent.Event{netEvent("event")}, wantReady: true}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + f := newFakeTicker() + polled := make(chan int, 2) + sentinel := errors.New("first poll failed") + ps := newTestPollSource("egress", time.Second, func(context.Context) ([]runtimeevent.Event, error) { + polled <- 1 + if tc.fail { + return nil, sentinel + } + return tc.events, nil + }, f) + ready := make(chan struct{}, 2) + ctx, cancel := context.WithCancel(runtimeevent.WithSourceReady(context.Background(), func() { ready <- struct{}{} })) + defer cancel() + done := make(chan error, 1) + go func() { done <- ps.Run(ctx, make(chan runtimeevent.Event, len(tc.events))) }() + + recvDuration(t, f.interval) + f.c <- time.Now() + recvInt(t, polled) + + if !tc.wantReady { + select { + case err := <-done: + if !errors.Is(err, sentinel) { + t.Errorf("Run error = %v, want %v", err, sentinel) + } + case <-time.After(testTimeout): + t.Fatal("Run did not return after a failed first poll") + } + select { + case <-ready: + t.Error("failed first poll announced readiness") + default: + } + return + } + + select { + case <-ready: + case <-time.After(testTimeout): + t.Fatal("successful first poll did not announce readiness") + } + f.c <- time.Now() + recvInt(t, polled) + cancel() + select { + case err := <-done: + if err != nil { + t.Errorf("Run after cancellation = %v, want nil", err) + } + case <-time.After(testTimeout): + t.Fatal("Run did not return after cancellation") + } + select { + case <-ready: + t.Error("second successful poll announced readiness again") + default: + } + }) + } +} + +// TestPollSourceRecoveryAnnouncesReadinessForNewRun requires each retry to +// establish readiness independently after the preceding run failed. +func TestPollSourceRecoveryAnnouncesReadinessForNewRun(t *testing.T) { + tickers := []*fakeTicker{newFakeTicker(), newFakeTicker()} + opened := 0 + ps := &pollSource{ + name: "egress", + interval: time.Second, + ticks: func(d time.Duration) (<-chan time.Time, func()) { + f := tickers[opened] + opened++ + return f.ticks(d) + }, + } + polls := 0 + ps.poll = func(context.Context) ([]runtimeevent.Event, error) { + polls++ + if polls == 1 { + return nil, errors.New("first reader failure") + } + return nil, nil + } + + type status struct { + State runtimeevent.SourceState + Reason string + } + statuses := make(chan status, 8) + backoff := make(chan time.Duration, 1) + fire := make(chan time.Time, 1) + c := New(logr.Discard(), 1, time.Second, nil, + WithSourceStatusFunc(func(_ string, state runtimeevent.SourceState, reason string) { + statuses <- status{State: state, Reason: reason} + })) + c.after = func(d time.Duration) <-chan time.Time { + backoff <- d + return fire + } + c.AddSource(ps) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { done <- c.Run(ctx) }() + recvDuration(t, tickers[0].interval) + tickers[0].c <- time.Now() + recvDuration(t, backoff) + fire <- time.Now() + recvDuration(t, tickers[1].interval) + tickers[1].c <- time.Now() + + want := []status{ + {runtimeevent.SourceStateStarting, runtimeevent.SourceReasonStarting}, + {runtimeevent.SourceStateUnavailable, runtimeevent.SourceReasonReaderFailed}, + {runtimeevent.SourceStateStarting, runtimeevent.SourceReasonStarting}, + {runtimeevent.SourceStateAvailable, runtimeevent.SourceReasonReady}, + } + var got []status + for range want { + select { + case s := <-statuses: + got = append(got, s) + case <-time.After(testTimeout): + t.Fatalf("missing lifecycle status; got %v", got) + } + } + if diff := cmp.Diff(want, got); diff != "" { + t.Errorf("statuses (-want +got):\n%s", diff) + } + + cancel() + select { + case err := <-done: + if err != nil { + t.Errorf("collector Run = %v, want nil", err) + } + case <-time.After(testTimeout): + t.Fatal("collector did not stop after cancellation") + } +} + func TestNewPollSourceWithNilPollFuncPanics(t *testing.T) { defer func() { if recover() == nil { diff --git a/pkg/controller/daemonplacement.go b/pkg/controller/daemonplacement.go new file mode 100644 index 00000000..f0233146 --- /dev/null +++ b/pkg/controller/daemonplacement.go @@ -0,0 +1,176 @@ +package controller + +import ( + "context" + "sort" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + apiequality "k8s.io/apimachinery/pkg/api/equality" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/fields" + "k8s.io/client-go/informers" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/tools/cache" +) + +type DaemonPlacement struct { + namespace string + name string + daemonSets cache.SharedIndexInformer + pods cache.SharedIndexInformer + nodes cache.SharedIndexInformer + onChange func() +} + +func NewDaemonPlacement(client kubernetes.Interface, namespace, name string, + nodes cache.SharedIndexInformer, onChange func()) (*DaemonPlacement, error) { + dsFactory := informers.NewSharedInformerFactoryWithOptions(client, 0, + informers.WithNamespace(namespace), informers.WithTweakListOptions(func(opts *metav1.ListOptions) { + opts.FieldSelector = fields.OneTermEqualSelector("metadata.name", name).String() + })) + podFactory := informers.NewSharedInformerFactoryWithOptions(client, 0, informers.WithNamespace(namespace)) + p := &DaemonPlacement{ + namespace: namespace, name: name, nodes: nodes, onChange: onChange, + daemonSets: dsFactory.Apps().V1().DaemonSets().Informer(), + pods: podFactory.Core().V1().Pods().Informer(), + } + if err := p.daemonSets.SetTransform(func(obj any) (any, error) { + if ds, ok := obj.(*appsv1.DaemonSet); ok { + return &appsv1.DaemonSet{ + ObjectMeta: metav1.ObjectMeta{Name: ds.Name, Namespace: ds.Namespace, UID: ds.UID, + Generation: ds.Generation, DeletionTimestamp: ds.DeletionTimestamp}, + Status: appsv1.DaemonSetStatus{DesiredNumberScheduled: ds.Status.DesiredNumberScheduled, + ObservedGeneration: ds.Status.ObservedGeneration}, + }, nil + } + return obj, nil + }); err != nil { + return nil, err + } + if err := p.pods.SetTransform(func(obj any) (any, error) { + if pod, ok := obj.(*corev1.Pod); ok { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: pod.Name, Namespace: pod.Namespace, UID: pod.UID, + OwnerReferences: pod.OwnerReferences, DeletionTimestamp: pod.DeletionTimestamp, + }, + Spec: corev1.PodSpec{NodeName: pod.Spec.NodeName, Affinity: pod.Spec.Affinity}, + Status: corev1.PodStatus{Phase: pod.Status.Phase}, + }, nil + } + return obj, nil + }); err != nil { + return nil, err + } + changed := func(obj any) { + if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok { + obj = tombstone.Obj + } + if pod, ok := obj.(*corev1.Pod); ok { + owner := metav1.GetControllerOf(pod) + if owner == nil || owner.Kind != "DaemonSet" || owner.Name != name { + return + } + } + onChange() + } + handler := cache.ResourceEventHandlerFuncs{ + AddFunc: changed, DeleteFunc: changed, + UpdateFunc: func(old, cur any) { + if !apiequality.Semantic.DeepEqual(old, cur) { + changed(old) + changed(cur) + } + }, + } + for _, informer := range []cache.SharedIndexInformer{p.daemonSets, p.pods, nodes} { + if _, err := informer.AddEventHandler(handler); err != nil { + return nil, err + } + } + return p, nil +} + +func (p *DaemonPlacement) Run(ctx context.Context) error { + go p.daemonSets.Run(ctx.Done()) + go p.pods.Run(ctx.Done()) + if cache.WaitForCacheSync(ctx.Done(), p.daemonSets.HasSynced, p.pods.HasSynced, p.nodes.HasSynced) { + p.onChange() + } + <-ctx.Done() + return nil +} + +func (p *DaemonPlacement) Snapshot() ExpectedSourceNodes { + if !p.daemonSets.HasSynced() || !p.pods.HasSynced() || !p.nodes.HasSynced() { + return ExpectedSourceNodes{} + } + obj, exists, err := p.daemonSets.GetStore().GetByKey(p.namespace + "/" + p.name) + if err != nil || !exists { + return ExpectedSourceNodes{} + } + ds, ok := obj.(*appsv1.DaemonSet) + if !ok { + return ExpectedSourceNodes{} + } + var pods []*corev1.Pod + for _, obj := range p.pods.GetStore().List() { + if pod, ok := obj.(*corev1.Pod); ok { + pods = append(pods, pod) + } + } + return daemonPlacementNodes(ds, pods, func(name string) bool { + _, exists, err := p.nodes.GetStore().GetByKey(name) + return err == nil && exists + }) +} + +func daemonPlacementNodes(ds *appsv1.DaemonSet, pods []*corev1.Pod, nodeExists func(string) bool) ExpectedSourceNodes { + result := ExpectedSourceNodes{ + Desired: int(ds.Status.DesiredNumberScheduled), + Synced: ds.Status.ObservedGeneration >= ds.Generation && ds.DeletionTimestamp == nil, + } + names := make(map[string]struct{}) + for _, pod := range pods { + owner := metav1.GetControllerOf(pod) + if owner == nil || owner.UID != ds.UID || owner.Kind != "DaemonSet" || + pod.DeletionTimestamp != nil || pod.Status.Phase == corev1.PodFailed || pod.Status.Phase == corev1.PodSucceeded { + continue + } + if name := daemonPodNodeName(pod); name != "" && nodeExists(name) { + names[name] = struct{}{} + } + } + for name := range names { + result.Names = append(result.Names, name) + } + sort.Strings(result.Names) + return result +} + +func daemonPodNodeName(pod *corev1.Pod) string { + if pod.Spec.NodeName != "" { + return pod.Spec.NodeName + } + affinity := pod.Spec.Affinity + if affinity == nil || affinity.NodeAffinity == nil || affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution == nil { + return "" + } + // DaemonSet pods name their target in required affinity before the scheduler + // binds them. An ambiguous selector is not evidence of one expected node. + var target string + for _, term := range affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms { + var name string + for _, field := range term.MatchFields { + if field.Key == "metadata.name" && field.Operator == corev1.NodeSelectorOpIn && len(field.Values) == 1 { + name = field.Values[0] + } + } + if name == "" || (target != "" && target != name) { + return "" + } + target = name + } + return target +} diff --git a/pkg/controller/daemonplacement_test.go b/pkg/controller/daemonplacement_test.go new file mode 100644 index 00000000..0452a0c8 --- /dev/null +++ b/pkg/controller/daemonplacement_test.go @@ -0,0 +1,110 @@ +package controller + +import ( + "testing" + "time" + + "github.com/google/go-cmp/cmp" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// TestDaemonPlacementIncludesPendingTargets keeps a new node visible before its +// daemon can report, without duplicating the scheduler's placement rules. +func TestDaemonPlacementIncludesPendingTargets(t *testing.T) { + controller := true + owned := func(node string) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{OwnerReferences: []metav1.OwnerReference{{ + Kind: "DaemonSet", Name: "runtime", UID: "runtime-uid", Controller: &controller, + }}}, + Spec: corev1.PodSpec{NodeName: node}, + } + } + pending := owned("") + pending.Spec.Affinity = targetNodeAffinity("node-b") + unrelated := owned("node-c") + unrelated.OwnerReferences[0].UID = "other-release" + terminating := owned("node-c") + now := metav1.NewTime(time.Unix(1, 0)) + terminating.DeletionTimestamp = &now + failed := owned("node-c") + failed.Status.Phase = corev1.PodFailed + cases := []struct { + name string + pods []*corev1.Pod + desired int32 + observed int64 + deleted bool + absentNode string + want ExpectedSourceNodes + }{ + {name: "pending target", pods: []*corev1.Pod{pending, owned("node-a")}, desired: 2, observed: 2, + want: ExpectedSourceNodes{Names: []string{"node-a", "node-b"}, Desired: 2, Synced: true}}, + {name: "controller has not created new pod", pods: []*corev1.Pod{owned("node-a")}, desired: 2, observed: 2, + want: ExpectedSourceNodes{Names: []string{"node-a"}, Desired: 2, Synced: true}}, + {name: "generation unobserved", pods: []*corev1.Pod{owned("node-a")}, desired: 1, observed: 1, + want: ExpectedSourceNodes{Names: []string{"node-a"}, Desired: 1}}, + {name: "rollout deduplicates", pods: []*corev1.Pod{owned("node-a"), owned("node-a"), unrelated, terminating, failed}, desired: 1, observed: 2, + want: ExpectedSourceNodes{Names: []string{"node-a"}, Desired: 1, Synced: true}}, + {name: "deleted node", pods: []*corev1.Pod{owned("node-a"), pending}, absentNode: "node-b", desired: 1, observed: 2, + want: ExpectedSourceNodes{Names: []string{"node-a"}, Desired: 1, Synced: true}}, + {name: "daemonset deleting", pods: []*corev1.Pod{owned("node-a")}, desired: 1, observed: 2, deleted: true, + want: ExpectedSourceNodes{Names: []string{"node-a"}, Desired: 1}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + ds := &appsv1.DaemonSet{ + ObjectMeta: metav1.ObjectMeta{UID: "runtime-uid", Generation: 2}, + Status: appsv1.DaemonSetStatus{DesiredNumberScheduled: tc.desired, ObservedGeneration: tc.observed}, + } + if tc.deleted { + ds.DeletionTimestamp = &now + } + got := daemonPlacementNodes(ds, tc.pods, func(name string) bool { return name != tc.absentNode }) + if diff := cmp.Diff(tc.want, got); diff != "" { + t.Fatalf("placement mismatch (-want +got):\n%s", diff) + } + }) + } +} + +// TestDaemonPodTargetRejectsAmbiguousAffinity refuses to turn a general affinity +// expression into evidence that a particular node should have a daemon. +func TestDaemonPodTargetRejectsAmbiguousAffinity(t *testing.T) { + ambiguous := targetNodeAffinity("node-a") + ambiguous.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms = append( + ambiguous.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms, + targetNodeAffinity("node-b").NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms...) + cases := []struct { + name string + spec corev1.PodSpec + want string + }{ + {name: "bound node", spec: corev1.PodSpec{NodeName: "node-a", Affinity: ambiguous}, want: "node-a"}, + {name: "pending target", spec: corev1.PodSpec{Affinity: targetNodeAffinity("node-b")}, want: "node-b"}, + {name: "no affinity"}, + {name: "ambiguous target", spec: corev1.PodSpec{Affinity: ambiguous}}, + {name: "empty terms", spec: corev1.PodSpec{Affinity: &corev1.Affinity{NodeAffinity: &corev1.NodeAffinity{ + RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{}, + }}}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := daemonPodNodeName(&corev1.Pod{Spec: tc.spec}); got != tc.want { + t.Fatalf("target = %q, want %q", got, tc.want) + } + }) + } +} + +func targetNodeAffinity(name string) *corev1.Affinity { + return &corev1.Affinity{NodeAffinity: &corev1.NodeAffinity{ + RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{ + NodeSelectorTerms: []corev1.NodeSelectorTerm{{MatchFields: []corev1.NodeSelectorRequirement{{ + Key: "metadata.name", Operator: corev1.NodeSelectorOpIn, Values: []string{name}, + }}}}, + }, + }} +} diff --git a/pkg/controller/source_dependencies_test.go b/pkg/controller/source_dependencies_test.go new file mode 100644 index 00000000..7f424b50 --- /dev/null +++ b/pkg/controller/source_dependencies_test.go @@ -0,0 +1,194 @@ +package controller + +import ( + "context" + "strings" + "testing" + + "github.com/nirmata/runtime/api/v1alpha1" + "github.com/nirmata/runtime/pkg/compiler" + "github.com/nirmata/runtime/pkg/events" + "github.com/nirmata/runtime/pkg/runtimeevent" + + "github.com/google/go-cmp/cmp" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// TestSourceDependenciesFollowDeclaredTargets covers every observation producer +// while keeping a behavior without values or expressions independent of it. +func TestSourceDependenciesFollowDeclaredTargets(t *testing.T) { + monitor, enforce, emptyMode := v1alpha1.PolicyModeMonitor, v1alpha1.PolicyModeEnforce, v1alpha1.RuntimePolicyMode("") + values := &v1alpha1.Behavior{Deny: &v1alpha1.BehaviorRule{Values: []string{"*"}}} + emptyRules := &v1alpha1.Behavior{Allow: &v1alpha1.BehaviorRule{}, Deny: &v1alpha1.BehaviorRule{Values: []string{}}} + expression := &v1alpha1.Behavior{Allow: &v1alpha1.BehaviorRule{Expression: `["/bin/sh"].filter(value, false)`}} + cases := []struct { + name string + mode *v1alpha1.RuntimePolicyMode + behaviors []v1alpha1.PolicyBehavior + want []string + }{ + {name: "nil mode bypassed API defaulting", behaviors: []v1alpha1.PolicyBehavior{{Exec: values}}}, + {name: "empty mode", mode: &emptyMode, behaviors: []v1alpha1.PolicyBehavior{{Exec: values}}}, + {name: "enforcement", mode: &enforce, behaviors: []v1alpha1.PolicyBehavior{{Exec: values}, {Network: values}}}, + {name: "empty behavior", mode: &monitor, behaviors: []v1alpha1.PolicyBehavior{{Exec: &v1alpha1.Behavior{}}}}, + {name: "empty rules", mode: &monitor, behaviors: []v1alpha1.PolicyBehavior{{DNS: emptyRules}, {Open: emptyRules}}}, + {name: "open", mode: &monitor, behaviors: []v1alpha1.PolicyBehavior{{Open: values}}, want: []string{openExecObserveSource}}, + {name: "exec", mode: &monitor, behaviors: []v1alpha1.PolicyBehavior{{Exec: values}}, want: []string{execTraceSource, openExecObserveSource}}, + {name: "network", mode: &monitor, behaviors: []v1alpha1.PolicyBehavior{{Network: values}}, want: []string{egressObserveSource}}, + {name: "protocol", mode: &monitor, behaviors: []v1alpha1.PolicyBehavior{{Protocol: values}}, want: []string{egressObserveSource}}, + {name: "dns", mode: &monitor, behaviors: []v1alpha1.PolicyBehavior{{DNS: values}}, want: []string{dnsQuerySource}}, + {name: "shared sources deduplicated", mode: &monitor, + behaviors: []v1alpha1.PolicyBehavior{{Open: values}, {Exec: values}, {Network: values}, {Protocol: values}, {DNS: values}}, + want: []string{dnsQuerySource, egressObserveSource, execTraceSource, openExecObserveSource}}, + {name: "expression can be reevaluated", mode: &monitor, behaviors: []v1alpha1.PolicyBehavior{{Exec: expression}}, + want: []string{execTraceSource, openExecObserveSource}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := sourceDependencies(v1alpha1.RuntimePolicySpec{Mode: tc.mode, Behaviors: tc.behaviors}) + if diff := cmp.Diff(tc.want, got); diff != "" { + t.Fatalf("source dependencies mismatch (-want +got):\n%s", diff) + } + }) + } +} + +// TestPollSourceFailureGatesMonitorApplied prevents an active policy from +// remaining applied when its only counter reader is retrying after failure. +func TestPollSourceFailureGatesMonitorApplied(t *testing.T) { + targets := &v1alpha1.Behavior{Deny: &v1alpha1.BehaviorRule{Values: []string{"*"}}} + cases := []struct { + name string + behavior v1alpha1.PolicyBehavior + source string + }{ + {name: "open", behavior: v1alpha1.PolicyBehavior{Open: targets}, source: openExecObserveSource}, + {name: "exec", behavior: v1alpha1.PolicyBehavior{Exec: targets}, source: openExecObserveSource}, + {name: "network", behavior: v1alpha1.PolicyBehavior{Network: targets}, source: egressObserveSource}, + {name: "protocol", behavior: v1alpha1.PolicyBehavior{Protocol: targets}, source: egressObserveSource}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + policy := monitorPolicyWithBehaviors("p", "uid-1", tc.behavior) + sw, client := newTestStatusWriter(t, "node-a", policy) + sw.SetExpectedSourceNodes(func() ExpectedSourceNodes { + return ExpectedSourceNodes{Names: []string{"node-a"}, Desired: 1, Synced: true} + }) + if err := sw.RuntimePolicyEvent(&compiler.EvaluationResult{UID: "uid-1", Name: "p", Mode: compiler.ModeMonitor}, events.EventTypeCreate); err != nil { + t.Fatal(err) + } + sw.RecordSourceStatus(execTraceSource, runtimeevent.SourceStateAvailable, runtimeevent.SourceReasonReady) + for _, step := range []struct { + state runtimeevent.SourceState + reason string + want metav1.ConditionStatus + }{ + {state: runtimeevent.SourceStateUnavailable, reason: runtimeevent.SourceReasonReaderFailed, want: metav1.ConditionFalse}, + {state: runtimeevent.SourceStateAvailable, reason: runtimeevent.SourceReasonReady, want: metav1.ConditionTrue}, + } { + sw.RecordSourceStatus(tc.source, step.state, step.reason) + if err := sw.Flush(context.Background()); err != nil { + t.Fatal(err) + } + got := getPolicy(t, client, "p") + for _, typ := range []string{v1alpha1.ConditionEventSourcesAvailable, v1alpha1.ConditionApplied} { + condition := conditionOfType(t, got.Status.Conditions, typ) + if condition.Status != step.want { + t.Errorf("%s after %s = %s, want %s", typ, step.state, condition.Status, step.want) + } + if step.want == metav1.ConditionFalse && !strings.Contains(condition.Message, tc.source) { + t.Errorf("%s message = %q, want failing source %q", typ, condition.Message, tc.source) + } + } + } + }) + } +} + +// TestEmptyRulesDoNotDependOnUnavailableSources keeps a no-op policy from +// reporting the state of readers it cannot use. +func TestEmptyRulesDoNotDependOnUnavailableSources(t *testing.T) { + for _, tc := range []struct { + name string + behavior *v1alpha1.Behavior + }{ + {name: "absent"}, + {name: "empty behavior", behavior: &v1alpha1.Behavior{}}, + {name: "empty allow and deny", behavior: &v1alpha1.Behavior{ + Allow: &v1alpha1.BehaviorRule{}, Deny: &v1alpha1.BehaviorRule{Values: []string{}}, + }}, + } { + t.Run(tc.name, func(t *testing.T) { + policy := monitorPolicyWithBehaviors("p", "uid-1", v1alpha1.PolicyBehavior{ + Open: tc.behavior, Exec: tc.behavior, Network: tc.behavior, Protocol: tc.behavior, DNS: tc.behavior, + }) + sw, client := newTestStatusWriter(t, "node-a", policy) + for _, source := range []string{openExecObserveSource, egressObserveSource, execTraceSource, dnsQuerySource} { + sw.RecordSourceStatus(source, runtimeevent.SourceStateUnavailable, runtimeevent.SourceReasonInitializationFailed) + } + if err := sw.RuntimePolicyEvent(&compiler.EvaluationResult{UID: "uid-1", Name: "p", Mode: compiler.ModeMonitor}, events.EventTypeCreate); err != nil { + t.Fatal(err) + } + if err := sw.Flush(context.Background()); err != nil { + t.Fatal(err) + } + got := getPolicy(t, client, "p") + if hasCondition(got.Status.Conditions, v1alpha1.ConditionEventSourcesAvailable) || len(got.Status.Nodes[0].EventSources) != 0 { + t.Errorf("empty rules produced source status: %+v", got.Status) + } + if applied := conditionOfType(t, got.Status.Conditions, v1alpha1.ConditionApplied); applied.Status != metav1.ConditionTrue { + t.Errorf("Applied = %+v, want True", applied) + } + }) + } +} + +// TestEmptyEvaluationRetainsDeclaredExpressionSources keeps a transient empty +// result from erasing coverage requirements before the next evaluation. +func TestEmptyEvaluationRetainsDeclaredExpressionSources(t *testing.T) { + policy := monitorPolicyWithBehaviors("p", "uid-1", v1alpha1.PolicyBehavior{Exec: &v1alpha1.Behavior{ + Deny: &v1alpha1.BehaviorRule{Expression: `["/bin/sh"].filter(value, false)`}, + }}) + sw, client := newTestStatusWriter(t, "node-a", policy) + sw.SetExpectedSourceNodes(func() ExpectedSourceNodes { + return ExpectedSourceNodes{Names: []string{"node-a"}, Desired: 1, Synced: true} + }) + sw.RecordSourceStatus(execTraceSource, runtimeevent.SourceStateAvailable, runtimeevent.SourceReasonReady) + sw.RecordSourceStatus(openExecObserveSource, runtimeevent.SourceStateUnavailable, runtimeevent.SourceReasonReaderFailed) + for _, eventType := range []string{events.EventTypeCreate, events.EventTypeUpdate} { + t.Run(eventType, func(t *testing.T) { + if err := sw.RuntimePolicyEvent(&compiler.EvaluationResult{UID: "uid-1", Name: "p", Mode: compiler.ModeMonitor, + Exec: &compiler.AllowDenyPair{}}, eventType); err != nil { + t.Fatal(err) + } + if err := sw.Flush(context.Background()); err != nil { + t.Fatal(err) + } + got := getPolicy(t, client, "p") + for _, typ := range []string{v1alpha1.ConditionEventSourcesAvailable, v1alpha1.ConditionApplied} { + if condition := conditionOfType(t, got.Status.Conditions, typ); condition.Status != metav1.ConditionFalse { + t.Errorf("%s = %+v, want False for declared expression", typ, condition) + } + } + }) + } +} + +// TestExecDependencyFailureReportsLostFilenameCoverage distinguishes a reader +// failure from loss of the manager that supplies both observation paths. +func TestExecDependencyFailureReportsLostFilenameCoverage(t *testing.T) { + for _, tc := range []struct { + reason string + want string + }{ + {reason: runtimeevent.SourceReasonReaderFailed, want: "exec filename observations may remain available"}, + {reason: runtimeevent.SourceReasonDependencyUnavailable, want: "argv and exec filename observations are unavailable"}, + } { + t.Run(tc.reason, func(t *testing.T) { + status := eventSourceStatus(execTraceSource, sourceStatus{state: runtimeevent.SourceStateUnavailable, reason: tc.reason}) + if status.Status != metav1.ConditionFalse || !strings.Contains(status.Message, tc.want) { + t.Fatalf("source status = %+v, want False and %q", status, tc.want) + } + }) + } +} diff --git a/pkg/controller/statuswriter.go b/pkg/controller/statuswriter.go index 6174554d..7916c4f8 100644 --- a/pkg/controller/statuswriter.go +++ b/pkg/controller/statuswriter.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "slices" "sort" "strings" "sync" @@ -13,6 +14,7 @@ import ( v1alpha1client "github.com/nirmata/runtime/pkg/client/clientset/versioned" "github.com/nirmata/runtime/pkg/compiler" "github.com/nirmata/runtime/pkg/events" + "github.com/nirmata/runtime/pkg/runtimeevent" "github.com/go-logr/logr" apiequality "k8s.io/apimachinery/pkg/api/equality" @@ -25,6 +27,27 @@ import ( // DefaultStatusFlushInterval is the flush cadence used by the daemon. const DefaultStatusFlushInterval = 30 * time.Second +const ( + execTraceSource = "exec-trace" + dnsQuerySource = "dnsquery" + openExecObserveSource = "openexec-observe" + egressObserveSource = "egress-observe" +) + +// ExpectedSourceNodes describes the DaemonSet nodes that should publish event +// source status. A non-synced or incomplete inventory cannot prove a source is +// available everywhere. +type ExpectedSourceNodes struct { + Names []string + Desired int + Synced bool +} + +type sourceStatus struct { + state runtimeevent.SourceState + reason string +} + // policyStatusState is this node's view of one policy's status. type policyStatusState struct { // name is needed to address the object. An entry whose name is still unknown @@ -34,7 +57,6 @@ type policyStatusState struct { // conditions is keyed by condition type; the last write wins. conditions map[string]metav1.Condition - // gen increments on every mutation. A flush records the gen it observed // and only clears dirty when nothing changed while the API call was in // flight. @@ -73,10 +95,15 @@ type StatusWriter struct { // against what was written to the API, not from RecordCondition: see // notifyConditionChanges. A nil func disables it. onConditionChanged func(policyUID, policyName string, cond metav1.Condition) + // expectedSourceNodes is nil when source status is aggregated over the + // reporting shards. + expectedSourceNodes func() ExpectedSourceNodes mu sync.Mutex // policies is keyed by policy UID. policies map[string]*policyStatusState + // sources is daemon-wide state, retained before a relevant policy appears. + sources map[string]sourceStatus } // NewStatusWriter builds a StatusWriter for this node. A non-positive interval @@ -96,6 +123,48 @@ func NewStatusWriter(client v1alpha1client.Interface, nodeName string, interval nodeGone: nodeGone, onConditionChanged: onConditionChanged, policies: make(map[string]*policyStatusState), + sources: make(map[string]sourceStatus), + } +} + +// SetExpectedSourceNodes installs the DaemonSet membership view used to +// aggregate event source status. It marks every policy dirty because placement +// changes can alter a cluster condition without a policy event. +func (s *StatusWriter) SetExpectedSourceNodes(f func() ExpectedSourceNodes) { + s.mu.Lock() + defer s.mu.Unlock() + s.expectedSourceNodes = f + for _, st := range s.policies { + st.touch() + } +} + +// MarkAllDirty makes the next flush recompute every policy's aggregate status. +func (s *StatusWriter) MarkAllDirty() { + s.mu.Lock() + defer s.mu.Unlock() + for _, st := range s.policies { + st.touch() + } +} + +// RecordSourceStatus records a daemon-wide source lifecycle transition. The +// source state is projected into relevant policy shards during flush. +func (s *StatusWriter) RecordSourceStatus(source string, state runtimeevent.SourceState, reason string) { + if source == "" || reason == "" { + return + } + s.mu.Lock() + defer s.mu.Unlock() + if previous, ok := s.sources[source]; ok && previous.state == state && previous.reason == reason { + return + } + s.sources[source] = sourceStatus{state: state, reason: reason} + if state == runtimeevent.SourceStateUnavailable { + s.log.V(0).Info("event source is unavailable", "source", source, "reason", reason) + } + for _, st := range s.policies { + st.touch() } } @@ -223,6 +292,14 @@ func (s *StatusWriter) appliedCondition(mode string, conditions map[string]metav cond.Message = gate.Message return cond } + if compiler.IsObserveMode(mode) { + if gate, ok := conditions[v1alpha1.ConditionEventSourcesAvailable]; ok && gate.Status != metav1.ConditionTrue { + cond.Status = gate.Status + cond.Reason = gate.Reason + cond.Message = gate.Message + return cond + } + } if gate, ok := conditions[v1alpha1.ConditionPodsMatched]; ok && gate.Status == metav1.ConditionFalse { cond.Status = metav1.ConditionFalse @@ -264,9 +341,8 @@ type flushItem struct { name string mode string conditions []metav1.Condition - // shard carries this node's compact signals; NodeName and - // LastEvaluatedTime are the flusher's to fill. - shard v1alpha1.NodePolicyStatus + signals map[string]metav1.Condition + sources map[string]sourceStatus // explicitApplied marks a directly recorded Applied, which the derived // cluster-scoped one must stand aside for. explicitApplied bool @@ -303,11 +379,18 @@ func (s *StatusWriter) snapshot() []flushItem { continue } item := flushItem{ - uid: uid, - name: st.name, - mode: st.mode, - shard: signalShard(st.conditions), - gen: st.gen, + uid: uid, + name: st.name, + mode: st.mode, + signals: make(map[string]metav1.Condition, len(st.conditions)), + sources: make(map[string]sourceStatus, len(s.sources)), + gen: st.gen, + } + for name, status := range s.sources { + item.sources[name] = status + } + for typ, condition := range st.conditions { + item.signals[typ] = condition } conds := make([]metav1.Condition, 0, len(st.conditions)) for t, c := range st.conditions { @@ -378,7 +461,8 @@ func (s *StatusWriter) flushOne(ctx context.Context, item flushItem) error { } s.pruneDeletedNodeShards(&updated.Status) - shard := item.shard + dependencies := sourceDependencies(cur.Spec) + shard := signalShard(item.signals, item.sources, dependencies) shard.NodeName = s.nodeName shard.LastEvaluatedTime = &now setNodeShard(&updated.Status, shard) @@ -393,7 +477,7 @@ func (s *StatusWriter) flushOne(ctx context.Context, item flushItem) error { } apimeta.SetStatusCondition(&updated.Status.Conditions, cond) } - s.setClusterConditions(&updated.Status, item.mode, item.explicitApplied) + s.setClusterConditions(&updated.Status, currentPolicyMode(cur.Spec, item.mode), item.explicitApplied, dependencies) if apiequality.Semantic.DeepEqual(before, updated.Status) { // nothing to say; skip the write entirely @@ -498,7 +582,7 @@ func recomputeLastEvaluated(status *v1alpha1.RuntimePolicyStatus) { // signalShard reduces recorded conditions to the compact per-node fields the // cluster-scoped aggregation reads. -func signalShard(conditions map[string]metav1.Condition) v1alpha1.NodePolicyStatus { +func signalShard(conditions map[string]metav1.Condition, sources map[string]sourceStatus, dependencies []string) v1alpha1.NodePolicyStatus { var shard v1alpha1.NodePolicyStatus if c, ok := conditions[v1alpha1.ConditionEnforcementAvailable]; ok { shard.EnforcementAvailable = conditionBool(c) @@ -515,6 +599,9 @@ func signalShard(conditions map[string]metav1.Condition) v1alpha1.NodePolicyStat break } } + for _, name := range dependencies { + shard.EventSources = append(shard.EventSources, eventSourceStatus(name, sources[name])) + } return shard } @@ -526,9 +613,9 @@ func conditionBool(c metav1.Condition) *bool { // setClusterConditions derives the cluster-scoped availability, pods-matched // and Applied conditions from the per-node shards, so every daemon publishes // the same top-level answer instead of its own node's. -func (s *StatusWriter) setClusterConditions(status *v1alpha1.RuntimePolicyStatus, mode string, explicitApplied bool) { +func (s *StatusWriter) setClusterConditions(status *v1alpha1.RuntimePolicyStatus, mode string, explicitApplied bool, sourceDependencies []string) { now := metav1.NewTime(s.clock()) - agg := make(map[string]metav1.Condition, 3) + agg := make(map[string]metav1.Condition, 4) // a type no shard reports is removed rather than left as written: keeping // it would preserve a value the shards no longer back record := func(c metav1.Condition, ok bool) { @@ -546,11 +633,236 @@ func (s *StatusWriter) setClusterConditions(status *v1alpha1.RuntimePolicyStatus v1alpha1.ReasonObservationAvailable, v1alpha1.ReasonObservationUnavailable, func(n *v1alpha1.NodePolicyStatus) *bool { return n.ObservationAvailable })) record(aggregatePodsMatched(status.Nodes, now)) + expected, hasExpected := s.sourceNodes() + record(aggregateEventSources(status.Nodes, now, sourceDependencies, expected, hasExpected)) if !explicitApplied { apimeta.SetStatusCondition(&status.Conditions, s.appliedCondition(mode, agg)) } } +func currentPolicyMode(spec v1alpha1.RuntimePolicySpec, fallback string) string { + if spec.Mode != nil { + return string(*spec.Mode) + } + return fallback +} + +func sourceDependencies(spec v1alpha1.RuntimePolicySpec) []string { + if spec.Mode == nil || !compiler.IsObserveMode(string(*spec.Mode)) { + return nil + } + var dependencies []string + for _, behavior := range spec.Behaviors { + if declaresTargets(behavior.Open) || declaresTargets(behavior.Exec) { + dependencies = append(dependencies, openExecObserveSource) + } + if declaresTargets(behavior.Network) || declaresTargets(behavior.Protocol) { + dependencies = append(dependencies, egressObserveSource) + } + if declaresTargets(behavior.Exec) { + dependencies = append(dependencies, execTraceSource) + } + if declaresTargets(behavior.DNS) { + dependencies = append(dependencies, dnsQuerySource) + } + } + sort.Strings(dependencies) + return slices.Compact(dependencies) +} + +func declaresTargets(behavior *v1alpha1.Behavior) bool { + if behavior == nil { + return false + } + // Expressions can become nonempty on re-evaluation without a spec update. + for _, rule := range []*v1alpha1.BehaviorRule{behavior.Allow, behavior.Deny} { + if rule != nil && (len(rule.Values) != 0 || rule.Expression != "") { + return true + } + } + return false +} + +func eventSourceStatus(name string, status sourceStatus) v1alpha1.EventSourceStatus { + result := v1alpha1.EventSourceStatus{Name: name, Status: metav1.ConditionUnknown, Reason: status.reason, Message: sourceMessage(name, status)} + if result.Reason == "" { + result.Reason = runtimeevent.SourceReasonStarting + } + switch status.state { + case runtimeevent.SourceStateAvailable: + result.Status = metav1.ConditionTrue + case runtimeevent.SourceStateUnavailable: + result.Status = metav1.ConditionFalse + } + return result +} + +func sourceMessage(name string, status sourceStatus) string { + switch name { + case execTraceSource: + if status.state == runtimeevent.SourceStateAvailable { + return "exec trace source is available" + } + if status.state == runtimeevent.SourceStateUnavailable { + if status.reason == runtimeevent.SourceReasonDependencyUnavailable { + return "exec trace source is unavailable because the open/exec manager could not load; argv and exec filename observations are unavailable" + } + return "exec trace source is unavailable; argv observations are unavailable, but exec filename observations may remain available" + } + return "exec trace source is starting" + case dnsQuerySource: + if status.state == runtimeevent.SourceStateAvailable { + return "DNS query source is available" + } + if status.state == runtimeevent.SourceStateUnavailable { + return "DNS query source is unavailable; DNS name observations are unavailable" + } + return "DNS query source is starting" + case openExecObserveSource: + if status.state == runtimeevent.SourceStateAvailable { + return "open/exec observation counter source is available" + } + if status.state == runtimeevent.SourceStateUnavailable { + return "open/exec observation counter source is unavailable; file-open observations and exec filename counter observations are unavailable" + } + return "open/exec observation counter source is starting" + case egressObserveSource: + if status.state == runtimeevent.SourceStateAvailable { + return "egress observation counter source is available" + } + if status.state == runtimeevent.SourceStateUnavailable { + return "egress observation counter source is unavailable; network and protocol observations are unavailable" + } + return "egress observation counter source is starting" + default: + return "event source status is unavailable" + } +} + +func aggregateEventSources(nodes []v1alpha1.NodePolicyStatus, now metav1.Time, dependencies []string, expected ExpectedSourceNodes, hasExpected bool) (metav1.Condition, bool) { + cond := metav1.Condition{Type: v1alpha1.ConditionEventSourcesAvailable, LastTransitionTime: now} + if len(dependencies) == 0 { + return cond, false + } + if !hasExpected { + for _, node := range nodes { + if len(node.EventSources) != 0 { + return aggregateEventSourceNodes(nodes, now, dependencies, len(nodes), false) + } + } + return cond, false + } + expectedNames := make(map[string]struct{}, len(expected.Names)) + for _, name := range expected.Names { + if name != "" { + expectedNames[name] = struct{}{} + } + } + if expected.Desired == 0 && expected.Synced { + cond.Status, cond.Reason, cond.Message = metav1.ConditionUnknown, v1alpha1.ReasonEventSourcesUnknown, "event source status is unknown because there are no expected daemon nodes" + return cond, true + } + incomplete := !expected.Synced || len(expectedNames) != expected.Desired + if incomplete { + return aggregateEventSourceNodes(nodes, now, dependencies, expected.Desired, true) + } + byName := make(map[string]v1alpha1.NodePolicyStatus, len(nodes)) + for _, node := range nodes { + byName[node.NodeName] = node + } + names := make([]string, 0, len(expectedNames)) + for name := range expectedNames { + names = append(names, name) + } + sort.Strings(names) + selected := make([]v1alpha1.NodePolicyStatus, 0, len(names)) + for _, name := range names { + if node, ok := byName[name]; ok { + selected = append(selected, node) + } else { + selected = append(selected, v1alpha1.NodePolicyStatus{NodeName: name}) + } + } + return aggregateEventSourceNodes(selected, now, dependencies, expected.Desired, false) +} + +func aggregateEventSourceNodes(nodes []v1alpha1.NodePolicyStatus, now metav1.Time, dependencies []string, desired int, inventoryIncomplete bool) (metav1.Condition, bool) { + cond := metav1.Condition{Type: v1alpha1.ConditionEventSourcesAvailable, LastTransitionTime: now} + if desired == 0 { + desired = len(nodes) + } + failures, unknown := make(map[string][]string), make(map[string][]string) + for _, node := range nodes { + for _, dependency := range dependencies { + status, ok := sourceStatusFor(node.EventSources, dependency) + if !ok || status.Status != metav1.ConditionTrue && status.Status != metav1.ConditionFalse { + unknown[node.NodeName] = append(unknown[node.NodeName], dependency) + continue + } + if status.Status == metav1.ConditionFalse { + entry := dependency + if status.Message != "" { + entry += ": " + status.Message + } + failures[node.NodeName] = append(failures[node.NodeName], entry) + } + } + } + if len(failures) != 0 { + reporting, scope := desired, "expected" + if inventoryIncomplete { + reporting, scope = len(nodes), "reporting" + } + cond.Status, cond.Reason = metav1.ConditionFalse, v1alpha1.ReasonEventSourcesUnavailable + cond.Message = fmt.Sprintf("required event sources are unavailable on %d of %d %s daemon node(s): %s", len(failures), reporting, scope, truncatedNodeList(sourceNodeEntries(failures))) + return cond, true + } + if inventoryIncomplete || len(unknown) != 0 || len(nodes) < desired { + cond.Status, cond.Reason = metav1.ConditionUnknown, v1alpha1.ReasonEventSourcesUnknown + if inventoryIncomplete { + cond.Message = "event source status is unknown while daemon membership is incomplete" + } else { + cond.Message = fmt.Sprintf("required event source status is pending on %d of %d expected daemon node(s): %s", len(unknown), desired, truncatedNodeList(sourceNodeEntries(unknown))) + } + return cond, true + } + cond.Status, cond.Reason = metav1.ConditionTrue, v1alpha1.ReasonEventSourcesAvailable + cond.Message = fmt.Sprintf("required event sources are available on all %d expected daemon node(s)", desired) + return cond, true +} + +func sourceNodeEntries(byNode map[string][]string) []string { + names := make([]string, 0, len(byNode)) + for name := range byNode { + names = append(names, name) + } + sort.Strings(names) + entries := make([]string, 0, len(names)) + for _, name := range names { + entries = append(entries, name+": "+strings.Join(byNode[name], ", ")) + } + return entries +} + +func sourceStatusFor(statuses []v1alpha1.EventSourceStatus, name string) (v1alpha1.EventSourceStatus, bool) { + for _, status := range statuses { + if status.Name == name { + return status, true + } + } + return v1alpha1.EventSourceStatus{}, false +} + +func (s *StatusWriter) sourceNodes() (ExpectedSourceNodes, bool) { + s.mu.Lock() + f := s.expectedSourceNodes + s.mu.Unlock() + if f == nil { + return ExpectedSourceNodes{}, false + } + return f(), true +} + // aggregateAvailability is all-true across the nodes reporting the value: one // node that cannot enforce or observe leaves that node's workloads uncovered // no matter how many other nodes can. It reports nothing when no node does. diff --git a/pkg/controller/statuswriter_test.go b/pkg/controller/statuswriter_test.go index c748f299..1a30f6e8 100644 --- a/pkg/controller/statuswriter_test.go +++ b/pkg/controller/statuswriter_test.go @@ -11,6 +11,7 @@ import ( fakeversioned "github.com/nirmata/runtime/pkg/client/clientset/versioned/fake" "github.com/nirmata/runtime/pkg/compiler" "github.com/nirmata/runtime/pkg/events" + "github.com/nirmata/runtime/pkg/runtimeevent" "github.com/go-logr/logr" "github.com/google/go-cmp/cmp" @@ -39,6 +40,13 @@ func policyObj(name, uid string) *v1alpha1.RuntimePolicy { } } +func monitorPolicyWithBehaviors(name, uid string, behaviors ...v1alpha1.PolicyBehavior) *v1alpha1.RuntimePolicy { + mode := v1alpha1.PolicyModeMonitor + policy := policyObj(name, uid) + policy.Spec = v1alpha1.RuntimePolicySpec{Mode: &mode, Behaviors: behaviors} + return policy +} + func evalResult(uid, name, mode string, sel labels.Selector) *compiler.EvaluationResult { return &compiler.EvaluationResult{ UID: uid, @@ -1361,6 +1369,288 @@ func TestNewStatusWriterDefaultsInterval(t *testing.T) { } } +// TestStatusWriterProjectsSourceStateRecordedBeforePolicy keeps startup source +// state visible to policies that the informer delivers later. +func TestStatusWriterProjectsSourceStateRecordedBeforePolicy(t *testing.T) { + mode := v1alpha1.PolicyModeMonitor + policy := policyObj("p", "uid-1") + policy.Spec = v1alpha1.RuntimePolicySpec{ + Mode: &mode, + Behaviors: []v1alpha1.PolicyBehavior{ + {Exec: &v1alpha1.Behavior{Deny: &v1alpha1.BehaviorRule{Values: []string{"/bin/sh"}}}}, + {DNS: &v1alpha1.Behavior{Deny: &v1alpha1.BehaviorRule{Values: []string{"*"}}}}, + }, + } + sw, client := newTestStatusWriter(t, "node-a", policy) + sw.SetExpectedSourceNodes(func() ExpectedSourceNodes { + return ExpectedSourceNodes{Names: []string{"node-a"}, Desired: 1, Synced: true} + }) + sw.RecordSourceStatus(execTraceSource, runtimeevent.SourceStateAvailable, runtimeevent.SourceReasonReady) + sw.RecordSourceStatus(openExecObserveSource, runtimeevent.SourceStateAvailable, runtimeevent.SourceReasonReady) + if err := sw.RuntimePolicyEvent(evalResult("uid-1", "p", compiler.ModeMonitor, labels.Everything()), events.EventTypeCreate); err != nil { + t.Fatal(err) + } + if err := sw.Flush(context.Background()); err != nil { + t.Fatal(err) + } + got := getPolicy(t, client, "p") + gate := conditionOfType(t, got.Status.Conditions, v1alpha1.ConditionEventSourcesAvailable) + if gate.Status != metav1.ConditionUnknown { + t.Errorf("EventSourcesAvailable = %s, want Unknown while dnsquery has not reported", gate.Status) + } + + sw.RecordSourceStatus(dnsQuerySource, runtimeevent.SourceStateAvailable, runtimeevent.SourceReasonReady) + if err := sw.Flush(context.Background()); err != nil { + t.Fatal(err) + } + got = getPolicy(t, client, "p") + gate = conditionOfType(t, got.Status.Conditions, v1alpha1.ConditionEventSourcesAvailable) + if gate.Status != metav1.ConditionTrue { + t.Errorf("EventSourcesAvailable = %s, want True", gate.Status) + } + + sw.RecordSourceStatus(execTraceSource, runtimeevent.SourceStateUnavailable, runtimeevent.SourceReasonReaderFailed) + if err := sw.Flush(context.Background()); err != nil { + t.Fatal(err) + } + got = getPolicy(t, client, "p") + gate = conditionOfType(t, got.Status.Conditions, v1alpha1.ConditionEventSourcesAvailable) + if gate.Status != metav1.ConditionFalse || gate.Reason != v1alpha1.ReasonEventSourcesUnavailable { + t.Errorf("EventSourcesAvailable = (%s, %s), want (False, %s)", gate.Status, gate.Reason, v1alpha1.ReasonEventSourcesUnavailable) + } + applied := conditionOfType(t, got.Status.Conditions, v1alpha1.ConditionApplied) + if applied.Status != metav1.ConditionFalse || applied.Reason != v1alpha1.ReasonEventSourcesUnavailable { + t.Errorf("Applied = (%s, %s), want (False, %s)", applied.Status, applied.Reason, v1alpha1.ReasonEventSourcesUnavailable) + } +} + +func TestAggregateEventSourcesExpectedNodes(t *testing.T) { + available := v1alpha1.EventSourceStatus{Name: execTraceSource, Status: metav1.ConditionTrue, Reason: runtimeevent.SourceReasonReady} + unavailable := v1alpha1.EventSourceStatus{Name: execTraceSource, Status: metav1.ConditionFalse, Reason: runtimeevent.SourceReasonReaderFailed} + tests := []struct { + name string + nodes []v1alpha1.NodePolicyStatus + expected ExpectedSourceNodes + depends []string + status metav1.ConditionStatus + message string + }{ + {name: "missing expected report", nodes: []v1alpha1.NodePolicyStatus{{NodeName: "node-a", EventSources: []v1alpha1.EventSourceStatus{available}}}, expected: ExpectedSourceNodes{Names: []string{"node-a", "node-b"}, Desired: 2, Synced: true}, status: metav1.ConditionUnknown}, + {name: "failed node wins over missing report", nodes: []v1alpha1.NodePolicyStatus{{NodeName: "node-a", EventSources: []v1alpha1.EventSourceStatus{unavailable}}}, expected: ExpectedSourceNodes{Names: []string{"node-a", "node-b"}, Desired: 2, Synced: true}, status: metav1.ConditionFalse}, + {name: "incomplete inventory retains known failure", nodes: []v1alpha1.NodePolicyStatus{{NodeName: "node-a", EventSources: []v1alpha1.EventSourceStatus{unavailable}}}, expected: ExpectedSourceNodes{Desired: 0, Synced: false}, status: metav1.ConditionFalse}, + {name: "no expected daemon nodes", expected: ExpectedSourceNodes{Desired: 0, Synced: true}, status: metav1.ConditionUnknown}, + {name: "outside placement ignored once synced", nodes: []v1alpha1.NodePolicyStatus{{NodeName: "node-a", EventSources: []v1alpha1.EventSourceStatus{available}}, {NodeName: "node-old", EventSources: []v1alpha1.EventSourceStatus{unavailable}}}, expected: ExpectedSourceNodes{Names: []string{"node-a"}, Desired: 1, Synced: true}, status: metav1.ConditionTrue}, + {name: "garbage status cannot claim readiness", nodes: []v1alpha1.NodePolicyStatus{{NodeName: "node-a", EventSources: []v1alpha1.EventSourceStatus{{Name: execTraceSource, Status: "Other", Reason: runtimeevent.SourceReasonReady}}}}, expected: ExpectedSourceNodes{Names: []string{"node-a"}, Desired: 1, Synced: true}, status: metav1.ConditionUnknown}, + {name: "two failing sources count one node", nodes: []v1alpha1.NodePolicyStatus{{NodeName: "node-a", EventSources: []v1alpha1.EventSourceStatus{unavailable, {Name: dnsQuerySource, Status: metav1.ConditionFalse, Reason: runtimeevent.SourceReasonReaderFailed}}}}, expected: ExpectedSourceNodes{Names: []string{"node-a"}, Desired: 1, Synced: true}, depends: []string{execTraceSource, dnsQuerySource}, status: metav1.ConditionFalse, message: "on 1 of 1 expected daemon node(s)"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + dependencies := tc.depends + if dependencies == nil { + dependencies = []string{execTraceSource} + } + got, ok := aggregateEventSources(tc.nodes, metav1.NewTime(fixedNow), dependencies, tc.expected, true) + if !ok || got.Status != tc.status { + t.Errorf("aggregateEventSources = (%v, %s), want (true, %s)", ok, got.Status, tc.status) + } + if tc.message != "" && !strings.Contains(got.Message, tc.message) { + t.Errorf("aggregateEventSources message = %q, want %q", got.Message, tc.message) + } + }) + } +} + +// TestSourceFailureBeforePolicyIsPublished ensures an initialization failure +// is retained until a policy needing that source appears. +func TestSourceFailureBeforePolicyIsPublished(t *testing.T) { + policy := monitorPolicyWithBehaviors("p", "uid-1", v1alpha1.PolicyBehavior{Exec: &v1alpha1.Behavior{Deny: &v1alpha1.BehaviorRule{Values: []string{"/bin/sh"}}}}) + sw, client := newTestStatusWriter(t, "node-a", policy) + sw.SetExpectedSourceNodes(func() ExpectedSourceNodes { + return ExpectedSourceNodes{Names: []string{"node-a"}, Desired: 1, Synced: true} + }) + sw.RecordSourceStatus(execTraceSource, runtimeevent.SourceStateUnavailable, runtimeevent.SourceReasonInitializationFailed) + if err := sw.RuntimePolicyEvent(&compiler.EvaluationResult{UID: "uid-1", Name: "p", Mode: compiler.ModeMonitor, + Exec: &compiler.AllowDenyPair{Deny: []string{"/bin/sh"}}}, events.EventTypeCreate); err != nil { + t.Fatal(err) + } + if err := sw.Flush(context.Background()); err != nil { + t.Fatal(err) + } + got := getPolicy(t, client, "p") + if status := conditionOfType(t, got.Status.Conditions, v1alpha1.ConditionEventSourcesAvailable).Status; status != metav1.ConditionFalse { + t.Errorf("EventSourcesAvailable = %s, want False", status) + } + if status := got.Status.Nodes[0].EventSources[0].Status; status != metav1.ConditionFalse { + t.Errorf("node event source status = %s, want False", status) + } +} + +// TestEventSourcePartialRecoveryStaysUnavailable ensures one recovered source +// cannot hide a second source failure needed by the same policy. +func TestEventSourcePartialRecoveryStaysUnavailable(t *testing.T) { + policy := monitorPolicyWithBehaviors("p", "uid-1", v1alpha1.PolicyBehavior{Exec: &v1alpha1.Behavior{Deny: &v1alpha1.BehaviorRule{Values: []string{"/bin/sh"}}}}, v1alpha1.PolicyBehavior{DNS: &v1alpha1.Behavior{Deny: &v1alpha1.BehaviorRule{Values: []string{"*"}}}}) + sw, client := newTestStatusWriter(t, "node-a", policy) + sw.SetExpectedSourceNodes(func() ExpectedSourceNodes { + return ExpectedSourceNodes{Names: []string{"node-a"}, Desired: 1, Synced: true} + }) + if err := sw.RuntimePolicyEvent(&compiler.EvaluationResult{UID: "uid-1", Name: "p", Mode: compiler.ModeMonitor, + Exec: &compiler.AllowDenyPair{Deny: []string{"/bin/sh"}}, DNS: &compiler.AllowDenyPair{Deny: []string{"*"}}}, events.EventTypeCreate); err != nil { + t.Fatal(err) + } + for _, source := range []string{execTraceSource, dnsQuerySource} { + sw.RecordSourceStatus(source, runtimeevent.SourceStateUnavailable, runtimeevent.SourceReasonReaderFailed) + } + sw.RecordSourceStatus(openExecObserveSource, runtimeevent.SourceStateAvailable, runtimeevent.SourceReasonReady) + if err := sw.Flush(context.Background()); err != nil { + t.Fatal(err) + } + sw.RecordSourceStatus(execTraceSource, runtimeevent.SourceStateAvailable, runtimeevent.SourceReasonReady) + if err := sw.Flush(context.Background()); err != nil { + t.Fatal(err) + } + got := getPolicy(t, client, "p") + gate := conditionOfType(t, got.Status.Conditions, v1alpha1.ConditionEventSourcesAvailable) + if gate.Status != metav1.ConditionFalse || !strings.Contains(gate.Message, dnsQuerySource) { + t.Errorf("EventSourcesAvailable = (%s, %q), want dnsquery failure", gate.Status, gate.Message) + } +} + +// TestPolicySourceDependenciesFollowCurrentSpec ensures a policy update drops +// source entries that its current mode and behaviors no longer require. +func TestPolicySourceDependenciesFollowCurrentSpec(t *testing.T) { + cases := []struct { + name string + mode v1alpha1.RuntimePolicyMode + behavior v1alpha1.PolicyBehavior + }{ + {name: "empty behavior", mode: v1alpha1.PolicyModeMonitor, behavior: v1alpha1.PolicyBehavior{Open: &v1alpha1.Behavior{}}}, + {name: "enforcement ignores observation sources", mode: v1alpha1.PolicyModeEnforce, behavior: v1alpha1.PolicyBehavior{Exec: &v1alpha1.Behavior{Deny: &v1alpha1.BehaviorRule{Values: []string{"/bin/sh"}}}}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + policy := monitorPolicyWithBehaviors("p", "uid-1", v1alpha1.PolicyBehavior{Exec: &v1alpha1.Behavior{Deny: &v1alpha1.BehaviorRule{Values: []string{"/bin/sh"}}}}) + sw, client := newTestStatusWriter(t, "node-a", policy) + sw.SetExpectedSourceNodes(func() ExpectedSourceNodes { + return ExpectedSourceNodes{Names: []string{"node-a"}, Desired: 1, Synced: true} + }) + sw.RecordSourceStatus(execTraceSource, runtimeevent.SourceStateUnavailable, runtimeevent.SourceReasonReaderFailed) + if err := sw.RuntimePolicyEvent(&compiler.EvaluationResult{UID: "uid-1", Name: "p", Mode: compiler.ModeMonitor, + Exec: &compiler.AllowDenyPair{Deny: []string{"/bin/sh"}}}, events.EventTypeCreate); err != nil { + t.Fatal(err) + } + if err := sw.Flush(context.Background()); err != nil { + t.Fatal(err) + } + current := getPolicy(t, client, "p") + current.Spec.Mode = &tc.mode + current.Spec.Behaviors = []v1alpha1.PolicyBehavior{tc.behavior} + if _, err := client.RuntimeV1alpha1().RuntimePolicies().Update(context.Background(), current, metav1.UpdateOptions{}); err != nil { + t.Fatal(err) + } + sw.MarkAllDirty() + if err := sw.Flush(context.Background()); err != nil { + t.Fatal(err) + } + got := getPolicy(t, client, "p") + if len(got.Status.Nodes[0].EventSources) != 0 { + t.Errorf("eventSources = %+v, want none after the spec update", got.Status.Nodes[0].EventSources) + } + if hasCondition(got.Status.Conditions, v1alpha1.ConditionEventSourcesAvailable) { + t.Error("EventSourcesAvailable survived after all source dependencies were removed") + } + if applied := conditionOfType(t, got.Status.Conditions, v1alpha1.ConditionApplied); applied.Status != metav1.ConditionTrue { + t.Errorf("Applied = %+v, want True after removing the optional source dependency", applied) + } + }) + } +} + +// TestCompileFailureOutranksEventSourceFailure keeps a compiler rejection as +// Applied's explanation even when a required observation source is down. +func TestCompileFailureOutranksEventSourceFailure(t *testing.T) { + policy := monitorPolicyWithBehaviors("p", "uid-1", v1alpha1.PolicyBehavior{Exec: &v1alpha1.Behavior{Deny: &v1alpha1.BehaviorRule{Values: []string{"/bin/sh"}}}}) + sw, client := newTestStatusWriter(t, "node-a", policy) + sw.SetExpectedSourceNodes(func() ExpectedSourceNodes { + return ExpectedSourceNodes{Names: []string{"node-a"}, Desired: 1, Synced: true} + }) + if err := sw.RuntimePolicyEvent(&compiler.EvaluationResult{UID: "uid-1", Name: "p", Mode: compiler.ModeMonitor, + Exec: &compiler.AllowDenyPair{Deny: []string{"/bin/sh"}}}, events.EventTypeCreate); err != nil { + t.Fatal(err) + } + sw.RecordCondition("uid-1", "p", metav1.Condition{Type: v1alpha1.ConditionApplied, Status: metav1.ConditionFalse, + Reason: v1alpha1.ReasonCompileFailed, Message: "policy compilation failed"}) + sw.RecordSourceStatus(execTraceSource, runtimeevent.SourceStateUnavailable, runtimeevent.SourceReasonInitializationFailed) + if err := sw.Flush(context.Background()); err != nil { + t.Fatal(err) + } + got := getPolicy(t, client, "p") + applied := conditionOfType(t, got.Status.Conditions, v1alpha1.ConditionApplied) + if applied.Status != metav1.ConditionFalse || applied.Reason != v1alpha1.ReasonCompileFailed { + t.Errorf("Applied = (%s, %s), want (False, %s)", applied.Status, applied.Reason, v1alpha1.ReasonCompileFailed) + } +} + +// TestHealthySourceShardCannotEraseFailedNode keeps the aggregate false when +// a healthy node flushes after a different daemon reports source failure. +func TestHealthySourceShardCannotEraseFailedNode(t *testing.T) { + policy := monitorPolicyWithBehaviors("p", "uid-1", v1alpha1.PolicyBehavior{Exec: &v1alpha1.Behavior{Deny: &v1alpha1.BehaviorRule{Values: []string{"/bin/sh"}}}}) + swA, client := newTestStatusWriter(t, "node-a", policy) + swB := NewStatusWriter(client, "node-b", time.Hour, logr.Discard(), nil, nil) + swB.clock = func() time.Time { return fixedNow } + for _, sw := range []*StatusWriter{swA, swB} { + sw.RecordSourceStatus(openExecObserveSource, runtimeevent.SourceStateAvailable, runtimeevent.SourceReasonReady) + sw.SetExpectedSourceNodes(func() ExpectedSourceNodes { + return ExpectedSourceNodes{Names: []string{"node-a", "node-b"}, Desired: 2, Synced: true} + }) + if err := sw.RuntimePolicyEvent(&compiler.EvaluationResult{UID: "uid-1", Name: "p", Mode: compiler.ModeMonitor, + Exec: &compiler.AllowDenyPair{Deny: []string{"/bin/sh"}}}, events.EventTypeCreate); err != nil { + t.Fatal(err) + } + } + swB.RecordSourceStatus(execTraceSource, runtimeevent.SourceStateUnavailable, runtimeevent.SourceReasonReaderFailed) + if err := swB.Flush(context.Background()); err != nil { + t.Fatal(err) + } + swA.RecordSourceStatus(execTraceSource, runtimeevent.SourceStateAvailable, runtimeevent.SourceReasonReady) + if err := swA.Flush(context.Background()); err != nil { + t.Fatal(err) + } + got := getPolicy(t, client, "p") + if status := conditionOfType(t, got.Status.Conditions, v1alpha1.ConditionEventSourcesAvailable).Status; status != metav1.ConditionFalse { + t.Errorf("EventSourcesAvailable = %s, want False after node-a flushes", status) + } +} + +// TestMembershipChangeReconcilesCleanPolicy ensures a placement update causes +// a new aggregate even when no policy or source transition occurred. +func TestMembershipChangeReconcilesCleanPolicy(t *testing.T) { + policy := monitorPolicyWithBehaviors("p", "uid-1", v1alpha1.PolicyBehavior{Exec: &v1alpha1.Behavior{Deny: &v1alpha1.BehaviorRule{Values: []string{"/bin/sh"}}}}) + sw, client := newTestStatusWriter(t, "node-a", policy) + expected := ExpectedSourceNodes{Names: []string{"node-a"}, Desired: 1, Synced: true} + sw.SetExpectedSourceNodes(func() ExpectedSourceNodes { return expected }) + sw.RecordSourceStatus(execTraceSource, runtimeevent.SourceStateAvailable, runtimeevent.SourceReasonReady) + sw.RecordSourceStatus(openExecObserveSource, runtimeevent.SourceStateAvailable, runtimeevent.SourceReasonReady) + if err := sw.RuntimePolicyEvent(&compiler.EvaluationResult{UID: "uid-1", Name: "p", Mode: compiler.ModeMonitor, + Exec: &compiler.AllowDenyPair{Deny: []string{"/bin/sh"}}}, events.EventTypeCreate); err != nil { + t.Fatal(err) + } + if err := sw.Flush(context.Background()); err != nil { + t.Fatal(err) + } + if status := conditionOfType(t, getPolicy(t, client, "p").Status.Conditions, v1alpha1.ConditionEventSourcesAvailable).Status; status != metav1.ConditionTrue { + t.Fatalf("EventSourcesAvailable = %s, want True", status) + } + + expected = ExpectedSourceNodes{Names: []string{"node-a", "node-b"}, Desired: 2, Synced: true} + sw.MarkAllDirty() + if err := sw.Flush(context.Background()); err != nil { + t.Fatal(err) + } + if status := conditionOfType(t, getPolicy(t, client, "p").Status.Conditions, v1alpha1.ConditionEventSourcesAvailable).Status; status != metav1.ConditionUnknown { + t.Errorf("EventSourcesAvailable = %s, want Unknown after node-b is expected", status) + } +} + func ptrTime(t time.Time) *metav1.Time { mt := metav1.NewTime(t) return &mt diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index 1bc2bd22..ef1a9d25 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -4,6 +4,8 @@ package metrics import ( + "github.com/nirmata/runtime/pkg/runtimeevent" + "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" ) @@ -18,6 +20,10 @@ type Metrics struct { // EventsDropped counts events dropped by the collector, labeled by // source and reason. EventsDropped *prometheus.CounterVec + // SourceAvailable reports whether a source has announced readiness. + SourceAvailable *prometheus.GaugeVec + // SourceFailures counts source lifecycle failures by stable reason. + SourceFailures *prometheus.CounterVec // AttributionMisses counts events that could not be attributed to a // pod (see pkg/attribution.Index.Annotate). AttributionMisses prometheus.Counter @@ -53,6 +59,18 @@ func New(reg prometheus.Registerer) *Metrics { Help: "Total number of runtime events dropped, by source and reason.", }, []string{"source", "reason"}), + SourceAvailable: f.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: namespace, + Name: "source_available", + Help: "Whether a runtime event source is available (1) or unavailable (0).", + }, []string{"source"}), + + SourceFailures: f.NewCounterVec(prometheus.CounterOpts{ + Namespace: namespace, + Name: "source_failures_total", + Help: "Total runtime event source failures, by source and reason.", + }, []string{"source", "reason"}), + AttributionMisses: f.NewCounter(prometheus.CounterOpts{ Namespace: namespace, Name: "attribution_misses_total", @@ -78,3 +96,17 @@ func New(reg prometheus.Registerer) *Metrics { }, []string{"result"}), } } + +// RecordSourceStatus updates the source lifecycle metrics. It accepts the +// source status seam directly so all production lifecycle writes share one +// path. +func (m *Metrics) RecordSourceStatus(source string, state runtimeevent.SourceState, reason string) { + if state == runtimeevent.SourceStateAvailable { + m.SourceAvailable.WithLabelValues(source).Set(1) + } else { + m.SourceAvailable.WithLabelValues(source).Set(0) + } + if state == runtimeevent.SourceStateUnavailable { + m.SourceFailures.WithLabelValues(source, reason).Inc() + } +} diff --git a/pkg/metrics/metrics_test.go b/pkg/metrics/metrics_test.go index b00f51af..29915070 100644 --- a/pkg/metrics/metrics_test.go +++ b/pkg/metrics/metrics_test.go @@ -3,6 +3,8 @@ package metrics import ( "testing" + "github.com/nirmata/runtime/pkg/runtimeevent" + "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/testutil" ) @@ -26,6 +28,20 @@ func TestNew_CountersIncrement(t *testing.T) { inc: func() { m.EventsDropped.WithLabelValues("lsm-observe", "buffer_full").Inc() }, coll: m.EventsDropped, }, + { + name: "SourceAvailable", + inc: func() { + m.RecordSourceStatus("dnsquery", runtimeevent.SourceStateAvailable, runtimeevent.SourceReasonReady) + }, + coll: m.SourceAvailable, + }, + { + name: "SourceFailures", + inc: func() { + m.RecordSourceStatus("dnsquery", runtimeevent.SourceStateUnavailable, runtimeevent.SourceReasonReaderFailed) + }, + coll: m.SourceFailures, + }, { name: "FindingsEmitted", inc: func() { m.FindingsEmitted.WithLabelValues("deny-egress", "network").Inc() }, @@ -72,9 +88,11 @@ func TestNew_MetricsAreRegisteredAgainstProvidedRegisterer(t *testing.T) { // Nothing registered yet reports zero families with data until a // label combination is observed; force one on every vec plus the - // plain counter, then confirm the registry gathers all five. + // plain counter, then confirm the registry gathers every family. m.EventsIngested.WithLabelValues("s", "k").Inc() m.EventsDropped.WithLabelValues("s", "buffer_full").Inc() + m.RecordSourceStatus("s", runtimeevent.SourceStateStarting, runtimeevent.SourceReasonStarting) + m.RecordSourceStatus("s", runtimeevent.SourceStateUnavailable, runtimeevent.SourceReasonReaderFailed) m.AttributionMisses.Inc() m.FindingsEmitted.WithLabelValues("p", "network").Inc() m.ReportWrites.WithLabelValues("ok").Inc() @@ -87,6 +105,8 @@ func TestNew_MetricsAreRegisteredAgainstProvidedRegisterer(t *testing.T) { want := map[string]bool{ namespace + "_events_ingested_total": false, namespace + "_events_dropped_total": false, + namespace + "_source_available": false, + namespace + "_source_failures_total": false, namespace + "_attribution_misses_total": false, namespace + "_findings_emitted_total": false, namespace + "_report_writes_total": false, @@ -102,3 +122,24 @@ func TestNew_MetricsAreRegisteredAgainstProvidedRegisterer(t *testing.T) { } } } + +func TestRecordSourceStatusTracksAvailabilityAndFailures(t *testing.T) { + m := New(prometheus.NewRegistry()) + + m.RecordSourceStatus("exec-trace", runtimeevent.SourceStateStarting, runtimeevent.SourceReasonStarting) + if got := testutil.ToFloat64(m.SourceAvailable.WithLabelValues("exec-trace")); got != 0 { + t.Fatalf("SourceAvailable while starting = %v, want 0", got) + } + if got := testutil.ToFloat64(m.SourceFailures.WithLabelValues("exec-trace", runtimeevent.SourceReasonStarting)); got != 0 { + t.Fatalf("SourceFailures while starting = %v, want 0", got) + } + + m.RecordSourceStatus("exec-trace", runtimeevent.SourceStateAvailable, runtimeevent.SourceReasonReady) + m.RecordSourceStatus("exec-trace", runtimeevent.SourceStateUnavailable, runtimeevent.SourceReasonReaderFailed) + if got := testutil.ToFloat64(m.SourceAvailable.WithLabelValues("exec-trace")); got != 0 { + t.Errorf("SourceAvailable after failure = %v, want 0", got) + } + if got := testutil.ToFloat64(m.SourceFailures.WithLabelValues("exec-trace", runtimeevent.SourceReasonReaderFailed)); got != 1 { + t.Errorf("SourceFailures after failure = %v, want 1", got) + } +} diff --git a/pkg/runtimeevent/iface.go b/pkg/runtimeevent/iface.go index 95ae413c..119b949e 100644 --- a/pkg/runtimeevent/iface.go +++ b/pkg/runtimeevent/iface.go @@ -6,10 +6,52 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) +// SourceState is the lifecycle state of an event source. +type SourceState string + +const ( + SourceStateStarting SourceState = "Starting" + SourceStateAvailable SourceState = "Available" + SourceStateUnavailable SourceState = "Unavailable" +) + +const ( + SourceReasonInitializationFailed = "InitializationFailed" + SourceReasonReaderFailed = "ReaderFailed" + SourceReasonUnexpectedExit = "UnexpectedExit" + SourceReasonDependencyUnavailable = "DependencyUnavailable" + SourceReasonStarting = "Starting" + SourceReasonReady = "Ready" +) + +// SourceStatusFunc observes source lifecycle changes. Reasons are stable +// diagnostic categories and must not include raw error text. +type SourceStatusFunc func(source string, state SourceState, reason string) + +type sourceReadyContextKey struct{} + +// WithSourceReady installs the callback a source calls once it can produce +// events. A nil callback leaves ctx unchanged. +func WithSourceReady(ctx context.Context, ready func()) context.Context { + if ready == nil { + return ctx + } + return context.WithValue(ctx, sourceReadyContextKey{}, ready) +} + +// SourceReady reports that a source initialized from ctx is ready to produce +// events. Contexts without a readiness callback are valid. +func SourceReady(ctx context.Context) { + if ready, ok := ctx.Value(sourceReadyContextKey{}).(func()); ok { + ready() + } +} + // Source produces events. type Source interface { Name() string - // Run blocks until ctx is done. Sends events on out. Must not close out. + // Run calls SourceReady after initialization, then sends events until ctx + // is done. It must not close out. Run(ctx context.Context, out chan<- Event) error } diff --git a/pkg/runtimeevent/iface_test.go b/pkg/runtimeevent/iface_test.go index a6608bbc..d5b7313b 100644 --- a/pkg/runtimeevent/iface_test.go +++ b/pkg/runtimeevent/iface_test.go @@ -78,6 +78,17 @@ func TestSourceSinkSeamsAreImplementable(t *testing.T) { } } +func TestSourceReadyCallsCallbackOnlyWhenInstalled(t *testing.T) { + called := 0 + ctx := WithSourceReady(context.Background(), func() { called++ }) + SourceReady(ctx) + SourceReady(context.Background()) + + if called != 1 { + t.Errorf("ready callback calls = %d, want 1", called) + } +} + func TestPolicyStatusRecorderSeam(t *testing.T) { var rec PolicyStatusRecorder = &fakeRecorder{} rec.RecordCondition("policy-uid", "policy-name", metav1.Condition{Type: "Applied", Status: metav1.ConditionTrue, Reason: "Monitoring"})