feat: added open telemetry metrics

raphael-goetz · raphael-goetz · commit fc8fc37c75a2 · 2026-07-09T19:13:49.000+02:00
diff --git a/crates/taurus/Cargo.toml b/crates/taurus/Cargo.toml
@@ -4,14 +4,14 @@ version.workspace = true
 edition.workspace = true
 
 [dependencies]
-code0-flow = { workspace = true, features = ["flow_service", "flow_health", "flow_config"] }
+code0-flow = { workspace = true, features = ["flow_service", "flow_health", "flow_config", "flow_telemetry"] }
 tucana = { workspace = true }
 tokio = { workspace = true }
 log = { workspace = true }
+opentelemetry = { workspace = true }
 futures-lite ={ workspace = true }
 rand = { workspace = true }
 base64 = { workspace = true }
-env_logger = { workspace = true }
 async-nats = { workspace = true }
 prost = { workspace = true }
 tonic-health = { workspace = true }
diff --git a/crates/taurus/src/app/mod.rs b/crates/taurus/src/app/mod.rs
@@ -1,7 +1,8 @@
 mod worker;
 
+use code0_flow::flow_config::environment::Environment;
 use code0_flow::flow_config::load_env_file;
-use code0_flow::flow_config::mode::Mode::DYNAMIC;
+use code0_flow::flow_config::mode::Mode::{DYNAMIC, STATIC};
 use code0_flow::flow_definition::Reader;
 use code0_flow::flow_service::FlowUpdateService;
 use std::sync::Arc;
@@ -18,12 +19,13 @@ use tucana::shared::module_status::StatusVariant;
 use crate::client::runtime_execution::TaurusRuntimeExecutionService;
 use crate::client::runtime_status::TaurusRuntimeStatusService;
 use crate::config::Config;
