feat(server): export gateway OCSF events as OTLP log records (#4500)

The gateway only persisted its OCSF events to the JSONL file, so collectors had
to tail that file. Adding `[openshell.gateway.otlp.ocsf]` now also exports each
gateway OCSF event as an OTLP log record, using the same resource identity as
gateway traces. `schema_version` optionally downgrades the exported JSON to
OCSF 1.1 or 1.3.

OTLP endpoints are now configured per signal, so OCSF events can go to a
collector that feeds a SIEM while spans go elsewhere. The existing
`[openshell.gateway.otlp]` `endpoint` is the default for every signal and on
its own still enables trace export. `[otlp.traces]` overrides the trace
endpoint, and `[otlp.ocsf]` takes its own `endpoint` or falls back to the
default.

Each record carries the OCSF JSON as its body, the event time as its timestamp,
a severity mapped from `severity_id`, and `ocsf.*` and `openshell.sandbox.id`
attributes for filtering without parsing the body.

Records flow through a queue bounded by count and bytes to the SDK OTLP log
exporter in batches bounded the same way. Failed exports retry with backoff;
records lost to a full queue, an oversized body, or exhausted retries are
counted in metrics and reported to the collector in an `openshell.ocsf.gap`
record on the next successful export. The exporter lives in `openshell-otel` so
other processes can share it.

Helm deployments set the `traces` and `ocsf` tables through `gatewayConfig`,
and the local development gateway tasks enable OCSF export.

Closes #4499
Refs #2892

Signed-off-by: Kris Hicks <khicks@nvidia.com>

Author: krishicks

Cargo.lock

@@ -41,6 +41,7 @@ use openshell_core::telemetry::TelemetryComputeDriver;
 ))]
 use openshell_server::ComputeDriverRegistration;
 use openshell_server::ComputeDriverRegistry;
+use openshell_server::config_file::OtlpConfig;
 
 #[cfg(all(
     not(target_os = "windows"),
