Pipeline metrics - Follow up of open-telemetry#1606 (open-telemetry#1632)

So far, we had hot-path support for **Counter**, **UpDownCounter**, and
**Gauge** metrics. These metrics are defined using macros that specify
the name, description, and unit, which lets us automatically generate a
semantic convention registry for our engine and for all pipeline
components (receiver, processor, exporter).

Some of these metrics are fundamentally **cumulative counters** or
**cumulative up-down counters** (for example, memory usage). In this PR,
I therefore split our `Counter` and `UpDownCounter` types into:

* **DeltaCounter / DeltaUpDownCounter**
* **ObserveCounter / ObserveUpDownCounter**

With the following API:

* delta counters and delta up-down counters expose an `add` method
* observe counters and observe up-down counters expose an `observe`
method
* gauges expose a `set` method

This, I believe, aligns with the OTel intrumentation API model.

Now I'm hitting a problem: I haven't found a way to automatically
represent these **cumulative** metrics without going through an SDK
exporter configuration file. I put a temporary workaround in
`dispatcher.rs`, but I’d like a better and more accurate solution to
connect the Rust Client SDK with these internal metrics.

Author: lquerel

rust/otap-dataflow/Cargo.toml

@@ -37,10 +37,12 @@ thiserror.workspace = true
 quiver = { workspace = true, optional = true }
 serde_json.workspace = true
 clap.workspace = true
-mimalloc.workspace = true
-tikv-jemallocator.workspace = true
+mimalloc = { workspace = true, optional = true }
 rustls = { workspace = true, optional = true }
 
+[target.'cfg(not(windows))'.dependencies]
+tikv-jemallocator = { workspace = true, optional = true }
+
 [workspace.dependencies]
 otap-df-pdata-otlp-macros = { path = "./crates/pdata/src/otlp/macros"}
 otap-df-pdata-otlp-model = { path = "./crates/pdata/src/otlp/model"}
@@ -176,7 +178,10 @@ time = "0.3.44"
 wiremock = "0.6.5"
 
 [features]
-default = []
+# ToDo When jemalloc is enabled, we could use jemalloc_pprof to profile where memory is allocated. See https://crates.io/crates/jemalloc_pprof
+default = ["jemalloc"]
+jemalloc = ["dep:tikv-jemallocator"]
+mimalloc = ["dep:mimalloc"]
 azure = ["otap-df-otap/azure"]
 unsafe-optimizations = ["unchecked-index", "unchecked-arithmetic"]
 unchecked-index = []

rust/otap-dataflow/benchmarks/Cargo.toml

@@ -19,7 +19,6 @@ criterion = { workspace = true, features = ["html_reports", "async_tokio"] }
 tonic = { workspace = true }
 tonic-prost = { workspace = true }
 prost = { workspace = true }
-tikv-jemallocator.workspace = true
 
 otap-df-config = { path = "../crates/config" }
 otap-df-channel = { path = "../crates/channel" }
@@ -39,6 +38,9 @@ unsync.workspace = true
 portpicker.workspace = true
 tokio-stream.workspace = true
 
+[target.'cfg(not(windows))'.dev-dependencies]
+tikv-jemallocator.workspace = true
+
 [lints]
 workspace = true
 

rust/otap-dataflow/benchmarks/benches/attribute_transform/main.rs

@@ -14,8 +14,11 @@ use otap_df_pdata::otap::transform::{
     AttributesTransform, DeleteTransform, RenameTransform, transform_attributes,
 };
 use otap_df_pdata::schema::consts;
+
+#[cfg(not(windows))]
 use tikv_jemallocator::Jemalloc;
 
+#[cfg(not(windows))]
 #[global_allocator]
 static GLOBAL: Jemalloc = Jemalloc;
 

rust/otap-dataflow/benchmarks/benches/channel/main.rs

@@ -12,9 +12,12 @@ use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_m
 use futures::{SinkExt, StreamExt};
 use futures_channel::mpsc as futures_mpsc;
 use std::rc::Rc;
