Telemetry refactor to support private events & Proto defined event data structs (#5715)

* remove unusued pbjson serde gen. add skip_serialize_none

pull protogen.toml path out of implementation

add glob support to protogen

better messages

move type attributes to protogen config

remove deserialize derive from vortex events

add message & enum attrs support to protogen

* add new tracing config creation api not relying on io_args
telemetry context
avoid rare flaky failures in tracing tests
update metadata crates
fix json payload inconsistency, simplify phase/node span_name
enusre stable sort in protogen
structure tracing tests and add jsonl filtering tests
fix metadata fix copybara structure module for manual impl for generated
proto files
add otlp testing infra hide arrow schema from public api's
make a private api to deserialize all event types from arrow including
private

* switch to macro derive for serde stuff in proto
add new method derive for protos

wip NodeEvaluated

wip

update goldie

wip

* proper skipped reporting

* fix evaluation detail serialization and update metadata sample

* add COPY fs/sa/crates/proto-rust-macros fs/sa/crates/proto-rust-macros for metadata

GitOrigin-RevId: 439a1d347344453817dd736ceaa0346931a7384f

Author: mishamsk

Cargo.toml

@@ -34,6 +34,7 @@ members = [
   "crates/vortex-client",
   "crates/vortex-events",
   "crates/proto-rust",
+  "crates/proto-rust-macros",
   "crates/dbt-env",
   "crates/dbt-test-utils",
 ]
@@ -178,6 +179,7 @@ dbt-test-containers = { path = "crates/dbt-test-containers" }
 dbt-test-primitives = { path = "crates/dbt-test-primitives" }
 dbt-test-utils = { path = "crates/dbt-test-utils" }
 proto-rust = { path = "crates/proto-rust" }
+proto-rust-macros = { path = "crates/proto-rust-macros" }
 vortex-client = { path = "crates/vortex-client" }
 vortex-events = { path = "crates/vortex-events" }
 
@@ -277,7 +279,7 @@ datafusion-physical-expr = { version = "48.0.1" }
 datafusion-physical-plan = { version = "48.0.1" }
 
 # uuid
-uuid = { version = "1.16.0", features = ["v4"] }
+uuid = { version = "1.16.0", features = ["v4", "serde"] }
 
 # networking & cache
 redis = { version = "0.31.0", features = [
@@ -386,6 +388,7 @@ rustyline = { version = "15.0.0", default-features = false, features = [
 # datatypes and algos
 base64 = "0.22.1"
 bigdecimal = "0.4.7"
+bitflags = "2.9.4"
 blake3 = "1.5.0"
 chrono = { version = "0.4.41", features = ["std", "clock", "serde"] }
 chrono-tz = { version = "0.10" }
@@ -461,7 +464,6 @@ log = { version = "0.4.27", features = ["kv_serde"] }
 
 # tracing (observability)
 tracing = "0.1.41"
-tracing-log = "0.2.0" # temporary until we migrate to tracing exclusively
 tracing-subscriber = { version = "0.3.20", features = ["env-filter", "json"] }
 
 # OpenTelemetry (optional embedded OTLP exporter for traces, see dbt-common)

crates/dbt-common/Cargo.toml

@@ -29,7 +29,7 @@ itertools = { workspace = true }
 rand = { workspace = true }
 regex = { workspace = true }
 unicode-segmentation = { version = "1.12" }
-uuid = { workspace = true }
+uuid = { workspace = true, features = ["v7"] }
 
 arrow = { workspace = true }
 arrow-schema = { workspace = true }
@@ -61,8 +61,8 @@ log = { workspace = true }
 rust-embed = { workspace = true }
 
 # Tracing
+proto-rust = { workspace = true }
 tracing = { workspace = true }
-tracing-log = { workspace = true } # temporary until we migrate to tracing exclusively
 tracing-subscriber = { workspace = true }
 
 # Embedded OTLP exporter for traces

crates/dbt-common/src/io_args.rs

@@ -1,5 +1,6 @@
 use clap::ValueEnum;
 use dbt_serde_yaml::{JsonSchema, Value};
+use dbt_telemetry::NodeType;
 use pathdiff::diff_paths;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
@@ -37,8 +38,8 @@ pub struct IoArgs {
     pub in_dir: PathBuf,
     pub out_dir: PathBuf,
     pub log_path: Option<PathBuf>,
-    pub otm_file_name: Option<String>,
-    pub otm_parquet_file_name: Option<String>,
+    pub otel_file_name: Option<String>,
+    pub otel_parquet_file_name: Option<String>,
     pub export_to_otlp: bool,
     pub log_format: LogFormat,
     pub log_level: Option<LevelFilter>,
@@ -75,7 +76,7 @@ impl fmt::Debug for IoArgs {
             .field("in_dir", &self.in_dir)
             .field("out_dir", &self.out_dir)
             .field("log_path", &self.log_path)
-            .field("otm_file_name", &self.otm_file_name)
+            .field("otel_file_name", &self.otel_file_name)
             .field("status_reporter", &self.status_reporter.is_some())
             .finish()
     }
@@ -111,8 +112,8 @@ impl IoArgs {
     /// This function takes an artifact path, which may either be a workspace
     /// resource, or some generated temp location, and returns a path to its
     /// corresponding location in the workspace
-    pub fn map_to_workspace_path(&self, path: &Path, resource_type: &str) -> PathBuf {
-        if resource_type == "unit_test" || resource_type == "snapshot" {
+    pub fn map_to_workspace_path(&self, path: &Path, resource_type: NodeType) -> PathBuf {
+        if resource_type == NodeType::UnitTest || resource_type == NodeType::Snapshot {
             let special_component_idx = path.components().position(|c| {
                 c.as_os_str() == DBT_GENERIC_TESTS_DIR_NAME
                     || c.as_os_str() == DBT_SNAPSHOTS_DIR_NAME
@@ -385,6 +386,19 @@ impl Display for ClapResourceType {
     }
 }
 
+impl From<&ClapResourceType> for NodeType {
+    fn from(value: &ClapResourceType) -> Self {
+        match value {
+            ClapResourceType::Model => NodeType::Model,
+            ClapResourceType::Source => NodeType::Source,
+            ClapResourceType::Seed => NodeType::Seed,
+            ClapResourceType::Snapshot => NodeType::Snapshot,
+            ClapResourceType::Test => NodeType::Test,
+            ClapResourceType::UnitTest => NodeType::UnitTest,
+        }
+    }
+}
+
 #[derive(
     Debug,
     Clone,

crates/dbt-common/src/logging/logger.rs

@@ -11,7 +11,6 @@ use std::fmt::Display;
 use std::io::{IsTerminal as _, Write};
 use std::path::PathBuf;
 use std::sync::{Arc, Mutex};
-use tracing_log::LogTracer;
 
 const QUERY_LOG_SQL: &str = "query_log.sql";
 const CACHE_LOG_FILE: &str = "beta_cache.log";
@@ -97,7 +96,6 @@ enum LogTarget {
     Stdout,
     Stderr,
     Writer(Arc<Mutex<Box<dyn Write + Send>>>),
-    TracingBridge(LogTracer),
 }
 
 // Individual logger that can be customized
@@ -135,9 +133,6 @@ macro_rules! locked_writeln {
                     writeln!(writer, $($arg)*).ok();
                 }
             }
-            LogTarget::TracingBridge(_) => {
-                // TracingBridge doesn't use writeln, handled separately in log() method
-            }
         }
     };
 }
@@ -153,7 +148,6 @@ impl Logger {
             LogTarget::Stdout => !std::io::stdout().is_terminal(),
             LogTarget::Stderr => !std::io::stderr().is_terminal(),
             LogTarget::Writer(_) => true, // Always remove ANSI codes for file writers
-            LogTarget::TracingBridge(_) => true, // Always remove ANSI codes for tracing bridge
         };
         Self {
             target: writer,
@@ -292,44 +286,24 @@ impl log::Log for Logger {
 
     fn log(&self, record: &Record) {
         if self.enabled(record.metadata()) && !super::term::is_term_control_only(record) {
-            match &self.target {
-                LogTarget::TracingBridge(tracer) => {
-                    // Check if this record is already handled by new tracing logic
-                    if record
-                        .key_values()
-                        .get(log::kv::Key::from_str("_TRACING_HANDLED_"))
-                        .is_some()
-                    {
-                        // This record is already handled by the new tracing logic, skip it
-                        return;
+            match self.config.format {
+                LogFormat::Text | LogFormat::Default => {
+                    let mut text = record.args().to_string();
+                    if self.remove_ansi_codes {
+                        text = remove_ansi_codes(&text);
                     }
-
-                    // For tracing bridge, we need to strip ANSI codes and pass through
-                    // the log level filtering logic, then delegate to the tracer
-                    let text = remove_ansi_codes(&record.args().to_string());
-
-                    // Pass a new record with stripped ANSI codes
-                    tracer.log(&record.to_builder().args(format_args!("{text}")).build());
+                    locked_writeln!(self, "{}", text);
                 }
-                _ => match self.config.format {
-                    LogFormat::Text | LogFormat::Default => {
-                        let mut text = record.args().to_string();
-                        if self.remove_ansi_codes {
-                            text = remove_ansi_codes(&text);
-                        }
-                        locked_writeln!(self, "{}", text);
-                    }
-                    LogFormat::Json => {
-                        let json = Self::format_json(
-                            record,
-                            &self.invocation_id.to_string(),
-                            self.remove_ansi_codes,
-                        );
-                        locked_writeln!(self, "{}", json);
-                    }
-                    // This is handled in new tracing infra
-                    LogFormat::Otel => {}
-                },
+                LogFormat::Json => {
+                    let json = Self::format_json(
+                        record,
+                        &self.invocation_id.to_string(),
+                        self.remove_ansi_codes,
+                    );
+                    locked_writeln!(self, "{}", json);
+                }
+                // This is handled in new tracing infra
+                LogFormat::Otel => {}
             }
         }
     }
@@ -347,9 +321,6 @@ impl log::Log for Logger {
                     let _ = writer.flush();
                 }
             }
-            LogTarget::TracingBridge(ref tracer) => {
-                tracer.flush();
-            }
         }
     }
 }
@@ -454,27 +425,6 @@ impl MultiLoggerBuilder {
         self
     }
 
-    fn add_tracing_bridge_logger(mut self, log_config: &FsLogConfig) -> Self {
-        let config = LoggerConfig {
-            level_filter: log_config.log_level,
-            format: LogFormat::Text, // Format doesn't matter for tracing bridge
-            min_level: None,
-            max_level: Some(LevelFilter::Debug),
-            includes: None,
-            excludes: None,
-        };
-
-        let logger = Logger::new(
-            "tracing_bridge",
-            LogTarget::TracingBridge(LogTracer::new()),
-            config,
-            self.invocation_id,
-        );
-
-        self.loggers.push(Box::new(logger));
-        self
-    }
-
     fn build(self) -> MultiLogger {
         MultiLogger {
             loggers: self.loggers,
@@ -644,9 +594,6 @@ pub fn init_logger(log_config: FsLogConfig) -> FsResult<()> {
     ) as Box<dyn Write + Send>));
     builder = builder.add_logger("cache_stats", file, cache_log_config);
 
-    // Add tracing bridge logger
-    builder = builder.add_tracing_bridge_logger(&log_config);
-
     // Build the logger
     let logger = builder.build();
     // Register the logger globally

crates/dbt-common/src/macros.rs

@@ -369,7 +369,7 @@ macro_rules! show_completed {
                 .signed_duration_since($start_time)
                 .to_std()
                 .unwrap_or_default();
-            let resource_type = $task.resource_type();
+            let resource_type_str = $task.resource_type();
             let materialization = $task.base().materialized.to_string();
             let schema = $task.base().schema.clone();
             let alias = $task.base().alias.clone();
@@ -389,7 +389,7 @@ macro_rules! show_completed {
 
             if $io.should_show(ShowOptions::Completed) {
                 let schema_alias = format_schema_alias(&schema, &alias);
-                let resource_type_formatted = format_resource_type_fixed_width(resource_type);
+                let resource_type_formatted = format_resource_type_fixed_width(resource_type_str.as_ref());
                 let materialization_suffix = format_materialization_suffix(Some(&materialization), desc.as_deref());
                 let duration = format_duration_fixed_width(duration);
                 let output = format!(
@@ -787,24 +787,28 @@ macro_rules! show_warning {
         let err = $err;
 
         // New tracing based logic
-        use $crate::tracing::{ToTracingValue, log_level_to_severity};
+        use $crate::tracing::log_level_to_severity;
         use $crate::tracing::emit::_tracing::Level as TracingLevel;
         use $crate::tracing::metrics::{increment_metric, MetricKey};
-        use $crate::tracing::constants::TRACING_ATTR_FIELD;
-        use $crate::macros::_dbt_telemetry::{LogEventInfo, TelemetryAttributes, RecordCodeLocation};
+        use $crate::macros::_dbt_telemetry::LogMessage;
         increment_metric(MetricKey::TotalWarnings, 1);
 
         let (original_severity_number, original_severity_text) = log_level_to_severity(&$crate::macros::log_adapter::log::Level::Warn);
 
         $crate::emit_tracing_event!(
             level: TracingLevel::WARN,
-            TelemetryAttributes::Log(LogEventInfo {
+            LogMessage {
                 code: Some(err.code as u16 as u32),
-                dbt_core_code: None,
-                original_severity_number,
+                dbt_core_event_code: None,
+                original_severity_number: original_severity_number as i32,
                 original_severity_text: original_severity_text.to_string(),
-                location: RecordCodeLocation::none(), // Will be auto injected
-            }),
+                // The rest will be auto injected
+                phase: None,
+                unique_id: None,
+                file: None,
+                line: None,
+            }
+            .into(),
             "{}",
             err.pretty().as_str()
         );