nucleus/
telemetry.rs

1use crate::container::{ContainerConfig, SeccompMode, WorkspaceMode};
2use crate::error::{NucleusError, Result};
3use crate::network::NetworkMode;
4use crate::resources::{ResourceLimits, ResourceStats};
5use serde::Serialize;
6use std::fs::{File, OpenOptions};
7use std::io::{self, Write};
8use std::os::fd::{FromRawFd, RawFd};
9use std::os::unix::fs::OpenOptionsExt;
10use std::path::Path;
11use std::sync::{Arc, Mutex};
12use std::time::{SystemTime, UNIX_EPOCH};
13use tracing_subscriber::fmt::MakeWriter;
14use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
15
16/// Initialize the tracing subscriber with env-filter.
17///
18/// RUST_LOG is respected but capped at `debug` to prevent `trace`-level
19/// output from leaking sensitive runtime data (syscall args, memory
20/// contents, etc.) in production.
21pub fn init_tracing(log: Option<&Path>, log_format: &str) -> Result<()> {
22    let env_filter = match tracing_subscriber::EnvFilter::try_from_default_env() {
23        Ok(filter) => filter,
24        Err(_) => tracing_subscriber::EnvFilter::new("info")
25            .add_directive("nucleus=debug".parse().expect("valid tracing directive")),
26    };
27    let writer = log_writer(log)?;
28
29    match log_format {
30        "text" => tracing_subscriber::registry()
31            .with(tracing_subscriber::fmt::layer().with_writer(writer))
32            .with(env_filter)
33            .with(tracing_subscriber::filter::LevelFilter::DEBUG)
34            .try_init(),
35        "json" => tracing_subscriber::registry()
36            .with(tracing_subscriber::fmt::layer().json().with_writer(writer))
37            .with(env_filter)
38            .with(tracing_subscriber::filter::LevelFilter::DEBUG)
39            .try_init(),
40        other => {
41            return Err(NucleusError::ConfigError(format!(
42                "Unsupported --log-format '{}'; expected text or json",
43                other
44            )));
45        }
46    }
47    .map_err(|e| NucleusError::ConfigError(format!("Failed to initialize telemetry: {}", e)))
48}
49
50fn log_writer(log: Option<&Path>) -> Result<SharedMakeWriter> {
51    let file = match log {
52        Some(path) => open_append_file(path).map_err(|e| {
53            NucleusError::ConfigError(format!("Failed to open log file {:?}: {}", path, e))
54        })?,
55        None => open_stderr_duplicate().map_err(|e| {
56            NucleusError::ConfigError(format!("Failed to duplicate stderr for logging: {}", e))
57        })?,
58    };
59    Ok(SharedMakeWriter::new(file))
60}
61
62fn open_append_file(path: &Path) -> io::Result<File> {
63    OpenOptions::new()
64        .create(true)
65        .append(true)
66        .mode(0o600)
67        .custom_flags(libc::O_CLOEXEC)
68        .open(path)
69}
70
71fn open_truncated_file(path: &Path) -> io::Result<File> {
72    OpenOptions::new()
73        .create(true)
74        .truncate(true)
75        .write(true)
76        .mode(0o600)
77        .custom_flags(libc::O_CLOEXEC)
78        .open(path)
79}
80
81fn open_stderr_duplicate() -> io::Result<File> {
82    // SAFETY: dup returns a new owned file descriptor for stderr on success.
83    let fd = unsafe { libc::dup(libc::STDERR_FILENO) };
84    if fd < 0 {
85        return Err(io::Error::last_os_error());
86    }
87    set_cloexec(fd)?;
88    // SAFETY: fd was returned by dup and is now owned by File.
89    Ok(unsafe { File::from_raw_fd(fd) })
90}
91
92fn set_cloexec(fd: RawFd) -> io::Result<()> {
93    // SAFETY: fcntl with F_GETFD/F_SETFD does not violate memory safety.
94    let flags = unsafe { libc::fcntl(fd, libc::F_GETFD) };
95    if flags < 0 {
96        return Err(io::Error::last_os_error());
97    }
98    let ret = unsafe { libc::fcntl(fd, libc::F_SETFD, flags | libc::FD_CLOEXEC) };
99    if ret < 0 {
100        return Err(io::Error::last_os_error());
101    }
102    Ok(())
103}
104
105#[derive(Clone)]
106struct SharedMakeWriter {
107    file: Arc<Mutex<File>>,
108}
109
110impl SharedMakeWriter {
111    fn new(file: File) -> Self {
112        Self {
113            file: Arc::new(Mutex::new(file)),
114        }
115    }
116}
117
118struct SharedWriter {
119    file: Arc<Mutex<File>>,
120}
121
122impl<'a> MakeWriter<'a> for SharedMakeWriter {
123    type Writer = SharedWriter;
124
125    fn make_writer(&'a self) -> Self::Writer {
126        SharedWriter {
127            file: self.file.clone(),
128        }
129    }
130}
131
132impl Write for SharedWriter {
133    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
134        self.file
135            .lock()
136            .map_err(|_| io::Error::other("log writer mutex poisoned"))?
137            .write(buf)
138    }
139
140    fn flush(&mut self) -> io::Result<()> {
141        self.file
142            .lock()
143            .map_err(|_| io::Error::other("log writer mutex poisoned"))?
144            .flush()
145    }
146}
147
148/// JSON Lines sink for machine-readable control-plane events.
149#[derive(Clone, Debug)]
150pub struct EventSink {
151    file: Arc<Mutex<File>>,
152}
153
154impl EventSink {
155    pub fn from_path(path: &Path) -> Result<Self> {
156        let file = open_truncated_file(path).map_err(|e| {
157            NucleusError::ConfigError(format!("Failed to open event stream {:?}: {}", path, e))
158        })?;
159        Ok(Self {
160            file: Arc::new(Mutex::new(file)),
161        })
162    }
163
164    pub fn from_fd(fd: RawFd) -> Result<Self> {
165        if fd <= libc::STDERR_FILENO {
166            return Err(NucleusError::ConfigError(format!(
167                "--events-fd must be greater than 2 to keep control events separate from stdio, got {}",
168                fd
169            )));
170        }
171        set_cloexec(fd).map_err(|e| {
172            NucleusError::ConfigError(format!(
173                "Failed to set close-on-exec on event stream fd {}: {}",
174                fd, e
175            ))
176        })?;
177        // SAFETY: The caller transfers ownership of this inherited fd to Nucleus.
178        let file = unsafe { File::from_raw_fd(fd) };
179        Ok(Self {
180            file: Arc::new(Mutex::new(file)),
181        })
182    }
183
184    pub fn from_cli(events_fd: Option<RawFd>, events_jsonl: Option<&Path>) -> Result<Option<Self>> {
185        match (events_fd, events_jsonl) {
186            (Some(_), Some(_)) => Err(NucleusError::ConfigError(
187                "--events-fd and --events-jsonl are mutually exclusive".to_string(),
188            )),
189            (Some(fd), None) => Self::from_fd(fd).map(Some),
190            (None, Some(path)) => Self::from_path(path).map(Some),
191            (None, None) => Ok(None),
192        }
193    }
194
195    pub fn emit<T: Serialize>(&self, event: &T) -> Result<()> {
196        let mut guard = self.file.lock().map_err(|_| {
197            NucleusError::ConfigError("Event stream writer mutex poisoned".to_string())
198        })?;
199        serde_json::to_writer(&mut *guard, event).map_err(|e| {
200            NucleusError::ConfigError(format!("Failed to serialize event stream record: {}", e))
201        })?;
202        guard.write_all(b"\n").map_err(|e| {
203            NucleusError::ConfigError(format!("Failed to write event stream record: {}", e))
204        })?;
205        guard.flush().map_err(|e| {
206            NucleusError::ConfigError(format!("Failed to flush event stream record: {}", e))
207        })
208    }
209}
210
211#[derive(Debug, Clone, Serialize)]
212#[serde(rename_all = "snake_case")]
213pub enum ControlEventType {
214    ContainerStarted,
215    ContainerSummary,
216}
217
218#[derive(Debug, Clone, Serialize)]
219pub struct ContainerControlEvent {
220    pub timestamp_unix_ms: u128,
221    #[serde(rename = "type")]
222    pub event_type: ControlEventType,
223    pub container: ContainerEventMetadata,
224    pub security: SecurityEventMetadata,
225    pub resource_limits: ResourceLimits,
226    #[serde(skip_serializing_if = "Option::is_none")]
227    pub exit_status: Option<ExitStatusEvent>,
228    #[serde(skip_serializing_if = "Option::is_none")]
229    pub resource_stats: Option<ResourceStats>,
230    #[serde(skip_serializing_if = "Option::is_none")]
231    pub resource_stats_error: Option<String>,
232    #[serde(skip_serializing_if = "Option::is_none")]
233    pub cleanup: Option<CleanupEvent>,
234}
235
236impl ContainerControlEvent {
237    pub fn started(config: &ContainerConfig, pid: u32, cgroup_path: Option<String>) -> Self {
238        Self::new(ControlEventType::ContainerStarted, config, pid, cgroup_path)
239    }
240
241    pub fn summary(
242        config: &ContainerConfig,
243        pid: u32,
244        cgroup_path: Option<String>,
245        exit_status: ExitStatusEvent,
246        resource_stats: Option<ResourceStats>,
247        resource_stats_error: Option<String>,
248        cleanup: CleanupEvent,
249    ) -> Self {
250        let mut event = Self::new(ControlEventType::ContainerSummary, config, pid, cgroup_path);
251        event.exit_status = Some(exit_status);
252        event.resource_stats = resource_stats;
253        event.resource_stats_error = resource_stats_error;
254        event.cleanup = Some(cleanup);
255        event
256    }
257
258    fn new(
259        event_type: ControlEventType,
260        config: &ContainerConfig,
261        pid: u32,
262        cgroup_path: Option<String>,
263    ) -> Self {
264        Self {
265            timestamp_unix_ms: now_unix_ms(),
266            event_type,
267            container: ContainerEventMetadata::from_config(config, pid, cgroup_path),
268            security: SecurityEventMetadata::from_config(config),
269            resource_limits: config.limits.clone(),
270            exit_status: None,
271            resource_stats: None,
272            resource_stats_error: None,
273            cleanup: None,
274        }
275    }
276}
277
278#[derive(Debug, Clone, Serialize)]
279pub struct ContainerEventMetadata {
280    pub id: String,
281    pub name: String,
282    pub pid: u32,
283    pub runtime: String,
284    pub cgroup_path: Option<String>,
285    pub workspace_mount: Option<WorkspaceMountEvent>,
286    pub network_mode: String,
287}
288
289impl ContainerEventMetadata {
290    fn from_config(config: &ContainerConfig, pid: u32, cgroup_path: Option<String>) -> Self {
291        Self {
292            id: config.id.clone(),
293            name: config.name.clone(),
294            pid,
295            runtime: if config.use_gvisor {
296                "gvisor".to_string()
297            } else {
298                "native".to_string()
299            },
300            cgroup_path,
301            workspace_mount: config.workspace.effective_host_path().map(|source| {
302                WorkspaceMountEvent {
303                    source: source.display().to_string(),
304                    destination: config.workspace.container_path.display().to_string(),
305                    mode: workspace_mode_label(config.workspace.mode).to_string(),
306                    allow_execute: config.workspace.allow_execute,
307                }
308            }),
309            network_mode: network_mode_label(&config.network).to_string(),
310        }
311    }
312}
313
314#[derive(Debug, Clone, Serialize)]
315pub struct WorkspaceMountEvent {
316    pub source: String,
317    pub destination: String,
318    pub mode: String,
319    pub allow_execute: bool,
320}
321
322#[derive(Debug, Clone, Serialize)]
323pub struct SecurityEventMetadata {
324    pub seccomp_mode: String,
325    pub landlock_status: String,
326    pub capabilities_status: String,
327    pub rootless: bool,
328    pub allow_degraded_security: bool,
329    #[serde(skip_serializing_if = "Option::is_none")]
330    pub gpu: Option<GpuEventSummary>,
331}
332
333/// GPU passthrough summary emitted in the container started/summary events.
334#[derive(Debug, Clone, Serialize)]
335pub struct GpuEventSummary {
336    pub vendor: String,
337    pub visible_devices: String,
338    pub driver_capabilities: String,
339    pub bind_driver_libraries: bool,
340    /// Whether the seccomp ioctl filter was relaxed for GPU driver ioctls.
341    pub relaxed_seccomp_ioctl: bool,
342}
343
344impl SecurityEventMetadata {
345    fn from_config(config: &ContainerConfig) -> Self {
346        let gpu = config.gpu.as_ref().map(|g| GpuEventSummary {
347            vendor: format!("{:?}", g.vendor).to_lowercase(),
348            visible_devices: g.visible_devices.clone(),
349            driver_capabilities: g.driver_capabilities.clone(),
350            bind_driver_libraries: g.bind_driver_libraries,
351            relaxed_seccomp_ioctl: true,
352        });
353        Self {
354            seccomp_mode: seccomp_mode_label(config),
355            landlock_status: landlock_status_label(config),
356            capabilities_status: capabilities_status_label(config),
357            rootless: config.user_ns_config.is_some(),
358            allow_degraded_security: config.allow_degraded_security,
359            gpu,
360        }
361    }
362}
363
364#[derive(Debug, Clone, Serialize)]
365pub struct ExitStatusEvent {
366    pub code: Option<i32>,
367    pub error: Option<String>,
368}
369
370impl ExitStatusEvent {
371    pub fn code(code: i32) -> Self {
372        Self {
373            code: Some(code),
374            error: None,
375        }
376    }
377
378    pub fn error(error: impl Into<String>) -> Self {
379        Self {
380            code: None,
381            error: Some(error.into()),
382        }
383    }
384}
385
386#[derive(Debug, Clone, Serialize)]
387pub struct CleanupEvent {
388    pub succeeded: bool,
389    pub errors: Vec<String>,
390}
391
392impl CleanupEvent {
393    pub fn from_errors(errors: Vec<String>) -> Self {
394        Self {
395            succeeded: errors.is_empty(),
396            errors,
397        }
398    }
399}
400
401fn network_mode_label(mode: &NetworkMode) -> &'static str {
402    match mode {
403        NetworkMode::None => "none",
404        NetworkMode::Host => "host",
405        NetworkMode::GVisorHost => "gvisor-host",
406        NetworkMode::Bridge(_) => "bridge",
407    }
408}
409
410fn workspace_mode_label(mode: WorkspaceMode) -> &'static str {
411    match mode {
412        WorkspaceMode::BindRw => "bind-rw",
413        WorkspaceMode::BindRo => "bind-ro",
414        WorkspaceMode::CopyInOut => "copy-in-out",
415    }
416}
417
418fn seccomp_mode_label(config: &ContainerConfig) -> String {
419    match config.seccomp_mode {
420        SeccompMode::Trace => "trace".to_string(),
421        SeccompMode::Enforce if config.seccomp_profile.is_some() => "profile".to_string(),
422        SeccompMode::Enforce => "enforce".to_string(),
423    }
424}
425
426fn landlock_status_label(config: &ContainerConfig) -> String {
427    if config.use_gvisor {
428        "managed_by_gvisor".to_string()
429    } else if config.landlock_policy.is_some() {
430        "policy_file".to_string()
431    } else {
432        "default_policy".to_string()
433    }
434}
435
436fn capabilities_status_label(config: &ContainerConfig) -> String {
437    if config.use_gvisor {
438        "managed_by_gvisor".to_string()
439    } else if config.caps_policy.is_some() {
440        "policy_file".to_string()
441    } else {
442        "drop_all".to_string()
443    }
444}
445
446fn now_unix_ms() -> u128 {
447    SystemTime::now()
448        .duration_since(UNIX_EPOCH)
449        .map(|duration| duration.as_millis())
450        .unwrap_or(0)
451}
452
453#[cfg(test)]
454mod tests {
455    use super::*;
456
457    #[test]
458    fn events_fd_rejects_stdio_descriptors() {
459        assert!(EventSink::from_cli(Some(1), None).is_err());
460        assert!(EventSink::from_cli(Some(2), None).is_err());
461    }
462
463    #[test]
464    fn events_cli_rejects_multiple_destinations() {
465        let err = EventSink::from_cli(Some(3), Some(Path::new("/tmp/events.jsonl"))).unwrap_err();
466        assert!(err.to_string().contains("mutually exclusive"));
467    }
468
469    #[test]
470    fn control_event_contains_required_runtime_fields() {
471        let workspace = crate::container::WorkspaceConfig::new()
472            .with_host_path(Path::new("/workspace-src").to_path_buf())
473            .with_mode(WorkspaceMode::BindRo)
474            .with_allow_execute(true);
475        let mut config = ContainerConfig::try_new_with_id(
476            Some("0123456789abcdef0123456789abcdef".to_string()),
477            Some("demo".to_string()),
478            vec!["/bin/true".to_string()],
479        )
480        .unwrap()
481        .with_gvisor(false)
482        .with_workspace(workspace);
483        config.seccomp_mode = SeccompMode::Enforce;
484
485        let event = ContainerControlEvent::started(
486            &config,
487            1234,
488            Some("/sys/fs/cgroup/nucleus-demo".to_string()),
489        );
490        let value = serde_json::to_value(event).unwrap();
491
492        assert_eq!(value["type"], "container_started");
493        assert_eq!(value["container"]["id"], "0123456789abcdef0123456789abcdef");
494        assert_eq!(value["container"]["pid"], 1234);
495        assert_eq!(value["container"]["network_mode"], "none");
496        assert_eq!(
497            value["container"]["cgroup_path"],
498            "/sys/fs/cgroup/nucleus-demo"
499        );
500        assert_eq!(
501            value["container"]["workspace_mount"]["destination"],
502            "/workspace"
503        );
504        assert_eq!(value["container"]["workspace_mount"]["mode"], "bind-ro");
505        assert_eq!(value["container"]["workspace_mount"]["allow_execute"], true);
506        assert_eq!(value["security"]["seccomp_mode"], "enforce");
507        assert_eq!(value["security"]["landlock_status"], "default_policy");
508        assert_eq!(value["security"]["capabilities_status"], "drop_all");
509        assert!(value.get("resource_limits").is_some());
510    }
511}
nucleus/telemetry.rs

nucleus/
telemetry.rs
