hyperactor_telemetry/sinks/
glog.rs

1/*
2 * Copyright (c) Meta Platforms, Inc. and affiliates.
3 * All rights reserved.
4 *
5 * This source code is licensed under the BSD-style license found in the
6 * LICENSE file in the root directory of this source tree.
7 */
8
9//! Glog-formatted text sink for trace events.
10//! Replicates the behavior of the fmt::Layer with glog formatting.
11
12use std::collections::HashMap;
13use std::fmt::Write as FmtWrite;
14use std::io::Write;
15use std::str::FromStr;
16
17use anyhow::Result;
18use tracing_core::LevelFilter;
19use tracing_subscriber::filter::Targets;
20
21use crate::config::MONARCH_FILE_LOG_LEVEL;
22use crate::trace_dispatcher::FieldValue;
23use crate::trace_dispatcher::TraceEvent;
24use crate::trace_dispatcher::TraceEventSink;
25use crate::trace_dispatcher::TraceFields;
26use crate::trace_dispatcher::get_field;
27
28const MAX_LINE_SIZE: usize = 65536;
29const TRUNCATION_SUFFIX_RESERVE: usize = 32;
30
31/// A string buffer that limits writes to a maximum size.
32/// Once the limit is reached, further writes are silently ignored and
33/// truncated chars are tracked for reporting.
34struct LimitedBuffer {
35    buffer: String,
36    /// Max bytes for content (excluding truncation suffix and newline).
37    limit: usize,
38    truncated_chars: usize,
39}
40
41impl LimitedBuffer {
42    fn new(limit: usize) -> Self {
43        Self {
44            buffer: String::with_capacity(limit + TRUNCATION_SUFFIX_RESERVE),
45            limit,
46            truncated_chars: 0,
47        }
48    }
49
50    fn clear(&mut self) {
51        self.buffer.clear();
52        self.truncated_chars = 0;
53    }
54
55    /// Write truncation suffix + add a newline
56    fn finish_line(&mut self) {
57        if self.truncated_chars > 0 {
58            use std::fmt::Write;
59            let _ = write!(
60                &mut self.buffer,
61                "...[truncated {} chars]",
62                self.truncated_chars
63            );
64        }
65        self.buffer.push('\n');
66    }
67
68    fn as_bytes(&self) -> &[u8] {
69        self.buffer.as_bytes()
70    }
71}
72
73impl FmtWrite for LimitedBuffer {
74    fn write_str(&mut self, s: &str) -> std::fmt::Result {
75        let remaining = self.limit.saturating_sub(self.buffer.len());
76        if remaining == 0 {
77            self.truncated_chars += s.chars().count();
78            return Ok(());
79        }
80        if s.len() <= remaining {
81            self.buffer.push_str(s);
82        } else {
83            let mut truncate_at = remaining;
84            while truncate_at > 0 && !s.is_char_boundary(truncate_at) {
85                truncate_at -= 1;
86            }
87            self.buffer.push_str(&s[..truncate_at]);
88            self.truncated_chars += s[truncate_at..].chars().count();
89        }
90        Ok(())
91    }
92}
93
94/// Glog sink that writes events in glog format to a file.
95/// This replaces the fmt::Layer that was previously used for text logging.
96///
97/// This only logs Events, not Spans (matching old fmt::Layer behavior).
98pub struct GlogSink {
99    writer: Box<dyn Write + Send>,
100    prefix: Option<String>,
101    /// Track active spans by ID with (name, fields, parent_id) to show span context in event logs
102    active_spans: HashMap<u64, (String, TraceFields, Option<u64>)>,
103    targets: Targets,
104    /// Reusable buffer for formatting log lines to ensure atomic writes.
105    /// We build the entire line in this buffer, then write it atomically to avoid
106    /// interleaving with other threads writing to the same fd (e.g., stderr).
107    line_buffer: LimitedBuffer,
108}
109
110impl GlogSink {
111    /// Create a new glog sink with the given writer.
112    ///
113    /// # Arguments
114    /// * `writer` - Writer to write log events to (used directly without buffering)
115    /// * `prefix_env_var` - Optional environment variable name to read prefix from (matching old impl)
116    /// * `file_log_level` - Minimum log level to capture (e.g., "info", "debug")
117    pub fn new(
118        writer: Box<dyn Write + Send>,
119        prefix_env_var: Option<String>,
120        file_log_level: &str,
121    ) -> Self {
122        let prefix = if let Some(ref env_var_name) = prefix_env_var {
123            std::env::var(env_var_name).ok()
124        } else {
125            None
126        };
127
128        Self {
129            writer,
130            prefix,
131            active_spans: HashMap::new(),
132            targets: Targets::new()
133                .with_default(LevelFilter::from_level({
134                    let log_level_str =
135                        hyperactor_config::global::try_get_cloned(MONARCH_FILE_LOG_LEVEL)
136                            .unwrap_or_else(|| file_log_level.to_string());
137                    tracing::Level::from_str(&log_level_str).unwrap_or_else(|_| {
138                        tracing::Level::from_str(file_log_level).expect("Invalid default log level")
139                    })
140                }))
141                .with_target("opentelemetry", LevelFilter::OFF), // otel has some log span under debug that we don't care about
142            line_buffer: LimitedBuffer::new(MAX_LINE_SIZE - TRUNCATION_SUFFIX_RESERVE),
143        }
144    }
145
146    fn write_event(&mut self, event: &TraceEvent) -> Result<()> {
147        self.line_buffer.clear();
148
149        let timestamp_str = match event {
150            TraceEvent::Event { timestamp, .. } => {
151                let datetime: chrono::DateTime<chrono::Local> = (*timestamp).into();
152                datetime.format("%m%d %H:%M:%S%.6f").to_string()
153            }
154            // write_event is only called for Events, but keep this for safety
155            _ => chrono::Local::now().format("%m%d %H:%M:%S%.6f").to_string(),
156        };
157
158        let prefix_str = if let Some(ref p) = self.prefix {
159            format!("[{}]", p)
160        } else {
161            "[-]".to_string()
162        };
163
164        match event {
165            TraceEvent::Event {
166                level,
167                fields,
168                parent_span,
169                file,
170                line,
171                ..
172            } => {
173                let level_char = match *level {
174                    tracing::Level::ERROR => 'E',
175                    tracing::Level::WARN => 'W',
176                    tracing::Level::INFO => 'I',
177                    tracing::Level::DEBUG => 'D',
178                    tracing::Level::TRACE => 'T',
179                };
180
181                // [prefix]LMMDD HH:MM:SS.ffffff file:line] message, key:value, key:value
182                write!(
183                    &mut self.line_buffer,
184                    "{}{}{} ",
185                    prefix_str, level_char, timestamp_str
186                )?;
187
188                if let (Some(f), Some(l)) = (file, line) {
189                    write!(&mut self.line_buffer, "{}:{}] ", f, l)?;
190                } else {
191                    write!(&mut self.line_buffer, "unknown:0] ")?;
192                }
193
194                // Render subject as a prefix. Check event fields first,
195                // then fall back to the enclosing span chain.
196                if let Some(subject) = get_field(fields, crate::SUBJECT_KEY) {
197                    Self::write_subject(&mut self.line_buffer, subject)?;
198                } else if let Some(parent_id) = parent_span {
199                    self.write_span_context(*parent_id)?;
200                }
201
202                if let Some(v) = get_field(fields, "message") {
203                    match v {
204                        FieldValue::Str(s) => write!(&mut self.line_buffer, "{}", s)?,
205                        FieldValue::Debug(s) => write!(&mut self.line_buffer, "{}", s)?,
206                        _ => write!(&mut self.line_buffer, "event")?,
207                    }
208                } else {
209                    write!(&mut self.line_buffer, "event")?;
210                }
211
212                for (k, v) in fields.iter() {
213                    if *k != "message" && *k != crate::SUBJECT_KEY {
214                        write!(&mut self.line_buffer, ", {k}: ")?;
215                        match v {
216                            FieldValue::Bool(b) => write!(&mut self.line_buffer, "{}", b)?,
217                            FieldValue::I64(i) => write!(&mut self.line_buffer, "{}", i)?,
218                            FieldValue::U64(u) => write!(&mut self.line_buffer, "{}", u)?,
219                            FieldValue::F64(f) => write!(&mut self.line_buffer, "{}", f)?,
220                            FieldValue::Str(s) => write!(&mut self.line_buffer, "{}", s)?,
221                            FieldValue::Debug(s) => write!(&mut self.line_buffer, "{}", s)?,
222                        }
223                    }
224                }
225
226                self.line_buffer.finish_line();
227
228                self.writer.write_all(self.line_buffer.as_bytes())?;
229            }
230
231            // write_event should only be called for Events, but handle gracefully
232            _ => {
233                self.line_buffer.clear();
234                write!(
235                    &mut self.line_buffer,
236                    "{}I{} - unknown:0] unexpected event type",
237                    prefix_str, timestamp_str
238                )?;
239                self.line_buffer.finish_line();
240                self.writer.write_all(self.line_buffer.as_bytes())?;
241            }
242        }
243
244        Ok(())
245    }
246
247    fn write_subject(buf: &mut LimitedBuffer, value: &FieldValue) -> Result<()> {
248        match value {
249            FieldValue::Str(s) => write!(buf, "{} ", s)?,
250            FieldValue::Debug(s) => write!(buf, "{} ", s)?,
251            _ => {}
252        }
253        Ok(())
254    }
255
256    /// Walks the span chain looking for a `subject` field. If found,
257    /// renders it as a prefix (e.g., `<actor id> `). If no subject is
258    /// found, no span context is rendered.
259    fn write_span_context(&mut self, span_id: u64) -> Result<()> {
260        let mut current_id = Some(span_id);
261        while let Some(id) = current_id {
262            if let Some((_, fields, parent_id)) = self.active_spans.get(&id) {
263                if let Some(subject) = get_field(fields, crate::SUBJECT_KEY) {
264                    Self::write_subject(&mut self.line_buffer, subject)?;
265                    return Ok(());
266                }
267                current_id = *parent_id;
268            } else {
269                break;
270            }
271        }
272        Ok(())
273    }
274}
275
276impl TraceEventSink for GlogSink {
277    fn consume(&mut self, event: &TraceEvent) -> Result<(), anyhow::Error> {
278        match event {
279            // Track span lifecycle for context display (must happen even if we don't export spans)
280            TraceEvent::NewSpan {
281                id,
282                name,
283                fields,
284                parent_id,
285                ..
286            } => {
287                self.active_spans
288                    .insert(*id, (name.to_string(), fields.clone(), *parent_id));
289            }
290            TraceEvent::SpanClose { id, .. } => {
291                self.active_spans.remove(id);
292            }
293            TraceEvent::Event { .. } => {
294                self.write_event(event)?;
295            }
296            _ => {}
297        }
298        Ok(())
299    }
300
301    fn flush(&mut self) -> Result<(), anyhow::Error> {
302        self.writer.flush()?;
303        Ok(())
304    }
305
306    fn name(&self) -> &str {
307        "GlogSink"
308    }
309
310    fn target_filter(&self) -> Option<&Targets> {
311        Some(&self.targets)
312    }
313}
314
315#[cfg(test)]
316mod test {
317    use super::*;
318
319    #[test]
320    fn test_limited_buffer_truncation() {
321        let mut buf = LimitedBuffer::new(20);
322
323        write!(
324            &mut buf,
325            "Hello, this is a very long message that exceeds the limit"
326        )
327        .unwrap();
328        buf.finish_line();
329
330        let output = std::str::from_utf8(buf.as_bytes()).unwrap();
331
332        assert_eq!(output, "Hello, this is a ver...[truncated 37 chars]\n");
333    }
334
335    #[test]
336    fn test_limited_buffer_no_truncation() {
337        let mut buf = LimitedBuffer::new(50);
338
339        write!(&mut buf, "Short message").unwrap();
340        buf.finish_line();
341
342        let output = std::str::from_utf8(buf.as_bytes()).unwrap();
343
344        assert_eq!(output, "Short message\n");
345    }
346}
hyperactor_telemetry/sinks/glog.rs

hyperactor_telemetry/sinks/
glog.rs
