Inline OTLP metrics exporter

LICENSE-3rdparty.csv

@@ -192,7 +192,6 @@ crc-catalog,https://github.com/akhilles/crc-catalog,MIT OR Apache-2.0,Akhil Vela
 crc-fast,https://github.com/awesomized/crc-fast-rust,MIT OR Apache-2.0,Don MacAskill
 crc32fast,https://github.com/srijs/rust-crc32fast,MIT OR Apache-2.0,"Sam Rijs <srijs@airpost.net>, Alex Crichton <alex@alexcrichton.com>"
 criterion-plot,https://github.com/criterion-rs/criterion.rs,Apache-2.0 OR MIT,"Jorge Aparicio <japaricious@gmail.com>, Brook Heisler <brookheisler@gmail.com>"
-critical-section,https://github.com/rust-embedded/critical-section,MIT OR Apache-2.0,The critical-section Authors
 cron,https://github.com/zslayton/cron,MIT OR Apache-2.0,Zack Slayton <zack.slayton@gmail.com>
 crossbeam-channel,https://github.com/crossbeam-rs/crossbeam,MIT OR Apache-2.0,The crossbeam-channel Authors
 crossbeam-deque,https://github.com/crossbeam-rs/crossbeam,MIT OR Apache-2.0,The crossbeam-deque Authors
@@ -470,7 +469,6 @@ measure_time,https://github.com/PSeitz/rust_measure_time,MIT,Pascal Seitz <pasca
 memchr,https://github.com/BurntSushi/memchr,Unlicense OR MIT,"Andrew Gallant <jamslam@gmail.com>, bluss"
 memmap2,https://github.com/RazrFalcon/memmap2-rs,MIT OR Apache-2.0,"Dan Burkert <dan@danburkert.com>, Yevhenii Reizner <razrfalcon@gmail.com>, The Contributors"
 metrics,https://github.com/metrics-rs/metrics,MIT,Toby Lawrence <toby@nuclearfurnace.com>
-metrics-exporter-otel,https://github.com/palindrom615/metrics,MIT,Whoemoon Jang <palindrom615@gmail.com>
 metrics-exporter-prometheus,https://github.com/metrics-rs/metrics,MIT AND Apache-2.0,Toby Lawrence <toby@nuclearfurnace.com>
 metrics-util,https://github.com/metrics-rs/metrics,MIT,Toby Lawrence <toby@nuclearfurnace.com>
 mime,https://github.com/hyperium/mime,MIT OR Apache-2.0,Sean McArthur <sean@seanmonstar.com>

quickwit/Cargo.toml

@@ -170,7 +170,6 @@ lru = "0.18"
 matches = "0.1"
 md5 = "0.8"
 metrics = "0.24"
-metrics-exporter-otel = "0.3"
 metrics-exporter-prometheus = { version = "0.18", default-features = false }
 metrics-util = "0.20"
 mime_guess = "2.0"
