Create a Perfetto Sink (#2169)

Summary:

Create a TraceDispatcherSink that writes perfetto traces in their native protobuf format directly to local files

The advantages include:
- No row limits (vs 400k for a scuba query)
- Zero ingestion latency (Traces are available immediately)
- No json -> protobuf conversion needed and very compact
- Compatibility with distributed file systems like OILFS
- Faster execution of expanse: No more querying scuba and conversion of scuba -> perfetto protobuf

Reviewed By: moonli

Differential Revision: D89403329

Author: thomasywang

hyperactor_telemetry/Cargo.toml

@@ -31,6 +31,7 @@ lazy_static = "1.5"
 libc = "0.2.139"
 opentelemetry = "0.29"
 opentelemetry_sdk = { version = "0.29.0", features = ["rt-tokio"] }
+prost = { version = "0.13.4", default-features = false }
 rand = { version = "0.8", features = ["small_rng"] }
 rusqlite = { version = "0.37.0", features = ["backup", "blob", "bundled", "column_decltype", "functions", "limits", "modern_sqlite", "serde_json"] }
 scuba = { version = "0.1.0", git = "https://github.com/facebookexperimental/rust-shed.git", branch = "main" }
@@ -43,6 +44,7 @@ tracing = { version = "0.1.41", features = ["attributes", "valuable"] }
 tracing-appender = "0.2.3"
 tracing-core = { version = "0.1.33", features = ["valuable"] }
 tracing-glog = { version = "0.4.1", features = ["ansi", "tracing-log"] }
+tracing-perfetto-sdk-schema = "0.12.0"
 tracing-subscriber = { version = "0.3.20", features = ["chrono", "env-filter", "json", "local-time", "parking_lot", "registry"] }
 urlencoding = "2.1.0"
 whoami = "1.5"

hyperactor_telemetry/src/lib.rs

@@ -704,6 +704,27 @@ fn initialize_logging_with_log_prefix_impl(
                 }
             }
 
+            if hyperactor_config::global::get(sinks::perfetto::PERFETTO_TRACE_MODE)
+                != sinks::perfetto::PerfettoTraceMode::Off
+            {
+                let exec_id = env::execution_id();
+                let process_name = std::env::var("HYPERACTOR_PROCESS_NAME")
+                    .unwrap_or_else(|_| "client".to_string());
+                match sinks::perfetto::PerfettoFileSink::new(
+                    sinks::perfetto::default_trace_dir(),
+                    &exec_id,
+                    &process_name,
+                ) {
+                    Ok(sink) => {
+                        max_level = Some(tracing::level_filters::LevelFilter::TRACE);
+                        sinks.push(Box::new(sink));
+                    }
+                    Err(e) => {
+                        tracing::warn!("failed to create PerfettoFileSink: {}", e);
+                    }
+                }
+            }
+
             {
                 if hyperactor_config::global::get(ENABLE_OTEL_TRACING) {
                     use crate::meta;

hyperactor_telemetry/src/sinks/mod.rs

@@ -11,4 +11,5 @@
 //! writing events to a specific backend (SQLite, Scuba, glog, etc).
 
 pub mod glog;
+pub mod perfetto;
 pub mod sqlite;