diff --git a/CHANGELOG.md b/CHANGELOG.md index b601f78..5912f9b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,27 @@ All notable changes to a3s-observer will be documented in this file. +## [Unreleased] + +### Added + +- A provider-neutral workload attribution contract: `WorkloadIdentity` requires stable workload, + deployment, immutable revision, logical replica, provider-unit, and node IDs. Each + `WorkloadIdentityValue` is bounded to 128 ASCII bytes and rejects whitespace, control + characters, free-form Unicode, and non-canonical boundary characters without echoing rejected + input in validation errors. +- Explicit `ObservationMetadata` for sampled signals, including observation timestamp, optional + sample timestamp and collection interval, and `fresh`, `stale`, `unavailable`, or `unknown` + state. Unavailable and unknown observations represent missing data without a zero-valued + sentinel. +- Optional `workload` and `observation` fields in NDJSON `EnrichedEvent` output. + `IdentityResolver::resolve_workload` defaults to `None`, preserving existing resolver + implementations and preventing incomplete attribution. + +This contract is the first slice of workload metrics support. Multi-replica Linux collection, +resource/restart/availability measurements, lifecycle fixtures, and OTLP/Prometheus metric parity +are not included yet. + ## [0.11.0] — SecurityAction: privesc / injection / open-port ### Added diff --git a/README.md b/README.md index 7e3e414..b915a9c 100644 --- a/README.md +++ b/README.md @@ -83,6 +83,49 @@ Filter with `jq`, e.g. every LLM call and its provider: Set `A3S_OBSERVER_COLLECTOR_ID` and `A3S_NODE_NAME` in DaemonSets when you want stable collector identity in downstream fleet-health views. If unset, the collector falls back to pod/host names. +### Workload attribution and freshness contract + +`EnrichedEvent` can also carry provider-neutral `workload` and `observation` objects for +per-replica signals: + +```json +{ + "workload": { + "workload_id": "workload-01HV7F5N", + "deployment_id": "deployment-01HV7F6A", + "revision_id": "revision-sha256:8f3a", + "replica_id": "replica-0007", + "provider_unit_id": "containerd:4f6c2d8a", + "node_id": "node-us-east-1a-03" + }, + "observation": { + "observed_at_unix_nanos": 1720000015000000000, + "sampled_at_unix_nanos": 1720000014000000000, + "collection_interval_nanos": 15000000000, + "freshness": "fresh" + } +} +``` + +`WorkloadIdentity` is complete by construction: a producer must provide stable workload, +deployment, immutable revision, logical replica, current provider-unit, and node IDs. Each ID is +an opaque platform identifier limited to 128 ASCII bytes and a label-safe alphabet. Producers +must normalize provider IDs and must never copy tenant secrets, display names, or raw user labels +into these fields. A logical `replica_id` survives process restart, adoption, and rescheduling; +`provider_unit_id` changes when the runtime unit is replaced. + +`ObservationMetadata` reports an observation time, an optional collection interval, and one of +`fresh`, `stale`, `unavailable`, or `unknown`. Fresh and stale data include the actual sample +timestamp. Unavailable and unknown observations omit it, giving producers an explicit +missing-data state instead of a zero-usage sentinel. + +This is the transport contract, not a claim that per-replica resource collection is complete. +Existing `IdentityResolver` implementations return no workload identity by default, and the +node-wide collector intentionally does not apply one static environment identity to every event. +Multi-replica Linux collection, CPU/memory/network/process/restart/availability samples, +restart/adoption fixtures, and equivalent OTLP and Prometheus metric exporters remain follow-up +work. + ## Intervene — egress / file / exec (opt-in) The same vantage point enforces an **external policy** — a plain file any controller writes; the diff --git a/a3s-observer-collector/src/main.rs b/a3s-observer-collector/src/main.rs index f380c54..3d2e0e3 100644 --- a/a3s-observer-collector/src/main.rs +++ b/a3s-observer-collector/src/main.rs @@ -260,6 +260,8 @@ async fn main() -> anyhow::Result<()> { if let Some(ev) = read_pod::(&item) { emit(exporter.as_ref(), &mut stats, EnrichedEvent { identity: identity_for(&resolver, ev.pid, &ev.comm), + workload: resolver.resolve_workload(ev.pid, 0, 0), + observation: None, process: Some(process_context(ev.pid, &ev.comm)), provider: None, event: AgentEvent::ToolExec { @@ -276,6 +278,8 @@ async fn main() -> anyhow::Result<()> { if let Some(ev) = read_pod::(&item) { emit(exporter.as_ref(), &mut stats, EnrichedEvent { identity: identity_for(&resolver, ev.pid, &ev.comm), + workload: resolver.resolve_workload(ev.pid, 0, 0), + observation: None, process: Some(process_context(ev.pid, &ev.comm)), provider: None, event: AgentEvent::ProcessExit { @@ -296,6 +300,8 @@ async fn main() -> anyhow::Result<()> { }; emit(exporter.as_ref(), &mut stats, EnrichedEvent { identity: identity_for(&resolver, ev.pid, &ev.comm), + workload: resolver.resolve_workload(ev.pid, 0, 0), + observation: None, process: Some(process_context(ev.pid, &ev.comm)), provider: None, event: AgentEvent::SecurityAction { @@ -316,6 +322,8 @@ async fn main() -> anyhow::Result<()> { peers.insert(sock_key(ev.pid, ev.fd), (peer, ev.port)); emit(exporter.as_ref(), &mut stats, EnrichedEvent { identity: identity_for(&resolver, ev.pid, &ev.comm), + workload: resolver.resolve_workload(ev.pid, 0, 0), + observation: None, process: Some(process_context(ev.pid, &ev.comm)), provider: None, event: AgentEvent::Egress { @@ -347,6 +355,8 @@ async fn main() -> anyhow::Result<()> { llm_meta.insert(sock_key(ev.pid, ev.fd), (sni.clone(), provider.clone(), peer)); emit(exporter.as_ref(), &mut stats, EnrichedEvent { identity: identity_for(&resolver, ev.pid, &ev.comm), + workload: resolver.resolve_workload(ev.pid, 0, 0), + observation: None, process: Some(process_context(ev.pid, &ev.comm)), provider, event: AgentEvent::Egress { @@ -365,6 +375,8 @@ async fn main() -> anyhow::Result<()> { if let Some(query) = parse_dns_qname(&ev.data[..len]) { emit(exporter.as_ref(), &mut stats, EnrichedEvent { identity: identity_for(&resolver, ev.pid, &ev.comm), + workload: resolver.resolve_workload(ev.pid, 0, 0), + observation: None, process: Some(process_context(ev.pid, &ev.comm)), provider: None, event: AgentEvent::Dns { pid: ev.pid, query }, @@ -388,6 +400,8 @@ async fn main() -> anyhow::Result<()> { }; emit(exporter.as_ref(), &mut stats, EnrichedEvent { identity: identity_for(&resolver, ev.pid, &ev.comm), + workload: resolver.resolve_workload(ev.pid, 0, 0), + observation: None, process: Some(process_context(ev.pid, &ev.comm)), provider: None, event, @@ -405,6 +419,8 @@ async fn main() -> anyhow::Result<()> { if provider.is_some() { emit(exporter.as_ref(), &mut stats, EnrichedEvent { identity: identity_for(&resolver, ev.pid, &ev.comm), + workload: resolver.resolve_workload(ev.pid, 0, 0), + observation: None, process: Some(process_context(ev.pid, &ev.comm)), provider, event: AgentEvent::LlmCall { @@ -428,12 +444,15 @@ async fn main() -> anyhow::Result<()> { let content = String::from_utf8_lossy(&ev.data[..len]).into_owned(); if !content.is_empty() { let identity = identity_for(&resolver, ev.pid, &ev.comm); + let workload = resolver.resolve_workload(ev.pid, 0, 0); // Structured LLM telemetry (model/tokens) alongside the raw content. if let Some((model, prompt_tokens, completion_tokens)) = parse_llm_meta(&content) { emit(exporter.as_ref(), &mut stats, EnrichedEvent { identity: identity.clone(), + workload: workload.clone(), + observation: None, process: Some(process_context(ev.pid, &ev.comm)), provider: None, event: AgentEvent::LlmApi { @@ -447,6 +466,8 @@ async fn main() -> anyhow::Result<()> { } emit(exporter.as_ref(), &mut stats, EnrichedEvent { identity, + workload, + observation: None, process: Some(process_context(ev.pid, &ev.comm)), provider: None, event: AgentEvent::SslContent { @@ -674,6 +695,8 @@ fn emit_collector_heartbeat( ) { exporter.export(&EnrichedEvent { identity: Identity::default(), + workload: None, + observation: None, process: None, provider: None, event: AgentEvent::CollectorHeartbeat { diff --git a/src/lib.rs b/src/lib.rs index 9365be4..a4c93ff 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -16,6 +16,7 @@ pub mod model; pub mod policy; pub mod traits; +pub mod workload; pub use model::{AgentEvent, EnrichedEvent, ProcessContext}; pub use policy::{parse_egress_policy, AllowAll, Policy, ProviderPolicy, Verdict}; @@ -23,3 +24,7 @@ pub use traits::{ read_ppid, Exporter, Identity, IdentityResolver, JsonExporter, KubeResolver, LogExporter, ProcResolver, Provider, ServiceClassifier, SniClassifier, }; +pub use workload::{ + Freshness, ObservationMetadata, ObservationMetadataError, WorkloadIdentity, + WorkloadIdentityValue, WorkloadIdentityValueError, MAX_WORKLOAD_IDENTITY_VALUE_LEN, +}; diff --git a/src/model.rs b/src/model.rs index ed31e89..9c84724 100644 --- a/src/model.rs +++ b/src/model.rs @@ -2,6 +2,7 @@ //! events the [`Exporter`](crate::Exporter) receives. use crate::traits::{Identity, Provider}; +use crate::workload::{ObservationMetadata, WorkloadIdentity}; use serde::Serialize; use std::net::IpAddr; use std::time::Duration; @@ -137,6 +138,12 @@ pub enum AgentEvent { #[derive(Debug, Clone, Serialize)] pub struct EnrichedEvent { pub identity: Identity, + /// Complete workload attribution, when a resolver can prove every stable identity field. + #[serde(skip_serializing_if = "Option::is_none")] + pub workload: Option, + /// Explicit timing and freshness for sampled signals. Consumers must not infer zero when absent. + #[serde(skip_serializing_if = "Option::is_none")] + pub observation: Option, #[serde(skip_serializing_if = "Option::is_none")] pub process: Option, pub provider: Option, diff --git a/src/traits.rs b/src/traits.rs index 66e380c..8d7bb47 100644 --- a/src/traits.rs +++ b/src/traits.rs @@ -5,6 +5,7 @@ //! probes. use crate::model::EnrichedEvent; +use crate::workload::WorkloadIdentity; use serde::Serialize; use std::net::IpAddr; @@ -21,6 +22,19 @@ pub struct Identity { /// Implementations: k8s (cgroup→pod), docker, a3s-box (pid/netns→box), bare pid-tree. pub trait IdentityResolver: Send + Sync { fn resolve(&self, pid: u32, cgroup_id: u64, netns: u64) -> Identity; + + /// Resolve a complete, provider-neutral workload identity when one is available. + /// + /// Existing process-only resolvers remain valid and default to no workload attribution. + /// Implementations must return `None` rather than inventing or partially filling identity. + fn resolve_workload( + &self, + _pid: u32, + _cgroup_id: u64, + _netns: u64, + ) -> Option { + None + } } /// Known service providers, identified language-agnostically from TLS SNI / DNS. @@ -346,6 +360,8 @@ mod tests { task: Some("1".into()), session: None, }, + workload: None, + observation: None, process: None, provider: None, event: AgentEvent::ProcessExit { diff --git a/src/workload.rs b/src/workload.rs new file mode 100644 index 0000000..a94b270 --- /dev/null +++ b/src/workload.rs @@ -0,0 +1,327 @@ +//! Provider-neutral workload attribution and sample freshness contracts. +//! +//! These types deliberately carry stable, platform-issued identifiers rather than display +//! names or arbitrary provider labels. A complete identity is required before a signal can be +//! attributed to a workload replica; partial identity must remain absent. + +use serde::{Deserialize, Deserializer, Serialize}; +use std::fmt; +use std::num::NonZeroU64; +use std::str::FromStr; + +/// Maximum encoded length of one workload identity component. +/// +/// This bound keeps downstream metric-label storage predictable. Producers must use canonical, +/// opaque platform identifiers and must not put tenant secrets or raw user labels in these +/// values. +pub const MAX_WORKLOAD_IDENTITY_VALUE_LEN: usize = 128; + +/// One validated component of a [`WorkloadIdentity`]. +/// +/// Values are ASCII, bounded, begin and end with an alphanumeric character, and may contain +/// alphanumerics plus `.`, `_`, `-`, `:`, and `/` internally. The restricted alphabet rejects +/// whitespace, control characters, and common free-form display labels; producers remain +/// responsible for supplying platform-issued IDs rather than user input. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize)] +#[serde(transparent)] +pub struct WorkloadIdentityValue(String); + +impl WorkloadIdentityValue { + /// Validate and construct an identity value. + pub fn new(value: impl Into) -> Result { + let value = value.into(); + validate_identity_value(&value)?; + Ok(Self(value)) + } + + /// Return the canonical string value. + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl AsRef for WorkloadIdentityValue { + fn as_ref(&self) -> &str { + self.as_str() + } +} + +impl fmt::Display for WorkloadIdentityValue { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(self.as_str()) + } +} + +impl FromStr for WorkloadIdentityValue { + type Err = WorkloadIdentityValueError; + + fn from_str(value: &str) -> Result { + Self::new(value) + } +} + +impl TryFrom for WorkloadIdentityValue { + type Error = WorkloadIdentityValueError; + + fn try_from(value: String) -> Result { + Self::new(value) + } +} + +impl TryFrom<&str> for WorkloadIdentityValue { + type Error = WorkloadIdentityValueError; + + fn try_from(value: &str) -> Result { + Self::new(value) + } +} + +impl<'de> Deserialize<'de> for WorkloadIdentityValue { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let value = String::deserialize(deserializer)?; + Self::new(value).map_err(serde::de::Error::custom) + } +} + +/// Why a workload identity component was rejected. +/// +/// Error messages intentionally never contain the rejected value because it may have originated +/// in an untrusted provider label. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum WorkloadIdentityValueError { + /// The value was empty. + Empty, + /// The value exceeded [`MAX_WORKLOAD_IDENTITY_VALUE_LEN`]. + TooLong, + /// The value did not use the canonical identity alphabet or boundary characters. + InvalidFormat, +} + +impl fmt::Display for WorkloadIdentityValueError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Empty => formatter.write_str("workload identity value cannot be empty"), + Self::TooLong => write!( + formatter, + "workload identity value exceeds {MAX_WORKLOAD_IDENTITY_VALUE_LEN} bytes" + ), + Self::InvalidFormat => { + formatter.write_str("workload identity value has an invalid format") + } + } + } +} + +impl std::error::Error for WorkloadIdentityValueError {} + +fn validate_identity_value(value: &str) -> Result<(), WorkloadIdentityValueError> { + if value.is_empty() { + return Err(WorkloadIdentityValueError::Empty); + } + if value.len() > MAX_WORKLOAD_IDENTITY_VALUE_LEN { + return Err(WorkloadIdentityValueError::TooLong); + } + + let bytes = value.as_bytes(); + let has_canonical_boundaries = bytes.first().is_some_and(u8::is_ascii_alphanumeric) + && bytes.last().is_some_and(u8::is_ascii_alphanumeric); + let uses_canonical_alphabet = bytes.iter().all(|byte| { + byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-' | b':' | b'/') + }); + if !has_canonical_boundaries || !uses_canonical_alphabet { + return Err(WorkloadIdentityValueError::InvalidFormat); + } + Ok(()) +} + +/// Complete, provider-neutral attribution for one observed workload replica. +/// +/// All fields are required so consumers never accept a partially attributed metric. Identity +/// lifecycle semantics: +/// +/// - `workload_id` identifies the long-lived workload. +/// - `deployment_id` identifies the deployment rollout. +/// - `revision_id` identifies immutable desired workload content. +/// - `replica_id` identifies the logical replica and survives process restart, adoption, and +/// rescheduling. +/// - `provider_unit_id` identifies the current runtime unit and changes when that unit is +/// replaced. +/// - `node_id` identifies the node that produced the observation. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct WorkloadIdentity { + pub workload_id: WorkloadIdentityValue, + pub deployment_id: WorkloadIdentityValue, + pub revision_id: WorkloadIdentityValue, + pub replica_id: WorkloadIdentityValue, + pub provider_unit_id: WorkloadIdentityValue, + pub node_id: WorkloadIdentityValue, +} + +impl WorkloadIdentity { + /// Construct a complete workload identity. + pub fn new( + workload_id: WorkloadIdentityValue, + deployment_id: WorkloadIdentityValue, + revision_id: WorkloadIdentityValue, + replica_id: WorkloadIdentityValue, + provider_unit_id: WorkloadIdentityValue, + node_id: WorkloadIdentityValue, + ) -> Self { + Self { + workload_id, + deployment_id, + revision_id, + replica_id, + provider_unit_id, + node_id, + } + } +} + +/// Whether an observation represents current, old, missing, or indeterminate data. +/// +/// `Unavailable` and `Unknown` observations have no sample timestamp, while `Fresh` and `Stale` +/// always do. Producers must use the missing-data states rather than substituting a zero-valued +/// resource sample. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum Freshness { + /// A current sample was collected successfully. + Fresh, + /// A prior sample exists but is outside the producer's freshness policy. + Stale, + /// The producer confirmed that the target or signal source was unavailable. + Unavailable, + /// The producer cannot currently determine sample availability or age. + Unknown, +} + +/// Timing and freshness metadata for one observation. +/// +/// Timestamps and intervals use Unix-epoch nanoseconds to map directly to OTLP without float +/// conversion. `observed_at_unix_nanos` records when the producer made the freshness +/// determination. A fresh or stale observation also carries the actual sample timestamp. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +pub struct ObservationMetadata { + observed_at_unix_nanos: u64, + #[serde(skip_serializing_if = "Option::is_none")] + sampled_at_unix_nanos: Option, + #[serde(skip_serializing_if = "Option::is_none")] + collection_interval_nanos: Option, + freshness: Freshness, +} + +impl ObservationMetadata { + /// Construct metadata for a current sample. + pub fn fresh( + observed_at_unix_nanos: u64, + sampled_at_unix_nanos: u64, + collection_interval_nanos: Option, + ) -> Result { + Self::sampled( + observed_at_unix_nanos, + sampled_at_unix_nanos, + collection_interval_nanos, + Freshness::Fresh, + ) + } + + /// Construct metadata for an old sample that must not be treated as current. + pub fn stale( + observed_at_unix_nanos: u64, + sampled_at_unix_nanos: u64, + collection_interval_nanos: Option, + ) -> Result { + Self::sampled( + observed_at_unix_nanos, + sampled_at_unix_nanos, + collection_interval_nanos, + Freshness::Stale, + ) + } + + /// Construct metadata for a confirmed missing sample. + pub fn unavailable( + observed_at_unix_nanos: u64, + collection_interval_nanos: Option, + ) -> Self { + Self { + observed_at_unix_nanos, + sampled_at_unix_nanos: None, + collection_interval_nanos, + freshness: Freshness::Unavailable, + } + } + + /// Construct metadata when sample availability or age cannot be determined. + pub fn unknown( + observed_at_unix_nanos: u64, + collection_interval_nanos: Option, + ) -> Self { + Self { + observed_at_unix_nanos, + sampled_at_unix_nanos: None, + collection_interval_nanos, + freshness: Freshness::Unknown, + } + } + + /// Time at which the producer made this observation. + pub fn observed_at_unix_nanos(&self) -> u64 { + self.observed_at_unix_nanos + } + + /// Actual sample time for fresh or stale data. + pub fn sampled_at_unix_nanos(&self) -> Option { + self.sampled_at_unix_nanos + } + + /// Intended collection interval, when known. + pub fn collection_interval_nanos(&self) -> Option { + self.collection_interval_nanos + } + + /// Explicit current state of the observation. + pub fn freshness(&self) -> Freshness { + self.freshness + } + + fn sampled( + observed_at_unix_nanos: u64, + sampled_at_unix_nanos: u64, + collection_interval_nanos: Option, + freshness: Freshness, + ) -> Result { + if sampled_at_unix_nanos > observed_at_unix_nanos { + return Err(ObservationMetadataError::SampleAfterObservation); + } + Ok(Self { + observed_at_unix_nanos, + sampled_at_unix_nanos: Some(sampled_at_unix_nanos), + collection_interval_nanos, + freshness, + }) + } +} + +/// Invalid timing relationship in [`ObservationMetadata`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ObservationMetadataError { + /// A sample claimed to have been captured after its observation was emitted. + SampleAfterObservation, +} + +impl fmt::Display for ObservationMetadataError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::SampleAfterObservation => { + formatter.write_str("sample timestamp cannot be later than observation timestamp") + } + } + } +} + +impl std::error::Error for ObservationMetadataError {} diff --git a/tests/workload_contract.rs b/tests/workload_contract.rs new file mode 100644 index 0000000..a9f46dd --- /dev/null +++ b/tests/workload_contract.rs @@ -0,0 +1,187 @@ +use a3s_observer::{ + AgentEvent, EnrichedEvent, Freshness, Identity, IdentityResolver, ObservationMetadata, + WorkloadIdentity, WorkloadIdentityValue, MAX_WORKLOAD_IDENTITY_VALUE_LEN, +}; +use serde_json::json; +use std::num::NonZeroU64; + +fn identity_value(value: &str) -> WorkloadIdentityValue { + WorkloadIdentityValue::new(value).expect("fixture identity must be valid") +} + +fn workload_identity() -> WorkloadIdentity { + WorkloadIdentity::new( + identity_value("workload-01HV7F5N"), + identity_value("deployment-01HV7F6A"), + identity_value("revision-sha256:8f3a"), + identity_value("replica-0007"), + identity_value("containerd:4f6c2d8a"), + identity_value("node-us-east-1a-03"), + ) +} + +#[test] +fn workload_identity_serializes_as_complete_stable_ids() { + assert_eq!( + serde_json::to_value(workload_identity()).unwrap(), + json!({ + "workload_id": "workload-01HV7F5N", + "deployment_id": "deployment-01HV7F6A", + "revision_id": "revision-sha256:8f3a", + "replica_id": "replica-0007", + "provider_unit_id": "containerd:4f6c2d8a", + "node_id": "node-us-east-1a-03" + }) + ); +} + +#[test] +fn workload_identity_values_are_bounded_and_safe_for_labels() { + let max = "a".repeat(MAX_WORKLOAD_IDENTITY_VALUE_LEN); + assert!(WorkloadIdentityValue::new(max).is_ok()); + + for rejected in [ + "", + "raw tenant label", + "line\nbreak", + "control\u{0007}byte", + "非规范标识", + "-leading-separator", + "trailing-separator-", + ] { + let error = WorkloadIdentityValue::new(rejected).unwrap_err(); + assert!( + rejected.is_empty() || !error.to_string().contains(rejected), + "validation errors must not echo rejected identity values" + ); + } + + let overlong = "s".repeat(MAX_WORKLOAD_IDENTITY_VALUE_LEN + 1); + let error = WorkloadIdentityValue::new(&overlong).unwrap_err(); + assert!(!error.to_string().contains(&overlong)); + + let decoded = serde_json::from_str::("\"raw tenant label\""); + assert!(decoded.is_err(), "deserialization must preserve validation"); +} + +#[test] +fn observation_metadata_distinguishes_samples_from_missing_data() { + let interval = NonZeroU64::new(15_000_000_000); + let fresh = ObservationMetadata::fresh( + 1_720_000_015_000_000_000, + 1_720_000_014_000_000_000, + interval, + ) + .unwrap(); + let stale = ObservationMetadata::stale( + 1_720_000_030_000_000_000, + 1_720_000_014_000_000_000, + interval, + ) + .unwrap(); + let unavailable = ObservationMetadata::unavailable(1_720_000_030_000_000_000, interval); + let unknown = ObservationMetadata::unknown(1_720_000_030_000_000_000, None); + + assert_eq!(fresh.freshness(), Freshness::Fresh); + assert_eq!(stale.freshness(), Freshness::Stale); + assert_eq!(unavailable.freshness(), Freshness::Unavailable); + assert_eq!(unknown.freshness(), Freshness::Unknown); + assert!(unavailable.sampled_at_unix_nanos().is_none()); + assert!(unknown.sampled_at_unix_nanos().is_none()); + + assert_eq!( + serde_json::to_value(unavailable).unwrap(), + json!({ + "observed_at_unix_nanos": 1_720_000_030_000_000_000u64, + "collection_interval_nanos": 15_000_000_000u64, + "freshness": "unavailable" + }) + ); +} + +#[test] +fn sample_timestamp_cannot_be_after_observation_timestamp() { + let error = ObservationMetadata::fresh(10, 11, None).unwrap_err(); + assert_eq!( + error.to_string(), + "sample timestamp cannot be later than observation timestamp" + ); +} + +struct WorkloadResolver; + +impl IdentityResolver for WorkloadResolver { + fn resolve(&self, _pid: u32, _cgroup_id: u64, _netns: u64) -> Identity { + Identity { + agent: Some("agent-7".into()), + task: Some("task-3".into()), + session: None, + } + } + + fn resolve_workload( + &self, + _pid: u32, + _cgroup_id: u64, + _netns: u64, + ) -> Option { + Some(workload_identity()) + } +} + +struct LegacyResolver; + +impl IdentityResolver for LegacyResolver { + fn resolve(&self, _pid: u32, _cgroup_id: u64, _netns: u64) -> Identity { + Identity::default() + } +} + +#[test] +fn resolver_workload_identity_and_observation_reach_ndjson() { + let resolver = WorkloadResolver; + let event = EnrichedEvent { + identity: resolver.resolve(7, 11, 13), + workload: resolver.resolve_workload(7, 11, 13), + observation: Some(ObservationMetadata::fresh(1_720_000_015, 1_720_000_014, None).unwrap()), + process: None, + provider: None, + event: AgentEvent::ProcessExit { + pid: 7, + exit_code: 0, + signal: 0, + }, + }; + + let line = serde_json::to_string(&event).unwrap(); + assert!(!line.contains('\n')); + let value: serde_json::Value = serde_json::from_str(&line).unwrap(); + assert_eq!(value["workload"]["replica_id"], "replica-0007"); + assert_eq!(value["observation"]["freshness"], "fresh"); + assert_eq!( + value["observation"]["sampled_at_unix_nanos"], + 1_720_000_014u64 + ); +} + +#[test] +fn existing_identity_resolvers_default_to_no_workload_identity() { + let resolver = LegacyResolver; + assert_eq!(resolver.resolve_workload(1, 2, 3), None); + + let event = EnrichedEvent { + identity: resolver.resolve(1, 2, 3), + workload: resolver.resolve_workload(1, 2, 3), + observation: None, + process: None, + provider: None, + event: AgentEvent::ProcessExit { + pid: 1, + exit_code: 0, + signal: 0, + }, + }; + let value = serde_json::to_value(event).unwrap(); + assert!(value.get("workload").is_none()); + assert!(value.get("observation").is_none()); +}