+use crate::telemetry::{self, TelemetrySettings};
 
 pub async fn run() {
-    init_logging();
     load_env_file();
 
     let config = Config::new();
+    let telemetry = init_telemetry(&config);
     let engine = ExecutionEngine::new();
     let client = connect_nats(&config).await;
 
@@ -42,6 +44,7 @@ pub async fn run() {
         nats_remote,
         runtime_emitter,
         runtime_execution_service,
+        mode_label(&config).to_string(),
     );
 
     wait_for_shutdown(&mut worker_task, &mut health_task).await;
@@ -56,12 +59,36 @@ pub async fn run() {
     update_stopped_status(runtime_status_service.as_ref()).await;
 
     log::info!("Taurus shutdown complete");
+    telemetry.shutdown();
 }
 
-fn init_logging() {
-    env_logger::Builder::from_default_env()
-        .filter_level(log::LevelFilter::Debug)
-        .init();
+fn init_telemetry(config: &Config) -> telemetry::Telemetry {
+    telemetry::Telemetry::initialize(
+        &config.opentelemetry,
+        TelemetrySettings {
+            environment: environment_label(&config.environment),
+            default_log_level: "debug",
+            service_version: env!("CARGO_PKG_VERSION"),
+            instrumentation_name: env!("CARGO_PKG_NAME"),
+            initialize_metrics: Some(telemetry::metrics::initialize),
+        },
+    )
+    .unwrap_or_else(|error| panic!("failed to initialize telemetry: {error}"))
+}
+
+fn environment_label(environment: &Environment) -> &'static str {
+    match environment {
+        Environment::Development => "development",
+        Environment::Staging => "staging",
+        Environment::Production => "production",
+    }
+}
+
+fn mode_label(config: &Config) -> &'static str {
+    match config.mode {
+        STATIC => "static",
+        DYNAMIC => "dynamic",
+    }
 }
 
 async fn connect_nats(config: &Config) -> async_nats::Client {
diff --git a/crates/taurus/src/app/worker.rs b/crates/taurus/src/app/worker.rs
@@ -1,5 +1,3 @@
-use std::time::Instant;
-
 use futures_lite::StreamExt;
 use prost::Message;
 use taurus_core::runtime::engine::{EmitType, ExecutionEngine, ExecutionId, RespondEmitter};
@@ -14,13 +12,15 @@ use tucana::shared::execution_result;
 use tucana::shared::{ExecutionFlow, ExecutionResult, NodeExecutionResult, Value};
 
 use crate::client::runtime_execution::TaurusRuntimeExecutionService;
+use crate::telemetry::metrics;
 
 pub fn spawn_worker(
     client: async_nats::Client,
     engine: ExecutionEngine,
     nats_remote: NATSRemoteRuntime,
     runtime_emitter: NATSRespondEmitter,
     mut runtime_execution_service: Option<TaurusRuntimeExecutionService>,
+    flow_type: String,
 ) -> JoinHandle<()> {
     tokio::spawn(async move {
         let mut execution_subscription = match client
@@ -50,6 +50,7 @@ pub fn spawn_worker(
                                 &nats_remote,
                                 &runtime_emitter,
                                 runtime_execution_service.as_mut(),
+                                flow_type.as_str(),
                             ).await;
                         }
                         None => {
@@ -71,6 +72,7 @@ async fn process_execution_message(
     nats_remote: &NATSRemoteRuntime,
     runtime_emitter: &NATSRespondEmitter,
     mut runtime_execution_service: Option<&mut TaurusRuntimeExecutionService>,
+    flow_type: &str,
 ) {
     let requested_execution_id = parse_execution_id_from_subject(&message.subject, "execution")
         .unwrap_or_else(|| {
@@ -101,6 +103,7 @@ async fn process_execution_message(
     };
 
     let flow_id = flow.flow_id;
+    let function_identifiers = function_identifiers_by_node_id(&flow);
     // Taurus app forwards all lifecycle events to emitter.
     // Direct request/reply responses remain disabled; delivery is emitter-only.
     let respond_emitter = |execution_id, emit_type: EmitType, value: Value| {
@@ -112,6 +115,8 @@ async fn process_execution_message(
         engine,
         Some(nats_remote),
         Some(&respond_emitter),
+        flow_type,
+        function_identifiers,
     )
     .await;
     log::debug!(
@@ -153,9 +158,12 @@ async fn execute_flow(
     engine: &ExecutionEngine,
     remote: Option<&dyn RemoteRuntime>,
     respond_emitter: Option<&dyn RespondEmitter>,
+    flow_type: &str,
+    function_identifiers: std::collections::HashMap<i64, String>,
 ) -> FlowRunResult {
     let started_at = now_unix_micros();
     let flow_id = flow.flow_id;
+    let project_id = flow.project_id;
     let input = flow.input_value.clone();
     let report = engine
         .execute_flow_with_execution_id_report_async(
@@ -167,6 +175,16 @@ async fn execute_flow(
         )
         .await;
     let finished_at = now_unix_micros();
+    record_flow_metrics(
+        flow_id,
+        project_id,
+        flow_type,
+        started_at,
+        finished_at,
+        &report.signal,
+        &report.node_execution_results,
+        &function_identifiers,
+    );
 
     FlowRunResult {
         execution_id,
@@ -179,6 +197,67 @@ async fn execute_flow(
     }
 }
 
+fn function_identifiers_by_node_id(flow: &ExecutionFlow) -> std::collections::HashMap<i64, String> {
+    flow.node_functions
+        .iter()
+        .filter_map(|function| {
+            function
+                .database_id
+                .map(|id| (id, function.runtime_function_id.clone()))
+        })
+        .collect()
+}
+
+fn record_flow_metrics(
+    flow_id: i64,
+    project_id: i64,
+    flow_type: &str,
+    started_at: i64,
+    finished_at: i64,
+    signal: &Signal,
+    node_execution_results: &[NodeExecutionResult],
+    function_identifiers: &std::collections::HashMap<i64, String>,
+) {
+    metrics::flow_execution(metrics::FlowExecution {
+        flow_id,
+        project_id,
+        flow_type,
+        outcome: signal_outcome(signal),
+        duration_seconds: metrics::duration_seconds(started_at, finished_at),
+    });
+
+    for result in node_execution_results {
+        let node_id = metrics::result_node_id(result);
+        let function_identifier = metrics::result_function_identifier(result)
+            .map(ToOwned::to_owned)
+            .or_else(|| node_id.and_then(|id| function_identifiers.get(&id).cloned()))
+            .unwrap_or_else(|| "unknown".to_string());
+        let (error_code, error_category) = metrics::node_result_error(result);
+
+        metrics::function_execution(metrics::FunctionExecution {
+            flow_id,
+            project_id,
+            flow_type,
+            function_identifier: function_identifier.as_str(),
+            node_id,
+            outcome: metrics::node_result_outcome(result),
+            duration_seconds: metrics::duration_seconds(result.started_at, result.finished_at),
+            error_code,
+            error_category,
+        });
+    }
+}
+
+fn signal_outcome(signal: &Signal) -> &'static str {
+    match signal {
+        Signal::Success(_) => "success",
+        Signal::Failure(_) => "failure",
+        Signal::Return(_) => "return",
+        Signal::Respond(_) => "respond",
+        Signal::Stop => "stop",
+    }
+}
+
 fn parse_execution_id_from_subject(
     subject: &async_nats::Subject,
     prefix: &str,
@@ -327,7 +406,17 @@ mod tests {
         let flow = execution_flow_from_fixture(fixture);
         let engine = ExecutionEngine::new();
 
-        let run_result = execute_flow(execution_id, flow, &engine, None, None).await;
+        let function_identifiers = function_identifiers_by_node_id(&flow);
+        let run_result = execute_flow(
+            execution_id,
+            flow,
+            &engine,
+            None,
+            None,
+            "test",
+            function_identifiers,
+        )
+        .await;
 
         println!(
             "started_at={} finished_at={} delta={}",
diff --git a/crates/taurus/src/config/mod.rs b/crates/taurus/src/config/mod.rs
@@ -2,6 +2,8 @@ use code0_flow::flow_config::env_with_default;
 use code0_flow::flow_config::environment::Environment;
 use code0_flow::flow_config::mode::Mode;
 
+use crate::telemetry::OpenTelemetry;
+
 /// Struct for all relevant `Taurus` startup configurations
 pub struct Config {
     pub environment: Environment,
@@ -39,6 +41,9 @@ pub struct Config {
 
     /// Timeout in seconds for remote runtime NATS flush and response waits.
     pub remote_runtime_timeout_secs: u64,
+
+    /// OpenTelemetry exporter configuration.
+    pub opentelemetry: OpenTelemetry,
 }
 
 /// Implementation for all relevant `Taurus` startup configurations
@@ -70,6 +75,23 @@ impl Config {
                 10_u64,
             ),
             remote_runtime_timeout_secs: env_with_default("REMOTE_RUNTIME_TIMEOUT_SECS", 30_u64),
+            opentelemetry: OpenTelemetry {
+                enabled: env_with_default("OPENTELEMETRY_ENABLED", false),
+                service_name: env_with_default(
+                    "OPENTELEMETRY_SERVICE_NAME",
+                    env!("CARGO_PKG_NAME").to_string(),
+                ),
+                logs_endpoint: optional_env("OPENTELEMETRY_LOGS_ENDPOINT"),
+                metrics_endpoint: optional_env("OPENTELEMETRY_METRICS_ENDPOINT"),
+                traces_endpoint: optional_env("OPENTELEMETRY_TRACES_ENDPOINT"),
+            },
         }
     }
 }
+
+fn optional_env(key: &str) -> Option<String> {
+    std::env::var(key)
+        .ok()
+        .map(|value| value.trim().to_owned())
+        .filter(|value| !value.is_empty())
+}
diff --git a/crates/taurus/src/main.rs b/crates/taurus/src/main.rs
@@ -1,6 +1,7 @@
 mod app;
 mod client;
 mod config;
+mod telemetry;
 
 #[tokio::main]
 async fn main() {
diff --git a/crates/taurus/src/telemetry/metrics.rs b/crates/taurus/src/telemetry/metrics.rs
diff --git a/crates/taurus/src/telemetry/mod.rs b/crates/taurus/src/telemetry/mod.rs
