remove OTEL dependency to keep system minimal

Author: agavra

Cargo.lock

@@ -18,7 +18,6 @@ thiserror.workspace = true
 uuid.workspace = true
 
 blake3 = "1.8.2"
-opentelemetry-proto = "0.30"
 promql-parser = "0.6"
 roaring = "0.7"
 tsz = "0.1"

open-tsdb/Cargo.toml

@@ -2,13 +2,14 @@ use std::collections::HashMap;
 use std::sync::atomic::{AtomicU32, Ordering};
 
 use dashmap::DashMap;
-use opentelemetry_proto::tonic::metrics::v1::{Metric, MetricsData};
 
 use crate::{
     index::{ForwardIndex, InvertedIndex},
-    model::{Attribute, MetricType, Sample, SeriesFingerprint, SeriesId, SeriesSpec, TimeBucket},
-    otel::{OtelUtil, collect_resource_attributes, collect_scope_attributes},
-    util::{Fingerprint, Result, normalize_str},
+    model::{
+        Attribute, MetricType, Sample, SampleWithAttributes, SeriesFingerprint, SeriesId,
+        SeriesSpec, TimeBucket,
+    },
+    util::Fingerprint,
 };
 
 /// The delta chunk is the current in-memory segment of OpenTSDB representing
@@ -40,44 +41,14 @@ impl<'a> TsdbDeltaBuilder<'a> {
         }
     }
 
-    /// Ingest a full MetricsData payload, processing all resource_metrics, scope_metrics, and metrics.
-    /// Resource and scope attributes are merged into each sample's attributes.
-    pub(crate) fn ingest_metrics_data(mut self, data: MetricsData) -> Result<Self> {
-        for resource_metrics in data.resource_metrics {
-            let resource_attrs = collect_resource_attributes(resource_metrics.resource.as_ref());
-
-            for scope_metrics in resource_metrics.scope_metrics {
-                let scope_attrs = collect_scope_attributes(scope_metrics.scope.as_ref());
-
-                for metric in scope_metrics.metrics {
-                    self.ingest_metric_internal(&metric, &resource_attrs, &scope_attrs)?;
-                }
-            }
-        }
-        Ok(self)
-    }
-
-    /// Internal method to ingest a single metric with pre-collected resource and scope attributes
-    fn ingest_metric_internal(
-        &mut self,
-        metric: &Metric,
-        resource_attrs: &[Attribute],
-        scope_attrs: &[Attribute],
-    ) -> Result<()> {
-        let metric_unit = normalize_str(&metric.unit);
-        let metric_type = MetricType::try_from(metric)?;
-        let samples_with_attrs = OtelUtil::samples(metric, resource_attrs, scope_attrs);
-
-        for sample_with_attrs in samples_with_attrs {
-            self.ingest_sample(
-                sample_with_attrs.attributes,
-                metric_unit.clone(),
-                metric_type,
-                sample_with_attrs.sample,
-            );
-        }
-
-        Ok(())
+    /// Ingest a sample with its attributes
+    pub(crate) fn ingest(&mut self, sample_with_attrs: SampleWithAttributes) {
+        self.ingest_sample(
+            sample_with_attrs.attributes,
+            sample_with_attrs.metric_unit,
+            sample_with_attrs.metric_type,
+            sample_with_attrs.sample,
+        );
     }
 
     fn ingest_sample(

open-tsdb/src/delta.rs

@@ -4,7 +4,6 @@ mod head;
 mod index;
 mod minitsdb;
 mod model;
-mod otel;
 mod promql;
 mod serde;
 mod storage;

open-tsdb/src/main.rs

@@ -23,16 +23,13 @@ use std::sync::{Arc, atomic::AtomicU32};
 
 use dashmap::DashMap;
 use opendata_common::{Storage, StorageRead};
-use opentelemetry_proto::tonic::metrics::v1::MetricsData;
 use tokio::sync::{Mutex, RwLock};
 
 use crate::delta::{TsdbDelta, TsdbDeltaBuilder};
+use crate::head::TsdbHead;
+use crate::model::{SampleWithAttributes, SeriesFingerprint, SeriesId, TimeBucket};
 use crate::storage::OpenTsdbStorageReadExt;
 use crate::util::Result;
-use crate::{
-    head::TsdbHead,
-    model::{SeriesFingerprint, SeriesId, TimeBucket},
-};
 
 pub(crate) struct MiniState {
     head: TsdbHead,
@@ -104,13 +101,16 @@ impl MiniTsdb {
         })
     }
 
-    /// Ingest an OLTP payload (this is the main entry point for ingestion)
-    pub(crate) async fn ingest(&self, data: MetricsData) -> Result<()> {
-        let delta =
-            TsdbDeltaBuilder::new(self.bucket.clone(), &self.series_dict, &self.next_series_id)
-                .ingest_metrics_data(data)?
-                .build();
+    /// Ingest samples with attributes (this is the main entry point for ingestion)
+    pub(crate) async fn ingest(&self, samples: Vec<SampleWithAttributes>) -> Result<()> {
+        let mut builder =
+            TsdbDeltaBuilder::new(self.bucket.clone(), &self.series_dict, &self.next_series_id);
 
+        for sample in samples {
+            builder.ingest(sample);
+        }
+
+        let delta = builder.build();
         self.ingest_delta(&delta).await
     }
 

open-tsdb/src/minitsdb.rs

@@ -84,6 +84,15 @@ pub(crate) struct Sample {
     pub(crate) value: f64,
 }
 
+/// A sample with all its attributes, including the metric name
+#[derive(Clone, Debug)]
+pub(crate) struct SampleWithAttributes {
+    pub(crate) attributes: Vec<Attribute>,
+    pub(crate) metric_unit: Option<String>,
+    pub(crate) metric_type: MetricType,
+    pub(crate) sample: Sample,
+}
+
 /// Convert TimeBucketSize to hours
 pub fn time_bucket_size_hours(size: BucketSize) -> u32 {
     if size == 0 || size > 15 {