fix: deepsource

Author: parmesant

src/handlers/http/query.rs

@@ -117,7 +117,7 @@ pub async fn get_records_and_fields(
     Ok((Some(records), Some(fields)))
 }
 
-#[tracing::instrument(name = "query", skip(req, query_request), fields(otel.kind = "server", query.sql = %query_request.query, query.streaming = query_request.streaming))]
+#[tracing::instrument(name = "query", skip(req, query_request), fields(otel.kind = "server", query.sql = %query_request.query, query.streaming = query_request.streaming, query.start_time = query_request.start_time, query.end_time = query_request.end_time))]
 pub async fn query(req: HttpRequest, query_request: Query) -> Result<HttpResponse, QueryError> {
     let mut session_state = QUERY_SESSION.get_ctx().state();
     let time_range =
@@ -497,8 +497,18 @@ pub async fn create_streams_for_distributed(
     }
 
     while let Some(result) = join_set.join_next().await {
-        if let Err(join_error) = result {
-            warn!("Task join error: {}", join_error);
+        match result {
+            Err(join_error) => {
+                warn!("Task join error: {}", join_error);
+            }
+            Ok((stream_name, Err(e))) => {
+                return Err(QueryError::Anyhow(anyhow::anyhow!(
+                    "Failed to create stream '{}': {}",
+                    stream_name,
+                    e
+                )));
+            }
+            Ok((_, Ok(_))) => {}
         }
     }
 

src/main.rs

@@ -31,7 +31,9 @@ use tracing::Level;
 use tracing::{info, warn};
 use tracing_subscriber::layer::SubscriberExt;
 use tracing_subscriber::util::SubscriberInitExt;
-use tracing_subscriber::{EnvFilter, Registry, fmt};
+use tracing_subscriber::{EnvFilter, Layer, Registry, fmt};
+
+const OTEL_TRACE_LEVEL: &str = "OTEL_TRACE_LEVEL";
 
 #[actix_web::main]
 async fn main() -> anyhow::Result<()> {
@@ -105,35 +107,35 @@ async fn main() -> anyhow::Result<()> {
 }
 
 pub fn init_logger() -> Option<SdkTracerProvider> {
-    let filter_layer = EnvFilter::try_from_default_env().unwrap_or_else(|_| {
-        let default_level = if cfg!(debug_assertions) {
-            Level::DEBUG
-        } else {
-            Level::WARN
-        };
-        EnvFilter::new(default_level.to_string())
-    });
+    let default_level = if cfg!(debug_assertions) {
+        Level::DEBUG
+    } else {
+        Level::WARN
+    };
+
+    let fmt_filter = EnvFilter::try_from_default_env()
+        .unwrap_or_else(|_| EnvFilter::new(default_level.to_string()));
 
     let fmt_layer = fmt::layer()
         .with_thread_names(true)
         .with_thread_ids(true)
         .with_line_number(true)
         .with_timer(tracing_subscriber::fmt::time::UtcTime::rfc_3339())
         .with_target(true)
-        .compact();
+        .compact()
+        .with_filter(fmt_filter.clone());
 
     let otel_provider = telemetry::init_otel_tracer();
 
     let otel_layer = otel_provider.as_ref().map(|provider| {
+        let otel_filter = EnvFilter::try_from_env(OTEL_TRACE_LEVEL).unwrap_or(fmt_filter);
         let tracer = provider.tracer("parseable");
-        tracing_opentelemetry::layer().with_tracer(tracer)
+        tracing_opentelemetry::layer()
+            .with_tracer(tracer)
+            .with_filter(otel_filter)
     });
 
-    Registry::default()
-        .with(filter_layer)
-        .with(fmt_layer)
-        .with(otel_layer)
-        .init();
+    Registry::default().with(fmt_layer).with(otel_layer).init();
 
     otel_provider
 }

src/query/mod.rs

@@ -78,7 +78,8 @@ use crate::utils::time::TimeRange;
 // pub static QUERY_SESSION: Lazy<SessionContext> =
 //     Lazy::new(|| Query::create_session_context(PARSEABLE.storage()));
 
-pub type RB = Either<
+/// Takes care of both streaming and non-streaming query flows
+pub type QueryResult = Either<
     Vec<RecordBatch>,
     Pin<
         Box<
@@ -159,7 +160,7 @@ pub async fn execute(
     query: Query,
     is_streaming: bool,
     tenant_id: &Option<String>,
-) -> Result<(RB, Vec<String>), ExecuteError> {
+) -> Result<(QueryResult, Vec<String>), ExecuteError> {
     let id = tenant_id.clone();
 
     // W3C TraceContext propagation across QUERY_RUNTIME (separate OS-thread runtime).
@@ -293,7 +294,7 @@ impl Query {
         &self,
         is_streaming: bool,
         tenant_id: &Option<String>,
-    ) -> Result<(RB, Vec<String>), ExecuteError> {
+    ) -> Result<(QueryResult, Vec<String>), ExecuteError> {
         let df = QUERY_SESSION
             .get_ctx()
             .execute_logical_plan(self.final_logical_plan(tenant_id))

src/telemetry.rs

@@ -25,6 +25,9 @@ use opentelemetry_sdk::{
     trace::{BatchSpanProcessor, SdkTracerProvider},
 };
 
+const EXPORTER_OTLP_ENDPOINT: &str = "OTEL_EXPORTER_OTLP_ENDPOINT";
+const EXPORTER_OTLP_PROTOCOL: &str = "OTEL_EXPORTER_OTLP_PROTOCOL";
+
 /// Initialise an OTLP tracer provider.
 ///
 /// **Required env var:**
@@ -48,10 +51,10 @@ use opentelemetry_sdk::{
 pub fn init_otel_tracer() -> Option<SdkTracerProvider> {
     // Only used to decide whether OTEL is enabled; the SDK reads it again
     // from env to build the exporter (which also appends /v1/traces for HTTP).
-    std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok()?;
+    std::env::var(EXPORTER_OTLP_ENDPOINT).ok()?;
 
     let protocol =
-        std::env::var("OTEL_EXPORTER_OTLP_PROTOCOL").unwrap_or_else(|_| "http/json".to_string());
+        std::env::var(EXPORTER_OTLP_PROTOCOL).unwrap_or_else(|_| "http/json".to_string());
 
     // Build the exporter using the SDK's env-var-aware builders.
     // We intentionally do NOT call .with_endpoint() / .with_headers() /
@@ -69,10 +72,18 @@ pub fn init_otel_tracer() -> Option<SdkTracerProvider> {
         // ── HTTP/JSON (default) ──────────────────────────────────────────────
         // Default when OTEL_EXPORTER_OTLP_PROTOCOL is unset.
         // Required for Parseable OSS — it only accepts application/json.
-        _ => SpanExporter::builder()
+        "http/json" => SpanExporter::builder()
             .with_http()
             .with_protocol(Protocol::HttpJson)
             .build(),
+        other => {
+            tracing::warn!(
+                "Unknown OTEL_EXPORTER_OTLP_PROTOCOL value '{}'; disabling OTEL tracing. \
+                 Supported values: grpc, http/protobuf, http/json",
+                other
+            );
+            return None;
+        }
     };
 
     let exporter = exporter