-use tikv_jemallocator::Jemalloc;
 use tokio::task::LocalSet;
 
+#[cfg(not(windows))]
+use tikv_jemallocator::Jemalloc;
+
+#[cfg(not(windows))]
 #[global_allocator]
 static GLOBAL: Jemalloc = Jemalloc;
 

rust/otap-dataflow/benchmarks/benches/exporter/main.rs

@@ -67,10 +67,13 @@ use otap_df_telemetry::MetricsSystem;
 use serde_json::json;
 use std::pin::Pin;
 use std::sync::Arc;
-use tikv_jemallocator::Jemalloc;
 use tokio_stream::Stream;
 use tokio_stream::wrappers::ReceiverStream;
 
+#[cfg(not(windows))]
+use tikv_jemallocator::Jemalloc;
+
+#[cfg(not(windows))]
 #[global_allocator]
 static GLOBAL: Jemalloc = Jemalloc;
 

rust/otap-dataflow/benchmarks/benches/item_count/main.rs

@@ -11,8 +11,11 @@ use std::hint::black_box;
 use otap_df_pdata::proto::opentelemetry::common::v1::*;
 use otap_df_pdata::proto::opentelemetry::logs::v1::*;
 use otap_df_pdata::proto::opentelemetry::resource::v1::*;
+
+#[cfg(not(windows))]
 use tikv_jemallocator::Jemalloc;
 
+#[cfg(not(windows))]
 #[global_allocator]
 static GLOBAL: Jemalloc = Jemalloc;
 

rust/otap-dataflow/benchmarks/benches/materialize_parent_id/main.rs

@@ -14,8 +14,11 @@ use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
 use otap_df_pdata::otap::transform::materialize_parent_id_for_attributes;
 use otap_df_pdata::otlp::attributes::AttributeValueType;
 use otap_df_pdata::schema::consts;
+
+#[cfg(not(windows))]
 use tikv_jemallocator::Jemalloc;
 
+#[cfg(not(windows))]
 #[global_allocator]
 static GLOBAL: Jemalloc = Jemalloc;
 

rust/otap-dataflow/benchmarks/benches/otap_encoder/main.rs

@@ -19,8 +19,11 @@ use otap_df_pdata::proto::opentelemetry::resource::v1::Resource;
 use otap_df_pdata::views::bench_helpers::visit_logs_data;
 use otap_df_pdata::views::otlp::bytes::logs::RawLogsData;
 use prost::Message;
+
+#[cfg(not(windows))]
 use tikv_jemallocator::Jemalloc;
 
+#[cfg(not(windows))]
 #[global_allocator]
 static GLOBAL: Jemalloc = Jemalloc;
 

rust/otap-dataflow/benchmarks/benches/otap_logs_view/main.rs

@@ -32,8 +32,11 @@ use arrow::datatypes::UInt8Type;
 use arrow::error::ArrowError;
 use otap_df_pdata::encode::record::attributes::StrKeysAttributesRecordBatchBuilder;
 use otap_df_pdata::encode::record::logs::LogsRecordBatchBuilder;
+
+#[cfg(not(windows))]
 use tikv_jemallocator::Jemalloc;
 
+#[cfg(not(windows))]
 #[global_allocator]
 static GLOBAL: Jemalloc = Jemalloc;
 

rust/otap-dataflow/benchmarks/benches/otlp_bytes/main.rs

@@ -11,8 +11,11 @@ use prost::Message;
 use otap_df_pdata::proto::opentelemetry::common::v1::*;
 use otap_df_pdata::proto::opentelemetry::logs::v1::*;
 use otap_df_pdata::proto::opentelemetry::resource::v1::*;
+
+#[cfg(not(windows))]
 use tikv_jemallocator::Jemalloc;
 
+#[cfg(not(windows))]
 #[global_allocator]
 static GLOBAL: Jemalloc = Jemalloc;