[zzylol]

feat(rust-sdk): sampling aggregations in a patched opentelemetry-rust SDK

Why

The data plane is moving to Rust: Rust OTel SDK, then the Rust collector
asap-precompute-rs (builds sketches per tumbling window with
asap_sketchlib), then a Rust sink. The Go SDK patch already samples raw
measurements in the application before they are serialized, so the collector
builds sketches from a fraction of the traffic. Upstream opentelemetry-rust
only has six fixed aggregations and no per-measurement hook, so a Rust
application has no way to do the same thing.

What

• opentelemetry-rust added as a submodule at opentelemetry_sdk-0.33.1,
  with an overlay opentelemetry-rust-patch/ (only the changed and added
  files) and restore_opentelemetry_rust_patches.sh /
  backup_opentelemetry_rust_patches.sh, written like the Go ones and run by
  restore_otel_patches.sh / backup_otel_patches.sh. setup.sh initialises
  the new submodule.
• Two new Aggregation variants in opentelemetry_sdk:
  • RowSampled { rows, sample_p } for Count-Min and Count Sketch.
  • Sampled { sample_p, tau } for DDSketch.
• Each kept measurement is exported as one OTLP Gauge point with the same
  reserved attributes as the Go SDK patch, so the collector reads one format.
• opentelemetry-proto's OTLP transform uses a Gauge point's own time when
  it has one.

The application uses the SDK as usual (Meter, instruments, add/record),
turns sampling on with a View, and exports with the standard
opentelemetry-otlp exporter.

How

• opentelemetry-sdk/src/metrics/internal/sampled.rs (new) is one aggregator
  for both variants: RowSampled maps to rows rows, Sampled to one row
  plus tau. It plugs into the existing plumbing (AggregateBuilder::sampled,
  pipeline::aggregate_fn, is_aggregator_compatible), next to LastValue.
• Recording: drop non-finite values; if tau is set and value > tau, keep
  it with sample_p = 1.0; otherwise ask the skip counter for the row mask.
  A non-zero mask stores one GaugeDataPoint with the filtered attributes,
  the three reserved attributes, the raw value and now().
• Collection moves all kept points into the Gauge and clears them (delta
  semantics whatever the reader's temporality, as in the Go patch, since a
  kept measurement is never re-exported).
• GaugeDataPoint gets a crate-private time: Option<SystemTime> with a
  public time() getter. LastValue sets None. The OTLP transform uses
  dp.time().unwrap_or(gauge.time()).
• Validation: StreamBuilder::build runs Aggregation::validate for the new
  variants (rows in 1..=64, sample_p in (0, 1], tau finite), so
  invalid parameters fail when the View's Stream is built.
• Instruments: Counter, UpDownCounter, Histogram and Gauge (f64/u64/i64).
  Observable instruments report one value per callback, so there is nothing
  to sample. They reject the variants, and as for any incompatible View the
  SDK logs a warning and falls back to the default aggregation.
• rand (already an optional SDK dependency) is now part of the metrics
  feature, for SmallRng.

Key code interfaces added / modified

// opentelemetry_sdk::metrics::Aggregation (#[non_exhaustive])
RowSampled { rows: u32, sample_p: f64 },
Sampled { sample_p: f64, tau: Option<f64> },

// opentelemetry_sdk::metrics::data::GaugeDataPoint
pub fn time(&self) -> Option<SystemTime>; // Some only for the sampling aggregations

