initial version of app-server observability (#972)

* initial version of app-server observability

* remove dbg

* unify tracer name

* add compatibility comment in Cargo.toml

* small fixes in main.rs

* add num bytes attribute

* fix bugbot comment

* extract tracing setup to a separate module

Author: dinmukhamedm

app-server/Cargo.lock

@@ -23,21 +23,24 @@ deadpool = "0.12.2"
 dotenv = "0.15"
 enum_delegate = "0.2.0"
 enum_dispatch = "0.3.13"
-env_logger = "0.11"
 futures-util = "0.3"
 indexmap = {version = "2.11.4", features = ["serde"]}
 itertools = "0.14.0"
 lapin = "3.0.0"
 log = "0.4.28"
 moka = {version = "0.12.10", features = ["sync", "future"]}
 num_cpus = "1.16.0"
+opentelemetry = {version = "0.29.1", features = ["trace"]}
+opentelemetry_sdk = {version = "0.29.0", features = ["trace"]}
 prost = "0.13"
 rand = "0.9.1"
 rayon = "1.10.0"
 redis = {version = "0.32.4", features = ["tokio-comp"]}
 regex = "1.11.1"
 reqwest = {version = "0.12.22", features = ["json"]}
 rustls = {version = "0.23", features = ["ring"]}
+# Sentry SDK must be compatible with the opentelemetry version. For sentry = 0.45.0, opentelemetry = 0.29
+sentry = {version = "0.45.0", features = ["opentelemetry"]}
 serde = "1.0"
 serde_json = {version = "1.0.140", features = ["preserve_order"]}
 sha3 = "0.10.8"
@@ -48,6 +51,10 @@ tikv-jemallocator = "0.6"
 tokio = {version = "1.46", features = ["macros", "rt-multi-thread"]}
 tokio-stream = {version = "0.1", features = ["net"]}
 tonic = {version = "0.13", features = ["gzip"]}
+tracing = {version = "0.1.41", features = ["attributes"]}
+# Tracing opentelemetry must be compatible with the opentelemetry version. For tracing-opentelemetry = 0.30.0, opentelemetry = 0.29
+tracing-opentelemetry = "0.30.0"
+tracing-subscriber = {version = "0.3", features = ["env-filter", "fmt"]}
 url = "2.5.4"
 uuid = {version = "1.18.1", features = ["v4", "fast-rng", "macro-diagnostics", "serde"]}
 

app-server/Cargo.toml

@@ -22,7 +22,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         .build_client(false)
         .build_server(true)
         .include_file("mod.rs")
-        .out_dir("./src/opentelemetry/")
+        .out_dir("./src/opentelemetry_proto/")
         .compile_protos(
             &[
                 "./proto/opentelemetry/common.proto",

app-server/build.rs

@@ -1,6 +1,10 @@
 use std::{collections::HashMap, sync::Arc};
 
 use actix_web::{HttpResponse, post, web};
+use opentelemetry::{
+    global,
+    trace::{Tracer, mark_span_as_active},
+};
 use serde::{Deserialize, Serialize};
 
 use crate::{
@@ -33,6 +37,10 @@ pub async fn execute_sql_query(
     let project_id = project_api_key.project_id;
     let SqlQueryRequest { query } = req.into_inner();
 
+    let tracer = global::tracer("tracer");
+    let span = tracer.start("api_sql_query");
+    let _guard = mark_span_as_active(span);
+
     match clickhouse_ro.as_ref() {
         Some(ro_client) => {
             match sql::execute_sql_query(

app-server/src/api/v1/sql.rs

@@ -8,7 +8,7 @@ use crate::{
     db::{DB, events::Event, project_api_keys::ProjectApiKey, spans::Span},
     features::{Feature, is_feature_enabled},
     mq::MessageQueue,
-    opentelemetry::opentelemetry::proto::collector::trace::v1::ExportTraceServiceRequest,
+    opentelemetry_proto::opentelemetry::proto::collector::trace::v1::ExportTraceServiceRequest,
     routes::types::ResponseResult,
     traces::{limits::get_workspace_limit_exceeded_by_project_id, producer::push_spans_to_queue},
 };

app-server/src/api/v1/traces.rs

@@ -1,6 +1,7 @@
 use anyhow::Result;
 use clickhouse::Row;
 use serde::{Deserialize, Serialize};
+use tracing::instrument;
 use uuid::Uuid;
 
 use crate::{
@@ -175,6 +176,7 @@ impl CHSpan {
     }
 }
 
+#[instrument(skip(clickhouse, spans))]
 pub async fn insert_spans_batch(clickhouse: clickhouse::Client, spans: &[CHSpan]) -> Result<()> {
     if spans.is_empty() {
         return Ok(());

app-server/src/ch/spans.rs

@@ -3,6 +3,7 @@ use std::collections::HashSet;
 use chrono::{DateTime, Utc};
 use clickhouse::Row;
 use serde::{Deserialize, Serialize};
+use tracing::instrument;
 use uuid::Uuid;
 
 use super::utils::chrono_to_nanoseconds;
@@ -218,6 +219,7 @@ impl TraceAggregation {
 }
 
 /// Insert or update traces in ClickHouse traces_replacing table
+#[instrument(skip(client, traces))]
 pub async fn upsert_traces_batch(
     client: clickhouse::Client,
     traces: &[CHTrace],

app-server/src/ch/traces.rs

@@ -4,7 +4,7 @@ use serde_json::Value;
 use uuid::Uuid;
 
 use crate::{
-    opentelemetry::opentelemetry_proto_trace_v1::span::Event as OtelEvent,
+    opentelemetry_proto::opentelemetry_proto_trace_v1::span::Event as OtelEvent,
     utils::estimate_json_size,
 };
 

app-server/src/db/events.rs

@@ -3,6 +3,7 @@ use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use sqlx::PgPool;
+use tracing::instrument;
 use uuid::Uuid;
 
 use crate::ch::traces::TraceAggregation;
@@ -111,6 +112,7 @@ impl Trace {
 
 /// Upsert trace statistics from aggregated span data
 /// Returns the updated trace statistics
+#[instrument(skip(pool, aggregations))]
 pub async fn upsert_trace_statistics_batch(
     pool: &PgPool,
     aggregations: &[TraceAggregation],

app-server/src/db/trace.rs

@@ -15,6 +15,7 @@ pub enum Feature {
     RabbitMQ,
     SqlQueryEngine,
     ClickhouseReadOnly,
+    Tracing,
 }
 
 pub fn is_feature_enabled(feature: Feature) -> bool {
@@ -40,5 +41,8 @@ pub fn is_feature_enabled(feature: Feature) -> bool {
         Feature::ClickhouseReadOnly => {
             env::var("CLICKHOUSE_RO_USER").is_ok() && env::var("CLICKHOUSE_RO_PASSWORD").is_ok()
         }
+        Feature::Tracing => {
+            env::var("SENTRY_DSN").is_ok() && env::var("ENABLE_TRACING").is_ok_and(|s| s == "true")
+        }
     }
 }