diff --git a/.spelling b/.spelling index f7bea220b..f63837118 100644 --- a/.spelling +++ b/.spelling @@ -631,3 +631,8 @@ ASTs DAGs representable rc +rescale +rescaled +rescaling +rescales +configurator diff --git a/AGENTS.md b/AGENTS.md index 770cd3e76..35a024fdc 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -48,6 +48,8 @@ packaged contents. Pull request titles must follow [Conventional Commits](https://www.conventionalcommits.org/) naming, e.g. `feat(bytesbuf): add new metric` or `fix(cachet): correct eviction logic`. +Pull request descriptions must be high-level and describe only the content the PR delivers and the problem it fixes. Do not narrate the history of how the PR came to be, or how it was created, designed, reviewed, or tested. + ## Feature-gated Doctests Doctests that reference items behind a Cargo feature must compile both with and without that feature; wrap their bodies in hidden `#[cfg(...)]` shims. See [AGENTS-feature-gated-doctests.md](AGENTS-feature-gated-doctests.md). diff --git a/Cargo.lock b/Cargo.lock index aee88eab4..aafa3ddb7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2684,6 +2684,15 @@ dependencies = [ "opentelemetry_sdk", ] +[[package]] +name = "opentelemetry_rescaled" +version = "0.1.0" +dependencies = [ + "foldhash 0.2.0", + "opentelemetry", + "opentelemetry_sdk", +] + [[package]] name = "opentelemetry_sdk" version = "0.32.1" diff --git a/crates/opentelemetry_rescaled/Cargo.toml b/crates/opentelemetry_rescaled/Cargo.toml new file mode 100644 index 000000000..29cdd8b7f --- /dev/null +++ b/crates/opentelemetry_rescaled/Cargo.toml @@ -0,0 +1,39 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +[package] +name = "opentelemetry_rescaled" +description = "Wraps an OpenTelemetry meter provider to emit rescaled side-by-side copies of selected instruments." +version = "0.1.0" +readme = "README.md" +keywords = ["opentelemetry", "metrics", "telemetry", "rescale"] +categories = ["development-tools::debugging"] +edition = { workspace = true } +rust-version = { workspace = true } +authors = { workspace = true } +license = { workspace = true } +homepage = { workspace = true } +include = { workspace = true } +repository = "https://github.com/microsoft/oxidizer/tree/main/crates/opentelemetry_rescaled" + +[package.metadata.cargo_check_external_types] +allowed_external_types = [ + "opentelemetry::metrics::meter::Meter", + "opentelemetry::metrics::meter::MeterProvider", +] + +[package.metadata.docs.rs] +all-features = true + +[features] +default = [] + +[lints] +workspace = true + +[dependencies] +foldhash = { workspace = true } +opentelemetry = { workspace = true, features = ["metrics"] } + +[dev-dependencies] +opentelemetry_sdk = { workspace = true, features = ["metrics", "testing"] } diff --git a/crates/opentelemetry_rescaled/README.md b/crates/opentelemetry_rescaled/README.md new file mode 100644 index 000000000..84df72a33 --- /dev/null +++ b/crates/opentelemetry_rescaled/README.md @@ -0,0 +1,64 @@ +
+ OpenTelemetry Rescaled Logo + +# OpenTelemetry Rescaled + +[![crate.io](https://img.shields.io/crates/v/opentelemetry_rescaled.svg)](https://crates.io/crates/opentelemetry_rescaled) +[![docs.rs](https://docs.rs/opentelemetry_rescaled/badge.svg)](https://docs.rs/opentelemetry_rescaled) +[![MSRV](https://img.shields.io/crates/msrv/opentelemetry_rescaled)](https://crates.io/crates/opentelemetry_rescaled) +[![CI](https://github.com/microsoft/oxidizer/actions/workflows/main.yml/badge.svg?event=push)](https://github.com/microsoft/oxidizer/actions/workflows/main.yml) +[![Coverage](https://codecov.io/gh/microsoft/oxidizer/graph/badge.svg?token=FCUG0EL5TI)](https://codecov.io/gh/microsoft/oxidizer) +[![License](https://img.shields.io/badge/license-MIT-blue.svg)](../../LICENSE) +This crate was developed as part of the Oxidizer project + +
+ +Wraps an inner OpenTelemetry meter provider to transparently emit *rescaled* side-by-side copies of selected instruments. + +For a chosen instrument in a chosen instrumentation scope, this layer creates +a second instrument whose measurements are the original values multiplied by a +fixed factor. For example, a `http.client.request.duration` instrument that +records seconds can gain a `http.client.request.duration.millis` sidecar that +records the same measurements multiplied by `1000.0`. + +The rescaling is invisible to instrument users — they interact only with their +original instrument — and the inner provider simply sees two independently +registered instruments. + +## Quick start + +```rust +use opentelemetry::metrics::MeterProvider; +use opentelemetry_rescaled::RescaledMetrics; + +// Any `MeterProvider` works as the inner provider. +let inner = opentelemetry::metrics::noop::NoopMeterProvider::new(); + +let outer = RescaledMetrics::builder(inner) + .scope("my_scope_name", |scope| { + // source name, target name, target unit (mandatory), factor + scope.rescale( + "http.client.request.duration", + "http.client.request.duration.millis", + "ms", + 1000.0, + ); + }) + .build(); + +// `outer` is itself a `MeterProvider`. +let meter = outer.meter("my_scope_name"); +let histogram = meter.f64_histogram("http.client.request.duration").build(); +histogram.record(1.5, &[]); // recorded as 1.5 s and, on the sidecar, 1500 ms +``` + +See [`docs/DESIGN.md`][__link0] +for the architecture and the resolved design decisions. + + +
+ +This crate was developed as part of The Oxidizer Project. Browse this crate's source code. + + + [__link0]: https://github.com/microsoft/oxidizer/blob/main/crates/opentelemetry_rescaled/docs/DESIGN.md diff --git a/crates/opentelemetry_rescaled/docs/DESIGN.md b/crates/opentelemetry_rescaled/docs/DESIGN.md new file mode 100644 index 000000000..5dab3dac8 --- /dev/null +++ b/crates/opentelemetry_rescaled/docs/DESIGN.md @@ -0,0 +1,178 @@ +# OpenTelemetry Rescaled — Architecture & Design + +This document describes the intended design of the crate so it can be reviewed +before implementation. For the user-facing summary see the crate-level rustdoc +(`src/lib.rs`). + +## Goal + +Provide a wrapping layer around an existing OpenTelemetry meter provider that, +for specific instruments in specific instrumentation scopes, transparently emits +a **rescaled sidecar** of each such instrument: a second instrument carrying the +same measurements multiplied by a fixed factor. + +The canonical motivating case: an instrument records a duration in seconds +(`http.client.request.duration`) and a downstream system expects milliseconds. +Rather than change the instrumented code, the operator configures a sidecar +`http.client.request.duration.millis` with factor `1000.0`. Both instruments are +exported independently by the underlying SDK. + +Scope: **metrics only**, all instrument kinds (synchronous and observable, for +every value type the API supports). + +## Tenets + +- **Transparent to the measurer.** Code recording measurements sees exactly one + instrument — its original. It has no knowledge that a sidecar exists, and its + hot path is unchanged apart from the fan-out described below. +- **Transparent to the inner provider.** The wrapped SDK sees two ordinary, + independently registered instruments. No SDK internals are touched; the layer + composes purely through the public OpenTelemetry API surface. +- **Configured once, at build time.** The set of scopes, source instruments, + targets, units, and factors is fixed when the provider is built and never + changes for the life of the provider. +- **Zero cost where unused.** Scopes and instruments that are not configured for + rescaling incur no wrapping and delegate directly to the inner provider. +- **Fail fast on nonsense.** A configuration that cannot produce a meaningful + sidecar (see [Configuration model](#configuration-model)) panics at build time + rather than silently emitting garbage. + +## Usage shape + +```rust,ignore +let inner = build_inner_meter_provider(); + +let outer = RescaledMetrics::builder(inner) + .scope("my_scope_name", |scope| { + // source name, target name, target unit (mandatory), factor + scope.rescale("http.client.request.duration", + "http.client.request.duration.millis", + "ms", + 1000.0); + }) + .build(); + +// `outer` is itself a `MeterProvider`; hand it wherever the inner one went +// (e.g. to instrumented libraries, or `global::set_meter_provider`). +``` + +## How interception works + +OpenTelemetry's Rust metrics API is layered as +`MeterProvider` → `Meter` → typed instrument builders → concrete instruments, and +a `Meter` is nothing more than a handle to an `InstrumentProvider`. This layering +is the seam the crate exploits: it substitutes its own `MeterProvider` and +`InstrumentProvider` while delegating all real work to the inner ones. + +### The provider wrapper + +`RescaledMetrics` implements `MeterProvider`. When a scoped meter is requested it +resolves the inner scoped meter and then decides, by matching the scope against +the configuration: + +- **Unconfigured scope** → return the inner meter unchanged (no wrapping, no + overhead). +- **Configured scope** → return a meter backed by a *rescaling instrument + provider* that holds the resolved inner meter plus that scope's rescale rules. + +### Synchronous instruments — fan-out + +A synchronous instrument (`Counter`, `UpDownCounter`, `Gauge`, `Histogram`) +delegates every measurement to an inner `SyncInstrument`. The rescaling provider +constructs, for a configured source instrument, **two** inner instruments — the +original and the sidecar — and returns to the caller a single instrument whose +backing `SyncInstrument` is a small **fan-out**: + +```text +caller.add(v, attrs) + │ + ▼ + fan-out.measure(v, attrs) + ├─────────────► original.measure(v, attrs) + └─────────────► sidecar .measure(scale(v), attrs) +``` + +The caller holds one handle; each recorded measurement reaches both inner +instruments. Because the fan-out records through the ordinary instrument handles, +no SDK internals are involved. + +### Observable instruments — dual registration + +Observable instruments (`ObservableCounter`, `ObservableUpDownCounter`, +`ObservableGauge`) carry user callbacks that the SDK invokes at collection time, +passing an observer bound to one specific instrument. There is no public +multi-instrument callback, so the layer registers the source instrument's +callbacks **twice** on the inner meter: + +- once on the original instrument, invoking the callbacks with the observer as-is; +- once on the sidecar instrument, invoking the same callbacks through a **scaling + observer** that multiplies every observed value before forwarding it. + +The user's callbacks are shared (behind an `Arc`) between the two registrations. +A consequence is that the callbacks run **twice per collection**. This is an +accepted cost: callbacks are expected to be cheap and idempotent, and a +replay-once cache would introduce its own staleness and correctness hazards for +no meaningful benefit. + +## Value rescaling + +A rescale factor is always a plain multiplicative `f64` — the only transform the +crate needs. Applying it depends on the instrument's value type: + +- **`f64` instruments** multiply directly. +- **`u64`/`i64` instruments** multiply in `f64`, **round** to the nearest + integer, and **saturate** at the type's bounds. Saturation rather than + wrap/overflow keeps a runaway sidecar bounded and obvious instead of silently + corrupt. + +Histograms need one extra step: the sidecar's **bucket boundaries** are scaled by +the same factor as the values, so the buckets stay meaningful. When the source +instrument supplies explicit boundaries the layer scales them; when it relies on +the SDK's default boundaries there is nothing to scale (the defaults are not +visible through the API), so the sidecar simply keeps the defaults. That yields +an obviously wrong bucketing that prompts the operator to configure real +boundaries — acceptable because default boundaries are not expected in real +production use. + +## Configuration model + +Configuration is a map from scope to a set of rescale rules. Each rule maps a +source instrument name to one or more targets; a target carries its **name**, its +**unit** (mandatory), and its **factor**. A single source may therefore feed +several sidecars. The sidecar inherits the source's description but **must** be +given a new unit at configuration time — rescaling almost always changes the unit +(`s` → `ms`), and inheriting the stale one would be misleading. + +Matching a source instrument is by name within its scope, and the same rule +applies to whichever value type the caller builds under that name. + +Scopes are matched **by name only** for now. If several instrumentation scopes +share a name, the rules apply to all of them. The configuration type is shaped so +that stricter matching (e.g. a future `scope_exact(...)` keyed on the full +instrumentation scope — name, version, schema URL, attributes) can be added later +without reworking the model. + +Duplicate target names across the process are **not** the crate's concern: +duplicate instrument registration is always possible in OpenTelemetry, and it is +the user's job to avoid collisions and the SDK's job to cope with them. + +Validation happens at build time, and a configuration that cannot yield a +meaningful sidecar **panics** — for example a factor that is `0.0`, `NaN`, +infinite, or negative, a rule whose source equals its target, or duplicate +targets within a scope. + +## Relationship to the wider system + +The crate depends only on the `opentelemetry` API crate — not on +`opentelemetry_sdk` — so it composes with any conforming provider, including the +SDK provider, the no-op provider, and other wrappers. It is itself a +`MeterProvider`, so wrappers may be stacked. + +The inner provider is taken **by value** but immediately **type-erased** behind a +trait object, so `RescaledMetrics` carries no generic parameter for it and does +not leak the inner provider's concrete type into callers' signatures. + +The public API surface exposes only the `opentelemetry` provider types +(`MeterProvider` and `Meter`, the latter through the `MeterProvider` trait +impl); these are enumerated in the crate's `allowed_external_types` allowlist, as +sibling crates do. diff --git a/crates/opentelemetry_rescaled/examples/print_metrics.rs b/crates/opentelemetry_rescaled/examples/print_metrics.rs new file mode 100644 index 000000000..502b9d7a5 --- /dev/null +++ b/crates/opentelemetry_rescaled/examples/print_metrics.rs @@ -0,0 +1,87 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +//! Records a handful of instruments through a [`RescaledMetrics`] provider and +//! prints the resulting metrics to the terminal, so you can see the rescaled +//! sidecars appear next to their originals — and confirm that an instrument +//! without a configured rule gets no sidecar at all. +//! +//! Run it with: +//! +//! ```text +//! cargo run -p opentelemetry_rescaled --example print_metrics +//! ``` + +use opentelemetry::metrics::MeterProvider as _; +use opentelemetry_rescaled::RescaledMetrics; +use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData}; +use opentelemetry_sdk::metrics::{InMemoryMetricExporter, PeriodicReader, SdkMeterProvider}; + +const SCOPE: &str = "example.http.client"; + +fn main() { + // A normal SDK provider with an in-memory exporter so we can read the + // metrics back and print them. + let exporter = InMemoryMetricExporter::default(); + let reader = PeriodicReader::builder(exporter.clone()).build(); + let sdk = SdkMeterProvider::builder().with_reader(reader).build(); + + // Wrap it: within `SCOPE`, mirror the seconds-based request duration into a + // millisecond sidecar (values multiplied by 1000). Everything else is left + // untouched. + let metrics = RescaledMetrics::builder(sdk.clone()) + .scope(SCOPE, |scope| { + scope.rescale("http.client.request.duration", "http.client.request.duration.millis", "ms", 1000.0); + }) + .build(); + + let meter = metrics.meter(SCOPE); + + // Configured instrument: gains a `.millis` sidecar automatically. + let duration = meter + .f64_histogram("http.client.request.duration") + .with_unit("s") + .with_boundaries(vec![0.1, 0.25, 0.5]) + .build(); + duration.record(0.2, &[]); + duration.record(0.4, &[]); + + // Unconfigured instrument: no sidecar is created for it. + let requests = meter.u64_counter("http.client.requests").with_unit("{request}").build(); + requests.add(2, &[]); + + // Flush and print whatever the exporter received. + sdk.force_flush().expect("force_flush should succeed"); + let resource_metrics = exporter.get_finished_metrics().expect("metrics should be available"); + + println!("Collected metrics:\n"); + for resource in &resource_metrics { + for scope in resource.scope_metrics() { + for metric in scope.metrics() { + println!(" {:<40} [{:>4}] {}", metric.name(), metric.unit(), summarize(metric.data()),); + } + } + } + + println!( + "\nNote how `http.client.request.duration.millis` appears alongside the\n\ + original (values x1000), while `http.client.requests` has no sidecar." + ); +} + +/// Renders a one-line summary of a metric's data points for display. +fn summarize(data: &AggregatedMetrics) -> String { + match data { + AggregatedMetrics::F64(MetricData::Histogram(hist)) => hist + .data_points() + .map(|dp| format!("histogram sum={} count={}", dp.sum(), dp.count())) + .collect::>() + .join(", "), + AggregatedMetrics::U64(MetricData::Sum(sum)) => sum + .data_points() + .map(|dp| format!("sum={}", dp.value())) + .collect::>() + .join(", "), + other => format!("{other:?}"), + } +} diff --git a/crates/opentelemetry_rescaled/logo.png b/crates/opentelemetry_rescaled/logo.png new file mode 100644 index 000000000..ae2ae5ced --- /dev/null +++ b/crates/opentelemetry_rescaled/logo.png @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:dfdc338b2229ac33c210aa8093128f2236c7a88d20eabce213b0f9b090fd24c8 +size 41426 diff --git a/crates/opentelemetry_rescaled/src/config.rs b/crates/opentelemetry_rescaled/src/config.rs new file mode 100644 index 000000000..f04f9bd90 --- /dev/null +++ b/crates/opentelemetry_rescaled/src/config.rs @@ -0,0 +1,161 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +//! Rescale configuration: the rules that map a source instrument, within a scope, to one or more rescaled sidecars. + +use std::borrow::Cow; +use std::collections::{HashMap, HashSet}; + +use foldhash::fast::RandomState; + +/// A single rescaled sidecar of a source instrument. +#[derive(Debug, Clone)] +pub(crate) struct RescaleRule { + /// Name of the sidecar instrument. + pub(crate) target_name: Cow<'static, str>, + /// Unit of the sidecar instrument (mandatory; rescaling changes the unit). + pub(crate) target_unit: Cow<'static, str>, + /// Multiplicative factor applied to each measurement. + pub(crate) factor: f64, +} + +/// The rescale rules for one instrumentation scope, keyed by source instrument name. +#[derive(Debug, Default)] +pub(crate) struct ScopeRules { + map: HashMap, Vec, RandomState>, +} + +impl ScopeRules { + /// Returns the rescale rules for an instrument name, if any are configured. + pub(crate) fn rules_for(&self, name: &str) -> Option<&[RescaleRule]> { + self.map.get(name).map(Vec::as_slice) + } + + /// Returns `true` if no rules are configured for this scope. + pub(crate) fn is_empty(&self) -> bool { + self.map.is_empty() + } +} + +/// Collects the rescale rules for a single instrumentation scope. +/// +/// An instance is passed to the closure given to +/// [`RescaledMetricsBuilder::scope`](crate::RescaledMetricsBuilder::scope); +/// call [`rescale`](Self::rescale) on it to register sidecars. +#[derive(Debug, Default)] +pub struct ScopeConfigurator { + rules: HashMap, Vec, RandomState>, + targets: HashSet, RandomState>, +} + +impl ScopeConfigurator { + /// Registers a rescaled sidecar for the `source` instrument. + /// + /// Whenever the `source` instrument is created in this scope, a second + /// instrument named `target` (carrying `unit`) is created alongside it, + /// recording every measurement multiplied by `factor`. + /// + /// # Panics + /// + /// Panics if the configuration cannot produce a meaningful sidecar: + /// - `factor` is not a finite, strictly positive number; + /// - `source` and `target` are equal; + /// - `target` is already used by another rule in this scope. + pub fn rescale( + &mut self, + source: impl Into>, + target: impl Into>, + unit: impl Into>, + factor: f64, + ) -> &mut Self { + let source = source.into(); + let target = target.into(); + let unit = unit.into(); + + assert!( + factor.is_finite() && factor > 0.0, + "rescale factor must be a finite, strictly positive number, got {factor}" + ); + assert!(source != target, "rescale source and target must differ, both are '{source}'"); + assert!( + self.targets.insert(target.clone()), + "duplicate rescale target '{target}' within a scope" + ); + + self.rules.entry(source).or_default().push(RescaleRule { + target_name: target, + target_unit: unit, + factor, + }); + self + } + + /// Consumes the configurator, yielding the collected per-scope rules. + pub(crate) fn into_rules(self) -> ScopeRules { + ScopeRules { map: self.rules } + } +} + +#[cfg(test)] +#[cfg_attr(coverage_nightly, coverage(off))] +mod tests { + use super::*; + + #[test] + fn collects_multiple_sidecars_per_source() { + let mut sc = ScopeConfigurator::default(); + sc.rescale("dur", "dur.ms", "ms", 1000.0) + .rescale("dur", "dur.us", "us", 1_000_000.0) + .rescale("size", "size.kb", "kB", 0.001); + + let rules = sc.into_rules(); + assert_eq!(rules.rules_for("dur").expect("dur has rules").len(), 2); + assert_eq!(rules.rules_for("size").expect("size has rules").len(), 1); + assert!(rules.rules_for("missing").is_none()); + assert!(!rules.is_empty()); + } + + #[test] + fn empty_configurator_is_empty() { + let rules = ScopeConfigurator::default().into_rules(); + assert!(rules.is_empty()); + } + + #[test] + #[should_panic(expected = "finite, strictly positive")] + fn rejects_zero_factor() { + ScopeConfigurator::default().rescale("a", "b", "u", 0.0); + } + + #[test] + #[should_panic(expected = "finite, strictly positive")] + fn rejects_negative_factor() { + ScopeConfigurator::default().rescale("a", "b", "u", -1.0); + } + + #[test] + #[should_panic(expected = "finite, strictly positive")] + fn rejects_nan_factor() { + ScopeConfigurator::default().rescale("a", "b", "u", f64::NAN); + } + + #[test] + #[should_panic(expected = "finite, strictly positive")] + fn rejects_infinite_factor() { + ScopeConfigurator::default().rescale("a", "b", "u", f64::INFINITY); + } + + #[test] + #[should_panic(expected = "source and target must differ")] + fn rejects_source_equal_target() { + ScopeConfigurator::default().rescale("same", "same", "u", 2.0); + } + + #[test] + #[should_panic(expected = "duplicate rescale target")] + fn rejects_duplicate_target() { + ScopeConfigurator::default() + .rescale("a", "shared", "u", 2.0) + .rescale("b", "shared", "u", 3.0); + } +} diff --git a/crates/opentelemetry_rescaled/src/instruments.rs b/crates/opentelemetry_rescaled/src/instruments.rs new file mode 100644 index 000000000..0c2315359 --- /dev/null +++ b/crates/opentelemetry_rescaled/src/instruments.rs @@ -0,0 +1,304 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +//! The rescaling [`InstrumentProvider`]: builds each configured source instrument together with its rescaled sidecars. + +use std::any::type_name; +use std::borrow::Cow; +use std::fmt; +use std::marker::PhantomData; +use std::sync::Arc; + +use opentelemetry::KeyValue; +use opentelemetry::metrics::{ + AsyncInstrument, AsyncInstrumentBuilder, Callback, Counter, Gauge, Histogram, HistogramBuilder, InstrumentBuilder, InstrumentProvider, + Meter, ObservableCounter, ObservableGauge, ObservableUpDownCounter, SyncInstrument, UpDownCounter, +}; + +use crate::config::ScopeRules; +use crate::rescale::Rescale; + +/// An [`InstrumentProvider`] that mirrors configured instruments into rescaled sidecars, delegating all real work to the wrapped inner [`Meter`]. +pub(crate) struct RescalingInstrumentProvider { + inner: Meter, + rules: Arc, +} + +impl RescalingInstrumentProvider { + pub(crate) fn new(inner: Meter, rules: Arc) -> Self { + Self { inner, rules } + } +} + +impl fmt::Debug for RescalingInstrumentProvider { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct(type_name::()).finish_non_exhaustive() + } +} + +/// A built synchronous instrument that records a single measurement. +/// +/// Unifies the differently named recording methods (`add`/`record`) so the +/// [`FanOut`] can drive any synchronous instrument uniformly. +pub(crate) trait Record { + fn record_value(&self, value: T, attributes: &[KeyValue]); +} + +impl Record for Counter { + fn record_value(&self, value: T, attributes: &[KeyValue]) { + self.add(value, attributes); + } +} + +impl Record for UpDownCounter { + fn record_value(&self, value: T, attributes: &[KeyValue]) { + self.add(value, attributes); + } +} + +impl Record for Gauge { + fn record_value(&self, value: T, attributes: &[KeyValue]) { + self.record(value, attributes); + } +} + +impl Record for Histogram { + fn record_value(&self, value: T, attributes: &[KeyValue]) { + self.record(value, attributes); + } +} + +/// A synchronous instrument backing that fans each measurement out to the original instrument and every rescaled sidecar. +struct FanOut { + original: R, + sidecars: Vec<(R, f64)>, + _value: PhantomData, +} + +impl SyncInstrument for FanOut +where + T: Rescale, + R: Record + Send + Sync, +{ + fn measure(&self, measurement: T, attributes: &[KeyValue]) { + self.original.record_value(measurement, attributes); + for (sidecar, factor) in &self.sidecars { + sidecar.record_value(measurement.rescale(*factor), attributes); + } + } +} + +/// An observer that scales every observation before forwarding it to the sidecar's inner observer. +struct ScalingObserver<'a, M> { + inner: &'a dyn AsyncInstrument, + factor: f64, +} + +impl AsyncInstrument for ScalingObserver<'_, M> +where + M: Rescale, +{ + fn observe(&self, measurement: M, attributes: &[KeyValue]) { + self.inner.observe(measurement.rescale(self.factor), attributes); + } +} + +impl RescalingInstrumentProvider { + /// Builds a synchronous instrument, returning either the plain inner instrument (no rules) or one whose backing is a [`FanOut`] over the original plus every configured sidecar. + fn make_sync( + &self, + name: Cow<'static, str>, + unit: Option>, + build_one: impl Fn(Cow<'static, str>, Option>) -> Inst, + wrap: impl FnOnce(Arc + Send + Sync>) -> Inst, + ) -> Inst + where + T: Rescale, + Inst: Record + Send + Sync + 'static, + { + let Some(rules) = self.rules.rules_for(&name) else { + return build_one(name, unit); + }; + + let sidecars = rules + .iter() + .map(|rule| { + let sidecar = build_one(rule.target_name.clone(), Some(rule.target_unit.clone())); + (sidecar, rule.factor) + }) + .collect(); + + let fan_out: Arc + Send + Sync> = Arc::new(FanOut { + original: build_one(name, unit), + sidecars, + _value: PhantomData, + }); + wrap(fan_out) + } +} + +/// Reconstructs a synchronous instrument builder on the inner meter, applying the inherited description and the given unit, then builds it. +macro_rules! sync_method { + ($method:ident, $inst:ident, $value:ty) => { + fn $method(&self, builder: InstrumentBuilder<'_, $inst<$value>>) -> $inst<$value> { + let description = builder.description; + self.make_sync( + builder.name, + builder.unit, + |name, unit| { + let mut inner = self.inner.$method(name); + if let Some(description) = description.clone() { + inner = inner.with_description(description); + } + if let Some(unit) = unit { + inner = inner.with_unit(unit); + } + inner.build() + }, + $inst::new, + ) + } + }; +} + +/// Reconstructs a histogram builder on the inner meter, scaling each sidecar's explicit bucket boundaries by its factor so the buckets stay meaningful. +macro_rules! histogram_method { + ($method:ident, $value:ty) => { + fn $method(&self, builder: HistogramBuilder<'_, Histogram<$value>>) -> Histogram<$value> { + let description = builder.description; + let source_unit = builder.unit; + let boundaries = builder.boundaries; + let name = builder.name; + + let build_one = |name: Cow<'static, str>, unit: Option>, boundaries: Option>| { + let mut inner = self.inner.$method(name); + if let Some(description) = description.clone() { + inner = inner.with_description(description); + } + if let Some(unit) = unit { + inner = inner.with_unit(unit); + } + if let Some(boundaries) = boundaries { + inner = inner.with_boundaries(boundaries); + } + inner.build() + }; + + let original = build_one(name.clone(), source_unit.clone(), boundaries.clone()); + let Some(rules) = self.rules.rules_for(&name) else { + return original; + }; + + let sidecars = rules + .iter() + .map(|rule| { + let scaled_boundaries = boundaries + .as_ref() + .map(|bounds| bounds.iter().map(|bound| bound * rule.factor).collect()); + let sidecar = build_one(rule.target_name.clone(), Some(rule.target_unit.clone()), scaled_boundaries); + (sidecar, rule.factor) + }) + .collect(); + + Histogram::new(Arc::new(FanOut { + original, + sidecars, + _value: PhantomData, + })) + } + }; +} + +/// Reconstructs an observable instrument on the inner meter, sharing the user callbacks between the original (identity) registration and one per sidecar (through a [`ScalingObserver`]). +/// +/// Because each registration is independent, the callbacks run once per +/// registered instrument per collection. +macro_rules! observable_method { + ($method:ident, $inst:ident, $value:ty) => { + fn $method(&self, builder: AsyncInstrumentBuilder<'_, $inst<$value>, $value>) -> $inst<$value> { + let name = builder.name; + let description = builder.description; + let unit = builder.unit; + let callbacks: Arc<[Callback<$value>]> = builder.callbacks.into(); + + { + let callbacks = Arc::clone(&callbacks); + let mut inner = self.inner.$method(name.clone()); + if let Some(description) = description.clone() { + inner = inner.with_description(description); + } + if let Some(unit) = unit.clone() { + inner = inner.with_unit(unit); + } + let _instrument = inner + .with_callback(move |observer| { + for callback in callbacks.iter() { + callback(observer); + } + }) + .build(); + } + + if let Some(rules) = self.rules.rules_for(&name) { + for rule in rules { + let callbacks = Arc::clone(&callbacks); + let factor = rule.factor; + let mut inner = self.inner.$method(rule.target_name.clone()); + if let Some(description) = description.clone() { + inner = inner.with_description(description); + } + inner = inner.with_unit(rule.target_unit.clone()); + let _instrument = inner + .with_callback(move |observer| { + let scaling = ScalingObserver { inner: observer, factor }; + for callback in callbacks.iter() { + callback(&scaling); + } + }) + .build(); + } + } + + $inst::new() + } + }; +} + +impl InstrumentProvider for RescalingInstrumentProvider { + sync_method!(u64_counter, Counter, u64); + sync_method!(f64_counter, Counter, f64); + sync_method!(i64_up_down_counter, UpDownCounter, i64); + sync_method!(f64_up_down_counter, UpDownCounter, f64); + sync_method!(u64_gauge, Gauge, u64); + sync_method!(i64_gauge, Gauge, i64); + sync_method!(f64_gauge, Gauge, f64); + + histogram_method!(u64_histogram, u64); + histogram_method!(f64_histogram, f64); + + observable_method!(u64_observable_counter, ObservableCounter, u64); + observable_method!(f64_observable_counter, ObservableCounter, f64); + observable_method!(i64_observable_up_down_counter, ObservableUpDownCounter, i64); + observable_method!(f64_observable_up_down_counter, ObservableUpDownCounter, f64); + observable_method!(u64_observable_gauge, ObservableGauge, u64); + observable_method!(i64_observable_gauge, ObservableGauge, i64); + observable_method!(f64_observable_gauge, ObservableGauge, f64); +} + +#[cfg(test)] +#[cfg_attr(coverage_nightly, coverage(off))] +mod tests { + use opentelemetry::metrics::MeterProvider as _; + use opentelemetry::metrics::noop::NoopMeterProvider; + + use super::*; + + #[test] + fn debug_is_non_exhaustive() { + let meter = NoopMeterProvider::new().meter("scope"); + let provider = RescalingInstrumentProvider::new(meter, Arc::new(ScopeRules::default())); + let rendered = format!("{provider:?}"); + assert!(rendered.contains("RescalingInstrumentProvider")); + assert!(rendered.contains("..")); + } +} diff --git a/crates/opentelemetry_rescaled/src/lib.rs b/crates/opentelemetry_rescaled/src/lib.rs new file mode 100644 index 000000000..ad42ee78f --- /dev/null +++ b/crates/opentelemetry_rescaled/src/lib.rs @@ -0,0 +1,54 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +#![cfg_attr(all(coverage_nightly, test), feature(coverage_attribute))] + +//! Wraps an inner OpenTelemetry meter provider to transparently emit *rescaled* side-by-side copies of selected instruments. +//! +//! For a chosen instrument in a chosen instrumentation scope, this layer creates +//! a second instrument whose measurements are the original values multiplied by a +//! fixed factor. For example, a `http.client.request.duration` instrument that +//! records seconds can gain a `http.client.request.duration.millis` sidecar that +//! records the same measurements multiplied by `1000.0`. +//! +//! The rescaling is invisible to instrument users — they interact only with their +//! original instrument — and the inner provider simply sees two independently +//! registered instruments. +//! +//! # Quick start +//! +//! ``` +//! use opentelemetry::metrics::MeterProvider; +//! use opentelemetry_rescaled::RescaledMetrics; +//! +//! // Any `MeterProvider` works as the inner provider. +//! let inner = opentelemetry::metrics::noop::NoopMeterProvider::new(); +//! +//! let outer = RescaledMetrics::builder(inner) +//! .scope("my_scope_name", |scope| { +//! // source name, target name, target unit (mandatory), factor +//! scope.rescale( +//! "http.client.request.duration", +//! "http.client.request.duration.millis", +//! "ms", +//! 1000.0, +//! ); +//! }) +//! .build(); +//! +//! // `outer` is itself a `MeterProvider`. +//! let meter = outer.meter("my_scope_name"); +//! let histogram = meter.f64_histogram("http.client.request.duration").build(); +//! histogram.record(1.5, &[]); // recorded as 1.5 s and, on the sidecar, 1500 ms +//! ``` +//! +//! See [`docs/DESIGN.md`](https://github.com/microsoft/oxidizer/blob/main/crates/opentelemetry_rescaled/docs/DESIGN.md) +//! for the architecture and the resolved design decisions. + +mod config; +mod instruments; +mod provider; +mod rescale; + +pub use config::ScopeConfigurator; +pub use provider::{RescaledMetrics, RescaledMetricsBuilder}; diff --git a/crates/opentelemetry_rescaled/src/provider.rs b/crates/opentelemetry_rescaled/src/provider.rs new file mode 100644 index 000000000..ea5c9cb72 --- /dev/null +++ b/crates/opentelemetry_rescaled/src/provider.rs @@ -0,0 +1,117 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +//! The [`RescaledMetrics`] meter provider and its builder. + +use std::any::type_name; +use std::borrow::Cow; +use std::collections::HashMap; +use std::fmt; +use std::sync::Arc; + +use foldhash::fast::RandomState; +use opentelemetry::InstrumentationScope; +use opentelemetry::metrics::{Meter, MeterProvider}; + +use crate::ScopeConfigurator; +use crate::config::ScopeRules; +use crate::instruments::RescalingInstrumentProvider; + +/// A meter provider that wraps an inner provider and emits rescaled side-by-side copies of selected instruments. +/// +/// Build one with [`RescaledMetrics::builder`]. The result is itself a +/// [`MeterProvider`], so it can be handed wherever the inner provider went — +/// including [`opentelemetry::global::set_meter_provider`]. +/// +/// Scopes that carry no rescale rules are returned untouched, so unconfigured +/// telemetry pays no wrapping cost. +#[derive(Clone)] +pub struct RescaledMetrics { + inner: Arc, + scopes: Arc, Arc, RandomState>>, +} + +impl RescaledMetrics { + /// Starts building a [`RescaledMetrics`] wrapping `inner`. + /// + /// The inner provider is taken by value and type-erased, so `RescaledMetrics` + /// carries no generic parameter for it. + pub fn builder(inner: impl MeterProvider + Send + Sync + 'static) -> RescaledMetricsBuilder { + RescaledMetricsBuilder { + inner: Arc::new(inner), + scopes: HashMap::default(), + } + } +} + +impl fmt::Debug for RescaledMetrics { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct(type_name::()) + .field("scopes", &self.scopes.keys().collect::>()) + .finish_non_exhaustive() + } +} + +impl MeterProvider for RescaledMetrics { + fn meter_with_scope(&self, scope: InstrumentationScope) -> Meter { + let inner_meter = self.inner.meter_with_scope(scope.clone()); + match self.scopes.get(scope.name()) { + Some(rules) => Meter::new(Arc::new(RescalingInstrumentProvider::new(inner_meter, Arc::clone(rules)))), + None => inner_meter, + } + } +} + +/// Builder for [`RescaledMetrics`]. +/// +/// Register rescaling per scope with [`scope`](Self::scope), then finish with +/// [`build`](Self::build). +pub struct RescaledMetricsBuilder { + inner: Arc, + scopes: HashMap, ScopeConfigurator, RandomState>, +} + +impl RescaledMetricsBuilder { + /// Configures rescaling for the instrumentation scope named `name`. + /// + /// The `configure` closure receives a [`ScopeConfigurator`] on which to + /// declare sidecars via [`ScopeConfigurator::rescale`]. Calling `scope` + /// more than once with the same name accumulates rules into that scope. + /// + /// Scopes are matched by name only; if several instrumentation scopes share + /// a name, the rules apply to all of them. + #[must_use] + pub fn scope(mut self, name: impl Into>, configure: impl FnOnce(&mut ScopeConfigurator)) -> Self { + let configurator = self.scopes.entry(name.into()).or_default(); + configure(configurator); + self + } + + /// Builds the [`RescaledMetrics`] provider. + /// + /// Scopes for which no rules were declared are dropped, so they pass through + /// to the inner provider with no wrapping. + #[must_use] + pub fn build(self) -> RescaledMetrics { + let scopes = self + .scopes + .into_iter() + .map(|(name, configurator)| (name, configurator.into_rules())) + .filter(|(_, rules)| !rules.is_empty()) + .map(|(name, rules)| (name, Arc::new(rules))) + .collect(); + + RescaledMetrics { + inner: self.inner, + scopes: Arc::new(scopes), + } + } +} + +impl fmt::Debug for RescaledMetricsBuilder { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct(type_name::()) + .field("scopes", &self.scopes.keys().collect::>()) + .finish_non_exhaustive() + } +} diff --git a/crates/opentelemetry_rescaled/src/rescale.rs b/crates/opentelemetry_rescaled/src/rescale.rs new file mode 100644 index 000000000..0d5f052ff --- /dev/null +++ b/crates/opentelemetry_rescaled/src/rescale.rs @@ -0,0 +1,85 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +//! Value rescaling: applying a multiplicative factor to a measurement, with +//! type-appropriate rounding and saturation for integer instruments. + +/// A measurement value that can be multiplied by an `f64` rescale factor. +/// +/// Floating-point values multiply directly. Integer values multiply in `f64`, +/// round to the nearest integer, and saturate at the type's bounds rather than +/// wrapping, so a runaway sidecar stays bounded and obvious. +pub(crate) trait Rescale: Copy + Send + Sync + 'static { + /// Returns `self` multiplied by `factor`, rounded and saturated as needed. + fn rescale(self, factor: f64) -> Self; +} + +impl Rescale for f64 { + fn rescale(self, factor: f64) -> Self { + self * factor + } +} + +impl Rescale for u64 { + #[expect( + clippy::cast_precision_loss, + clippy::cast_possible_truncation, + clippy::cast_sign_loss, + reason = "saturating round is the intended behavior; float->int `as` casts saturate at the type bounds and map NaN to 0" + )] + fn rescale(self, factor: f64) -> Self { + (self as f64 * factor).round() as Self + } +} + +impl Rescale for i64 { + #[expect( + clippy::cast_precision_loss, + clippy::cast_possible_truncation, + reason = "saturating round is the intended behavior; float->int `as` casts saturate at the type bounds and map NaN to 0" + )] + fn rescale(self, factor: f64) -> Self { + (self as f64 * factor).round() as Self + } +} + +#[cfg(test)] +#[cfg_attr(coverage_nightly, coverage(off))] +mod tests { + use super::*; + + #[test] + fn f64_multiplies_directly() { + assert!((1.5_f64.rescale(1000.0) - 1500.0).abs() < f64::EPSILON); + assert!((2.0_f64.rescale(0.001) - 0.002).abs() < f64::EPSILON); + } + + #[test] + fn u64_rounds_to_nearest() { + assert_eq!(1_u64.rescale(1000.0), 1000); + // 3 * 0.5 = 1.5 -> rounds to 2 (round half away from zero). + assert_eq!(3_u64.rescale(0.5), 2); + // 2 * 0.5 = 1.0 -> exactly 1. + assert_eq!(2_u64.rescale(0.5), 1); + // 1 * 0.4 = 0.4 -> rounds to 0. + assert_eq!(1_u64.rescale(0.4), 0); + } + + #[test] + fn u64_saturates_on_overflow() { + assert_eq!(u64::MAX.rescale(1000.0), u64::MAX); + } + + #[test] + fn i64_rounds_and_handles_sign() { + assert_eq!((-2_i64).rescale(1000.0), -2000); + assert_eq!(3_i64.rescale(0.5), 2); + assert_eq!((-3_i64).rescale(0.5), -2); + } + + #[test] + fn i64_saturates_at_both_bounds() { + assert_eq!(i64::MAX.rescale(1000.0), i64::MAX); + assert_eq!(i64::MIN.rescale(1000.0), i64::MIN); + } +} diff --git a/crates/opentelemetry_rescaled/tests/integration.rs b/crates/opentelemetry_rescaled/tests/integration.rs new file mode 100644 index 000000000..74fe3b7a5 --- /dev/null +++ b/crates/opentelemetry_rescaled/tests/integration.rs @@ -0,0 +1,734 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +//! End-to-end tests driving the rescaling provider through a real +//! `SdkMeterProvider` with an in-memory exporter, verifying that every +//! instrument kind produces a correctly rescaled sidecar alongside its original. + +#![allow( + clippy::unwrap_used, + clippy::panic, + clippy::float_cmp, + reason = "unwrap, panic, and exact float comparisons keep tests concise and readable" +)] + +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; + +use opentelemetry::metrics::MeterProvider as _; +use opentelemetry::{InstrumentationScope, KeyValue}; +use opentelemetry_rescaled::{RescaledMetrics, RescaledMetricsBuilder}; +use opentelemetry_sdk::metrics::data::{AggregatedMetrics, Metric, MetricData, ResourceMetrics, ScopeMetrics, SumDataPoint}; +use opentelemetry_sdk::metrics::{InMemoryMetricExporter, PeriodicReader, SdkMeterProvider}; + +const SCOPE: &str = "test_scope"; + +struct Harness { + outer: RescaledMetrics, + sdk: SdkMeterProvider, + exporter: InMemoryMetricExporter, +} + +impl Harness { + fn new(configure: impl FnOnce(RescaledMetricsBuilder) -> RescaledMetrics) -> Self { + let exporter = InMemoryMetricExporter::default(); + let reader = PeriodicReader::builder(exporter.clone()).build(); + let sdk = SdkMeterProvider::builder().with_reader(reader).build(); + let outer = configure(RescaledMetrics::builder(sdk.clone())); + Self { outer, sdk, exporter } + } + + /// Configures a single scope named [`SCOPE`]. + fn with_scope(configure: impl FnOnce(&mut opentelemetry_rescaled::ScopeConfigurator)) -> Self { + Self::new(|builder| builder.scope(SCOPE, configure).build()) + } + + fn collect(&self) -> Vec { + self.sdk.force_flush().unwrap(); + self.exporter.get_finished_metrics().unwrap() + } +} + +fn find<'a>(metrics: &'a [ResourceMetrics], name: &str) -> Option<&'a Metric> { + metrics + .iter() + .flat_map(ResourceMetrics::scope_metrics) + .flat_map(ScopeMetrics::metrics) + .find(|metric| metric.name() == name) +} + +fn metric<'a>(metrics: &'a [ResourceMetrics], name: &str) -> &'a Metric { + find(metrics, name).unwrap_or_else(|| panic!("metric '{name}' not found")) +} + +fn sum_u64(metric: &Metric) -> u64 { + match metric.data() { + AggregatedMetrics::U64(MetricData::Sum(sum)) => sum.data_points().map(SumDataPoint::value).sum(), + other => panic!("expected u64 sum, got {other:?}"), + } +} + +fn sum_f64(metric: &Metric) -> f64 { + match metric.data() { + AggregatedMetrics::F64(MetricData::Sum(sum)) => sum.data_points().map(SumDataPoint::value).sum(), + other => panic!("expected f64 sum, got {other:?}"), + } +} + +fn sum_i64(metric: &Metric) -> i64 { + match metric.data() { + AggregatedMetrics::I64(MetricData::Sum(sum)) => sum.data_points().map(SumDataPoint::value).sum(), + other => panic!("expected i64 sum, got {other:?}"), + } +} + +fn gauge_u64(metric: &Metric) -> u64 { + match metric.data() { + AggregatedMetrics::U64(MetricData::Gauge(gauge)) => gauge.data_points().next().unwrap().value(), + other => panic!("expected u64 gauge, got {other:?}"), + } +} + +fn gauge_i64(metric: &Metric) -> i64 { + match metric.data() { + AggregatedMetrics::I64(MetricData::Gauge(gauge)) => gauge.data_points().next().unwrap().value(), + other => panic!("expected i64 gauge, got {other:?}"), + } +} + +fn gauge_f64(metric: &Metric) -> f64 { + match metric.data() { + AggregatedMetrics::F64(MetricData::Gauge(gauge)) => gauge.data_points().next().unwrap().value(), + other => panic!("expected f64 gauge, got {other:?}"), + } +} + +fn histogram_f64_sum(metric: &Metric) -> f64 { + match metric.data() { + AggregatedMetrics::F64(MetricData::Histogram(hist)) => hist.data_points().next().unwrap().sum(), + other => panic!("expected f64 histogram, got {other:?}"), + } +} + +fn histogram_u64_sum(metric: &Metric) -> u64 { + match metric.data() { + AggregatedMetrics::U64(MetricData::Histogram(hist)) => hist.data_points().next().unwrap().sum(), + other => panic!("expected u64 histogram, got {other:?}"), + } +} + +fn histogram_f64_bounds(metric: &Metric) -> Vec { + match metric.data() { + AggregatedMetrics::F64(MetricData::Histogram(hist)) => hist.data_points().next().unwrap().bounds().collect(), + other => panic!("expected f64 histogram, got {other:?}"), + } +} + +// ------------------------------------------------------------------------- +// Synchronous instruments +// ------------------------------------------------------------------------- + +#[test] +fn u64_counter_fans_out_with_metadata() { + let harness = Harness::with_scope(|scope| { + scope.rescale("bytes", "kilobytes", "kB", 0.001); + }); + + let counter = harness + .outer + .meter(SCOPE) + .u64_counter("bytes") + .with_description("bytes transferred") + .with_unit("By") + .build(); + counter.add(4000, &[]); + + let metrics = harness.collect(); + + let original = metric(&metrics, "bytes"); + assert_eq!(sum_u64(original), 4000); + assert_eq!(original.unit(), "By"); + assert_eq!(original.description(), "bytes transferred"); + + let sidecar = metric(&metrics, "kilobytes"); + assert_eq!(sum_u64(sidecar), 4); // 4000 * 0.001 + assert_eq!(sidecar.unit(), "kB", "sidecar carries the configured unit"); + assert_eq!( + sidecar.description(), + "bytes transferred", + "sidecar inherits the source description" + ); +} + +#[test] +fn f64_counter_fans_out() { + let harness = Harness::with_scope(|scope| { + scope.rescale("seconds", "millis", "ms", 1000.0); + }); + + let counter = harness.outer.meter(SCOPE).f64_counter("seconds").build(); + counter.add(1.5, &[]); + counter.add(0.5, &[]); + + let metrics = harness.collect(); + assert_eq!(sum_f64(metric(&metrics, "seconds")), 2.0); + assert_eq!(sum_f64(metric(&metrics, "millis")), 2000.0); +} + +#[test] +fn i64_up_down_counter_fans_out_with_negatives() { + let harness = Harness::with_scope(|scope| { + scope.rescale("delta", "delta.k", "k", 1000.0); + }); + + let updown = harness.outer.meter(SCOPE).i64_up_down_counter("delta").build(); + updown.add(5, &[]); + updown.add(-2, &[]); + + let metrics = harness.collect(); + assert_eq!(sum_i64(metric(&metrics, "delta")), 3); + assert_eq!(sum_i64(metric(&metrics, "delta.k")), 3000); +} + +#[test] +fn f64_up_down_counter_fans_out() { + let harness = Harness::with_scope(|scope| { + scope.rescale("balance", "balance.milli", "m", 1000.0); + }); + + let updown = harness.outer.meter(SCOPE).f64_up_down_counter("balance").build(); + updown.add(2.5, &[]); + updown.add(-1.0, &[]); + + let metrics = harness.collect(); + assert_eq!(sum_f64(metric(&metrics, "balance")), 1.5); + assert_eq!(sum_f64(metric(&metrics, "balance.milli")), 1500.0); +} + +#[test] +fn u64_gauge_fans_out() { + let harness = Harness::with_scope(|scope| { + scope.rescale("temp", "temp.milli", "m", 1000.0); + }); + + let gauge = harness.outer.meter(SCOPE).u64_gauge("temp").build(); + gauge.record(42, &[]); + + let metrics = harness.collect(); + assert_eq!(gauge_u64(metric(&metrics, "temp")), 42); + assert_eq!(gauge_u64(metric(&metrics, "temp.milli")), 42000); +} + +#[test] +fn i64_gauge_fans_out() { + let harness = Harness::with_scope(|scope| { + scope.rescale("level", "level.k", "k", 1000.0); + }); + + let gauge = harness.outer.meter(SCOPE).i64_gauge("level").build(); + gauge.record(-7, &[]); + + let metrics = harness.collect(); + assert_eq!(gauge_i64(metric(&metrics, "level")), -7); + assert_eq!(gauge_i64(metric(&metrics, "level.k")), -7000); +} + +#[test] +fn f64_gauge_fans_out() { + let harness = Harness::with_scope(|scope| { + scope.rescale("ratio", "ratio.pct", "%", 100.0); + }); + + let gauge = harness.outer.meter(SCOPE).f64_gauge("ratio").build(); + gauge.record(0.25, &[]); + + let metrics = harness.collect(); + assert_eq!(gauge_f64(metric(&metrics, "ratio")), 0.25); + assert_eq!(gauge_f64(metric(&metrics, "ratio.pct")), 25.0); +} + +// ------------------------------------------------------------------------- +// Histograms — including bucket-boundary scaling +// ------------------------------------------------------------------------- + +#[test] +fn f64_histogram_scales_explicit_boundaries() { + let harness = Harness::with_scope(|scope| { + scope.rescale("dur", "dur.ms", "ms", 1000.0); + }); + + let histogram = harness + .outer + .meter(SCOPE) + .f64_histogram("dur") + .with_description("request duration") + .with_boundaries(vec![1.0, 2.0, 3.0]) + .build(); + histogram.record(1.5, &[]); + + let metrics = harness.collect(); + + assert_eq!(histogram_f64_sum(metric(&metrics, "dur")), 1.5); + assert_eq!(metric(&metrics, "dur").description(), "request duration"); + assert_eq!(histogram_f64_bounds(metric(&metrics, "dur")), vec![1.0, 2.0, 3.0]); + + let sidecar = metric(&metrics, "dur.ms"); + assert_eq!(histogram_f64_sum(sidecar), 1500.0); + assert_eq!(sidecar.unit(), "ms"); + assert_eq!(sidecar.description(), "request duration", "sidecar inherits the source description"); + assert_eq!( + histogram_f64_bounds(sidecar), + vec![1000.0, 2000.0, 3000.0], + "sidecar boundaries are scaled by the same factor" + ); +} + +#[test] +fn f64_histogram_without_boundaries_keeps_defaults() { + let harness = Harness::with_scope(|scope| { + scope.rescale("dur", "dur.ms", "ms", 1000.0); + }); + + let histogram = harness.outer.meter(SCOPE).f64_histogram("dur").build(); + histogram.record(1.5, &[]); + + let metrics = harness.collect(); + assert_eq!(histogram_f64_sum(metric(&metrics, "dur")), 1.5); + assert_eq!(histogram_f64_sum(metric(&metrics, "dur.ms")), 1500.0); + // Both keep the SDK default boundaries (nothing to scale). + assert_eq!( + histogram_f64_bounds(metric(&metrics, "dur")), + histogram_f64_bounds(metric(&metrics, "dur.ms")), + ); +} + +#[test] +fn u64_histogram_fans_out() { + let harness = Harness::with_scope(|scope| { + scope.rescale("size", "size.kb", "kB", 0.001); + }); + + let histogram = harness + .outer + .meter(SCOPE) + .u64_histogram("size") + .with_boundaries(vec![1000.0, 2000.0]) + .build(); + histogram.record(3000, &[]); + + let metrics = harness.collect(); + assert_eq!(histogram_u64_sum(metric(&metrics, "size")), 3000); + assert_eq!(histogram_u64_sum(metric(&metrics, "size.kb")), 3); // 3000 * 0.001 +} + +// ------------------------------------------------------------------------- +// Observable instruments — dual registration +// ------------------------------------------------------------------------- + +#[test] +fn u64_observable_counter_fans_out() { + let harness = Harness::with_scope(|scope| { + scope.rescale("obs", "obs.k", "k", 1000.0); + }); + + let _instrument = harness + .outer + .meter(SCOPE) + .u64_observable_counter("obs") + .with_callback(|observer| observer.observe(100, &[])) + .build(); + + let metrics = harness.collect(); + assert_eq!(sum_u64(metric(&metrics, "obs")), 100); + assert_eq!(sum_u64(metric(&metrics, "obs.k")), 100_000); +} + +#[test] +fn f64_observable_gauge_fans_out() { + let harness = Harness::with_scope(|scope| { + scope.rescale("g", "g.milli", "m", 1000.0); + }); + + let _instrument = harness + .outer + .meter(SCOPE) + .f64_observable_gauge("g") + .with_callback(|observer| observer.observe(1.5, &[])) + .build(); + + let metrics = harness.collect(); + assert_eq!(gauge_f64(metric(&metrics, "g")), 1.5); + assert_eq!(gauge_f64(metric(&metrics, "g.milli")), 1500.0); +} + +#[test] +fn i64_observable_gauge_fans_out() { + let harness = Harness::with_scope(|scope| { + scope.rescale("g", "g.k", "k", 1000.0); + }); + + let _instrument = harness + .outer + .meter(SCOPE) + .i64_observable_gauge("g") + .with_callback(|observer| observer.observe(-3, &[])) + .build(); + + let metrics = harness.collect(); + assert_eq!(gauge_i64(metric(&metrics, "g")), -3); + assert_eq!(gauge_i64(metric(&metrics, "g.k")), -3000); +} + +#[test] +fn u64_observable_gauge_fans_out() { + let harness = Harness::with_scope(|scope| { + scope.rescale("g", "g.k", "k", 1000.0); + }); + + let _instrument = harness + .outer + .meter(SCOPE) + .u64_observable_gauge("g") + .with_callback(|observer| observer.observe(7, &[])) + .build(); + + let metrics = harness.collect(); + assert_eq!(gauge_u64(metric(&metrics, "g")), 7); + assert_eq!(gauge_u64(metric(&metrics, "g.k")), 7000); +} + +#[test] +fn i64_observable_up_down_counter_fans_out() { + let harness = Harness::with_scope(|scope| { + scope.rescale("obs", "obs.k", "k", 1000.0); + }); + + let _instrument = harness + .outer + .meter(SCOPE) + .i64_observable_up_down_counter("obs") + .with_callback(|observer| observer.observe(-4, &[])) + .build(); + + let metrics = harness.collect(); + assert_eq!(sum_i64(metric(&metrics, "obs")), -4); + assert_eq!(sum_i64(metric(&metrics, "obs.k")), -4000); +} + +#[test] +fn f64_observable_counter_fans_out() { + let harness = Harness::with_scope(|scope| { + scope.rescale("obs", "obs.milli", "m", 1000.0); + }); + + let _instrument = harness + .outer + .meter(SCOPE) + .f64_observable_counter("obs") + .with_callback(|observer| observer.observe(2.0, &[])) + .build(); + + let metrics = harness.collect(); + assert_eq!(sum_f64(metric(&metrics, "obs")), 2.0); + assert_eq!(sum_f64(metric(&metrics, "obs.milli")), 2000.0); +} + +#[test] +fn f64_observable_up_down_counter_fans_out() { + let harness = Harness::with_scope(|scope| { + scope.rescale("obs", "obs.milli", "m", 1000.0); + }); + + let _instrument = harness + .outer + .meter(SCOPE) + .f64_observable_up_down_counter("obs") + .with_callback(|observer| observer.observe(1.25, &[])) + .build(); + + let metrics = harness.collect(); + assert_eq!(sum_f64(metric(&metrics, "obs")), 1.25); + assert_eq!(sum_f64(metric(&metrics, "obs.milli")), 1250.0); +} + +#[test] +fn observable_callback_runs_once_per_registered_instrument() { + let calls = Arc::new(AtomicUsize::new(0)); + let harness = Harness::with_scope(|scope| { + scope.rescale("obs", "obs.k", "k", 1000.0); + }); + + let calls_in_cb = Arc::clone(&calls); + let _instrument = harness + .outer + .meter(SCOPE) + .u64_observable_counter("obs") + .with_callback(move |observer| { + calls_in_cb.fetch_add(1, Ordering::Relaxed); + observer.observe(1, &[]); + }) + .build(); + + let _ = harness.collect(); + + // Registered twice (original + sidecar), so the callback runs twice per collection. + assert_eq!(calls.load(Ordering::Relaxed), 2); +} + +// ------------------------------------------------------------------------- +// Multiple sidecars, attributes, rounding +// ------------------------------------------------------------------------- + +#[test] +fn single_source_feeds_multiple_sidecars() { + let harness = Harness::with_scope(|scope| { + scope + .rescale("dur", "dur.ms", "ms", 1000.0) + .rescale("dur", "dur.us", "us", 1_000_000.0); + }); + + let counter = harness.outer.meter(SCOPE).f64_counter("dur").build(); + counter.add(2.0, &[]); + + let metrics = harness.collect(); + assert_eq!(sum_f64(metric(&metrics, "dur")), 2.0); + assert_eq!(sum_f64(metric(&metrics, "dur.ms")), 2000.0); + assert_eq!(sum_f64(metric(&metrics, "dur.us")), 2_000_000.0); +} + +#[test] +fn attributes_are_forwarded_to_sidecar() { + let harness = Harness::with_scope(|scope| { + scope.rescale("req", "req.k", "k", 1000.0); + }); + + let counter = harness.outer.meter(SCOPE).u64_counter("req").build(); + counter.add(1, &[KeyValue::new("route", "/health")]); + + let metrics = harness.collect(); + let sidecar = metric(&metrics, "req.k"); + let has_attr = match sidecar.data() { + AggregatedMetrics::U64(MetricData::Sum(sum)) => sum.data_points().any(|dp| { + dp.attributes() + .any(|kv| kv.key.as_str() == "route" && kv.value.as_str() == "/health") + }), + other => panic!("expected u64 sum, got {other:?}"), + }; + assert!(has_attr, "sidecar data point retains the recorded attributes"); +} + +#[test] +fn integer_rescale_rounds_and_saturates_end_to_end() { + let harness = Harness::with_scope(|scope| { + scope.rescale("halve", "halved", "x", 0.5).rescale("huge", "huge.big", "x", 1e30); + }); + + let meter = harness.outer.meter(SCOPE); + let halve = meter.u64_counter("halve").build(); + halve.add(3, &[]); // 3 * 0.5 = 1.5 -> rounds to 2 + + let huge = meter.u64_counter("huge").build(); + huge.add(u64::MAX, &[]); // saturates + + let metrics = harness.collect(); + assert_eq!(sum_u64(metric(&metrics, "halved")), 2); + assert_eq!(sum_u64(metric(&metrics, "huge.big")), u64::MAX); +} + +// ------------------------------------------------------------------------- +// Pass-through behavior +// ------------------------------------------------------------------------- + +#[test] +fn unconfigured_instrument_in_configured_scope_has_no_sidecar() { + let harness = Harness::with_scope(|scope| { + scope.rescale("configured", "configured.k", "k", 1000.0); + }); + + let counter = harness.outer.meter(SCOPE).u64_counter("other").build(); + counter.add(5, &[]); + + let metrics = harness.collect(); + assert_eq!(sum_u64(metric(&metrics, "other")), 5); + assert!( + find(&metrics, "configured.k").is_none(), + "no sidecar for an instrument that was never created" + ); + assert!(find(&metrics, "other.k").is_none()); +} + +#[test] +fn unconfigured_scope_passes_through() { + let harness = Harness::with_scope(|scope| { + scope.rescale("x", "x.k", "k", 1000.0); + }); + + let counter = harness.outer.meter("some_other_scope").u64_counter("x").build(); + counter.add(9, &[]); + + let metrics = harness.collect(); + assert_eq!(sum_u64(metric(&metrics, "x")), 9); + assert!(find(&metrics, "x.k").is_none(), "rules do not apply outside their configured scope"); +} + +#[test] +fn name_only_matching_applies_to_all_scopes_sharing_a_name() { + let harness = Harness::with_scope(|scope| { + scope.rescale("hits", "hits.k", "k", 1000.0); + }); + + // Two meters whose scopes share the name but differ in version. + let scope_v1 = InstrumentationScope::builder(SCOPE).with_version("1.0").build(); + let scope_v2 = InstrumentationScope::builder(SCOPE).with_version("2.0").build(); + + harness.outer.meter_with_scope(scope_v1).u64_counter("hits").build().add(1, &[]); + harness.outer.meter_with_scope(scope_v2).u64_counter("hits").build().add(2, &[]); + + let metrics = harness.collect(); + // Both scopes emit a sidecar; summing across scopes gives (1 + 2) * 1000. + let sidecar_total: u64 = metrics + .iter() + .flat_map(ResourceMetrics::scope_metrics) + .flat_map(ScopeMetrics::metrics) + .filter(|m| m.name() == "hits.k") + .map(sum_u64) + .sum(); + assert_eq!(sidecar_total, 3000); +} + +#[test] +fn multiple_scope_calls_accumulate_rules() { + let harness = Harness::new(|builder| { + builder + .scope(SCOPE, |scope| { + scope.rescale("a", "a.k", "k", 1000.0); + }) + .scope(SCOPE, |scope| { + scope.rescale("b", "b.k", "k", 1000.0); + }) + .build() + }); + + let meter = harness.outer.meter(SCOPE); + meter.u64_counter("a").build().add(1, &[]); + meter.u64_counter("b").build().add(2, &[]); + + let metrics = harness.collect(); + assert_eq!(sum_u64(metric(&metrics, "a.k")), 1000); + assert_eq!(sum_u64(metric(&metrics, "b.k")), 2000); +} + +#[test] +fn scope_without_rules_is_dropped_and_passes_through() { + let harness = Harness::new(|builder| { + builder + .scope("empty_scope", |_scope| { + // No rescale calls: this scope must not be wrapped. + }) + .build() + }); + + let counter = harness.outer.meter("empty_scope").u64_counter("plain").build(); + counter.add(3, &[]); + + let metrics = harness.collect(); + assert_eq!(sum_u64(metric(&metrics, "plain")), 3); +} + +// ------------------------------------------------------------------------- +// No-rule instruments inside a configured scope (early-return branches) +// ------------------------------------------------------------------------- + +#[test] +fn unconfigured_histogram_in_configured_scope_has_no_sidecar() { + let harness = Harness::with_scope(|scope| { + scope.rescale("configured", "configured.ms", "ms", 1000.0); + }); + + let histogram = harness + .outer + .meter(SCOPE) + .f64_histogram("other") + .with_boundaries(vec![1.0, 2.0]) + .build(); + histogram.record(1.5, &[]); + + let metrics = harness.collect(); + assert_eq!(histogram_f64_sum(metric(&metrics, "other")), 1.5); + assert!( + find(&metrics, "other.ms").is_none(), + "an unconfigured histogram gets no sidecar even inside a configured scope" + ); +} + +#[test] +fn unconfigured_observable_in_configured_scope_has_no_sidecar() { + let harness = Harness::with_scope(|scope| { + scope.rescale("configured", "configured.k", "k", 1000.0); + }); + + let _instrument = harness + .outer + .meter(SCOPE) + .u64_observable_counter("other") + .with_callback(|observer| observer.observe(5, &[])) + .build(); + + let metrics = harness.collect(); + assert_eq!(sum_u64(metric(&metrics, "other")), 5); + assert!( + find(&metrics, "other.k").is_none(), + "an unconfigured observable gets no sidecar even inside a configured scope" + ); +} + +// ------------------------------------------------------------------------- +// Metadata inheritance for observables +// ------------------------------------------------------------------------- + +#[test] +fn observable_sidecar_inherits_description_and_carries_unit() { + let harness = Harness::with_scope(|scope| { + scope.rescale("obs", "obs.k", "k", 1000.0); + }); + + let _instrument = harness + .outer + .meter(SCOPE) + .u64_observable_counter("obs") + .with_description("observed count") + .with_unit("things") + .with_callback(|observer| observer.observe(3, &[])) + .build(); + + let metrics = harness.collect(); + + let original = metric(&metrics, "obs"); + assert_eq!(sum_u64(original), 3); + assert_eq!(original.description(), "observed count"); + assert_eq!(original.unit(), "things"); + + let sidecar = metric(&metrics, "obs.k"); + assert_eq!(sum_u64(sidecar), 3000); + assert_eq!(sidecar.description(), "observed count", "sidecar inherits the source description"); + assert_eq!(sidecar.unit(), "k", "sidecar carries its configured unit"); +} + +// ------------------------------------------------------------------------- +// Debug formatting +// ------------------------------------------------------------------------- + +#[test] +fn debug_impls_render_scope_names() { + let builder = RescaledMetrics::builder(SdkMeterProvider::builder().build()).scope(SCOPE, |scope| { + scope.rescale("a", "a.k", "k", 1000.0); + }); + let builder_debug = format!("{builder:?}"); + assert!(builder_debug.contains("RescaledMetricsBuilder")); + assert!(builder_debug.contains(SCOPE)); + + let provider = builder.build(); + let provider_debug = format!("{provider:?}"); + assert!(provider_debug.contains("RescaledMetrics")); + assert!(provider_debug.contains(SCOPE)); +}