@@ -192,6 +191,7 @@ parquet = { version = "58", default-features = false, features = ["arrow", "expe
 percent-encoding = "2.3"
 pin-project = "1.1"
 pnet = { version = "0.35", features = ["std"] }
+portable-atomic = { version = "1", features = ["float"] }
 postcard = { version = "1.1", features = [
   "use-std",
 ], default-features = false }

quickwit/quickwit-telemetry-exporters/Cargo.toml

@@ -13,13 +13,13 @@ license.workspace = true
 [dependencies]
 anyhow = { workspace = true }
 metrics = { workspace = true }
-metrics-exporter-otel = { workspace = true }
 metrics-exporter-prometheus = { workspace = true }
 metrics-util = { workspace = true }
 opentelemetry = { workspace = true }
 opentelemetry-appender-tracing = { workspace = true }
 opentelemetry-otlp = { workspace = true }
 opentelemetry_sdk = { workspace = true }
+portable-atomic = { workspace = true }
 serde_json = { workspace = true }
 time = { workspace = true, features = ["parsing"] }
 tracing = { workspace = true }
@@ -28,3 +28,6 @@ tracing-subscriber = { workspace = true }
 
 quickwit-common = { workspace = true }
 quickwit-metrics = { workspace = true }
+
+[dev-dependencies]
+opentelemetry_sdk = { workspace = true, features = ["testing"] }

quickwit/quickwit-telemetry-exporters/src/otlp/metrics.rs

@@ -13,11 +13,11 @@
 // limitations under the License.
 
 use anyhow::Context;
-use metrics_exporter_otel::OpenTelemetryRecorder;
 use opentelemetry::metrics::MeterProvider;
 use opentelemetry_otlp::{MetricExporter, Protocol as OtlpWireProtocol, WithExportConfig};
 use opentelemetry_sdk::metrics::{SdkMeterProvider, Temporality};
 
+use super::metrics_exporter::OtlpMetricsRecorder;
 use crate::otlp::{OtlpExporterConfig, OtlpProtocol, quickwit_resource};
 
 impl OtlpProtocol {
@@ -48,7 +48,7 @@ impl OtlpProtocol {
 pub(crate) fn build_recorder(
     service_version: &str,
     otlp_config: &OtlpExporterConfig,
-) -> anyhow::Result<(OpenTelemetryRecorder, SdkMeterProvider)> {
+) -> anyhow::Result<(OtlpMetricsRecorder, SdkMeterProvider)> {
     let metrics_protocol = otlp_config.metrics_protocol()?;
     let temporality = otlp_config.metrics_temporality()?;
     let metric_exporter = metrics_protocol.metric_exporter(temporality)?;
@@ -58,7 +58,7 @@ pub(crate) fn build_recorder(
         .build();
     let meter = metrics_provider.meter("quickwit");
 
-    let recorder = OpenTelemetryRecorder::new(meter);
+    let recorder = OtlpMetricsRecorder::new(meter);
     for (name, buckets) in quickwit_metrics::histogram_buckets() {
         recorder.set_histogram_bounds(&metrics::KeyName::from(name), buckets);
     }

quickwit/quickwit-telemetry-exporters/src/otlp/metrics_exporter/mod.rs

@@ -0,0 +1,22 @@
+// Copyright 2021-Present Datadog, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+mod recorder;
+mod state;
+mod storage;
+
+pub(crate) use recorder::OtlpMetricsRecorder;
+
+#[cfg(test)]
+mod tests;

quickwit/quickwit-telemetry-exporters/src/otlp/metrics_exporter/recorder.rs

@@ -0,0 +1,72 @@
+// Copyright 2021-Present Datadog, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use metrics::{Counter, Gauge, Histogram, Key, KeyName, Metadata, Recorder, SharedString, Unit};
+use metrics_util::MetricKind;
+
+use super::state::OtlpMetricsState;
+
+#[derive(Clone)]
+pub(crate) struct OtlpMetricsRecorder {
+    state: Arc<OtlpMetricsState>,
+}
+
+impl OtlpMetricsRecorder {
+    pub(crate) fn new(meter: opentelemetry::metrics::Meter) -> Self {
+        Self {
+            state: Arc::new(OtlpMetricsState::new(meter)),
+        }
+    }
+
+    pub(crate) fn set_histogram_bounds(&self, key_name: &KeyName, bounds: Vec<f64>) {
+        self.state.set_histogram_bounds(key_name, bounds);
+    }
+}
+
+impl Recorder for OtlpMetricsRecorder {
+    fn describe_counter(&self, key_name: KeyName, unit: Option<Unit>, description: SharedString) {
+        self.state
+            .set_description(key_name, MetricKind::Counter, unit, description);
+    }
+
+    fn describe_gauge(&self, key_name: KeyName, unit: Option<Unit>, description: SharedString) {
+        self.state
+            .set_description(key_name, MetricKind::Gauge, unit, description);
+    }
+
+    fn describe_histogram(&self, key_name: KeyName, unit: Option<Unit>, description: SharedString) {
+        self.state
+            .set_description(key_name, MetricKind::Histogram, unit, description);
+    }
+
+    fn register_counter(&self, key: &Key, _metadata: &Metadata<'_>) -> Counter {
+        self.state
+            .registry()
+            .get_or_create_counter(key, |counter| Counter::from_arc(counter.clone()))
+    }
+
+    fn register_gauge(&self, key: &Key, _metadata: &Metadata<'_>) -> Gauge {
+        self.state
+            .registry()
+            .get_or_create_gauge(key, |gauge| Gauge::from_arc(gauge.clone()))
+    }
+
+    fn register_histogram(&self, key: &Key, _metadata: &Metadata<'_>) -> Histogram {
+        self.state
+            .registry()
+            .get_or_create_histogram(key, |histogram| Histogram::from_arc(histogram.clone()))
+    }
+}

quickwit/quickwit-telemetry-exporters/src/otlp/metrics_exporter/state.rs

@@ -0,0 +1,121 @@
+// Copyright 2021-Present Datadog, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::collections::HashMap;
+use std::sync::{Arc, RwLock};
+
+use metrics::{Key, KeyName, SharedString, Unit};
+use metrics_util::MetricKind;
+use metrics_util::registry::Registry;
+
+use super::storage::OtlpMetricStorage;
+
+pub(super) struct OtlpMetricsState {
+    registry: Registry<Key, OtlpMetricStorage>,
+    metadata: OtlpMetricMetadata,
+}
+
+impl OtlpMetricsState {
+    pub(super) fn new(meter: opentelemetry::metrics::Meter) -> Self {
+        let metadata = OtlpMetricMetadata::default();
+        let registry = Registry::new(OtlpMetricStorage::new(meter, metadata.clone()));
+        Self { registry, metadata }
+    }
+
+    pub(super) fn registry(&self) -> &Registry<Key, OtlpMetricStorage> {
+        &self.registry
+    }
+
+    pub(super) fn set_description(
+        &self,
+        key_name: KeyName,
+        metric_kind: MetricKind,
+        unit: Option<Unit>,
+        description: SharedString,
+    ) {
+        self.metadata
+            .set_description(key_name, metric_kind, unit, description);
+    }
+
+    pub(super) fn set_histogram_bounds(&self, key_name: &KeyName, bounds: Vec<f64>) {
+        self.metadata
+            .set_histogram_bounds(key_name.to_retained(), bounds);
+    }
+}
+
+#[derive(Clone, Default)]
+pub(super) struct OtlpMetricMetadata {
+    inner: Arc<OtlpMetricMetadataInner>,
+}
+
+#[derive(Default)]
+struct OtlpMetricMetadataInner {
+    descriptions: RwLock<HashMap<(KeyName, MetricKind), OtlpMetricDescription>>,
+    histogram_bounds: RwLock<HashMap<KeyName, Vec<f64>>>,
+}
+
+impl OtlpMetricMetadata {
+    pub(super) fn set_description(
+        &self,
+        key_name: KeyName,
+        metric_kind: MetricKind,
+        unit: Option<Unit>,
+        description: SharedString,
+    ) {
+        self.inner
+            .descriptions
+            .write()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .insert(
+                (key_name.to_retained(), metric_kind),
+                OtlpMetricDescription { unit, description },
+            );
+    }
+
+    pub(super) fn get_description(
+        &self,
+        key_name: &KeyName,
+        metric_kind: MetricKind,
+    ) -> Option<OtlpMetricDescription> {
+        self.inner
+            .descriptions
+            .read()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .get(&(key_name.to_retained(), metric_kind))
+            .cloned()
+    }
+
+    fn set_histogram_bounds(&self, key_name: KeyName, bounds: Vec<f64>) {
+        self.inner
+            .histogram_bounds
+            .write()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .insert(key_name, bounds);
+    }
+
+    pub(super) fn get_histogram_bounds(&self, key_name: &KeyName) -> Option<Vec<f64>> {
+        self.inner
+            .histogram_bounds
+            .read()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .get(&key_name.to_retained())
+            .cloned()
+    }
+}
+
+#[derive(Clone)]
+pub(super) struct OtlpMetricDescription {
+    pub(super) unit: Option<Unit>,
+    pub(super) description: SharedString,
+}