Reserved OTLP attributes on each exported point (same keys and types as the Go
patch and asapedgeprocessor's extractRowSampledMeta):

key	type	value
__asap_row_sampled_admitted_rows	int	bit r set when row r kept the measurement (uint64 bits as int64)
__asap_row_sampled_rows	int	rows (1 for Sampled)
__asap_row_sampled_sample_p	double	sample_p (1.0 for a tau tail value)

Usage

# Cargo.toml of the application
[dependencies]
opentelemetry = "0.33"
opentelemetry_sdk = "0.33"
opentelemetry-otlp = { version = "0.33", features = ["grpc-tonic", "metrics"] }

[patch.crates-io]
opentelemetry       = { path = "<ASAPCollector>/opentelemetry-rust/opentelemetry" }
opentelemetry_sdk   = { path = "<ASAPCollector>/opentelemetry-rust/opentelemetry-sdk" }
opentelemetry-proto = { path = "<ASAPCollector>/opentelemetry-rust/opentelemetry-proto" }
opentelemetry-otlp  = { path = "<ASAPCollector>/opentelemetry-rust/opentelemetry-otlp" }
opentelemetry-http  = { path = "<ASAPCollector>/opentelemetry-rust/opentelemetry-http" }

Run ./restore_otel_patches.sh (or ./restore_opentelemetry_rust_patches.sh)
first so the submodule contains the overlay.

use opentelemetry::{metrics::MeterProvider as _, KeyValue};
use opentelemetry_otlp::MetricExporter;
use opentelemetry_sdk::metrics::{Aggregation, Instrument, SdkMeterProvider, Stream};

let exporter = MetricExporter::builder().with_tonic().build()?;
let provider = SdkMeterProvider::builder()
    .with_periodic_exporter(exporter)
    // In 0.33 a View is a closure returning Option<Stream>.
    .with_view(|i: &Instrument| {
        let aggregation = match i.name() {
            // Count-Min / Count Sketch with 4 rows built in the collector.
            "http.requests" => Aggregation::RowSampled { rows: 4, sample_p: 0.25 },
            // DDSketch: keep 1% of values, every value above 500 ms.
            "http.latency" => Aggregation::Sampled { sample_p: 0.01, tau: Some(500.0) },
            _ => return None,
        };
        Stream::builder().with_aggregation(aggregation).build().ok()
    })
    .build();

let meter = provider.meter("app");
let requests = meter.u64_counter("http.requests").build();
requests.add(3, &[KeyValue::new("method", "GET")]);
let latency = meter.f64_histogram("http.latency").build();
latency.record(812.0, &[KeyValue::new("route", "/a")]);

Major algorithm changes

Geometric skip sampling (NitroSketch). Instead of a Bernoulli(p) coin per
(measurement, row) pair, the sampler draws the number of pairs to skip before
the next kept one, gap ~ Geometric(p) with P(gap = k) = (1 - p)^k p, by
inverting its CDF: gap = floor(ln(U) / ln(1 - p)), U uniform in (0, 1].
The kept positions have the same distribution as independent Bernoulli(p)
coins, so the collector's ×1/p rescale stays unbiased.

Per-row sampling uses one counter over the flattened stream of
(measurement, row) pairs, as the Go common.GeometricSampler.AdmitRows does.
For a measurement with d rows:

• skip >= d: no row keeps it; skip -= d and the measurement is dropped
  (one comparison and one subtraction, no RNG draw, no allocation).
• otherwise: set bit skip, draw the next gap, jump ahead, and repeat until
  the position passes d; the remainder becomes the new skip.

Each (measurement, row) pair is then kept independently with probability
p, so row r keeps each measurement with probability p and a measurement
is exported with probability 1 - (1 - p)^d. sample_p = 1 skips the RNG and
keeps every row. Sampled is the d = 1 case. Values above tau are kept
outside the sampled stream and do not move the counter.

There is one counter per stream (instrument + View), shared by all attribute
sets, seeded with a fixed constant so runs and tests are reproducible.

Before this PR

A Rust application could only export the six upstream aggregations. To feed a
collector-side Count-Min sketch it had to export every raw measurement, or
pre-aggregate and lose the per-measurement keys.

After this PR

With RowSampled { rows: 4, sample_p: 0.25 }, requests.add(3, [method=GET])
is exported (when at least one row keeps it) as:

Metric "http.requests" Gauge
  NumberDataPoint
    as_int: 3                                   # raw value, not rescaled
    time_unix_nano: <time of the add() call>
    start_time_unix_nano: <start of the collection interval>
    attributes:
      method = "GET"
      __asap_row_sampled_admitted_rows = 5      # 0b0101: rows 0 and 2 kept it
      __asap_row_sampled_rows = 4
      __asap_row_sampled_sample_p = 0.25

A Sampled { sample_p: 0.01, tau: Some(500.0) } value of 812.0 is always
exported with admitted_rows = 1, rows = 1, sample_p = 1.0. A kept value
below tau has sample_p = 0.01. Measurements that no row keeps are never
exported.

Verification

Run from the restored submodule (./restore_opentelemetry_rust_patches.sh,
then cd opentelemetry-rust):

• cargo test -p opentelemetry_sdk --lib --all-features: 420 passed, 25 ignored
  (upstream suite plus the new tests).
  • metrics::internal::sampled::tests (100k measurements each; counts within
    5 standard deviations of the binomial mean): admission rate for
    sample_p = 0.1; every value above tau kept with sample_p = 1.0 and the
    rest sampled at 0.01; per-row frequency p and export rate
    1 - (1 - p)^d for rows = 4, p = 0.25; sample_p = 1 with 64 rows gives
    mask -1; collect clears and NaN/inf are skipped; raw value, filtered
    attributes and per-point time are kept.
  • aggregation::tests::validate_aggregation,
    instrument::tests::stream_sampling_aggregation_validation: invalid
    rows / sample_p / tau rejected.
• cargo test -p opentelemetry-proto --all-features --test asap_sampling:
  MeterProvider with Views selecting both variants, in-memory exporter, OTLP
  encoding; checks values, attributes, per-point times, the three reserved
  attributes, and that an observable gauge falls back to its default
  aggregation.
• cargo fmt --all -- --check;
  cargo clippy -p opentelemetry_sdk -p opentelemetry-proto --all-targets --all-features -- -Dwarnings: clean.
• cargo check -p opentelemetry_sdk --no-default-features --features metrics
  and cargo check -p opentelemetry-otlp -p opentelemetry-stdout --all-features: build.
• Overlay round trip: clean submodule + restore_otel_patches.sh reproduces
  the same diff.
• [patch.crates-io] resolution checked with cargo tree in a scratch crate
  that uses crates.io opentelemetry-otlp 0.33: every opentelemetry crate
  resolves to the submodule, with no duplicates.