Merge pull request #254 from code0-tech/#228-add-open-telemetry

add open telemetry

Author: raphael-goetz

Cargo.lock

@@ -12,6 +12,7 @@ code0-flow = { version = "0.0.40" }
 tucana = { version = "0.0.75" }
 tokio = { version = "1.44.1", features = ["rt-multi-thread", "signal"] }
 log = "0.4.27"
+opentelemetry = { version = "0.32.0", features = ["metrics"] }
 futures-lite = "2.6.0"
 rand = "0.10.0"
 base64 = "0.22.1"

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 }

crates/taurus/Cargo.toml

@@ -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,14 @@ 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, errors};
 
 pub async fn run() {
-    init_logging();
     load_env_file();
 
     let config = Config::new();
+    let telemetry = init_telemetry(&config);
+    install_panic_logging();
     let engine = ExecutionEngine::new();
     let client = connect_nats(&config).await;
 
@@ -42,6 +45,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;
@@ -51,17 +55,72 @@ pub async fn run() {
             && !err.is_cancelled()
         {
             log::warn!("Runtime status heartbeat task ended unexpectedly: {}", err);
+            errors::record(
+                "task",
+                "runtime_status_heartbeat.task",
+                &err,
+                "mode=dynamic",
+            );
         }
     }
     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 install_panic_logging() {
+    std::panic::set_hook(Box::new(move |panic_info| {
+        let message = if let Some(message) = panic_info.payload().downcast_ref::<&str>() {
+            *message
+        } else if let Some(message) = panic_info.payload().downcast_ref::<String>() {
+            message.as_str()
+        } else {
+            "<non-string panic payload>"
+        };
+
+        let location = panic_info
+            .location()
+            .map(|location| {
+                format!(
+                    "{}:{}:{}",
+                    location.file(),
+                    location.line(),
+                    location.column()
+                )
+            })
+            .unwrap_or_else(|| "unknown".into());
+        errors::panic(message, &location);
+    }));
+}
+
+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 {
@@ -71,6 +130,7 @@ async fn connect_nats(config: &Config) -> async_nats::Client {
             client
         }
         Err(err) => {
+            errors::record("transport", "nats.connect", &err, "component=nats");
             panic!("Failed to connect to NATS server: {}", err);
         }
     }
@@ -86,6 +146,12 @@ fn spawn_health_task(config: &Config) -> Option<JoinHandle<()>> {
         Ok(address) => address,
         Err(err) => {
             log::error!("Failed to parse gRPC address: {:?}", err);
+            errors::record(
+                "configuration",
+                "health.address.parse",
+                &err,
+                "service=health",
+            );
             return None;
         }
     };
@@ -98,6 +164,7 @@ fn spawn_health_task(config: &Config) -> Option<JoinHandle<()>> {
             .await
         {
             log::error!("Health server error: {:?}", err);
+            errors::record("server", "health.serve", &err, "service=health");
         } else {
             log::info!("Health server stopped gracefully");
         }
@@ -196,6 +263,12 @@ fn read_module_status_identifiers(definition_path: &str) -> Vec<String> {
                 "Failed to read module definitions for runtime status: {:?}",
                 err
             );
+            errors::record_message(
+                "configuration",
+                "definitions.read_modules",
+                format!("{err:?}"),
+                format!("path={definition_path}"),
+            );
             Vec::new()
         }
     }

crates/taurus/src/app/mod.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::{errors, 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
@@ -33,6 +33,12 @@ pub fn spawn_worker(
             }
             Err(err) => {
                 log::error!("Failed to subscribe to 'execution.*': {:?}", err);
+                errors::record(
+                    "transport",
+                    "nats.subscribe",
+                    &err,
+                    "subject=execution.* queue=taurus",
+                );
                 return;
             }
         };
@@ -50,6 +56,7 @@ pub fn spawn_worker(
                                 &nats_remote,
                                 &runtime_emitter,
                                 runtime_execution_service.as_mut(),
+                                flow_type.as_str(),
                             ).await;
                         }
                         None => {
@@ -71,6 +78,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(|| {
@@ -91,6 +99,16 @@ async fn process_execution_message(
                 err,
                 &message.payload
             );
+            errors::record(
+                "serialization",
+                "execution_flow.decode",
+                &err,
+                format!(
+                    "subject={} payload_bytes={}",
+                    message.subject,
+                    message.payload.len()
+                ),
+            );
             if let Some(execution_service) = runtime_execution_service.as_mut() {
                 execution_service
                     .update_runtime_execution(build_decode_error_result(requested_execution_id))
@@ -101,6 +119,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 +131,8 @@ async fn process_execution_message(
         engine,
         Some(nats_remote),
         Some(&respond_emitter),
+        flow_type,
+        function_identifiers,
     )
     .await;
     log::debug!(
@@ -153,9 +174,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 +191,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 +213,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 +422,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={}",