@@ -265,7 +266,10 @@ impl openshell_server::ComputeDriverFactory for KubernetesFactory {
         context: openshell_server::ComputeDriverBuildContext<'_>,
     ) -> openshell_core::Result<openshell_server::ComputeDriverInstance> {
         let mut config = kubernetes_config(context.config_context())?;
-        config.supervisor_otlp_endpoint = context.otlp_config().map(|otlp| otlp.endpoint.clone());
+        config.supervisor_otlp_endpoint = context
+            .otlp_config()
+            .and_then(OtlpConfig::traces_endpoint)
+            .map(str::to_string);
         let driver = openshell_driver_kubernetes::KubernetesComputeDriver::new(
             config,
             context.shutdown_receiver(),
@@ -326,7 +330,10 @@ impl openshell_server::ComputeDriverFactory for DockerFactory {
         context: openshell_server::ComputeDriverBuildContext<'_>,
     ) -> openshell_core::Result<openshell_server::ComputeDriverInstance> {
         let mut config = docker_config(context.config_context())?;
-        config.supervisor_otlp_endpoint = context.otlp_config().map(|otlp| otlp.endpoint.clone());
+        config.supervisor_otlp_endpoint = context
+            .otlp_config()
+            .and_then(OtlpConfig::traces_endpoint)
+            .map(str::to_string);
         require_guest_tls_for_local_driver(&context, "docker")?;
         apply_guest_tls(&mut config.guest_tls_ca, context.guest_tls_ca());
         let driver = openshell_driver_docker::DockerComputeDriver::new(
@@ -388,7 +395,10 @@ impl openshell_server::ComputeDriverFactory for PodmanFactory {
         context: openshell_server::ComputeDriverBuildContext<'_>,
     ) -> openshell_core::Result<openshell_server::ComputeDriverInstance> {
         let mut config = podman_config(context.config_context())?;
-        config.supervisor_otlp_endpoint = context.otlp_config().map(|otlp| otlp.endpoint.clone());
+        config.supervisor_otlp_endpoint = context
+            .otlp_config()
+            .and_then(OtlpConfig::traces_endpoint)
+            .map(str::to_string);
         require_guest_tls_for_local_driver(&context, "podman")?;
         apply_guest_tls(&mut config.guest_tls_ca, context.guest_tls_ca());
         let driver = openshell_driver_podman::PodmanComputeDriver::new(config)

crates/openshell-gateway/src/lib.rs

@@ -660,8 +660,8 @@ fn append_vm_proxy_and_spiffe_args(command: &mut Command, config: &VmComputeConf
 }
 
 fn append_otlp_args(command: &mut Command, otlp_config: Option<&OtlpConfig>, gateway_name: &str) {
-    if let Some(config) = otlp_config {
-        command.arg("--otlp-endpoint").arg(&config.endpoint);
+    if let Some(endpoint) = otlp_config.and_then(OtlpConfig::traces_endpoint) {
+        command.arg("--otlp-endpoint").arg(endpoint);
         command.arg("--gateway-name").arg(gateway_name);
     }
 }
@@ -760,7 +760,7 @@ mod tests {
         validate_vm_sandbox_identity,
     };
     use openshell_core::UpstreamProxyConfig;
-    use openshell_server::config_file::OtlpConfig;
+    use openshell_server::config_file::{OtlpConfig, OtlpTracesConfig};
     use std::os::unix::fs::PermissionsExt;
     use std::os::unix::net::UnixListener as StdUnixListener;
     use std::path::PathBuf;
@@ -808,8 +808,11 @@ mod tests {
         append_otlp_args(
             &mut command,
             Some(&OtlpConfig {
-                endpoint: "http://collector.internal:4317".to_string(),
                 service_name: Some("custom-gateway".to_string()),
+                traces: Some(OtlpTracesConfig {
+                    endpoint: "http://collector.internal:4317".to_string(),
+                }),
+                ocsf: None,
             }),
             "production-us-west",
         );

crates/openshell-gateway/src/vm.rs

@@ -6,8 +6,41 @@
 //! Transforms serialized OCSF JSON events to conform to older schema versions
 //! by stripping fields and profiles that don't exist in the target version.
 
+use serde::{Deserialize, Serialize};
 use serde_json::Value;
 
+/// An older OCSF schema version that events can be downgraded to.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub enum OcsfSchemaVersion {
+    #[serde(rename = "1.1")]
+    V1_1,
+    #[serde(rename = "1.3")]
+    V1_3,
+}
+
+impl OcsfSchemaVersion {
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::V1_1 => "1.1",
+            Self::V1_3 => "1.3",
+        }
+    }
+}
+
+impl std::str::FromStr for OcsfSchemaVersion {
+    type Err = String;
+
+    fn from_str(version: &str) -> Result<Self, Self::Err> {
+        match version {
+            "1.1" => Ok(Self::V1_1),
+            "1.3" => Ok(Self::V1_3),
+            other => Err(format!(
+                "unsupported OCSF schema version {other:?}; expected 1.1 or 1.3"
+            )),
+        }
+    }
+}
+
 /// Fields to strip when downgrading to v1.3.0 or earlier.
 const STRIP_FOR_V1_3: &[&str] = &["ai_model", "container", "observation_point_id"];
 
@@ -96,6 +129,15 @@ fn parse_version(v: &str) -> (u32, u32, u32) {
 mod tests {
     use super::*;
 
+    #[test]
+    fn schema_versions_parse_from_their_names() {
+        for version in [OcsfSchemaVersion::V1_1, OcsfSchemaVersion::V1_3] {
+            assert_eq!(version.as_str().parse(), Ok(version));
+        }
+        assert!("1.6".parse::<OcsfSchemaVersion>().is_err());
+        assert!("".parse::<OcsfSchemaVersion>().is_err());
+    }
+
     fn test_event() -> Value {
         serde_json::json!({
             "class_uid": 4002,

crates/openshell-ocsf/src/format/downgrade.rs

@@ -59,6 +59,8 @@ pub use objects::{
 // --- Sanitizers ---
 pub use sanitize::sanitize_reason_for_audit;
 
+pub use format::downgrade::OcsfSchemaVersion;
+
 // --- Builders ---
 pub use builders::{
     ApiActivityBuilder, AppLifecycleBuilder, BaseEventBuilder, ConfigStateChangeBuilder,

crates/openshell-ocsf/src/lib.rs

@@ -12,7 +12,7 @@ repository.workspace = true
 
 [dependencies]
 openshell-otel = { path = "../openshell-otel" }
-opentelemetry-proto = { version = "0.32", default-features = false, features = ["gen-tonic", "trace"] }
+opentelemetry-proto = { version = "0.32", default-features = false, features = ["gen-tonic", "logs", "trace"] }
 tokio = { workspace = true }
 tokio-stream = { workspace = true }
 tonic = { workspace = true, features = ["gzip", "tls-aws-lc"] }

crates/openshell-otel-test-support/Cargo.toml

@@ -6,10 +6,16 @@
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
+use opentelemetry_proto::tonic::collector::logs::v1::{
+    ExportLogsServiceRequest, ExportLogsServiceResponse,
+    logs_service_server::{LogsService, LogsServiceServer},
+};
 use opentelemetry_proto::tonic::collector::trace::v1::{
     ExportTraceServiceRequest, ExportTraceServiceResponse,
     trace_service_server::{TraceService, TraceServiceServer},
 };
+use opentelemetry_proto::tonic::common::v1::any_value::Value;
+use opentelemetry_proto::tonic::resource::v1::Resource;
 use opentelemetry_proto::tonic::trace::v1::Span;
 
 /// Serialize tracing tests and keep their callsites enabled process-wide.
@@ -68,12 +74,34 @@ pub async fn assert_compute_driver_tracing(
 #[derive(Clone, Debug, Default)]
 pub struct ReceivedTraces {
     pub spans: Vec<Span>,
+    /// String bodies of exported log records.
+    pub log_bodies: Vec<String>,
     pub service_names: Vec<String>,
     pub gateway_names: Vec<String>,
     pub compute_drivers: Vec<String>,
     pub encodings: Vec<String>,
 }
 
+impl ReceivedTraces {
+    fn record_resource(&mut self, resource: Option<Resource>) {
+        for attribute in resource
+            .into_iter()
+            .flat_map(|resource| resource.attributes)
+        {
+            let Some(Value::StringValue(value)) = attribute.value.and_then(|value| value.value)
+            else {
+                continue;
+            };
+            match attribute.key.as_str() {
+                "service.name" => self.service_names.push(value),
+                "openshell.gateway.name" => self.gateway_names.push(value),
+                "openshell.gateway.compute_driver" => self.compute_drivers.push(value),
+                _ => {}
+            }
+        }
+    }
+}
+
 #[derive(Clone)]
 struct Collector {
     received: Arc<Mutex<ReceivedTraces>>,
@@ -99,27 +127,7 @@ impl TraceService for Collector {
                 received.encodings.push(encoding.to_string());
             }
             for resource_span in request.into_inner().resource_spans {
-                if let Some(resource) = resource_span.resource {
-                    for attribute in resource.attributes {
-                        let Some(value) = attribute.value.and_then(|value| value.value) else {
-                            continue;
-                        };
-                        let opentelemetry_proto::tonic::common::v1::any_value::Value::StringValue(
-                            value,
-                        ) = value
-                        else {
-                            continue;
-                        };
-                        match attribute.key.as_str() {
-                            "service.name" => received.service_names.push(value),
-                            "openshell.gateway.name" => received.gateway_names.push(value),
-                            "openshell.gateway.compute_driver" => {
-                                received.compute_drivers.push(value);
-                            }
-                            _ => {}
-                        }
-                    }
-                }
+                received.record_resource(resource_span.resource);
                 for scope_span in resource_span.scope_spans {
                     received.spans.extend(scope_span.spans);
                 }
@@ -130,7 +138,36 @@ impl TraceService for Collector {
     }
 }
 
-/// Loopback OTLP/gRPC server that captures exported spans and resources.
+#[tonic::async_trait]
+impl LogsService for Collector {
+    async fn export(
+        &self,
+        request: tonic::Request<ExportLogsServiceRequest>,
+    ) -> Result<tonic::Response<ExportLogsServiceResponse>, tonic::Status> {
+        {
+            let mut received = self
+                .received
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner);
+            for resource_logs in request.into_inner().resource_logs {
+                received.record_resource(resource_logs.resource);
+                for scope_logs in resource_logs.scope_logs {
+                    for record in scope_logs.log_records {
+                        if let Some(Value::StringValue(body)) =
+                            record.body.and_then(|body| body.value)
+                        {
+                            received.log_bodies.push(body);
+                        }
+                    }
+                }
+            }
+        }
+        self.exported.notify_one();
+        Ok(tonic::Response::new(ExportLogsServiceResponse::default()))
+    }
+}
+
+/// Loopback OTLP/gRPC server that captures exported spans, logs, and resources.
 pub struct OtlpTestServer {
     endpoint: String,
     received: Arc<Mutex<ReceivedTraces>>,
@@ -171,12 +208,15 @@ impl OtlpTestServer {
                 .tls_config(tonic::transport::ServerTlsConfig::new().identity(identity))
                 .expect("OTLP test collector should accept its TLS identity");
         }
-        let service = TraceServiceServer::new(collector)
+        let trace_service = TraceServiceServer::new(collector.clone())
+            .accept_compressed(tonic::codec::CompressionEncoding::Gzip);
+        let logs_service = LogsServiceServer::new(collector)
             .accept_compressed(tonic::codec::CompressionEncoding::Gzip);
         let (shutdown, shutdown_rx) = tokio::sync::oneshot::channel();
         let task = tokio::spawn(async move {
             server
-                .add_service(service)
+                .add_service(trace_service)
+                .add_service(logs_service)
                 .serve_with_incoming_shutdown(
                     tokio_stream::wrappers::TcpListenerStream::new(listener),
                     async {

crates/openshell-otel-test-support/src/lib.rs

@@ -3,7 +3,7 @@
 
 //! Shared TLS export scenario for the one-test-per-binary TLS tests.
 
-use openshell_otel::{OtlpTraceConfig, ServiceName, provider_for};
+use openshell_otel::{OtlpExportConfig, ServiceName, provider_for};
 use openshell_otel_test_support::{OtlpTestServer, ReceivedTraces};
 use rcgen::{
     BasicConstraints, Certificate, CertificateParams, ExtendedKeyUsagePurpose, IsCa, KeyPair,
@@ -55,7 +55,7 @@ pub async fn export_span_over_tls(server_trusted: bool) -> (bool, ReceivedTraces
         server_key.serialize_pem(),
     ))
     .await;
-    let (provider, error) = provider_for(Some(OtlpTraceConfig {
+    let (provider, error) = provider_for(Some(OtlpExportConfig {
         endpoint: collector.endpoint(),
         service_name: ServiceName::Fixed("tls-test"),
         service_version: None,

crates/openshell-otel-test-support/tests/common/mod.rs

@@ -3,7 +3,7 @@
 
 [package]
 name = "openshell-otel"
-description = "Shared OpenTelemetry trace export support for OpenShell services"
+description = "Shared OpenTelemetry trace and log export support for OpenShell services"
 version.workspace = true
 edition.workspace = true
 rust-version.workspace = true
@@ -13,10 +13,14 @@ repository.workspace = true
 [dependencies]
 futures = { workspace = true }
 http = { workspace = true }
+metrics = { workspace = true }
+openshell-ocsf = { path = "../openshell-ocsf" }
 opentelemetry = { workspace = true }
-opentelemetry_sdk = { workspace = true }
-opentelemetry-otlp = { workspace = true }
+opentelemetry_sdk = { workspace = true, features = ["logs"] }
+opentelemetry-otlp = { workspace = true, features = ["logs"] }
+serde_json = { workspace = true }
 thiserror = { workspace = true }
+tokio = { workspace = true }
 tracing = { workspace = true }
 tracing-opentelemetry = { workspace = true }
 tracing-subscriber = { workspace = true }
@@ -25,7 +29,7 @@ tower-http = { workspace = true }
 
 [dev-dependencies]
 opentelemetry_sdk = { workspace = true, features = ["testing"] }
-tokio = { workspace = true }
+tokio = { workspace = true, features = ["test-util"] }
 
 [lints]
 workspace = true

crates/openshell-otel/Cargo.toml

@@ -12,7 +12,7 @@ use tracing_subscriber::EnvFilter;
 use tracing_subscriber::prelude::*;
 use tracing_subscriber::registry::LookupSpan;
 
-use crate::{OtlpTraceConfig, SdkTracerProvider, ServiceName, SetupError};
+use crate::{OtlpExportConfig, SdkTracerProvider, ServiceName, SetupError};
 
 /// Target used by every in-process compute-driver RPC boundary.
 pub const IN_PROCESS_COMPUTE_DRIVER_TARGET: &str = "openshell_otel::in_process_compute_driver";
@@ -77,7 +77,7 @@ impl ComputeDriverTracing {
         gateway_name: Option<&str>,
         compute_driver: Option<&str>,
     ) -> (Option<SdkTracerProvider>, Option<SetupError>) {
-        crate::provider_for(endpoint.map(|endpoint| OtlpTraceConfig {
+        crate::provider_for(endpoint.map(|endpoint| OtlpExportConfig {
             endpoint,
             service_name: ServiceName::Fixed(self.service_name),
             service_version: Some(service_version),