eggserve_core/
ops.rs

1use std::borrow::Cow;
2use std::fmt;
3use std::sync::atomic::{AtomicU64, Ordering};
4use std::sync::OnceLock;
5use std::time::SystemTime;
6
7pub const SCHEMA_VERSION: u32 = 1;
8
9#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd)]
10pub enum Severity {
11    Debug,
12    Info,
13    Warn,
14    Error,
15}
16
17impl fmt::Display for Severity {
18    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
19        match self {
20            Severity::Debug => write!(f, "DEBUG"),
21            Severity::Info => write!(f, "INFO"),
22            Severity::Warn => write!(f, "WARN"),
23            Severity::Error => write!(f, "ERROR"),
24        }
25    }
26}
27
28#[derive(Debug, Clone, Copy, PartialEq, Eq)]
29pub enum EventKind {
30    // Process/Config
31    ProcessStarting,
32    RootInitialized,
33    ListenerReady,
34    ShutdownRequested,
35    DrainingStarted,
36    ForcedShutdownStarted,
37    ShutdownComplete,
38
39    // Connection
40    ConnectionAccepted,
41    ConnectionRejected,
42    TlsHandshakeSuccess,
43    TlsHandshakeFailure,
44    TlsHandshakeTimeout,
45    HeaderTimeout,
46    BodyReadTimeout,
47    ParserRejection,
48    KeepAliveClosed,
49    ConnectionTotalTimeout,
50    ClientDisconnect,
51    ConnectionPanic,
52
53    // Request/Service
54    RequestCompleted,
55    FileNotFound,
56    FileDenied,
57    FileError,
58    DotfileDenied,
59    SymlinkDenied,
60    RootEscapeDenied,
61    BodyPolicyRejection,
62    IncompleteBodyClose,
63    ServiceInvocationSuppressed,
64    ServiceTimeout,
65    ServiceError,
66    DirectoryListingLimit,
67
68    // Operational
69    ListenerTransientError,
70    ListenerPersistentError,
71    ResourceExhaustion,
72    BlockingWorkerSaturation,
73    LogSinkFailure,
74}
75
76impl fmt::Display for EventKind {
77    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
78        let name = match self {
79            EventKind::ProcessStarting => "process_starting",
80            EventKind::RootInitialized => "root_initialized",
81            EventKind::ListenerReady => "listener_ready",
82            EventKind::ShutdownRequested => "shutdown_requested",
83            EventKind::DrainingStarted => "draining_started",
84            EventKind::ForcedShutdownStarted => "forced_shutdown_started",
85            EventKind::ShutdownComplete => "shutdown_complete",
86
87            EventKind::ConnectionAccepted => "connection_accepted",
88            EventKind::ConnectionRejected => "connection_rejected",
89            EventKind::TlsHandshakeSuccess => "tls_handshake_success",
90            EventKind::TlsHandshakeFailure => "tls_handshake_failure",
91            EventKind::TlsHandshakeTimeout => "tls_handshake_timeout",
92            EventKind::HeaderTimeout => "header_timeout",
93            EventKind::BodyReadTimeout => "body_read_timeout",
94            EventKind::ParserRejection => "parser_rejection",
95            EventKind::KeepAliveClosed => "keep_alive_closed",
96            EventKind::ConnectionTotalTimeout => "connection_total_timeout",
97            EventKind::ClientDisconnect => "client_disconnect",
98            EventKind::ConnectionPanic => "connection_panic",
99
100            EventKind::RequestCompleted => "request_completed",
101            EventKind::FileNotFound => "file_not_found",
102            EventKind::FileDenied => "file_denied",
103            EventKind::FileError => "file_error",
104            EventKind::DotfileDenied => "dotfile_denied",
105            EventKind::SymlinkDenied => "symlink_denied",
106            EventKind::RootEscapeDenied => "root_escape_denied",
107            EventKind::BodyPolicyRejection => "body_policy_rejection",
108            EventKind::IncompleteBodyClose => "incomplete_body_close",
109            EventKind::ServiceInvocationSuppressed => "service_invocation_suppressed",
110            EventKind::ServiceTimeout => "service_timeout",
111            EventKind::ServiceError => "service_error",
112            EventKind::DirectoryListingLimit => "directory_listing_limit",
113
114            EventKind::ListenerTransientError => "listener_transient_error",
115            EventKind::ListenerPersistentError => "listener_persistent_error",
116            EventKind::ResourceExhaustion => "resource_exhaustion",
117            EventKind::BlockingWorkerSaturation => "blocking_worker_saturation",
118            EventKind::LogSinkFailure => "log_sink_failure",
119        };
120        write!(f, "{}", name)
121    }
122}
123
124#[derive(Debug, Clone, PartialEq, Eq)]
125pub enum Field {
126    Bool(String, bool),
127    I64(String, i64),
128    U64(String, u64),
129    Str(String, String),
130}
131
132impl fmt::Display for Field {
133    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
134        match self {
135            Field::Bool(k, v) => write!(f, "\"{}\": {}", k, v),
136            Field::I64(k, v) => write!(f, "\"{}\": {}", k, v),
137            Field::U64(k, v) => write!(f, "\"{}\": {}", k, v),
138            Field::Str(k, v) => write!(f, "\"{}\": \"{}\"", k, escape_json_string(v)),
139        }
140    }
141}
142
143#[derive(Debug, Clone)]
144pub struct Event {
145    pub schema_version: u32,
146    pub severity: Severity,
147    pub event: EventKind,
148    pub timestamp: String,
149    pub message: String,
150    pub connection_id: Option<u64>,
151    pub request_seq: Option<u32>,
152    pub fields: Vec<Field>,
153}
154
155impl Event {
156    pub fn new(severity: Severity, event: EventKind, message: impl Into<String>) -> Self {
157        Self {
158            schema_version: SCHEMA_VERSION,
159            severity,
160            event,
161            timestamp: rfc3339_now(),
162            message: message.into(),
163            connection_id: None,
164            request_seq: None,
165            fields: Vec::new(),
166        }
167    }
168
169    pub fn field(mut self, field: Field) -> Self {
170        self.fields.push(field);
171        self
172    }
173
174    pub fn connection_id(mut self, id: u64) -> Self {
175        self.connection_id = Some(id);
176        self
177    }
178
179    pub fn request_seq(mut self, seq: u32) -> Self {
180        self.request_seq = Some(seq);
181        self
182    }
183}
184
185fn rfc3339_now() -> String {
186    let dur = SystemTime::now()
187        .duration_since(SystemTime::UNIX_EPOCH)
188        .unwrap_or_default();
189    let secs = dur.as_secs();
190
191    let days_since_epoch = secs / 86400;
192    let time_of_day = secs % 86400;
193    let hours = time_of_day / 3600;
194    let minutes = (time_of_day % 3600) / 60;
195    let seconds = time_of_day % 60;
196    let millis = dur.subsec_millis();
197
198    // Civil date from days since 1970-01-01
199    let (year, month, day) = days_to_civil(days_since_epoch);
200
201    format!(
202        "{:04}-{:02}-{:02}T{:02}:{:02}:{:02}.{:03}Z",
203        year, month, day, hours, minutes, seconds, millis
204    )
205}
206
207fn days_to_civil(days: u64) -> (u64, u64, u64) {
208    let z = days + 719468;
209    let era = z / 146097;
210    let doe = z - era * 146097;
211    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146096) / 365;
212    let y = yoe + era * 400;
213    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
214    let mp = (5 * doy + 2) / 153;
215    let d = doy - (153 * mp + 2) / 5 + 1;
216    let m = if mp < 10 { mp + 3 } else { mp - 9 };
217    let y = if m <= 2 { y + 1 } else { y };
218    (y, m, d)
219}
220
221pub fn sanitize_text_field(text: &str) -> String {
222    let filtered: String = text
223        .chars()
224        .filter(|c| {
225            let code = *c as u32;
226            if code < 0x20 {
227                return false;
228            }
229            if code == 0x7F {
230                return false;
231            }
232            if code == 0x1B {
233                return false;
234            }
235            if code > 0x7E {
236                return false;
237            }
238            true
239        })
240        .collect();
241    truncate_str(&filtered, 512).into_owned()
242}
243
244pub fn sanitize_path(path: &str) -> String {
245    let last_component = path.rsplit('/').next().unwrap_or(path);
246    let without_query = last_component.split('?').next().unwrap_or(last_component);
247    let sanitized: String = without_query
248        .chars()
249        .filter(|c| {
250            let code = *c as u32;
251            (0x20..0x7F).contains(&code) && code != 0x1B
252        })
253        .collect();
254    truncate_str(&sanitized, 127).into_owned()
255}
256
257pub fn truncate(text: &str, max_len: usize) -> Cow<'_, str> {
258    truncate_str(text, max_len)
259}
260
261fn truncate_str(text: &str, max_len: usize) -> Cow<'_, str> {
262    if text.len() <= max_len {
263        return Cow::Borrowed(text);
264    }
265    // Find a char boundary at or before max_len
266    let mut end = max_len;
267    while end > 0 && !text.is_char_boundary(end) {
268        end -= 1;
269    }
270    Cow::Owned(format!("{}…", &text[..end]))
271}
272
273pub trait LogSink: Send + Sync {
274    fn emit(&self, event: &Event);
275    fn flush(&self);
276}
277
278pub struct NopLogSink;
279
280impl LogSink for NopLogSink {
281    fn emit(&self, _event: &Event) {}
282    fn flush(&self) {}
283}
284
285/// A log sink that wraps another sink and only forwards events at or above
286/// a minimum severity level. Used for `--quiet` mode.
287pub struct FilteredLogSink {
288    inner: Box<dyn LogSink>,
289    min_severity: Severity,
290}
291
292impl FilteredLogSink {
293    pub fn new(inner: Box<dyn LogSink>, min_severity: Severity) -> Self {
294        Self {
295            inner,
296            min_severity,
297        }
298    }
299}
300
301impl LogSink for FilteredLogSink {
302    fn emit(&self, event: &Event) {
303        if event.severity >= self.min_severity {
304            self.inner.emit(event);
305        }
306    }
307
308    fn flush(&self) {
309        self.inner.flush();
310    }
311}
312
313pub struct CompositeLogSink {
314    sinks: Vec<Box<dyn LogSink>>,
315}
316
317impl CompositeLogSink {
318    pub fn new(sinks: Vec<Box<dyn LogSink>>) -> Self {
319        Self { sinks }
320    }
321}
322
323impl LogSink for CompositeLogSink {
324    fn emit(&self, event: &Event) {
325        for sink in &self.sinks {
326            let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
327                sink.emit(event);
328            }));
329            if result.is_err() {
330                global_counters()
331                    .dropped_log_events
332                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
333                // Emit a LogSinkFailure event to surface sink panics.
334                // Use catch_unwind to prevent recursive failure.
335                let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
336                    Logger::global().emit(Event::new(
337                        Severity::Error,
338                        EventKind::LogSinkFailure,
339                        "log sink panicked",
340                    ));
341                }));
342            }
343        }
344    }
345    fn flush(&self) {
346        for sink in &self.sinks {
347            let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
348                sink.flush();
349            }));
350        }
351    }
352}
353
354#[derive(Debug, Clone, Copy, PartialEq, Eq)]
355pub enum LogFormat {
356    Text,
357    Json,
358}
359
360pub struct StderrLogSink {
361    pub log_format: LogFormat,
362}
363
364impl LogSink for StderrLogSink {
365    fn emit(&self, event: &Event) {
366        match self.log_format {
367            LogFormat::Text => {
368                let mut line = format!("[{}] {}: {}", event.severity, event.event, event.message);
369                if let Some(cid) = event.connection_id {
370                    line.push_str(&format!(" conn={}", cid));
371                }
372                if let Some(seq) = event.request_seq {
373                    line.push_str(&format!(" seq={}", seq));
374                }
375                for f in &event.fields {
376                    line.push_str(&format!(" {}", f));
377                }
378                eprintln!("{}", line);
379            }
380            LogFormat::Json => {
381                let json = event_to_json(event);
382                eprintln!("{}", json);
383            }
384        }
385    }
386
387    fn flush(&self) {}
388}
389
390pub fn event_to_json(event: &Event) -> String {
391    let mut out = String::with_capacity(256);
392    out.push('{');
393
394    out.push_str("\"schema_version\":");
395    out.push_str(&event.schema_version.to_string());
396
397    out.push_str(",\"severity\":\"");
398    out.push_str(&format!("{}", event.severity));
399    out.push('"');
400
401    out.push_str(",\"event\":\"");
402    out.push_str(&format!("{}", event.event));
403    out.push('"');
404
405    out.push_str(",\"timestamp\":\"");
406    out.push_str(&escape_json_string(&event.timestamp));
407    out.push('"');
408
409    out.push_str(",\"message\":\"");
410    out.push_str(&escape_json_string(&event.message));
411    out.push('"');
412
413    if let Some(cid) = event.connection_id {
414        out.push_str(",\"connection_id\":");
415        out.push_str(&cid.to_string());
416    }
417
418    if let Some(seq) = event.request_seq {
419        out.push_str(",\"request_seq\":");
420        out.push_str(&seq.to_string());
421    }
422
423    if !event.fields.is_empty() {
424        out.push_str(",\"fields\":[");
425        for (i, f) in event.fields.iter().enumerate() {
426            if i > 0 {
427                out.push(',');
428            }
429            out.push('{');
430            match f {
431                Field::Bool(k, v) => {
432                    out.push('"');
433                    out.push_str(&escape_json_string(k));
434                    out.push_str("\":");
435                    out.push_str(if *v { "true" } else { "false" });
436                }
437                Field::I64(k, v) => {
438                    out.push('"');
439                    out.push_str(&escape_json_string(k));
440                    out.push_str("\":");
441                    out.push_str(&v.to_string());
442                }
443                Field::U64(k, v) => {
444                    out.push('"');
445                    out.push_str(&escape_json_string(k));
446                    out.push_str("\":");
447                    out.push_str(&v.to_string());
448                }
449                Field::Str(k, v) => {
450                    out.push('"');
451                    out.push_str(&escape_json_string(k));
452                    out.push_str("\":\"");
453                    out.push_str(&escape_json_string(v));
454                    out.push('"');
455                }
456            }
457            out.push('}');
458        }
459        out.push(']');
460    }
461
462    out.push('}');
463    out
464}
465
466pub(crate) fn escape_json_string(s: &str) -> String {
467    let mut out = String::with_capacity(s.len());
468    for c in s.chars() {
469        match c {
470            '"' => out.push_str("\\\""),
471            '\\' => out.push_str("\\\\"),
472            '\n' => out.push_str("\\n"),
473            '\r' => out.push_str("\\r"),
474            '\t' => out.push_str("\\t"),
475            c if (c as u32) < 0x20 => {
476                out.push_str(&format!("\\u{:04x}", c as u32));
477            }
478            _ => out.push(c),
479        }
480    }
481    out
482}
483
484#[allow(dead_code)]
485pub struct Logger {
486    sink: Box<dyn LogSink>,
487}
488
489#[allow(dead_code)]
490static GLOBAL_LOGGER: OnceLock<Logger> = OnceLock::new();
491
492static GLOBAL_COUNTERS: OnceLock<OpsCounters> = OnceLock::new();
493
494pub fn global_counters() -> &'static OpsCounters {
495    GLOBAL_COUNTERS.get_or_init(OpsCounters::new)
496}
497
498#[allow(dead_code)]
499impl Logger {
500    pub fn init(sink: Box<dyn LogSink>) {
501        GLOBAL_LOGGER
502            .set(Logger { sink })
503            .ok()
504            .expect("Logger::init called more than once");
505    }
506
507    #[allow(clippy::result_unit_err)]
508    pub fn try_init(sink: Box<dyn LogSink>) -> Result<(), ()> {
509        GLOBAL_LOGGER.set(Logger { sink }).map_err(|_| ())
510    }
511
512    pub fn global() -> &'static Logger {
513        GLOBAL_LOGGER.get_or_init(|| Logger {
514            sink: Box::new(NopLogSink),
515        })
516    }
517
518    pub fn emit(&self, event: Event) {
519        self.sink.emit(&event);
520    }
521
522    pub fn emit_if(&self, condition: bool, event: Event) {
523        if condition {
524            self.sink.emit(&event);
525        }
526    }
527}
528
529pub struct CorrelationId {
530    connection_id: AtomicU64,
531}
532
533impl Default for CorrelationId {
534    fn default() -> Self {
535        Self::new()
536    }
537}
538
539impl CorrelationId {
540    pub fn new() -> Self {
541        Self {
542            connection_id: AtomicU64::new(1),
543        }
544    }
545
546    pub fn next(&self) -> u64 {
547        self.connection_id.fetch_add(1, Ordering::Relaxed)
548    }
549}
550
551#[derive(Debug)]
552pub struct OpsCounters {
553    pub connections_accepted: AtomicU64,
554    pub connections_rejected: AtomicU64,
555    pub active_connections: AtomicU64,
556    pub active_file_streams: AtomicU64,
557    pub connection_panics: AtomicU64,
558    pub parser_rejects: AtomicU64,
559    pub body_rejections: AtomicU64,
560    pub header_timeouts: AtomicU64,
561    pub body_read_timeouts: AtomicU64,
562    pub connection_total_timeouts: AtomicU64,
563    pub bytes_sent: AtomicU64,
564    pub graceful_shutdowns: AtomicU64,
565    pub forced_shutdowns: AtomicU64,
566    pub listener_errors: AtomicU64,
567    pub dropped_log_events: AtomicU64,
568}
569
570impl Default for OpsCounters {
571    fn default() -> Self {
572        Self::new()
573    }
574}
575
576impl OpsCounters {
577    pub fn new() -> Self {
578        Self {
579            connections_accepted: AtomicU64::new(0),
580            connections_rejected: AtomicU64::new(0),
581            active_connections: AtomicU64::new(0),
582            active_file_streams: AtomicU64::new(0),
583            connection_panics: AtomicU64::new(0),
584            parser_rejects: AtomicU64::new(0),
585            body_rejections: AtomicU64::new(0),
586            header_timeouts: AtomicU64::new(0),
587            body_read_timeouts: AtomicU64::new(0),
588            connection_total_timeouts: AtomicU64::new(0),
589            bytes_sent: AtomicU64::new(0),
590            graceful_shutdowns: AtomicU64::new(0),
591            forced_shutdowns: AtomicU64::new(0),
592            listener_errors: AtomicU64::new(0),
593            dropped_log_events: AtomicU64::new(0),
594        }
595    }
596
597    pub fn snapshot(&self) -> OpsSnapshot {
598        OpsSnapshot {
599            connections_accepted: self.connections_accepted.load(Ordering::Relaxed),
600            connections_rejected: self.connections_rejected.load(Ordering::Relaxed),
601            active_connections: self.active_connections.load(Ordering::Relaxed),
602            active_file_streams: self.active_file_streams.load(Ordering::Relaxed),
603            connection_panics: self.connection_panics.load(Ordering::Relaxed),
604            parser_rejects: self.parser_rejects.load(Ordering::Relaxed),
605            body_rejections: self.body_rejections.load(Ordering::Relaxed),
606            header_timeouts: self.header_timeouts.load(Ordering::Relaxed),
607            body_read_timeouts: self.body_read_timeouts.load(Ordering::Relaxed),
608            connection_total_timeouts: self.connection_total_timeouts.load(Ordering::Relaxed),
609            bytes_sent: self.bytes_sent.load(Ordering::Relaxed),
610            graceful_shutdowns: self.graceful_shutdowns.load(Ordering::Relaxed),
611            forced_shutdowns: self.forced_shutdowns.load(Ordering::Relaxed),
612            listener_errors: self.listener_errors.load(Ordering::Relaxed),
613            dropped_log_events: self.dropped_log_events.load(Ordering::Relaxed),
614        }
615    }
616}
617
618#[derive(Debug, Clone, PartialEq, Eq)]
619pub struct OpsSnapshot {
620    pub connections_accepted: u64,
621    pub connections_rejected: u64,
622    pub active_connections: u64,
623    pub active_file_streams: u64,
624    pub connection_panics: u64,
625    pub parser_rejects: u64,
626    pub body_rejections: u64,
627    pub header_timeouts: u64,
628    pub body_read_timeouts: u64,
629    pub connection_total_timeouts: u64,
630    pub bytes_sent: u64,
631    pub graceful_shutdowns: u64,
632    pub forced_shutdowns: u64,
633    pub listener_errors: u64,
634    pub dropped_log_events: u64,
635}
636
637#[cfg(test)]
638mod tests {
639    use super::*;
640
641    #[test]
642    fn sanitize_text_removes_control_chars() {
643        assert_eq!(sanitize_text_field("hello\r\nworld"), "helloworld");
644        assert_eq!(sanitize_text_field("tab\there"), "tabhere");
645        assert_eq!(sanitize_text_field("esc\x1B[31mred"), "esc[31mred");
646        assert_eq!(sanitize_text_field("null\x00byte\x7Fdel"), "nullbytedel");
647        assert_eq!(sanitize_text_field("normal text"), "normal text");
648    }
649
650    #[test]
651    fn sanitize_path_extracts_last_component() {
652        assert_eq!(sanitize_path("/foo/bar/baz.txt"), "baz.txt");
653        assert_eq!(sanitize_path("no/slash/here/"), "");
654        assert_eq!(sanitize_path("only-one"), "only-one");
655        assert_eq!(sanitize_path("/a/b/c/d/e/f.txt"), "f.txt");
656    }
657
658    #[test]
659    fn sanitize_path_truncates_long_paths() {
660        let long_name: String = "a".repeat(200);
661        let result = sanitize_path(&format!("/prefix/{}", long_name));
662        assert!(result.chars().count() <= 128);
663        assert!(result.ends_with('…'));
664    }
665
666    #[test]
667    fn event_timestamp_is_valid() {
668        let ev = Event::new(Severity::Info, EventKind::ProcessStarting, "test");
669        // Format: YYYY-MM-DDTHH:MM:SS.mmmZ
670        assert!(ev.timestamp.ends_with('Z'));
671        assert_eq!(ev.timestamp.len(), 24);
672        assert!(ev.timestamp.contains('T'));
673        // Dashes in date part
674        assert_eq!(ev.timestamp.matches('-').count(), 2);
675        // Colons in time part
676        assert_eq!(ev.timestamp.matches(':').count(), 2);
677    }
678
679    #[test]
680    fn correlation_id_increments() {
681        let cid = CorrelationId::new();
682        assert_eq!(cid.next(), 1);
683        assert_eq!(cid.next(), 2);
684        assert_eq!(cid.next(), 3);
685    }
686
687    #[test]
688    fn ops_counters_snapshot() {
689        let counters = OpsCounters::new();
690        counters
691            .connections_accepted
692            .fetch_add(5, Ordering::Relaxed);
693        counters.bytes_sent.fetch_add(1024, Ordering::Relaxed);
694        counters.listener_errors.fetch_add(1, Ordering::Relaxed);
695
696        let snap = counters.snapshot();
697        assert_eq!(snap.connections_accepted, 5);
698        assert_eq!(snap.bytes_sent, 1024);
699        assert_eq!(snap.listener_errors, 1);
700        assert_eq!(snap.connections_rejected, 0);
701        assert_eq!(snap.active_connections, 0);
702    }
703}
eggserve_core/ops.rs

eggserve_core/
ops.rs
