diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS index b74a4dc341..da71790663 100644 --- a/.github/CODEOWNERS +++ b/.github/CODEOWNERS @@ -55,6 +55,7 @@ libdd-http-client @DataDog/apm-common-components-core libdd-agent-client @DataDog/apm-common-components-core libdd-library-config*/ @DataDog/apm-sdk-capabilities-rust libdd-log*/ @DataDog/apm-common-components-core +libdd-metrics-v3/ @DataDog/libdatadog @DataDog/agent-metric-pipelines libdd-otel-thread-ctx/ @DataDog/apm-common-components-core libdd-otel-thread-ctx-ffi/ @DataDog/apm-common-components-core libdd-profiling*/ @DataDog/libdatadog-profiling diff --git a/.github/workflows/diff-agent-payload-proto.sh b/.github/workflows/diff-agent-payload-proto.sh new file mode 100755 index 0000000000..95aa237b53 --- /dev/null +++ b/.github/workflows/diff-agent-payload-proto.sh @@ -0,0 +1,43 @@ +#!/usr/bin/env bash + +# Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/ +# SPDX-License-Identifier: Apache-2.0 + +set -eux + +PROTO_FILE="" +AGENT_PAYLOAD_COMMIT="" + +while [[ $# -gt 0 ]]; do + case $1 in + --file) + PROTO_FILE=$2 + shift && shift # past argument and value + ;; + --commit) + AGENT_PAYLOAD_COMMIT=$2 + shift && shift # past argument and value + ;; + *) + echo "Unknown option $1" + exit 1 + ;; + esac +done + +if [ -z "$PROTO_FILE" ]; then + echo "Missing --file argument" + exit 1 +fi + +if [ -z "$AGENT_PAYLOAD_COMMIT" ]; then + echo "Missing --commit argument" + exit 1 +fi + +# Vendored files must stay byte-for-byte identical to their pinned commit in agent-payload from +# `syntax = ...;` onward, so unlike diff-proto-files.sh (which fixes up import/package names for +# datadog-agent's proto layout) this only strips the local "Vendored from:" preamble comment +# before diffing, with no other rewriting. +curl -sf "https://raw.githubusercontent.com/DataDog/agent-payload/$AGENT_PAYLOAD_COMMIT/proto/metrics/$PROTO_FILE" | + diff -u <(sed -n '/^syntax = /,$p' "$PROTO_FILE") - diff --git a/.github/workflows/no-std-metrics-v3.yml b/.github/workflows/no-std-metrics-v3.yml new file mode 100644 index 0000000000..3e5dc622ff --- /dev/null +++ b/.github/workflows/no-std-metrics-v3.yml @@ -0,0 +1,40 @@ +name: 'No-std check (libdd-metrics-v3)' +on: + pull_request: + push: + branches: + - main + - mq-working-branch-* +env: + CARGO_TERM_COLOR: always + CARGO_INCREMENTAL: 0 +jobs: + no-std-check: + # `#![no_std]` alone only stops this crate's own code from referencing `std`; a dependency + # that quietly requires std would still link fine on a normal host target. Building for a + # target with no OS (so no std is even available to link against) is what actually proves the + # whole dependency graph is no_std-compatible. + name: "libdd-metrics-v3 builds for a target with no OS and no std" + runs-on: ubuntu-latest + steps: + - name: Checkout sources + uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # 4.2.2 + - name: Read Rust version from rust-toolchain.toml + id: rust-version + run: echo "version=$(grep -Po '^channel = "\K[^"]+' rust-toolchain.toml)" >> $GITHUB_OUTPUT + - name: Install ${{ steps.rust-version.outputs.version }} toolchain + run: | + rustup set profile minimal + rustup install ${{ steps.rust-version.outputs.version }} + rustup default ${{ steps.rust-version.outputs.version }} + rustup target add x86_64-unknown-none + rustup component add clippy + - name: Cache [rust] + uses: Swatinem/rust-cache@f13886b937689c021905a6b90929199931d60db1 # 2.8.1 + with: + cache-targets: true # cache build artifacts + cache-bin: true # cache the ~/.cargo/bin directory + - name: Build for x86_64-unknown-none + run: cargo build -p libdd-metrics-v3 --target x86_64-unknown-none + - name: Clippy for x86_64-unknown-none + run: cargo clippy -p libdd-metrics-v3 --target x86_64-unknown-none -- -D warnings diff --git a/.github/workflows/verify-metrics-v3-proto.yml b/.github/workflows/verify-metrics-v3-proto.yml new file mode 100644 index 0000000000..3448001379 --- /dev/null +++ b/.github/workflows/verify-metrics-v3-proto.yml @@ -0,0 +1,37 @@ +name: 'Verify metrics-v3 proto' +on: + pull_request: + types: [ opened, synchronize, reopened ] +env: + # Pinned commit of https://github.com/DataDog/agent-payload/blob/master/proto/metrics/intake_v3.proto + AGENT_PAYLOAD_COMMIT: "512d523e386f75077efe9e8d878284104d491f46" + CARGO_TERM_COLOR: always + CARGO_INCREMENTAL: 0 +jobs: + verify-proto-files: + name: "Verify libdd-metrics-v3's vendored intake_v3.proto and generated bindings are in sync" + runs-on: ubuntu-latest + permissions: + pull-requests: write + steps: + - name: Checkout sources + uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # 4.2.2 + - name: diff intake_v3.proto against agent-payload + working-directory: libdd-metrics-v3/proto + run: | + ../../.github/workflows/diff-agent-payload-proto.sh --file intake_v3.proto --commit "$AGENT_PAYLOAD_COMMIT" + - name: Read Rust version from rust-toolchain.toml + id: rust-version + run: echo "version=$(grep -Po '^channel = "\K[^"]+' rust-toolchain.toml)" >> $GITHUB_OUTPUT + - name: Install ${{ steps.rust-version.outputs.version }} toolchain + run: rustup set profile minimal && rustup install ${{ steps.rust-version.outputs.version }} && rustup default ${{ steps.rust-version.outputs.version }} + - name: Cache [rust] + uses: Swatinem/rust-cache@f13886b937689c021905a6b90929199931d60db1 # 2.8.1 + with: + cache-targets: true # cache build artifacts + cache-bin: true # cache the ~/.cargo/bin directory + - name: diff tests/pb/mod.rs + working-directory: libdd-metrics-v3 + run: | + cargo build --features generate-protobuf + git diff --exit-code -- tests/pb/mod.rs diff --git a/Cargo.lock b/Cargo.lock index 2917cbba83..45243307ae 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3179,6 +3179,18 @@ dependencies = [ "libdd-log", ] +[[package]] +name = "libdd-metrics-v3" +version = "0.1.0" +dependencies = [ + "bolero", + "foldhash 0.2.0", + "hashbrown 0.16.1", + "prost", + "prost-build", + "protoc-bin-vendored", +] + [[package]] name = "libdd-otel-thread-ctx" version = "1.0.0" diff --git a/Cargo.toml b/Cargo.toml index 20d3714041..20787f74eb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -61,6 +61,7 @@ members = [ "libdd-log", "libdd-log-ffi", "libdd-sampling", + "libdd-metrics-v3", ] # https://doc.rust-lang.org/cargo/reference/resolver.html diff --git a/libdd-metrics-v3/Cargo.toml b/libdd-metrics-v3/Cargo.toml new file mode 100644 index 0000000000..b4cba91bc7 --- /dev/null +++ b/libdd-metrics-v3/Cargo.toml @@ -0,0 +1,42 @@ +# Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/ +# SPDX-License-Identifier: Apache-2.0 + +[package] +name = "libdd-metrics-v3" +version = "0.1.0" +description = "V3 columnar protobuf codec for Datadog metrics" +homepage = "https://github.com/DataDog/libdatadog/tree/main/libdd-metrics-v3" +repository = "https://github.com/DataDog/libdatadog/tree/main/libdd-metrics-v3" +edition.workspace = true +rust-version.workspace = true +license.workspace = true +autobenches = false + +[lints] +workspace = true + +[lib] +bench = false + +[dependencies] +# default-features = false: enabling foldhash's default "std" feature would pull `std` back into +# this otherwise `no_std` crate. +foldhash = { version = "0.2", default-features = false } +hashbrown = { version = "0.16", default-features = false } + +[dev-dependencies] +# Reference implementation used by the wire-format parity tests in `tests/`: encodes the same +# columnar data via prost's generated bindings for `proto/intake_v3.proto` and checks that it's +# byte-for-byte identical to this crate's hand-rolled encoder output. +bolero = "0.13" +prost = "0.14" + +[build-dependencies] +prost-build = { version = "0.14", optional = true } +protoc-bin-vendored = { version = "3.0", optional = true } + +[features] +# Regenerates `tests/pb/mod.rs` from `proto/intake_v3.proto`. Not needed for normal builds or test +# runs: the generated bindings are checked into version control, and CI verifies they're in sync +# with the vendored .proto file (see `.github/workflows/verify-metrics-v3-proto.yml`). +generate-protobuf = ["dep:prost-build", "dep:protoc-bin-vendored"] diff --git a/libdd-metrics-v3/build.rs b/libdd-metrics-v3/build.rs new file mode 100644 index 0000000000..d6b4c82203 --- /dev/null +++ b/libdd-metrics-v3/build.rs @@ -0,0 +1,76 @@ +// Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +// This build script always runs on the host with std available (the `no_std` library it builds +// has nothing to do with the build script's own environment), so the workspace-wide +// std-vs-core/alloc and expect-used lints (meant to guard the library itself) don't apply here. +#![allow( + clippy::std_instead_of_core, + clippy::std_instead_of_alloc, + clippy::expect_used +)] + +fn main() { + #[cfg(feature = "generate-protobuf")] + generate::run().expect("failed to regenerate protobuf bindings from proto/intake_v3.proto"); + #[cfg(not(feature = "generate-protobuf"))] + println!("cargo:rerun-if-changed=build.rs"); +} + +/// Regenerates `tests/pb/mod.rs` from `proto/intake_v3.proto`. Only compiled when the +/// `generate-protobuf` feature is enabled, since it's the only fallible part of this build +/// script: normal builds and test runs use the checked-in `tests/pb/mod.rs` and never reach it. +#[cfg(feature = "generate-protobuf")] +mod generate { + use std::error::Error; + use std::{env, fs, path::Path}; + + const HEADER: &str = "// Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +// This file is @generated by prost-build from `proto/intake_v3.proto`. Do not edit it directly: +// regenerate it with `cargo build -p libdd-metrics-v3 --features generate-protobuf` after +// changing the vendored .proto file. +#![allow(dead_code, clippy::all, clippy::pedantic, clippy::nursery)] + +"; + + pub fn run() -> Result<(), Box> { + // protoc is required to compile proto files. This uses protoc-bin-vendored to obtain a + // protoc binary, setting the env var to tell prost-build where to find it. + env::set_var("PROTOC", protoc_bin_vendored::protoc_bin_path()?); + + let cur_working_dir = env::var("CARGO_MANIFEST_DIR")?; + let out_dir = Path::new(&cur_working_dir).join("tests/pb"); + fs::create_dir_all(&out_dir)?; + + let mut config = prost_build::Config::new(); + config.out_dir(&out_dir); + + println!("cargo:rerun-if-changed=proto/intake_v3.proto"); + config.compile_protos(&["proto/intake_v3.proto"], &["proto"])?; + + // prost-build names the output file after the proto's package + // (`datadoghq.api.metrics.v3.rs`); rename it to `mod.rs` so `tests/parity.rs` can + // pull it in as a plain submodule. + let generated = out_dir.join("datadoghq.api.metrics.v3.rs"); + let mod_rs = out_dir.join("mod.rs"); + fs::rename(&generated, &mod_rs)?; + + prepend_to_file(HEADER.as_bytes(), &mod_rs)?; + + Ok(()) + } + + fn prepend_to_file(data: &[u8], file_path: &Path) -> Result<(), Box> { + use std::io::{Read, Write}; + + let mut f = fs::File::open(file_path)?; + let mut content = data.to_owned(); + f.read_to_end(&mut content)?; + + let mut f = fs::File::create(file_path)?; + f.write_all(content.as_slice())?; + Ok(()) + } +} diff --git a/libdd-metrics-v3/proto/intake_v3.proto b/libdd-metrics-v3/proto/intake_v3.proto new file mode 100644 index 0000000000..49300f87df --- /dev/null +++ b/libdd-metrics-v3/proto/intake_v3.proto @@ -0,0 +1,90 @@ +// Vendored from: +// https://github.com/DataDog/agent-payload/blob/512d523e386f75077efe9e8d878284104d491f46/proto/metrics/intake_v3.proto +// +// This is the wire format that `libdd-metrics-v3`'s hand-rolled encoder (see `src/writer.rs`) +// implements without a Protocol Buffers library. `.github/workflows/verify-metrics-v3-proto.yml` +// checks in CI that the content below stays byte-for-byte identical to the pinned commit above, +// and `tests/parity.rs` checks that our hand-rolled encoder's output is byte-identical to what +// prost's generated bindings for this file produce from the same data. + +syntax = "proto3"; + +package datadoghq.api.metrics.v3; + +option go_package = "github.com/DataDog/agent-payload/v5/metrics/intake_v3"; + +message Payload { + reserved 1; // for compatibility with agentpayload.MetricPayload.series + Metadata metadata = 2; + MetricData metricData = 3; +} + +message Metadata { + repeated string tags = 1; + repeated string resources = 2; // even number of elements, [Type, Name] pairs +} + +message MetricData { + // Dictionaries + // All dictionary indexes are base-1, zero implicitly represents an empty value. + bytes dictNameStr = 1; // varint length + value + bytes dictTagStr = 2; // varint length + value + repeated sint64 dictTagsets = 3; // length, delta encoded set of indexes into dictTagsStr + + bytes dictResourceStr = 4; // varint length + value + repeated int64 dictResourceLen = 5; // number of elements in Type and Name arrays + repeated sint64 dictResourceType = 6; // delta encoded set of indexes into dictResourceStr + repeated sint64 dictResourceName = 7; // delta encoded set of indexes into dictResourceStr + + bytes dictSourceTypeName = 8; // varint length + value + repeated int32 dictOriginInfo = 9; // (product, category, service) tuples + bytes dictUnitStr = 25; // varint length + value + + // One entry per time series + repeated uint64 types = 10; // type = metricType | valueType | metricFlags + repeated sint64 nameRefs = 11; // index into dictNameStr, entire array is delta encoded + repeated sint64 tagsetRefs = 12; // index into dictTagsets, entire array is delta encoded + repeated sint64 resourcesRefs = 13; // index into dictResourceLen, entire array is delta encoded + repeated uint64 intervals = 14; + repeated uint64 numPoints = 15; + repeated sint64 sourceTypeNameRefs = 23; // index into dictSourceTypeName, entire array is delta encoded + repeated sint64 originInfoRefs = 24; // index into dictOriginInfo, entire array is delta encoded + repeated sint64 unitRefs = 26; // index into dictUnitStr, value present if flagHasUnit is set, entire array is delta encoded + + // each metric has numPoints values in this section + repeated sint64 timestamps = 16; // entire array delta encoded + repeated sint64 valsSint64 = 17; // or + repeated float valsFloat32 = 18; // or + repeated double valsFloat64 = 19; // based on valueType + repeated uint64 sketchNumBins = 20; + repeated sint32 sketchBinKeys = 21; // per-metric sequence is delta encoded + repeated uint32 sketchBinCnts = 22; + // sketch summary Sum, Min, Max are encoded as three consecutive elements in one of vals using valueType + // sketch summary Cnt is always encoded in valInt64 + // sketch summary Avg is reconstructed as Sum/Cnt in the intake +} + +enum metricType { + UNUSED = 0; + Count = 1; + Rate = 2; + Gauge = 3; + Sketch = 4; +} + +enum valueType { + Zero = 0x00; // value is zero, not stored explicitly + Sint64 = 0x10; // value is stored in valsSint64 + Float32 = 0x20; // value is stored in valsFloat32 + Float64 = 0x30; // value is stored in valsFloat64 +} + +enum metricFlags { + flagNone = 0; + flagNoIndex = 0x100; // metric should not be indexed (equivalent to origin metric type == agent_hidden in v2) + flagHasUnit = 0x200; // timeseries has a unit in the unitRefs column +} + +message Response { + string error = 1; +} diff --git a/libdd-metrics-v3/src/constants.rs b/libdd-metrics-v3/src/constants.rs new file mode 100644 index 0000000000..e3a081586d --- /dev/null +++ b/libdd-metrics-v3/src/constants.rs @@ -0,0 +1,137 @@ +// Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +// Protocol Buffers field numbers for the `MetricData` message in the V3 format. +// +// These field numbers come from the Protocol Buffers definitions in `proto/intake_v3.proto`, +// vendored from https://github.com/DataDog/agent-payload/blob/master/proto/metrics/intake_v3.proto. +/// Field number for the `DictNameStr` column. +pub const DICT_NAME_STR_FIELD_NUMBER: u32 = 1; +/// Field number for the `DictTagsStr` column. +pub const DICT_TAGS_STR_FIELD_NUMBER: u32 = 2; +/// Field number for the `DictTagsets` column. +pub const DICT_TAGSETS_FIELD_NUMBER: u32 = 3; +/// Field number for the `DictResourceStr` column. +pub const DICT_RESOURCE_STR_FIELD_NUMBER: u32 = 4; +/// Field number for the `DictResourcesLen` column. +pub const DICT_RESOURCE_LEN_FIELD_NUMBER: u32 = 5; +/// Field number for the `DictResourceType` column. +pub const DICT_RESOURCE_TYPE_FIELD_NUMBER: u32 = 6; +/// Field number for the `DictResourceName` column. +pub const DICT_RESOURCE_NAME_FIELD_NUMBER: u32 = 7; +/// Field number for the `DictSourceTypeName` column. +pub const DICT_SOURCE_TYPE_NAME_FIELD_NUMBER: u32 = 8; +/// Field number for the `DictOriginInfo` column. +pub const DICT_ORIGIN_INFO_FIELD_NUMBER: u32 = 9; +/// Field number for the `Type` column. +pub const TYPES_FIELD_NUMBER: u32 = 10; +/// Field number for the `Name` column. +pub const NAMES_FIELD_NUMBER: u32 = 11; +/// Field number for the `Tags` column. +pub const TAGS_FIELD_NUMBER: u32 = 12; +/// Field number for the `Resources` column. +pub const RESOURCES_FIELD_NUMBER: u32 = 13; +/// Field number for the `Interval` column. +pub const INTERVALS_FIELD_NUMBER: u32 = 14; +/// Field number for the `NumPoints` column. +pub const NUM_POINTS_FIELD_NUMBER: u32 = 15; +/// Field number for the `Timestamp` column. +pub const TIMESTAMPS_FIELD_NUMBER: u32 = 16; +/// Field number for the `ValueSint64` column. +pub const VALS_SINT64_FIELD_NUMBER: u32 = 17; +/// Field number for the `ValueFloat32` column. +pub const VALS_FLOAT32_FIELD_NUMBER: u32 = 18; +/// Field number for the `ValueFloat64` column. +pub const VALS_FLOAT64_FIELD_NUMBER: u32 = 19; +/// Field number for the `SketchNBins` column. +pub const SKETCH_NUM_BINS_FIELD_NUMBER: u32 = 20; +/// Field number for the `SketchBinKeys` column. +pub const SKETCH_BIN_KEYS_FIELD_NUMBER: u32 = 21; +/// Field number for the `SketchBinCounts` column. +pub const SKETCH_BIN_CNTS_FIELD_NUMBER: u32 = 22; +/// Field number for the `SourceTypeName` column. +pub const SOURCE_TYPE_NAME_FIELD_NUMBER: u32 = 23; +/// Field number for the `OriginInfo` column. +pub const ORIGIN_INFO_FIELD_NUMBER: u32 = 24; +/// Field number for the `DictUnitStr` column. +pub const DICT_UNIT_STR_FIELD_NUMBER: u32 = 25; +/// Field number for the `UnitRef` column. +pub const UNIT_REFS_FIELD_NUMBER: u32 = 26; + +/// Display names for the V3 columns, indexed by their Protocol Buffers field number. +/// +/// Field numbers come from `proto/intake_v3.proto`, also mirrored in `crate::writer` as the +/// `*_FIELD_NUMBER` constants. Index 0 is unused since field numbers start at 1. +pub const COLUMN_NAMES: [&str; 27] = [ + "reserved", + "DictNameStr", + "DictTagsStr", + "DictTagsets", + "DictResourceStr", + "DictResourcesLen", + "DictResourceType", + "DictResourceName", + "DictSourceTypeName", + "DictOriginInfo", + "Type", + "Name", + "Tags", + "Resources", + "Interval", + "NumPoints", + "Timestamp", + "ValueSint64", + "ValueFloat32", + "ValueFloat64", + "SketchNBins", + "SketchBinKeys", + "SketchBinCounts", + "SourceTypeName", + "OriginInfo", + "DictUnitStr", + "UnitRef", +]; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn field_numbers_match_column_names() { + let pairs: &[(u32, &str)] = &[ + (DICT_NAME_STR_FIELD_NUMBER, "DictNameStr"), + (DICT_TAGS_STR_FIELD_NUMBER, "DictTagsStr"), + (DICT_TAGSETS_FIELD_NUMBER, "DictTagsets"), + (DICT_RESOURCE_STR_FIELD_NUMBER, "DictResourceStr"), + (DICT_RESOURCE_LEN_FIELD_NUMBER, "DictResourcesLen"), + (DICT_RESOURCE_TYPE_FIELD_NUMBER, "DictResourceType"), + (DICT_RESOURCE_NAME_FIELD_NUMBER, "DictResourceName"), + (DICT_SOURCE_TYPE_NAME_FIELD_NUMBER, "DictSourceTypeName"), + (DICT_ORIGIN_INFO_FIELD_NUMBER, "DictOriginInfo"), + (TYPES_FIELD_NUMBER, "Type"), + (NAMES_FIELD_NUMBER, "Name"), + (TAGS_FIELD_NUMBER, "Tags"), + (RESOURCES_FIELD_NUMBER, "Resources"), + (INTERVALS_FIELD_NUMBER, "Interval"), + (NUM_POINTS_FIELD_NUMBER, "NumPoints"), + (TIMESTAMPS_FIELD_NUMBER, "Timestamp"), + (VALS_SINT64_FIELD_NUMBER, "ValueSint64"), + (VALS_FLOAT32_FIELD_NUMBER, "ValueFloat32"), + (VALS_FLOAT64_FIELD_NUMBER, "ValueFloat64"), + (SKETCH_NUM_BINS_FIELD_NUMBER, "SketchNBins"), + (SKETCH_BIN_KEYS_FIELD_NUMBER, "SketchBinKeys"), + (SKETCH_BIN_CNTS_FIELD_NUMBER, "SketchBinCounts"), + (SOURCE_TYPE_NAME_FIELD_NUMBER, "SourceTypeName"), + (ORIGIN_INFO_FIELD_NUMBER, "OriginInfo"), + (DICT_UNIT_STR_FIELD_NUMBER, "DictUnitStr"), + (UNIT_REFS_FIELD_NUMBER, "UnitRef"), + ]; + + for (field_number, expected_name) in pairs { + assert_eq!( + COLUMN_NAMES[*field_number as usize], *expected_name, + "field number {field_number} should index COLUMN_NAMES to \"{expected_name}\"" + ); + } + } +} diff --git a/libdd-metrics-v3/src/interner.rs b/libdd-metrics-v3/src/interner.rs new file mode 100644 index 0000000000..a9b1d24d79 --- /dev/null +++ b/libdd-metrics-v3/src/interner.rs @@ -0,0 +1,99 @@ +// Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +//! Generic interning for dictionary deduplication. + +use alloc::borrow::ToOwned; +use core::{borrow::Borrow, hash::Hash}; + +type FastBuildHasher = foldhash::quality::RandomState; +type FastHashMap = hashbrown::HashMap; + +/// Generic interning structure for dictionary deduplication. +/// +/// Assigns unique 1-based IDs to values, returning the same ID for duplicate values. +/// ID 0 is reserved for "empty/none" in the V3 format. +#[derive(Debug)] +pub struct Interner { + index: FastHashMap, + last_id: i64, +} + +impl Default for Interner { + fn default() -> Self { + Self::new() + } +} + +impl Interner { + /// Creates a new empty interner. + pub fn new() -> Self { + Self { + index: FastHashMap::default(), + last_id: 0, + } + } + + /// Gets the ID for a key, inserting it if not present. + /// + /// Returns `(id, is_new)` where `is_new` is true if the key was newly inserted. + /// IDs are 1-based (0 is reserved for empty/none values). + pub fn get_or_insert(&mut self, key: &Q) -> (i64, bool) + where + K: Borrow, + Q: ToOwned + Hash + Eq + ?Sized, + { + if let Some(&id) = self.index.get(key) { + (id, false) + } else { + self.last_id += 1; + self.index.insert(key.to_owned(), self.last_id); + (self.last_id, true) + } + } + + /// Returns the number of interned values. + #[cfg(test)] + pub fn len(&self) -> usize { + self.index.len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_interner_basic() { + let mut interner: Interner = Interner::new(); + + // First insertion returns ID 1 and is_new=true + let (id1, is_new1) = interner.get_or_insert("hello"); + assert_eq!(id1, 1); + assert!(is_new1); + + // Second insertion of same value returns same ID and is_new=false + let (id2, is_new2) = interner.get_or_insert("hello"); + assert_eq!(id2, 1); + assert!(!is_new2); + + // New value gets next ID + let (id3, is_new3) = interner.get_or_insert("world"); + assert_eq!(id3, 2); + assert!(is_new3); + + assert_eq!(interner.len(), 2); + } + + #[test] + fn test_interner_tuples() { + let mut interner: Interner<(i32, i32, i32)> = Interner::new(); + + let (id1, _) = interner.get_or_insert(&(1, 2, 3)); + let (id2, _) = interner.get_or_insert(&(1, 2, 3)); + let (id3, _) = interner.get_or_insert(&(4, 5, 6)); + + assert_eq!(id1, id2); + assert_ne!(id1, id3); + } +} diff --git a/libdd-metrics-v3/src/lib.rs b/libdd-metrics-v3/src/lib.rs new file mode 100644 index 0000000000..baf18836a8 --- /dev/null +++ b/libdd-metrics-v3/src/lib.rs @@ -0,0 +1,47 @@ +// Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +//! V3 columnar protobuf codec for Datadog metrics. +//! +//! This crate implements the V3 format for Datadog metrics payloads which uses +//! a columnar layout with dictionary-based string deduplication for efficient encoding. +//! +//! [`V3Writer`] accumulates metrics one at a time via [`V3Writer::write`], then +//! [`V3Writer::into_columns`] produces the encoded columns. [`V3Writer::finalize`] +//! serializes those columns into a protobuf payload. +//! +//! Consumers with their own Protocol Buffers implementation can instead serialize [`V3EncodedData`] +//! directly using the `*_FIELD_NUMBER` constants. + +#![cfg_attr(not(test), no_std)] +#![deny(missing_docs)] +#![deny( + clippy::std_instead_of_core, + clippy::std_instead_of_alloc, + clippy::alloc_instead_of_core +)] + +extern crate alloc; + +mod constants; +mod interner; +mod types; +mod writer; + +pub use constants::{ + COLUMN_NAMES, DICT_NAME_STR_FIELD_NUMBER, DICT_ORIGIN_INFO_FIELD_NUMBER, + DICT_RESOURCE_LEN_FIELD_NUMBER, DICT_RESOURCE_NAME_FIELD_NUMBER, + DICT_RESOURCE_STR_FIELD_NUMBER, DICT_RESOURCE_TYPE_FIELD_NUMBER, + DICT_SOURCE_TYPE_NAME_FIELD_NUMBER, DICT_TAGSETS_FIELD_NUMBER, DICT_TAGS_STR_FIELD_NUMBER, + DICT_UNIT_STR_FIELD_NUMBER, INTERVALS_FIELD_NUMBER, NAMES_FIELD_NUMBER, + NUM_POINTS_FIELD_NUMBER, ORIGIN_INFO_FIELD_NUMBER, RESOURCES_FIELD_NUMBER, + SKETCH_BIN_CNTS_FIELD_NUMBER, SKETCH_BIN_KEYS_FIELD_NUMBER, SKETCH_NUM_BINS_FIELD_NUMBER, + SOURCE_TYPE_NAME_FIELD_NUMBER, TAGS_FIELD_NUMBER, TIMESTAMPS_FIELD_NUMBER, TYPES_FIELD_NUMBER, + UNIT_REFS_FIELD_NUMBER, VALS_FLOAT32_FIELD_NUMBER, VALS_FLOAT64_FIELD_NUMBER, + VALS_SINT64_FIELD_NUMBER, +}; +pub use types::V3MetricType; +pub use writer::{ + V3EncodedData, V3EncodedMetrics, V3EncoderStats, V3MetricBuilder, V3ValueEncodingStats, + V3Writer, V3WriterError, +}; diff --git a/libdd-metrics-v3/src/types.rs b/libdd-metrics-v3/src/types.rs new file mode 100644 index 0000000000..a8ff3b1e74 --- /dev/null +++ b/libdd-metrics-v3/src/types.rs @@ -0,0 +1,260 @@ +// Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +//! V3 payload type definitions and protocol buffer field numbers. + +/// V3 metric type values. +/// +/// These match the `metricType` enum in `intake_v3.proto`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(u8)] +pub enum V3MetricType { + /// A monotonically increasing counter, submitted as per-interval deltas. + Count = 1, + /// A count normalized to a per-second rate. + Rate = 2, + /// A point-in-time value. + Gauge = 3, + /// A distribution summarized as a `DDSketch`. + Sketch = 4, +} + +impl V3MetricType { + /// Returns the numeric value for encoding in the types column. + #[must_use] + pub const fn as_u64(self) -> u64 { + self as u64 + } +} + +/// V3 value type values. +/// +/// These are encoded in bits 4-7 of the types column and indicate which +/// value array contains the metric's points. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(u8)] +pub enum V3ValueType { + /// Value is zero, not stored explicitly. + Zero = 0x00, + + /// Value is stored in `vals_sint64`. + Sint64 = 0x10, + + /// Value is stored in `vals_float32`. + Float32 = 0x20, + + /// Value is stored in `vals_float64`. + Float64 = 0x30, +} + +impl V3ValueType { + /// Returns the numeric value for encoding in the types column. + #[must_use] + pub const fn as_u64(self) -> u64 { + self as u64 + } +} + +/// Intermediate point classification for value type compaction. +/// +/// This provides finer-grained classification than [`V3ValueType`] to avoid +/// precision loss when combining different value types. In particular, it +/// distinguishes small integers (that fit losslessly in f32) from large integers +/// (that don't), so that mixing a large integer with a Float32 value correctly +/// escalates to Float64 rather than silently truncating the integer. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +#[repr(u8)] +enum PointKind { + /// Value is zero. + Zero = 0, + /// Integer with |v| <= 2^24, fits losslessly in both sint64 and f32. + Int24 = 1, + /// Integer with |v| > 2^24, fits in sint64 varint but NOT losslessly in f32. + Int48 = 2, + /// Fractional value exactly representable as f32. + Float32 = 3, + /// Everything else - requires full f64 precision. + Float64 = 4, +} + +/// Maximum integer magnitude that fits losslessly in f32 (2^24). +const F32_INT_MAX: i64 = 1 << 24; + +impl PointKind { + /// Classifies a single f64 value. + // Casts round-trip `v` through a narrower type to check whether the conversion is lossless. + #[allow( + clippy::cast_possible_truncation, + clippy::cast_precision_loss, + clippy::float_cmp + )] + fn for_value(v: f64) -> Self { + // Varint range that fits in 7 bytes or less (49 bits). + const VARINT_WIDTH: i32 = 7 * 7 - 1; + const MAX_INT: i64 = 1 << VARINT_WIDTH; + const MIN_INT: i64 = -MAX_INT; + + if v == 0.0 { + return Self::Zero; + } + + let i = v as i64; + if (MIN_INT..MAX_INT).contains(&i) && (i as f64) == v { + if (-F32_INT_MAX..=F32_INT_MAX).contains(&i) { + return Self::Int24; + } + return Self::Int48; + } + + if f64::from(v as f32) == v { + return Self::Float32; + } + + Self::Float64 + } + + /// Combines two point kinds into the smallest kind that can represent both. + /// + /// This is `max(self, other)` in all cases **except**: + /// - `Int48 + Float32 = Float64` (and vice versa), because large integers lose precision in + /// f32, and fractional values can't be stored as sint64. + fn union(self, other: Self) -> Self { + match (self, other) { + (Self::Int48, Self::Float32) | (Self::Float32, Self::Int48) => Self::Float64, + _ => self.max(other), + } + } + + /// Converts to the wire-format value type. + const fn to_value_type(self) -> V3ValueType { + match self { + Self::Zero => V3ValueType::Zero, + Self::Int24 | Self::Int48 => V3ValueType::Sint64, + Self::Float32 => V3ValueType::Float32, + Self::Float64 => V3ValueType::Float64, + } + } +} + +/// Determines the best [`V3ValueType`] for a set of f64 values. +/// +/// Uses [`PointKind`] internally to avoid precision loss when mixing +/// large integers with fractional float32 values. +pub fn value_type_for_values(values: impl Iterator) -> V3ValueType { + let mut kind = PointKind::Zero; + for v in values { + kind = kind.union(PointKind::for_value(v)); + } + kind.to_value_type() +} + +#[cfg(test)] +#[allow(clippy::cast_precision_loss, clippy::cast_lossless)] +mod tests { + use super::*; + + #[test] + fn test_point_kind_classification() { + // Zero + assert_eq!(PointKind::for_value(0.0), PointKind::Zero); + + // Small integers (fit in f32) + assert_eq!(PointKind::for_value(100.0), PointKind::Int24); + assert_eq!(PointKind::for_value(-100.0), PointKind::Int24); + assert_eq!(PointKind::for_value((1 << 24) as f64), PointKind::Int24); + assert_eq!(PointKind::for_value(-((1 << 24) as f64)), PointKind::Int24); + + // Large integers (don't fit losslessly in f32) + assert_eq!( + PointKind::for_value(((1 << 24) + 1) as f64), + PointKind::Int48 + ); + assert_eq!(PointKind::for_value((1i64 << 30) as f64), PointKind::Int48); + + // Float32 + assert_eq!(PointKind::for_value(1.5), PointKind::Float32); + assert_eq!(PointKind::for_value(2.75), PointKind::Float32); + + // Float64 + assert_eq!( + PointKind::for_value(core::f64::consts::PI), + PointKind::Float64 + ); + let large = ((1i64 << 50) + 1) as f64; + assert_eq!(PointKind::for_value(large), PointKind::Float64); + } + + #[test] + fn test_point_kind_union() { + // Standard widening (max) + assert_eq!(PointKind::Zero.union(PointKind::Int24), PointKind::Int24); + assert_eq!(PointKind::Int24.union(PointKind::Int48), PointKind::Int48); + assert_eq!( + PointKind::Int24.union(PointKind::Float32), + PointKind::Float32 + ); + assert_eq!( + PointKind::Float32.union(PointKind::Float64), + PointKind::Float64 + ); + assert_eq!( + PointKind::Float64.union(PointKind::Zero), + PointKind::Float64 + ); + + // The critical case: large integer + float32 must escalate to float64 + assert_eq!( + PointKind::Int48.union(PointKind::Float32), + PointKind::Float64 + ); + assert_eq!( + PointKind::Float32.union(PointKind::Int48), + PointKind::Float64 + ); + } + + #[test] + fn test_value_type_for_values() { + // All zeros + assert_eq!( + value_type_for_values([0.0, 0.0].into_iter()), + V3ValueType::Zero + ); + + // Small integers + assert_eq!( + value_type_for_values([100.0, 200.0].into_iter()), + V3ValueType::Sint64 + ); + + // Large integers + assert_eq!( + value_type_for_values([(1i64 << 30) as f64, 200.0].into_iter()), + V3ValueType::Sint64 + ); + + // Small integer + float32 → Float32 (safe, small int fits in f32) + assert_eq!( + value_type_for_values([100.0, 1.5].into_iter()), + V3ValueType::Float32 + ); + + // Large integer + float32 → Float64 (the bug fix!) + assert_eq!( + value_type_for_values([(1i64 << 30) as f64, 1.5].into_iter()), + V3ValueType::Float64 + ); + + // Float64 value forces Float64 + assert_eq!( + value_type_for_values([100.0, core::f64::consts::PI].into_iter()), + V3ValueType::Float64 + ); + + // Empty iterator + assert_eq!( + value_type_for_values(core::iter::empty()), + V3ValueType::Zero + ); + } +} diff --git a/libdd-metrics-v3/src/writer.rs b/libdd-metrics-v3/src/writer.rs new file mode 100644 index 0000000000..61577e7d72 --- /dev/null +++ b/libdd-metrics-v3/src/writer.rs @@ -0,0 +1,1488 @@ +// Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +//! V3 columnar metrics writer. +//! +//! [`V3Writer`] accumulates metrics in columnar format with dictionary deduplication, then +//! produces [`V3EncodedData`] via [`V3Writer::into_columns`]. [`V3Writer::finalize`] +//! serializes that columnar data to protobuf wire format, using a hand-rolled encoder. + +use alloc::{string::String, vec::Vec}; + +use crate::{ + constants::{ + DICT_NAME_STR_FIELD_NUMBER, DICT_ORIGIN_INFO_FIELD_NUMBER, DICT_RESOURCE_LEN_FIELD_NUMBER, + DICT_RESOURCE_NAME_FIELD_NUMBER, DICT_RESOURCE_STR_FIELD_NUMBER, + DICT_RESOURCE_TYPE_FIELD_NUMBER, DICT_SOURCE_TYPE_NAME_FIELD_NUMBER, + DICT_TAGSETS_FIELD_NUMBER, DICT_TAGS_STR_FIELD_NUMBER, DICT_UNIT_STR_FIELD_NUMBER, + INTERVALS_FIELD_NUMBER, NAMES_FIELD_NUMBER, NUM_POINTS_FIELD_NUMBER, + ORIGIN_INFO_FIELD_NUMBER, RESOURCES_FIELD_NUMBER, SKETCH_BIN_CNTS_FIELD_NUMBER, + SKETCH_BIN_KEYS_FIELD_NUMBER, SKETCH_NUM_BINS_FIELD_NUMBER, SOURCE_TYPE_NAME_FIELD_NUMBER, + TAGS_FIELD_NUMBER, TIMESTAMPS_FIELD_NUMBER, TYPES_FIELD_NUMBER, UNIT_REFS_FIELD_NUMBER, + VALS_FLOAT32_FIELD_NUMBER, VALS_FLOAT64_FIELD_NUMBER, VALS_SINT64_FIELD_NUMBER, + }, + interner::Interner, + types::{value_type_for_values, V3MetricType, V3ValueType}, +}; + +pub const FLAG_NO_INDEX: u64 = 0x100; +pub const FLAG_HAS_UNIT: u64 = 0x200; + +/// Bitmask for the base [`V3MetricType`] stored in bits 0-3 of a `types` column entry. +const METRIC_TYPE_MASK: u64 = 0x0F; + +/// Errors returned by [`V3MetricBuilder`] methods when a caller violates one of their +/// preconditions. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum V3WriterError { + /// [`add_point`](V3MetricBuilder::add_point) was called on a builder created with + /// [`V3MetricType::Sketch`]; use [`add_sketch`](V3MetricBuilder::add_sketch) instead. + PointOnSketchMetric, + /// [`add_sketch`](V3MetricBuilder::add_sketch) was called on a builder created with a + /// non-sketch metric type; use [`add_point`](V3MetricBuilder::add_point) instead. + SketchOnNonSketchMetric, + /// [`add_sketch`](V3MetricBuilder::add_sketch)'s `bin_keys` and `bin_counts` slices had + /// different lengths. + SketchBinLengthMismatch { + /// Length of the `bin_keys` slice. + bin_keys_len: usize, + /// Length of the `bin_counts` slice. + bin_counts_len: usize, + }, +} + +impl core::fmt::Display for V3WriterError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + match self { + Self::PointOnSketchMetric => { + write!( + f, + "add_point called on a Sketch metric; use add_sketch instead" + ) + } + Self::SketchOnNonSketchMetric => { + write!( + f, + "add_sketch called on a non-Sketch metric; use add_point instead" + ) + } + Self::SketchBinLengthMismatch { + bin_keys_len, + bin_counts_len, + } => write!( + f, + "bin_keys length ({bin_keys_len}) does not match bin_counts length \ + ({bin_counts_len})" + ), + } + } +} + +impl core::error::Error for V3WriterError {} + +/// Encoded V3 payload data, prior to wire-format serialization. +/// +/// This is the columnar representation produced by [`V3Writer::into_columns`], after delta +/// encoding but before any protobuf framing. Column names are given by [`crate::COLUMN_NAMES`] +/// and field numbers by the `*_FIELD_NUMBER` constants in this crate. Consumers with their own +/// Protocol Buffers implementation can serialize these columns directly instead of using +/// [`V3Writer::finalize`]. +#[derive(Debug, Default)] +pub struct V3EncodedData { + /// Dictionary of interned metric names, as varint-length-prefixed strings concatenated + /// together. + pub dict_name_bytes: Vec, + /// Dictionary of interned tag strings, as varint-length-prefixed strings concatenated + /// together. + pub dict_tags_bytes: Vec, + /// Dictionary of interned tag sets. Each entry is a count of tags followed by that many + /// sorted, delta-encoded tag dictionary IDs. + pub dict_tagsets: Vec, + /// Dictionary of interned resource type/name strings, as varint-length-prefixed strings + /// concatenated together. + pub dict_resource_str_bytes: Vec, + /// Number of (type, name) pairs in each interned resource set. + pub dict_resource_len: Vec, + /// Delta-encoded resource type dictionary IDs, grouped contiguously per interned resource set. + pub dict_resource_type: Vec, + /// Delta-encoded resource name dictionary IDs, grouped contiguously per interned resource set. + pub dict_resource_name: Vec, + /// Dictionary of interned source type name strings, as varint-length-prefixed strings + /// concatenated together. + pub dict_source_type_bytes: Vec, + /// Dictionary of interned origin metadata, as flattened (product, category, service) triples. + pub dict_origin_info: Vec, + /// Dictionary of interned unit strings, as varint-length-prefixed strings concatenated + /// together. + pub dict_unit_bytes: Vec, + + /// Per-metric type and flags column. + pub types: Vec, + /// Per-metric delta-encoded name dictionary IDs. + pub names: Vec, + /// Per-metric delta-encoded tag set dictionary IDs. + pub tags: Vec, + /// Per-metric delta-encoded resource set dictionary IDs. + pub resources: Vec, + /// Per-metric interval, in seconds, used for rate metrics. + pub intervals: Vec, + /// Per-metric number of points. + pub num_points: Vec, + /// Per-metric delta-encoded source type name dictionary IDs. + pub source_type_names: Vec, + /// Per-metric delta-encoded origin metadata dictionary IDs. + pub origin_infos: Vec, + /// Per-metric delta-encoded unit dictionary IDs. Present only for metrics with `FLAG_HAS_UNIT` + /// set. + pub unit_refs: Vec, + + /// Delta-encoded point timestamps, across all metrics. + pub timestamps: Vec, + /// Point values stored as signed 64-bit integers. + pub vals_sint64: Vec, + /// Point values stored as 32-bit floats. + pub vals_float32: Vec, + /// Point values stored as 64-bit floats. + pub vals_float64: Vec, + + /// Number of bins in each sketch. + pub sketch_num_bins: Vec, + /// Delta-encoded sketch bin keys, grouped contiguously per sketch. + pub sketch_bin_keys: Vec, + /// Sketch bin counts, grouped contiguously per sketch. + pub sketch_bin_cnts: Vec, + + /// Telemetry produced while encoding the columns. + pub value_encoding_stats: V3ValueEncodingStats, +} + +/// Encoded V3 metrics payload with telemetry data. +pub struct V3EncodedMetrics { + /// Serialized `MetricData` protobuf payload. + pub payload: Vec, + /// Telemetry produced while encoding the payload. + pub stats: V3EncoderStats, +} + +/// Telemetry data produced while encoding a V3 metrics payload. +pub struct V3EncoderStats { + /// Counts of how many point values were compacted into each value column. + pub value_encoding_stats: V3ValueEncodingStats, + /// Raw bytes written for each present column, keyed by field number. + pub columns: Vec, +} + +/// Counts of how many point values were encoded into each value column. +#[derive(Clone, Copy, Debug, Default)] +pub struct V3ValueEncodingStats { + /// Number of point values that were zero and required no explicit storage. + pub zero: u64, + /// Number of point values stored as signed integers. + pub sint64: u64, + /// Number of point values stored as 32-bit floats. + pub float32: u64, + /// Number of point values stored as 64-bit floats. + pub float64: u64, +} + +/// Raw stream bytes for a single V3 column before protobuf field framing. +pub struct V3ColumnBytes { + /// Protocol Buffers field number this column corresponds to. + pub field_number: u32, + /// Column contents, framed as an unwrapped (no field tag) protobuf value. + pub bytes: Vec, + /// Reserved for the compressed length of `bytes`; currently always `0`. + pub compressed_len: usize, +} + +/// V3 columnar metrics writer. +/// +/// Accumulates metrics in columnar format with dictionary deduplication. +/// Call [`V3Writer::write`] for each metric, then [`V3Writer::finalize`] to finalize +/// and get the encoded data. +#[derive(Debug, Default)] +pub struct V3Writer { + // Interners for dictionary deduplication + name_interner: Interner, + tag_interner: Interner, + tagset_interner: Interner>, + resource_str_interner: Interner, + resource_interner: Interner>, + source_type_interner: Interner, + origin_interner: Interner<(i32, i32, i32)>, + unit_interner: Interner, + + // Dictionary encoded bytes + dict_name_bytes: Vec, + dict_tags_bytes: Vec, + dict_tagsets: Vec, + dict_resource_str_bytes: Vec, + dict_resource_len: Vec, + dict_resource_type: Vec, + dict_resource_name: Vec, + dict_source_type_bytes: Vec, + dict_origin_info: Vec, + dict_unit_bytes: Vec, + + // Per-metric columns (one entry per metric, except conditional columns) + types: Vec, + names: Vec, + tags: Vec, + resources: Vec, + intervals: Vec, + num_points: Vec, + source_type_names: Vec, + origin_infos: Vec, + unit_refs: Vec, // Present only for metrics with FLAG_HAS_UNIT set. + + // Point data + timestamps: Vec, + vals_sint64: Vec, + vals_float32: Vec, + vals_float64: Vec, + + // Sketch data + sketch_num_bins: Vec, + sketch_bin_keys: Vec, + sketch_bin_cnts: Vec, + + // Scratch data + tag_ids: Vec, + resource_ids: Vec<(i64, i64)>, + value_encoding_stats: V3ValueEncodingStats, +} + +impl V3Writer { + /// Creates a new V3 writer. + #[must_use] + pub fn new() -> Self { + Self::default() + } + + /// Begins writing a new metric. + /// + /// Returns a [`V3MetricBuilder`] that must be used to set the metric's + /// properties and add points, then closed with [`V3MetricBuilder::close`]. + pub fn write(&mut self, metric_type: V3MetricType, name: &str) -> V3MetricBuilder<'_> { + let name_id = self.intern_name(name); + let metric_idx = self.types.len(); + let point_start_idx = self.vals_float64.len(); + let sint64_start_idx = self.vals_sint64.len(); + + // Initialize the per-metric columns with default values + self.types.push(metric_type.as_u64()); + self.names.push(name_id); + self.tags.push(0); + self.resources.push(0); + self.intervals.push(0); + self.num_points.push(0); + self.source_type_names.push(0); + self.origin_infos.push(0); + + V3MetricBuilder { + writer: self, + point_start_idx, + sint64_start_idx, + metric_idx, + unit_ref_idx: None, + closed: false, + } + } + + /// Finalizes the writer and returns the encoded columnar data. + #[must_use] + pub fn into_columns(mut self) -> V3EncodedData { + // Delta encode all of the index arrays first. + delta_encode(&mut self.names); + delta_encode(&mut self.tags); + delta_encode(&mut self.resources); + delta_encode(&mut self.source_type_names); + delta_encode(&mut self.origin_infos); + delta_encode(&mut self.unit_refs); + delta_encode(&mut self.timestamps); + + V3EncodedData { + dict_name_bytes: self.dict_name_bytes, + dict_tags_bytes: self.dict_tags_bytes, + dict_tagsets: self.dict_tagsets, + dict_resource_str_bytes: self.dict_resource_str_bytes, + dict_resource_len: self.dict_resource_len, + dict_resource_type: self.dict_resource_type, + dict_resource_name: self.dict_resource_name, + dict_source_type_bytes: self.dict_source_type_bytes, + dict_origin_info: self.dict_origin_info, + dict_unit_bytes: self.dict_unit_bytes, + types: self.types, + names: self.names, + tags: self.tags, + resources: self.resources, + intervals: self.intervals, + num_points: self.num_points, + source_type_names: self.source_type_names, + origin_infos: self.origin_infos, + unit_refs: self.unit_refs, + timestamps: self.timestamps, + vals_sint64: self.vals_sint64, + vals_float32: self.vals_float32, + vals_float64: self.vals_float64, + sketch_num_bins: self.sketch_num_bins, + sketch_bin_keys: self.sketch_bin_keys, + sketch_bin_cnts: self.sketch_bin_cnts, + value_encoding_stats: self.value_encoding_stats, + } + } + + /// Finalizes the writer and serializes the data to the given output buffer. + #[allow(clippy::too_many_lines)] + pub fn finalize(self) -> V3EncodedMetrics { + let data = self.into_columns(); + let mut output = Vec::new(); + let mut columns = Vec::new(); + + // Dictionary fields (bytes - varint-length-prefixed strings concatenated) + write_bytes_column( + &mut output, + &mut columns, + DICT_NAME_STR_FIELD_NUMBER, + &data.dict_name_bytes, + ); + write_bytes_column( + &mut output, + &mut columns, + DICT_TAGS_STR_FIELD_NUMBER, + &data.dict_tags_bytes, + ); + + // Packed repeated fields for dictionaries + write_packed_column( + &mut output, + &mut columns, + DICT_TAGSETS_FIELD_NUMBER, + &data.dict_tagsets, + write_sint64_value, + ); + + write_bytes_column( + &mut output, + &mut columns, + DICT_RESOURCE_STR_FIELD_NUMBER, + &data.dict_resource_str_bytes, + ); + + write_packed_column( + &mut output, + &mut columns, + DICT_RESOURCE_LEN_FIELD_NUMBER, + &data.dict_resource_len, + write_int64_value, + ); + write_packed_column( + &mut output, + &mut columns, + DICT_RESOURCE_TYPE_FIELD_NUMBER, + &data.dict_resource_type, + write_sint64_value, + ); + write_packed_column( + &mut output, + &mut columns, + DICT_RESOURCE_NAME_FIELD_NUMBER, + &data.dict_resource_name, + write_sint64_value, + ); + + write_bytes_column( + &mut output, + &mut columns, + DICT_SOURCE_TYPE_NAME_FIELD_NUMBER, + &data.dict_source_type_bytes, + ); + + write_packed_column( + &mut output, + &mut columns, + DICT_ORIGIN_INFO_FIELD_NUMBER, + &data.dict_origin_info, + write_int32_value, + ); + write_bytes_column( + &mut output, + &mut columns, + DICT_UNIT_STR_FIELD_NUMBER, + &data.dict_unit_bytes, + ); + + // Per-metric columns + write_packed_column( + &mut output, + &mut columns, + TYPES_FIELD_NUMBER, + &data.types, + write_uint64_value, + ); + write_packed_column( + &mut output, + &mut columns, + NAMES_FIELD_NUMBER, + &data.names, + write_sint64_value, + ); + write_packed_column( + &mut output, + &mut columns, + TAGS_FIELD_NUMBER, + &data.tags, + write_sint64_value, + ); + write_packed_column( + &mut output, + &mut columns, + RESOURCES_FIELD_NUMBER, + &data.resources, + write_sint64_value, + ); + write_packed_column( + &mut output, + &mut columns, + INTERVALS_FIELD_NUMBER, + &data.intervals, + write_uint64_value, + ); + write_packed_column( + &mut output, + &mut columns, + NUM_POINTS_FIELD_NUMBER, + &data.num_points, + write_uint64_value, + ); + write_packed_column( + &mut output, + &mut columns, + SOURCE_TYPE_NAME_FIELD_NUMBER, + &data.source_type_names, + write_sint64_value, + ); + write_packed_column( + &mut output, + &mut columns, + ORIGIN_INFO_FIELD_NUMBER, + &data.origin_infos, + write_sint64_value, + ); + write_packed_column( + &mut output, + &mut columns, + UNIT_REFS_FIELD_NUMBER, + &data.unit_refs, + write_sint64_value, + ); + + // Point data + write_packed_column( + &mut output, + &mut columns, + TIMESTAMPS_FIELD_NUMBER, + &data.timestamps, + write_sint64_value, + ); + write_packed_column( + &mut output, + &mut columns, + VALS_SINT64_FIELD_NUMBER, + &data.vals_sint64, + write_sint64_value, + ); + write_packed_column( + &mut output, + &mut columns, + VALS_FLOAT32_FIELD_NUMBER, + &data.vals_float32, + write_f32_value, + ); + write_packed_column( + &mut output, + &mut columns, + VALS_FLOAT64_FIELD_NUMBER, + &data.vals_float64, + write_f64_value, + ); + + // Sketch data + write_packed_column( + &mut output, + &mut columns, + SKETCH_NUM_BINS_FIELD_NUMBER, + &data.sketch_num_bins, + write_uint64_value, + ); + write_packed_column( + &mut output, + &mut columns, + SKETCH_BIN_KEYS_FIELD_NUMBER, + &data.sketch_bin_keys, + write_sint32_value, + ); + write_packed_column( + &mut output, + &mut columns, + SKETCH_BIN_CNTS_FIELD_NUMBER, + &data.sketch_bin_cnts, + write_uint32_value, + ); + + V3EncodedMetrics { + payload: output, + stats: V3EncoderStats { + value_encoding_stats: data.value_encoding_stats, + columns, + }, + } + } + + fn intern_name(&mut self, name: &str) -> i64 { + if name.is_empty() { + return 0; + } + let (id, is_new) = self.name_interner.get_or_insert(name); + if is_new { + append_len_str(&mut self.dict_name_bytes, name); + } + id + } + + fn intern_tag(&mut self, tag: &str) { + if tag.is_empty() { + self.tag_ids.push(0); + return; + } + + let (id, is_new) = self.tag_interner.get_or_insert(tag); + if is_new { + append_len_str(&mut self.dict_tags_bytes, tag); + } + self.tag_ids.push(id); + } + + fn intern_tagset(&mut self, tags: I) -> i64 + where + I: Iterator, + S: AsRef, + { + self.tag_ids.clear(); + for tag in tags { + self.intern_tag(tag.as_ref()); + } + + if self.tag_ids.is_empty() { + return 0; + } + + let (id, is_new) = self.tagset_interner.get_or_insert(&self.tag_ids); + if is_new { + self.encode_tagset(); + } + id + } + + fn encode_tagset(&mut self) { + // Push the length. `tag_ids.len()` can never approach `i64::MAX`. + #[allow(clippy::cast_possible_wrap)] + self.dict_tagsets.push(self.tag_ids.len() as i64); + + let start = self.dict_tagsets.len(); + + // Add all tag IDs + self.dict_tagsets.extend_from_slice(&self.tag_ids); + + // Sort and delta-encode the tagset portion + self.dict_tagsets[start..].sort_unstable(); + delta_encode(&mut self.dict_tagsets[start..]); + } + + fn intern_resource_str(&mut self, s: &str) -> i64 { + if s.is_empty() { + return 0; + } + let (id, is_new) = self.resource_str_interner.get_or_insert(s); + if is_new { + append_len_str(&mut self.dict_resource_str_bytes, s); + } + id + } + + fn intern_resources(&mut self, resources: &[(&str, &str)]) -> i64 { + self.resource_ids.clear(); + for (resource_type, resource_name) in resources { + let type_id = self.intern_resource_str(resource_type); + let name_id = self.intern_resource_str(resource_name); + self.resource_ids.push((type_id, name_id)); + } + + if self.resource_ids.is_empty() { + return 0; + } + + let (id, is_new) = self.resource_interner.get_or_insert(&self.resource_ids); + if is_new { + self.encode_resources(); + } + id + } + + fn encode_resources(&mut self) { + // `resource_ids.len()` can never approach `i64::MAX`. + #[allow(clippy::cast_possible_wrap)] + self.dict_resource_len.push(self.resource_ids.len() as i64); + + let type_start = self.dict_resource_type.len(); + let name_start = self.dict_resource_name.len(); + + for (type_id, name_id) in &self.resource_ids { + self.dict_resource_type.push(*type_id); + self.dict_resource_name.push(*name_id); + } + + delta_encode(&mut self.dict_resource_type[type_start..]); + delta_encode(&mut self.dict_resource_name[name_start..]); + } + + fn intern_source_type(&mut self, s: &str) -> i64 { + if s.is_empty() { + return 0; + } + let (id, is_new) = self.source_type_interner.get_or_insert(s); + if is_new { + append_len_str(&mut self.dict_source_type_bytes, s); + } + id + } + + fn intern_origin(&mut self, product: i32, category: i32, service: i32) -> i64 { + if product == 0 && category == 0 && service == 0 { + return 0; + } + + let (id, is_new) = self + .origin_interner + .get_or_insert(&(product, category, service)); + if is_new { + self.dict_origin_info.push(product); + self.dict_origin_info.push(category); + self.dict_origin_info.push(service); + } + id + } + + fn intern_unit(&mut self, unit: &str) -> i64 { + if unit.is_empty() { + return 0; + } + let (id, is_new) = self.unit_interner.get_or_insert(unit); + if is_new { + append_len_str(&mut self.dict_unit_bytes, unit); + } + id + } +} + +/// Builder for a single metric within a V3 payload. +/// +/// Use the setter methods to configure the metric, add points with [`add_point`](Self::add_point), +/// then call [`close`](Self::close) to finalize. If dropped without calling `close`, the metric +/// is finalized automatically. +pub struct V3MetricBuilder<'a> { + writer: &'a mut V3Writer, + point_start_idx: usize, + sint64_start_idx: usize, + metric_idx: usize, + unit_ref_idx: Option, + closed: bool, +} + +impl Drop for V3MetricBuilder<'_> { + /// Finalizes the metric if [`close`](Self::close) was never called. + fn drop(&mut self) { + if !self.closed { + self.compact_values(); + } + } +} + +impl V3MetricBuilder<'_> { + /// Sets the tags for this metric. + /// + /// Tags should be in "key:value" format. + pub fn set_tags(&mut self, tags: I) + where + I: Iterator, + S: AsRef, + { + let tagset_id = self.writer.intern_tagset(tags); + self.writer.tags[self.metric_idx] = tagset_id; + } + + /// Sets the resources for this metric. + /// + /// Resources are (type, name) pairs, for example, (`host`, `server1`). + pub fn set_resources(&mut self, resources: &[(&str, &str)]) { + let res_id = self.writer.intern_resources(resources); + self.writer.resources[self.metric_idx] = res_id; + } + + /// Sets the interval for this metric (used for rate metrics). + pub fn set_interval(&mut self, interval: u64) { + self.writer.intervals[self.metric_idx] = interval; + } + + /// Sets the source type name for this metric. + pub fn set_source_type(&mut self, source_type: &str) { + if source_type.is_empty() { + self.writer.source_type_names[self.metric_idx] = 0; + return; + } + let id = self.writer.intern_source_type(source_type); + self.writer.source_type_names[self.metric_idx] = id; + } + + /// Sets the origin metadata for this metric. + pub fn set_origin(&mut self, product: u32, category: u32, service: u32, no_index: bool) { + let id = self.writer.intern_origin( + product.cast_signed(), + category.cast_signed(), + service.cast_signed(), + ); + self.writer.origin_infos[self.metric_idx] = id; + if no_index { + self.writer.types[self.metric_idx] |= FLAG_NO_INDEX; + } else { + self.writer.types[self.metric_idx] &= !FLAG_NO_INDEX; + } + } + + /// Sets the unit for this metric. + pub fn set_unit(&mut self, unit: &str) { + if unit.is_empty() { + self.writer.types[self.metric_idx] &= !FLAG_HAS_UNIT; + if let Some(unit_ref_idx) = self.unit_ref_idx.take() { + self.writer.unit_refs.remove(unit_ref_idx); + } + return; + } + + let id = self.writer.intern_unit(unit); + if let Some(unit_ref_idx) = self.unit_ref_idx { + self.writer.unit_refs[unit_ref_idx] = id; + } else { + self.unit_ref_idx = Some(self.writer.unit_refs.len()); + self.writer.unit_refs.push(id); + } + self.writer.types[self.metric_idx] |= FLAG_HAS_UNIT; + } + + /// Adds a data point to this metric. + /// + /// # Errors + /// + /// Returns [`V3WriterError::PointOnSketchMetric`] if this builder was created with + /// [`V3MetricType::Sketch`]; use [`add_sketch`](Self::add_sketch) for sketch metrics instead. + pub fn add_point(&mut self, timestamp: i64, value: f64) -> Result<(), V3WriterError> { + if (self.writer.types[self.metric_idx] & METRIC_TYPE_MASK) == V3MetricType::Sketch as u64 { + return Err(V3WriterError::PointOnSketchMetric); + } + + self.writer.timestamps.push(timestamp); + self.writer.vals_float64.push(value); + self.writer.num_points[self.metric_idx] += 1; + Ok(()) + } + + /// Adds sketch data for a distribution metric. + /// + /// For sketches, the summary values (count, sum, min, max) are stored as points, + /// and the bin keys/counts are stored separately. + /// + /// # Errors + /// + /// Returns [`V3WriterError::SketchOnNonSketchMetric`] if this builder was not created with + /// [`V3MetricType::Sketch`], or [`V3WriterError::SketchBinLengthMismatch`] if `bin_keys` and + /// `bin_counts` have different lengths. + #[allow(clippy::too_many_arguments)] + pub fn add_sketch( + &mut self, + timestamp: i64, + count: i64, + sum: f64, + min: f64, + max: f64, + bin_keys: &[i32], + bin_counts: &[u32], + ) -> Result<(), V3WriterError> { + if (self.writer.types[self.metric_idx] & METRIC_TYPE_MASK) != V3MetricType::Sketch as u64 { + return Err(V3WriterError::SketchOnNonSketchMetric); + } + if bin_keys.len() != bin_counts.len() { + return Err(V3WriterError::SketchBinLengthMismatch { + bin_keys_len: bin_keys.len(), + bin_counts_len: bin_counts.len(), + }); + } + + self.writer.timestamps.push(timestamp); + + // Count goes in sint64, sum/min/max go in float64 + self.writer.vals_sint64.push(count); + self.writer.vals_float64.push(sum); + self.writer.vals_float64.push(min); + self.writer.vals_float64.push(max); + + // Store bin data + self.writer.sketch_num_bins.push(bin_keys.len() as u64); + + let key_start = self.writer.sketch_bin_keys.len(); + self.writer.sketch_bin_keys.extend_from_slice(bin_keys); + self.writer.sketch_bin_cnts.extend_from_slice(bin_counts); + + // Delta-encode this sketch's bin keys + delta_encode_i32(&mut self.writer.sketch_bin_keys[key_start..]); + + self.writer.num_points[self.metric_idx] += 1; + Ok(()) + } + + /// Finalizes this metric. + pub fn close(mut self) { + // Compacts the point values to use the smallest representation that can hold + // all values without loss. + self.compact_values(); + self.closed = true; + } + + #[allow(clippy::cast_possible_truncation)] + fn compact_values(&mut self) { + let count = self.writer.num_points[self.metric_idx] as usize; + if count == 0 { + return; + } + + let start = self.point_start_idx; + let end = self.writer.vals_float64.len(); + + // Determine the best value type for all points in this metric. + let val_ty = value_type_for_values(self.writer.vals_float64[start..end].iter().copied()); + let is_sketch = + (self.writer.types[self.metric_idx] & METRIC_TYPE_MASK) == V3MetricType::Sketch as u64; + let float_values_len = end - start; + if is_sketch { + // Sketches always carry one integer count per point in addition to sum/min/max values. + self.writer.value_encoding_stats.sint64 += count as u64; + } + + // Update the type field + self.writer.types[self.metric_idx] |= val_ty.as_u64(); + + // Convert values to the appropriate storage + match val_ty { + V3ValueType::Zero => { + self.writer.value_encoding_stats.zero += float_values_len as u64; + // Values are all zero, don't store anything + self.writer.vals_float64.truncate(start); + } + V3ValueType::Sint64 => { + self.writer.value_encoding_stats.sint64 += float_values_len as u64; + if is_sketch { + // For sketches, vals_sint64 already has one count per point (pushed by + // add_sketch), and vals_float64 has 3 values per point + // (sum, min, max). When compacting to Sint64, we need to + // interleave them as: sum, min, max, cnt per point. + let counts: Vec = + self.writer.vals_sint64[self.sint64_start_idx..].to_vec(); + self.writer.vals_sint64.truncate(self.sint64_start_idx); + for (i, cnt) in counts.into_iter().enumerate() { + let f_off = start + i * 3; + self.writer + .vals_sint64 + .push(self.writer.vals_float64[f_off] as i64); + self.writer + .vals_sint64 + .push(self.writer.vals_float64[f_off + 1] as i64); + self.writer + .vals_sint64 + .push(self.writer.vals_float64[f_off + 2] as i64); + self.writer.vals_sint64.push(cnt); + } + } else { + for i in start..end { + self.writer + .vals_sint64 + .push(self.writer.vals_float64[i] as i64); + } + } + self.writer.vals_float64.truncate(start); + } + V3ValueType::Float32 => { + self.writer.value_encoding_stats.float32 += float_values_len as u64; + for i in start..end { + self.writer + .vals_float32 + .push(self.writer.vals_float64[i] as f32); + } + self.writer.vals_float64.truncate(start); + } + V3ValueType::Float64 => { + self.writer.value_encoding_stats.float64 += float_values_len as u64; + // Already stored in vals_float64, keep them + } + } + } +} + +/// Protobuf wire type for length-delimited fields (bytes, strings, packed repeated fields). +const WIRE_LEN: u32 = 2; + +/// Writes a raw protobuf varint (LEB128, 7 bits per byte). +fn write_varint(buf: &mut Vec, mut value: u64) { + loop { + let byte = (value & 0x7f) as u8; + value >>= 7; + if value == 0 { + buf.push(byte); + return; + } + buf.push(byte | 0x80); + } +} + +/// Writes a protobuf field tag: `(field_number << 3) | wire_type`, as a varint. +fn write_tag(buf: &mut Vec, field_number: u32, wire_type: u32) { + write_varint(buf, (u64::from(field_number) << 3) | u64::from(wire_type)); +} + +/// Bit-reinterprets (not arithmetically converts) `v` as unsigned; this is the standard protobuf +/// zigzag transform, not a lossy narrowing. +const fn zigzag64(v: i64) -> u64 { + ((v << 1) ^ (v >> 63)).cast_unsigned() +} + +/// See [`zigzag64`]. +const fn zigzag32(v: i32) -> u32 { + ((v << 1) ^ (v >> 31)).cast_unsigned() +} + +// Scalar value encoders, one per protobuf field type used by the V3 payload. Each writes a +// single value with no tag or length prefix, matching the layout of a packed repeated field's +// payload (and reused as-is for `V3ColumnBytes::bytes`, which is exactly that payload). + +fn write_uint64_value(buf: &mut Vec, v: u64) { + write_varint(buf, v); +} + +fn write_sint64_value(buf: &mut Vec, v: i64) { + write_varint(buf, zigzag64(v)); +} + +/// Protobuf's `int64` wire type is a plain varint of the value's two's-complement bit pattern +/// (inefficient for negative numbers, but that's the wire format). +fn write_int64_value(buf: &mut Vec, v: i64) { + write_varint(buf, v.cast_unsigned()); +} + +/// See [`write_int64_value`]; `int32` is sign-extended to 64 bits before the same treatment. +fn write_int32_value(buf: &mut Vec, v: i32) { + write_varint(buf, i64::from(v).cast_unsigned()); +} + +fn write_sint32_value(buf: &mut Vec, v: i32) { + write_varint(buf, u64::from(zigzag32(v))); +} + +fn write_uint32_value(buf: &mut Vec, v: u32) { + write_varint(buf, u64::from(v)); +} + +fn write_f32_value(buf: &mut Vec, v: f32) { + buf.extend_from_slice(&v.to_le_bytes()); +} + +fn write_f64_value(buf: &mut Vec, v: f64) { + buf.extend_from_slice(&v.to_le_bytes()); +} + +/// Writes a `bytes` field (tag + varint length + data). No-op if `bytes` is empty. +fn write_bytes_column( + output: &mut Vec, + columns: &mut Vec, + field_number: u32, + bytes: &[u8], +) { + if bytes.is_empty() { + return; + } + + write_tag(output, field_number, WIRE_LEN); + write_varint(output, bytes.len() as u64); + output.extend_from_slice(bytes); + + columns.push(V3ColumnBytes { + field_number, + bytes: bytes.to_vec(), + compressed_len: 0, + }); +} + +/// Writes a packed repeated field (tag + varint byte-length + concatenated encoded values). +/// No-op if `values` is empty. +fn write_packed_column( + output: &mut Vec, + columns: &mut Vec, + field_number: u32, + values: &[T], + encode_one: fn(&mut Vec, T), +) { + if values.is_empty() { + return; + } + + let mut raw = Vec::new(); + for &v in values { + encode_one(&mut raw, v); + } + + write_tag(output, field_number, WIRE_LEN); + write_varint(output, raw.len() as u64); + output.extend_from_slice(&raw); + + columns.push(V3ColumnBytes { + field_number, + bytes: raw, + compressed_len: 0, + }); +} + +fn append_len_str(dst: &mut Vec, s: &str) { + let mut len = s.len() as u64; + loop { + let mut byte = (len & 0x7F) as u8; + len >>= 7; + if len != 0 { + byte |= 0x80; + } + dst.push(byte); + if len == 0 { + break; + } + } + dst.extend_from_slice(s.as_bytes()); +} + +fn delta_encode(s: &mut [i64]) { + if s.len() < 2 { + return; + } + for i in (1..s.len()).rev() { + s[i] -= s[i - 1]; + } +} + +fn delta_encode_i32(s: &mut [i32]) { + if s.len() < 2 { + return; + } + for i in (1..s.len()).rev() { + s[i] -= s[i - 1]; + } +} + +#[cfg(test)] +#[allow(clippy::cast_precision_loss, clippy::cast_lossless)] +mod tests { + use super::*; + + #[test] + fn test_delta_encode() { + let mut data = vec![100, 110, 130, 145]; + delta_encode(&mut data); + assert_eq!(data, vec![100, 10, 20, 15]); + } + + #[test] + fn test_delta_encode_empty() { + let mut data: Vec = vec![]; + delta_encode(&mut data); + assert!(data.is_empty()); + } + + #[test] + fn test_delta_encode_single() { + let mut data = vec![42]; + delta_encode(&mut data); + assert_eq!(data, vec![42]); + } + + #[test] + fn test_append_len_str() { + let mut buf = Vec::new(); + append_len_str(&mut buf, "hello"); + // Length 5 = 0x05, then "hello" + assert_eq!(buf, vec![5, b'h', b'e', b'l', b'l', b'o']); + } + + #[test] + fn test_varint_encoding() { + let mut buf = Vec::new(); + write_varint(&mut buf, 0); + assert_eq!(buf, [0x00]); + buf.clear(); + write_varint(&mut buf, 127); + assert_eq!(buf, [0x7f]); + buf.clear(); + write_varint(&mut buf, 128); + assert_eq!(buf, [0x80, 0x01]); + buf.clear(); + write_varint(&mut buf, 300); + assert_eq!(buf, [0xac, 0x02]); + } + + #[test] + fn test_zigzag64() { + assert_eq!(zigzag64(0), 0); + assert_eq!(zigzag64(-1), 1); + assert_eq!(zigzag64(1), 2); + assert_eq!(zigzag64(-2), 3); + assert_eq!(zigzag64(2147483647), 4294967294); + assert_eq!(zigzag64(-2147483648), 4294967295); + } + + #[test] + fn test_zigzag32() { + assert_eq!(zigzag32(0), 0); + assert_eq!(zigzag32(-1), 1); + assert_eq!(zigzag32(1), 2); + assert_eq!(zigzag32(-2), 3); + } + + #[test] + fn test_write_f32_value_little_endian() { + let mut buf = Vec::new(); + write_f32_value(&mut buf, 1.0); + // 1.0f32 = 0x3f800000; LE bytes = [0x00, 0x00, 0x80, 0x3f] + assert_eq!(buf, [0x00, 0x00, 0x80, 0x3f]); + } + + #[test] + fn test_write_f64_value_little_endian() { + let mut buf = Vec::new(); + write_f64_value(&mut buf, 1.0); + assert_eq!(buf, 1.0f64.to_le_bytes()); + } + + #[test] + fn test_writer_basic() { + let mut writer = V3Writer::new(); + + { + let mut metric = writer.write(V3MetricType::Gauge, "test.metric"); + metric.set_tags(["env:prod", "service:web"].iter().copied()); + metric.add_point(1000, 42.0).unwrap(); + metric.add_point(1010, 43.5).unwrap(); + metric.close(); + } + + let data = writer.into_columns(); + + assert_eq!(data.types.len(), 1); + assert_eq!(data.names.len(), 1); + assert_eq!(data.timestamps.len(), 2); + } + + #[test] + fn test_writer_unit() { + let mut writer = V3Writer::new(); + + { + let mut metric = writer.write(V3MetricType::Gauge, "has.unit"); + metric.set_unit("millisecond"); + metric.add_point(1000, 42.0).unwrap(); + metric.close(); + } + { + let mut metric = writer.write(V3MetricType::Gauge, "no.unit"); + metric.add_point(1000, 43.0).unwrap(); + metric.close(); + } + { + let mut metric = writer.write(V3MetricType::Gauge, "same.unit"); + metric.set_unit("millisecond"); + metric.add_point(1000, 44.0).unwrap(); + metric.close(); + } + + let data = writer.into_columns(); + + assert_eq!(data.unit_refs, vec![1, 0]); + assert_eq!(data.dict_unit_bytes, b"\x0bmillisecond"); + assert_eq!(data.types[0] & FLAG_HAS_UNIT, FLAG_HAS_UNIT); + assert_eq!(data.types[1] & FLAG_HAS_UNIT, 0); + assert_eq!(data.types[2] & FLAG_HAS_UNIT, FLAG_HAS_UNIT); + } + + #[test] + fn test_writer_multiple_metrics() { + let mut writer = V3Writer::new(); + + { + let mut m1 = writer.write(V3MetricType::Count, "metric1"); + m1.add_point(1000, 10.0).unwrap(); + m1.close(); + } + + { + let mut m2 = writer.write(V3MetricType::Rate, "metric2"); + m2.set_interval(60); + m2.add_point(2000, 20.0).unwrap(); + m2.close(); + } + + let data = writer.into_columns(); + + assert_eq!(data.types.len(), 2); + assert_eq!(data.names.len(), 2); + assert_eq!(data.intervals[0], 0); + // Second metric's interval won't be 60 directly since names is delta-encoded, + // but we can verify the structure is correct + } + + #[test] + fn test_value_compaction_zero() { + let mut writer = V3Writer::new(); + + { + let mut metric = writer.write(V3MetricType::Gauge, "zero.metric"); + metric.add_point(1000, 0.0).unwrap(); + metric.add_point(2000, 0.0).unwrap(); + metric.close(); + } + + let data = writer.into_columns(); + + // Values should be compacted - zero values don't need storage + assert!(data.vals_float64.is_empty()); + assert!(data.vals_sint64.is_empty()); + assert!(data.vals_float32.is_empty()); + } + + #[test] + fn test_value_compaction_int() { + let mut writer = V3Writer::new(); + + { + let mut metric = writer.write(V3MetricType::Count, "int.metric"); + metric.add_point(1000, 100.0).unwrap(); + metric.add_point(2000, 200.0).unwrap(); + metric.close(); + } + + let data = writer.into_columns(); + + // Integer values should be stored in sint64 + assert!(data.vals_float64.is_empty()); + assert_eq!(data.vals_sint64, vec![100, 200]); + assert!(data.vals_float32.is_empty()); + } + + #[test] + fn test_serialize_empty() { + let writer = V3Writer::new(); + let encoded = writer.finalize(); + assert!(encoded.payload.is_empty()); + } + + #[test] + fn test_value_compaction_large_int_plus_float32() { + // Regression test: a large integer (> 2^24) mixed with a fractional + // float32 value must use Float64, not Float32, to avoid precision loss. + let mut writer = V3Writer::new(); + + { + let mut metric = writer.write(V3MetricType::Gauge, "mixed.metric"); + metric.add_point(1000, (1i64 << 30) as f64).unwrap(); // large int, doesn't fit in f32 + metric.add_point(2000, 1.5).unwrap(); // fractional, fits in f32 + metric.close(); + } + + let data = writer.into_columns(); + + // Must be stored in float64, not float32 + assert!( + data.vals_float32.is_empty(), + "large int should not be stored as float32" + ); + assert_eq!(data.vals_float64, vec![(1i64 << 30) as f64, 1.5]); + assert!(data.vals_sint64.is_empty()); + } + + #[test] + fn test_value_compaction_small_int_plus_float32() { + // Small integers (|v| <= 2^24) mixed with float32 values should + // compact to Float32, since small ints fit losslessly in f32. + let mut writer = V3Writer::new(); + + { + let mut metric = writer.write(V3MetricType::Gauge, "small.mixed"); + metric.add_point(1000, 100.0).unwrap(); + metric.add_point(2000, 1.5).unwrap(); + metric.close(); + } + + let data = writer.into_columns(); + + assert!(data.vals_float64.is_empty()); + assert_eq!(data.vals_float32, vec![100.0, 1.5]); + assert!(data.vals_sint64.is_empty()); + } + + #[test] + fn test_serialize_basic_metric() { + let mut writer = V3Writer::new(); + + { + let mut metric = writer.write(V3MetricType::Gauge, "test.metric"); + metric.add_point(1000, 42.0).unwrap(); + metric.close(); + } + + let encoded = writer.finalize(); + + // Should produce non-empty output + assert!(!encoded.payload.is_empty()); + assert_eq!(encoded.stats.value_encoding_stats.sint64, 1); + } + + #[test] + fn test_column_stats_for_bytes_column_use_raw_column_stream() { + let mut writer = V3Writer::new(); + + { + let mut metric = writer.write(V3MetricType::Gauge, "test.metric"); + metric.add_point(1000, 42.0).unwrap(); + metric.close(); + } + + let encoded = writer.finalize(); + let name_column = encoded + .stats + .columns + .iter() + .find(|column| column.field_number == DICT_NAME_STR_FIELD_NUMBER) + .expect("name dictionary column should be present"); + + let mut expected = Vec::new(); + append_len_str(&mut expected, "test.metric"); + assert_eq!(name_column.bytes, expected); + } + + #[test] + fn test_column_stats_for_packed_column_use_raw_column_stream() { + let mut writer = V3Writer::new(); + + { + let mut metric = writer.write(V3MetricType::Gauge, "test.metric"); + metric.add_point(1000, 42.0).unwrap(); + metric.close(); + } + + let encoded = writer.finalize(); + let timestamps_column = encoded + .stats + .columns + .iter() + .find(|column| column.field_number == TIMESTAMPS_FIELD_NUMBER) + .expect("timestamp column should be present"); + + let mut expected = Vec::new(); + write_sint64_value(&mut expected, 1000); + assert_eq!(timestamps_column.bytes, expected); + } + + #[test] + fn test_column_stats_do_not_include_absent_columns() { + let mut writer = V3Writer::new(); + + { + let mut metric = writer.write(V3MetricType::Gauge, "test.metric"); + metric.add_point(1000, 42.0).unwrap(); + metric.close(); + } + + let encoded = writer.finalize(); + assert!(!encoded + .stats + .columns + .iter() + .any(|column| column.field_number == UNIT_REFS_FIELD_NUMBER)); + } + + #[test] + fn test_set_origin_clears_no_index_flag_when_reset() { + let mut writer = V3Writer::new(); + + { + let mut metric = writer.write(V3MetricType::Gauge, "origin.metric"); + metric.set_origin(1, 2, 3, true); + metric.set_origin(1, 2, 3, false); + metric.add_point(1000, 1.0).unwrap(); + metric.close(); + } + + let data = writer.into_columns(); + assert_eq!(data.types[0] & FLAG_NO_INDEX, 0); + } + + #[test] + fn test_builder_finalizes_on_drop_without_close() { + let mut writer = V3Writer::new(); + + { + let mut metric = writer.write(V3MetricType::Gauge, "dropped.metric"); + metric.add_point(1000, 42.0).unwrap(); + // Deliberately not calling `close`; `Drop` must finalize the metric anyway. + } + + let data = writer.into_columns(); + assert_eq!(data.vals_sint64, vec![42]); + assert!(data.vals_float64.is_empty()); + assert_eq!(data.types[0] & 0x30, V3ValueType::Sint64.as_u64()); + } + + #[test] + fn test_add_point_rejected_on_sketch_metric() { + let mut writer = V3Writer::new(); + let mut metric = writer.write(V3MetricType::Sketch, "wrong.method"); + assert_eq!( + metric.add_point(1000, 1.0), + Err(V3WriterError::PointOnSketchMetric) + ); + } + + #[test] + fn test_add_sketch_rejected_on_non_sketch_metric() { + let mut writer = V3Writer::new(); + let mut metric = writer.write(V3MetricType::Gauge, "wrong.method"); + assert_eq!( + metric.add_sketch(1000, 1, 1.0, 1.0, 1.0, &[0], &[1]), + Err(V3WriterError::SketchOnNonSketchMetric) + ); + } + + #[test] + fn test_add_sketch_rejected_on_bin_length_mismatch() { + let mut writer = V3Writer::new(); + let mut metric = writer.write(V3MetricType::Sketch, "mismatched.bins"); + assert_eq!( + metric.add_sketch(1000, 1, 1.0, 1.0, 1.0, &[0, 1], &[1]), + Err(V3WriterError::SketchBinLengthMismatch { + bin_keys_len: 2, + bin_counts_len: 1, + }) + ); + } +} diff --git a/libdd-metrics-v3/tests/parity.rs b/libdd-metrics-v3/tests/parity.rs new file mode 100644 index 0000000000..ff2ee729a8 --- /dev/null +++ b/libdd-metrics-v3/tests/parity.rs @@ -0,0 +1,836 @@ +// Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +//! Wire-format parity tests. +//! +//! `libdd-metrics-v3` hand-rolls its own protobuf wire-format encoder (see `src/writer.rs`) +//! instead of depending on a Protocol Buffers library for efficiency purposes and to keep this +//! crate `no_std`. We check the encoder agains generated bindings in two dimensions: +//! +//! - [`assert_wire_parity`] checks *wire framing*: it decodes the hand-rolled encoder's bytes with +//! `prost`'s generated decoder and checks the result is exactly the [`V3EncodedData`] columns we +//! intended to encode. +//! - [`assert_points_round_trip`] checks *columnar correctness*: it decodes point/sketch values +//! straight from the wire bytes' raw `types`/`timestamps`/`vals_*`/`sketch_*` arrays using its +//! own delta-decoding and value-type dispatch that shares no code with `V3Writer`, and compares +//! the result against the exact values that were passed to `add_point`/`add_sketch`. + +// To check the encoder's value-type compaction boundaries +#![allow(clippy::cast_precision_loss)] +// For reference decoder check +#![allow(clippy::expect_used)] + +mod pb; + +use libdd_metrics_v3::{V3EncodedData, V3MetricType, V3Writer}; +use prost::Message as _; + +/// Builds a [`pb::MetricData`] holding the exact same columnar data as `data`. +fn to_reference_message(data: V3EncodedData) -> pb::MetricData { + pb::MetricData { + dict_name_str: data.dict_name_bytes, + dict_tag_str: data.dict_tags_bytes, + dict_tagsets: data.dict_tagsets, + dict_resource_str: data.dict_resource_str_bytes, + dict_resource_len: data.dict_resource_len, + dict_resource_type: data.dict_resource_type, + dict_resource_name: data.dict_resource_name, + dict_source_type_name: data.dict_source_type_bytes, + dict_origin_info: data.dict_origin_info, + dict_unit_str: data.dict_unit_bytes, + types: data.types, + name_refs: data.names, + tagset_refs: data.tags, + resources_refs: data.resources, + intervals: data.intervals, + num_points: data.num_points, + source_type_name_refs: data.source_type_names, + origin_info_refs: data.origin_infos, + unit_refs: data.unit_refs, + timestamps: data.timestamps, + vals_sint64: data.vals_sint64, + vals_float32: data.vals_float32, + vals_float64: data.vals_float64, + sketch_num_bins: data.sketch_num_bins, + sketch_bin_keys: data.sketch_bin_keys, + sketch_bin_cnts: data.sketch_bin_cnts, + } +} + +/// Runs `build` against two independent [`V3Writer`]s: one is encoded with the hand-rolled encoder +/// under test, the other becomes the expected [`pb::MetricData`] via [`to_reference_message`]. +/// Asserts that decoding the hand-rolled bytes with `prost`'s generated decoder reproduces the +/// expected message exactly, and that the two encoded lengths match (see the module docs for why +/// this checks decoded equality plus length rather than raw byte equality). +#[track_caller] +fn assert_wire_parity(build: impl Fn(&mut V3Writer)) { + let mut ours = V3Writer::new(); + build(&mut ours); + let ours_bytes = ours.finalize().payload; + + let mut reference = V3Writer::new(); + build(&mut reference); + let expected = to_reference_message(reference.into_columns()); + + let decoded = pb::MetricData::decode(ours_bytes.as_slice()) + .expect("hand-rolled encoder output must be valid protobuf wire format"); + assert_eq!( + decoded, expected, + "decoding the hand-rolled encoder's bytes must reproduce the intended message" + ); + assert_eq!( + ours_bytes.len(), + expected.encoded_len(), + "hand-rolled and prost-encoded payloads must be the same length" + ); +} + +/// A point or sketch value decoded independently of `V3Writer`, identified by the bit pattern of +/// its f64 value (so NaN, which is unequal to itself under `==`, still compares correctly). +#[derive(Debug, Clone, PartialEq, Eq)] +enum DecodedPoint { + Value { + timestamp: i64, + bits: u64, + }, + Sketch { + timestamp: i64, + count: i64, + sum_bits: u64, + min_bits: u64, + max_bits: u64, + bin_keys: Vec, + bin_counts: Vec, + }, +} + +/// Decodes the point/sketch values of every metric in `data`, in writer-call order, straight from +/// the raw `types`/`timestamps`/`vals_*`/`sketch_*` columns — using this function's own +/// delta-decoding and value-type dispatch rather than anything from `src/writer.rs` or +/// `src/types.rs`. This is deliberately independent of `V3Writer` so that a bug in its +/// delta-encoding or value-type compaction changes this function's output without also changing +/// what it's compared against (see the module docs). +#[allow( + clippy::cast_sign_loss, + clippy::cast_possible_truncation, + clippy::too_many_lines, + clippy::panic +)] +fn decode_points(data: &pb::MetricData) -> Vec> { + const METRIC_TYPE_MASK: u64 = 0x0F; + const VALUE_TYPE_MASK: u64 = 0xF0; + const METRIC_TYPE_SKETCH: u64 = 4; + + fn delta_decode(values: &mut [i64]) { + for i in 1..values.len() { + values[i] += values[i - 1]; + } + } + + fn delta_decode_i32(values: &mut [i32]) { + for i in 1..values.len() { + values[i] += values[i - 1]; + } + } + + fn read_value( + data: &pb::MetricData, + value_type: u64, + sint_cursor: &mut usize, + f32_cursor: &mut usize, + f64_cursor: &mut usize, + ) -> f64 { + match value_type { + 0x00 => 0.0, + 0x10 => { + let v = data.vals_sint64[*sint_cursor]; + *sint_cursor += 1; + v as f64 + } + 0x20 => { + let v = data.vals_float32[*f32_cursor]; + *f32_cursor += 1; + f64::from(v) + } + 0x30 => { + let v = data.vals_float64[*f64_cursor]; + *f64_cursor += 1; + v + } + other => panic!("unknown v3 value type {other:#x}"), + } + } + + let mut timestamps = data.timestamps.clone(); + delta_decode(&mut timestamps); + let mut sketch_bin_keys = data.sketch_bin_keys.clone(); + + let (mut ts_cursor, mut sint_cursor, mut f32_cursor, mut f64_cursor) = (0, 0, 0, 0); + let (mut sketch_point_cursor, mut sketch_key_cursor, mut sketch_cnt_cursor) = (0, 0, 0); + + data.types + .iter() + .enumerate() + .map(|(i, &type_field)| { + let metric_type = type_field & METRIC_TYPE_MASK; + let value_type = type_field & VALUE_TYPE_MASK; + let num_points = data.num_points[i] as usize; + + (0..num_points) + .map(|_| { + let timestamp = timestamps[ts_cursor]; + ts_cursor += 1; + + if metric_type == METRIC_TYPE_SKETCH { + let sum = read_value( + data, + value_type, + &mut sint_cursor, + &mut f32_cursor, + &mut f64_cursor, + ); + let min = read_value( + data, + value_type, + &mut sint_cursor, + &mut f32_cursor, + &mut f64_cursor, + ); + let max = read_value( + data, + value_type, + &mut sint_cursor, + &mut f32_cursor, + &mut f64_cursor, + ); + let count = data.vals_sint64[sint_cursor]; + sint_cursor += 1; + + let num_bins = data.sketch_num_bins[sketch_point_cursor] as usize; + sketch_point_cursor += 1; + + let start = sketch_key_cursor; + delta_decode_i32(&mut sketch_bin_keys[start..start + num_bins]); + let bin_keys = sketch_bin_keys[start..start + num_bins].to_vec(); + sketch_key_cursor += num_bins; + + let bin_counts = data.sketch_bin_cnts + [sketch_cnt_cursor..sketch_cnt_cursor + num_bins] + .to_vec(); + sketch_cnt_cursor += num_bins; + + DecodedPoint::Sketch { + timestamp, + count, + sum_bits: sum.to_bits(), + min_bits: min.to_bits(), + max_bits: max.to_bits(), + bin_keys, + bin_counts, + } + } else { + let value = read_value( + data, + value_type, + &mut sint_cursor, + &mut f32_cursor, + &mut f64_cursor, + ); + DecodedPoint::Value { + timestamp, + bits: value.to_bits(), + } + } + }) + .collect() + }) + .collect() +} + +/// Runs `build` against a [`V3Writer`], and checks that independently decoding the resulting +/// hand-rolled bytes (via [`decode_points`]) reproduces exactly `expected_points` — the values +/// `build` is expected to have passed to `add_point`/`add_sketch`, in the same per-metric, +/// per-point order. Unlike [`assert_wire_parity`], this does not go through `V3Writer` on the +/// "expected" side at all, so it actually exercises delta-encoding and value-type compaction (see +/// the module docs). +#[track_caller] +fn assert_points_round_trip(build: impl Fn(&mut V3Writer), expected_points: &[Vec]) { + let mut writer = V3Writer::new(); + build(&mut writer); + let bytes = writer.finalize().payload; + + let message = pb::MetricData::decode(bytes.as_slice()) + .expect("hand-rolled encoder output must be valid protobuf"); + let decoded = decode_points(&message); + + assert_eq!( + decoded, expected_points, + "decoding the hand-rolled encoder's bytes must reproduce the original point values" + ); +} + +#[test] +fn empty_payload() { + assert_wire_parity(|_writer| {}); +} + +#[test] +fn single_gauge_zero_value() { + let build = |writer: &mut V3Writer| { + let mut m = writer.write(V3MetricType::Gauge, "zero.metric"); + m.add_point(1_000, 0.0).unwrap(); + m.close(); + }; + assert_wire_parity(build); + assert_points_round_trip( + build, + &[vec![DecodedPoint::Value { + timestamp: 1_000, + bits: 0.0_f64.to_bits(), + }]], + ); +} + +#[test] +fn single_count_small_int_value() { + let build = |writer: &mut V3Writer| { + let mut m = writer.write(V3MetricType::Count, "small.int"); + m.add_point(1_000, 42.0).unwrap(); + m.close(); + }; + assert_wire_parity(build); + assert_points_round_trip( + build, + &[vec![DecodedPoint::Value { + timestamp: 1_000, + bits: 42.0_f64.to_bits(), + }]], + ); +} + +#[test] +fn single_gauge_large_int_value() { + // Larger than 2^24 but still losslessly representable as sint64. + let value = (1i64 << 40) as f64; + let build = move |writer: &mut V3Writer| { + let mut m = writer.write(V3MetricType::Gauge, "large.int"); + m.add_point(1_000, value).unwrap(); + m.close(); + }; + assert_wire_parity(build); + assert_points_round_trip( + build, + &[vec![DecodedPoint::Value { + timestamp: 1_000, + bits: value.to_bits(), + }]], + ); +} + +#[test] +fn single_gauge_float32_value() { + let build = |writer: &mut V3Writer| { + let mut m = writer.write(V3MetricType::Gauge, "float32.metric"); + m.add_point(1_000, 1.5).unwrap(); + m.close(); + }; + assert_wire_parity(build); + assert_points_round_trip( + build, + &[vec![DecodedPoint::Value { + timestamp: 1_000, + bits: 1.5_f64.to_bits(), + }]], + ); +} + +#[test] +fn single_gauge_float64_value() { + let build = |writer: &mut V3Writer| { + let mut m = writer.write(V3MetricType::Gauge, "float64.metric"); + m.add_point(1_000, core::f64::consts::PI).unwrap(); + m.close(); + }; + assert_wire_parity(build); + assert_points_round_trip( + build, + &[vec![DecodedPoint::Value { + timestamp: 1_000, + bits: core::f64::consts::PI.to_bits(), + }]], + ); +} + +#[test] +fn nan_value_round_trips_as_float64() { + // NaN is a legitimate (if unusual) point value: producers can submit it, and this crate must + // encode it deterministically rather than panicking or silently substituting another value. + // `assert_wire_parity` can't express this case: `MetricData`'s derived `PartialEq` (like IEEE + // 754 itself) treats NaN as unequal to itself, so it would fail even on a correct encoding. + // `assert_points_round_trip` compares bit patterns instead, so it can actually check this. + let build = |writer: &mut V3Writer| { + let mut m = writer.write(V3MetricType::Gauge, "nan.metric"); + m.add_point(1_000, f64::NAN).unwrap(); + m.close(); + }; + assert_points_round_trip( + build, + &[vec![DecodedPoint::Value { + timestamp: 1_000, + bits: f64::NAN.to_bits(), + }]], + ); +} + +#[test] +fn mixed_large_int_and_float32_promotes_to_float64() { + // Regression case: a large integer mixed with a fractional float32 value must be stored (and + // therefore wire-encoded) as float64 to avoid precision loss. `large` is deliberately not a + // power of two: 2^30 itself happens to survive an f32 round-trip losslessly (only the + // exponent is used, the mantissa is all zero), so it wouldn't actually detect a regression + // where this promotion is missing and everything gets compacted to float32 instead. 2^30 + 1 + // needs 31 significant bits and does not fit in f32's 24-bit mantissa, so a missing promotion + // would visibly corrupt this value. + let large = ((1i64 << 30) + 1) as f64; + let build = move |writer: &mut V3Writer| { + let mut m = writer.write(V3MetricType::Gauge, "mixed.metric"); + m.add_point(1_000, large).unwrap(); + m.add_point(2_000, 1.5).unwrap(); + m.close(); + }; + assert_wire_parity(build); + assert_points_round_trip( + build, + &[vec![ + DecodedPoint::Value { + timestamp: 1_000, + bits: large.to_bits(), + }, + DecodedPoint::Value { + timestamp: 2_000, + bits: 1.5_f64.to_bits(), + }, + ]], + ); +} + +#[test] +fn rate_metric_with_interval() { + let build = |writer: &mut V3Writer| { + let mut m = writer.write(V3MetricType::Rate, "rate.metric"); + m.set_interval(60); + m.add_point(1_000, 3.5).unwrap(); + m.close(); + }; + assert_wire_parity(build); + assert_points_round_trip( + build, + &[vec![DecodedPoint::Value { + timestamp: 1_000, + bits: 3.5_f64.to_bits(), + }]], + ); +} + +#[test] +fn multiple_points_per_metric() { + let build = |writer: &mut V3Writer| { + let mut m = writer.write(V3MetricType::Gauge, "multi.point"); + for i in 0..10 { + m.add_point(1_000 + i * 10, i as f64).unwrap(); + } + m.close(); + }; + assert_wire_parity(build); + assert_points_round_trip( + build, + &[(0..10) + .map(|i| DecodedPoint::Value { + timestamp: 1_000 + i * 10, + bits: (i as f64).to_bits(), + }) + .collect()], + ); +} + +#[test] +fn multiple_metrics_share_interned_name() { + assert_wire_parity(|writer| { + for i in 0u32..3 { + let mut m = writer.write(V3MetricType::Count, "shared.name"); + m.add_point(1_000 + i64::from(i), f64::from(i)).unwrap(); + m.close(); + } + }); +} + +#[test] +fn tags_are_deduplicated_across_metrics() { + assert_wire_parity(|writer| { + { + let mut m = writer.write(V3MetricType::Gauge, "a"); + m.set_tags(["env:prod", "service:web"].into_iter()); + m.add_point(1_000, 1.0).unwrap(); + m.close(); + } + { + // Overlapping tag (env:prod) plus a new one; exercises both dictionary reuse and a + // brand-new tagset. + let mut m = writer.write(V3MetricType::Gauge, "b"); + m.set_tags(["env:prod", "service:api"].into_iter()); + m.add_point(2_000, 2.0).unwrap(); + m.close(); + } + { + // Exact same tagset as the first metric; exercises tagset (not just tag) dedup. + let mut m = writer.write(V3MetricType::Gauge, "c"); + m.set_tags(["env:prod", "service:web"].into_iter()); + m.add_point(3_000, 3.0).unwrap(); + m.close(); + } + { + let mut m = writer.write(V3MetricType::Gauge, "no.tags"); + m.add_point(4_000, 4.0).unwrap(); + m.close(); + } + }); +} + +#[test] +fn resources_host_and_device_pairs() { + assert_wire_parity(|writer| { + { + let mut m = writer.write(V3MetricType::Gauge, "with.resources"); + m.set_resources(&[("host", "server-1"), ("device", "eth0")]); + m.add_point(1_000, 1.0).unwrap(); + m.close(); + } + { + // Same resource set again: exercises resource-set dedup. + let mut m = writer.write(V3MetricType::Gauge, "same.resources"); + m.set_resources(&[("host", "server-1"), ("device", "eth0")]); + m.add_point(2_000, 2.0).unwrap(); + m.close(); + } + { + let mut m = writer.write(V3MetricType::Gauge, "no.resources"); + m.add_point(3_000, 3.0).unwrap(); + m.close(); + } + }); +} + +#[test] +fn source_type_name_is_encoded() { + assert_wire_parity(|writer| { + let mut m = writer.write(V3MetricType::Count, "with.source.type"); + m.set_source_type("nginx"); + m.add_point(1_000, 1.0).unwrap(); + m.close(); + }); +} + +#[test] +fn origin_metadata_with_no_index_flag() { + assert_wire_parity(|writer| { + let mut m = writer.write(V3MetricType::Gauge, "with.origin"); + m.set_origin(1, 2, 3, true); + m.add_point(1_000, 1.0).unwrap(); + m.close(); + }); +} + +#[test] +fn unit_toggled_on_off_on() { + assert_wire_parity(|writer| { + { + let mut m = writer.write(V3MetricType::Gauge, "has.unit"); + m.set_unit("millisecond"); + m.add_point(1_000, 42.0).unwrap(); + m.close(); + } + { + let mut m = writer.write(V3MetricType::Gauge, "no.unit"); + m.add_point(1_000, 43.0).unwrap(); + m.close(); + } + { + // Reuses the "millisecond" unit dictionary entry. + let mut m = writer.write(V3MetricType::Gauge, "same.unit"); + m.set_unit("millisecond"); + m.add_point(1_000, 44.0).unwrap(); + m.close(); + } + }); +} + +#[test] +fn unit_set_then_cleared() { + assert_wire_parity(|writer| { + let mut m = writer.write(V3MetricType::Gauge, "cleared.unit"); + m.set_unit("byte"); + m.set_unit(""); // clears it back out + m.add_point(1_000, 1.0).unwrap(); + m.close(); + }); +} + +#[test] +fn sketch_with_integer_summary() { + let build = |writer: &mut V3Writer| { + let mut m = writer.write(V3MetricType::Sketch, "sketch.int"); + m.add_sketch( + 1_000, + 5, + 15.0, + 1.0, + 9.0, + &[-2, -1, 0, 1, 2], + &[1, 1, 1, 1, 1], + ) + .unwrap(); + m.close(); + }; + assert_wire_parity(build); + assert_points_round_trip( + build, + &[vec![DecodedPoint::Sketch { + timestamp: 1_000, + count: 5, + sum_bits: 15.0_f64.to_bits(), + min_bits: 1.0_f64.to_bits(), + max_bits: 9.0_f64.to_bits(), + bin_keys: vec![-2, -1, 0, 1, 2], + bin_counts: vec![1, 1, 1, 1, 1], + }]], + ); +} + +#[test] +fn sketch_with_float_summary() { + let build = |writer: &mut V3Writer| { + let mut m = writer.write(V3MetricType::Sketch, "sketch.float"); + m.add_sketch( + 1_000, + 3, + 4.5, + 0.5, + core::f64::consts::E, + &[-1, 0, 1], + &[2, 3, 1], + ) + .unwrap(); + m.close(); + }; + assert_wire_parity(build); + assert_points_round_trip( + build, + &[vec![DecodedPoint::Sketch { + timestamp: 1_000, + count: 3, + sum_bits: 4.5_f64.to_bits(), + min_bits: 0.5_f64.to_bits(), + max_bits: core::f64::consts::E.to_bits(), + bin_keys: vec![-1, 0, 1], + bin_counts: vec![2, 3, 1], + }]], + ); +} + +#[test] +fn sketch_with_multiple_points() { + let build = |writer: &mut V3Writer| { + let mut m = writer.write(V3MetricType::Sketch, "sketch.multi"); + m.add_sketch(1_000, 2, 3.0, 1.0, 2.0, &[0, 1], &[1, 1]) + .unwrap(); + m.add_sketch(2_000, 4, 20.0, 1.0, 15.0, &[-3, -2, -1, 0], &[1, 1, 1, 1]) + .unwrap(); + m.close(); + }; + assert_wire_parity(build); + assert_points_round_trip( + build, + &[vec![ + DecodedPoint::Sketch { + timestamp: 1_000, + count: 2, + sum_bits: 3.0_f64.to_bits(), + min_bits: 1.0_f64.to_bits(), + max_bits: 2.0_f64.to_bits(), + bin_keys: vec![0, 1], + bin_counts: vec![1, 1], + }, + DecodedPoint::Sketch { + timestamp: 2_000, + count: 4, + sum_bits: 20.0_f64.to_bits(), + min_bits: 1.0_f64.to_bits(), + max_bits: 15.0_f64.to_bits(), + bin_keys: vec![-3, -2, -1, 0], + bin_counts: vec![1, 1, 1, 1], + }, + ]], + ); +} + +#[test] +fn kitchen_sink_payload() { + // Combines every dimension above into one payload: multiple metric types, shared and + // divergent tag/resource/unit/origin/source-type dictionaries, every value-type compaction + // path, and both point and sketch metrics. + assert_wire_parity(|writer| { + { + let mut m = writer.write(V3MetricType::Count, "requests.count"); + m.set_tags(["env:prod", "service:web", "region:us-east"].into_iter()); + m.set_resources(&[("host", "server-1")]); + m.set_source_type("nginx"); + m.add_point(1_000, 0.0).unwrap(); + m.add_point(1_010, 12.0).unwrap(); + m.close(); + } + { + let mut m = writer.write(V3MetricType::Rate, "requests.rate"); + m.set_tags(["env:prod", "service:web"].into_iter()); + m.set_interval(10); + m.set_unit("request"); + m.add_point(1_000, 1.2).unwrap(); + m.close(); + } + { + let mut m = writer.write(V3MetricType::Gauge, "memory.usage"); + m.set_tags(["env:prod", "service:api"].into_iter()); + m.set_resources(&[("host", "server-1"), ("container", "abc123")]); + m.set_unit("byte"); + m.set_origin(7, 2, 1, false); + m.add_point(1_000, (1i64 << 32) as f64).unwrap(); + m.close(); + } + { + let mut m = writer.write(V3MetricType::Gauge, "cpu.usage"); + m.set_tags(core::iter::once("env:staging")); + m.set_origin(7, 2, 1, true); // reuses origin dict entry, sets no-index flag + m.add_point(1_000, 0.42).unwrap(); + m.add_point(1_010, (1i64 << 30) as f64).unwrap(); // forces float64 alongside a fraction below + m.add_point(1_020, 0.5).unwrap(); + m.close(); + } + { + let mut m = writer.write(V3MetricType::Sketch, "latency.distribution"); + m.set_tags(["env:prod", "service:web"].into_iter()); // reuses an earlier tagset + m.set_unit("millisecond"); + m.add_sketch( + 1_000, + 5, + 25.0, + 1.0, + 12.0, + &[-2, -1, 0, 1, 2], + &[1, 1, 1, 1, 1], + ) + .unwrap(); + m.add_sketch( + 2_000, + 3, + 4.5, + 0.5, + core::f64::consts::E, + &[-1, 0, 1], + &[2, 3, 1], + ) + .unwrap(); + m.close(); + } + { + // No tags, no resources, no unit, no origin, no source type: exercises every "0 = + // empty" dictionary reference path in the same payload as everything above. + let mut m = writer.write(V3MetricType::Gauge, "bare.metric"); + m.add_point(1_000, 99.0).unwrap(); + m.close(); + } + }); +} + +/// Randomized coverage of the writer/encoder's most intricate logic: name and tag interning +/// (dictionary reuse across metrics), delta encoding of the resulting reference columns, and +/// value-type compaction across the zero/int24/int48/float32/float64 boundaries. Each generated +/// case is checked for the same byte-for-byte parity as the examples above. +#[test] +fn wire_bytes_match_prost_for_randomized_metrics() { + use bolero::TypeGenerator as _; + + const NAME_POOL: &[&str] = &["requests", "latency", "errors", "cpu.usage", "memory.usage"]; + const TAG_POOL: &[&str] = &[ + "env:prod", + "env:staging", + "service:web", + "service:api", + "region:us-east", + ]; + + let metric_type_idx = 0u8..=2; // Count, Rate, Gauge (sketches are covered by dedicated tests above) + let name_idx = 0usize..NAME_POOL.len(); + let tag_idx = 0usize..TAG_POOL.len(); + let tags = Vec::::produce().with().values(tag_idx); + let values = Vec::::produce(); + let metrics = Vec::<(u8, usize, Vec, Vec)>::produce() + .with() + .values((metric_type_idx, name_idx, tags, values)); + + bolero::check!() + .with_generator(metrics) + .for_each(|metrics| { + // NaN is scrubbed out up front (and shared between the writer and the expectations + // below): it's encoded deterministically like any other value, but `MetricData`'s + // derived `PartialEq` (like IEEE 754 itself) treats NaN as unequal to itself, which + // would make the equality checks below spuriously fail on a + // correctly-encoded payload. + let metrics: Vec<(u8, usize, Vec, Vec)> = metrics + .iter() + .map(|(type_idx, name_idx, tag_idxs, values)| { + let values = values + .iter() + .map(|v| if v.is_nan() { 0.0 } else { *v }) + .collect(); + (*type_idx, *name_idx, tag_idxs.clone(), values) + }) + .collect(); + + let build = |writer: &mut V3Writer| { + for (type_idx, name_idx, tag_idxs, values) in &metrics { + let metric_type = match type_idx % 3 { + 0 => V3MetricType::Count, + 1 => V3MetricType::Rate, + _ => V3MetricType::Gauge, + }; + let mut m = writer.write(metric_type, NAME_POOL[*name_idx]); + m.set_tags(tag_idxs.iter().map(|&i| TAG_POOL[i])); + m.set_interval(7); + for (i, &value) in values.iter().enumerate() { + // Non-negative, monotonically increasing timestamps (`i` is bounded by the + // generated `Vec`'s length, nowhere near overflowing); the actual values + // are what's under test here. + #[allow(clippy::cast_possible_wrap)] + m.add_point(1_000 + i as i64, value).unwrap(); + } + m.close(); + } + }; + assert_wire_parity(build); + + let expected_points: Vec> = metrics + .iter() + .map(|(_, _, _, values)| { + values + .iter() + .enumerate() + .map(|(i, value)| DecodedPoint::Value { + #[allow(clippy::cast_possible_wrap)] + timestamp: 1_000 + i as i64, + bits: value.to_bits(), + }) + .collect() + }) + .collect(); + assert_points_round_trip(build, &expected_points); + }); +} diff --git a/libdd-metrics-v3/tests/pb/mod.rs b/libdd-metrics-v3/tests/pb/mod.rs new file mode 100644 index 0000000000..769a9ca92b --- /dev/null +++ b/libdd-metrics-v3/tests/pb/mod.rs @@ -0,0 +1,218 @@ +// Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +// This file is @generated by prost-build from `proto/intake_v3.proto`. Do not edit it directly: +// regenerate it with `cargo build -p libdd-metrics-v3 --features generate-protobuf` after +// changing the vendored .proto file. +#![allow(dead_code, clippy::all, clippy::pedantic, clippy::nursery)] + +// This file is @generated by prost-build. +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct Payload { + #[prost(message, optional, tag = "2")] + pub metadata: ::core::option::Option, + #[prost(message, optional, tag = "3")] + pub metric_data: ::core::option::Option, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct Metadata { + #[prost(string, repeated, tag = "1")] + pub tags: ::prost::alloc::vec::Vec<::prost::alloc::string::String>, + /// even number of elements, \[Type, Name\] pairs + #[prost(string, repeated, tag = "2")] + pub resources: ::prost::alloc::vec::Vec<::prost::alloc::string::String>, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct MetricData { + /// Dictionaries + /// All dictionary indexes are base-1, zero implicitly represents an empty value. + /// + /// varint length + value + #[prost(bytes = "vec", tag = "1")] + pub dict_name_str: ::prost::alloc::vec::Vec, + /// varint length + value + #[prost(bytes = "vec", tag = "2")] + pub dict_tag_str: ::prost::alloc::vec::Vec, + /// length, delta encoded set of indexes into dictTagsStr + #[prost(sint64, repeated, tag = "3")] + pub dict_tagsets: ::prost::alloc::vec::Vec, + /// varint length + value + #[prost(bytes = "vec", tag = "4")] + pub dict_resource_str: ::prost::alloc::vec::Vec, + /// number of elements in Type and Name arrays + #[prost(int64, repeated, tag = "5")] + pub dict_resource_len: ::prost::alloc::vec::Vec, + /// delta encoded set of indexes into dictResourceStr + #[prost(sint64, repeated, tag = "6")] + pub dict_resource_type: ::prost::alloc::vec::Vec, + /// delta encoded set of indexes into dictResourceStr + #[prost(sint64, repeated, tag = "7")] + pub dict_resource_name: ::prost::alloc::vec::Vec, + /// varint length + value + #[prost(bytes = "vec", tag = "8")] + pub dict_source_type_name: ::prost::alloc::vec::Vec, + /// (product, category, service) tuples + #[prost(int32, repeated, tag = "9")] + pub dict_origin_info: ::prost::alloc::vec::Vec, + /// varint length + value + #[prost(bytes = "vec", tag = "25")] + pub dict_unit_str: ::prost::alloc::vec::Vec, + /// One entry per time series + /// + /// type = metricType | valueType | metricFlags + #[prost(uint64, repeated, tag = "10")] + pub types: ::prost::alloc::vec::Vec, + /// index into dictNameStr, entire array is delta encoded + #[prost(sint64, repeated, tag = "11")] + pub name_refs: ::prost::alloc::vec::Vec, + /// index into dictTagsets, entire array is delta encoded + #[prost(sint64, repeated, tag = "12")] + pub tagset_refs: ::prost::alloc::vec::Vec, + /// index into dictResourceLen, entire array is delta encoded + #[prost(sint64, repeated, tag = "13")] + pub resources_refs: ::prost::alloc::vec::Vec, + #[prost(uint64, repeated, tag = "14")] + pub intervals: ::prost::alloc::vec::Vec, + #[prost(uint64, repeated, tag = "15")] + pub num_points: ::prost::alloc::vec::Vec, + /// index into dictSourceTypeName, entire array is delta encoded + #[prost(sint64, repeated, tag = "23")] + pub source_type_name_refs: ::prost::alloc::vec::Vec, + /// index into dictOriginInfo, entire array is delta encoded + #[prost(sint64, repeated, tag = "24")] + pub origin_info_refs: ::prost::alloc::vec::Vec, + /// index into dictUnitStr, value present if flagHasUnit is set, entire array is delta encoded + #[prost(sint64, repeated, tag = "26")] + pub unit_refs: ::prost::alloc::vec::Vec, + /// each metric has numPoints values in this section + /// + /// entire array delta encoded + #[prost(sint64, repeated, tag = "16")] + pub timestamps: ::prost::alloc::vec::Vec, + /// or + #[prost(sint64, repeated, tag = "17")] + pub vals_sint64: ::prost::alloc::vec::Vec, + /// or + #[prost(float, repeated, tag = "18")] + pub vals_float32: ::prost::alloc::vec::Vec, + /// based on valueType + #[prost(double, repeated, tag = "19")] + pub vals_float64: ::prost::alloc::vec::Vec, + #[prost(uint64, repeated, tag = "20")] + pub sketch_num_bins: ::prost::alloc::vec::Vec, + /// per-metric sequence is delta encoded + #[prost(sint32, repeated, tag = "21")] + pub sketch_bin_keys: ::prost::alloc::vec::Vec, + /// sketch summary Sum, Min, Max are encoded as three consecutive elements in one of vals using valueType + /// sketch summary Cnt is always encoded in valInt64 + /// sketch summary Avg is reconstructed as Sum/Cnt in the intake + #[prost(uint32, repeated, tag = "22")] + pub sketch_bin_cnts: ::prost::alloc::vec::Vec, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct Response { + #[prost(string, tag = "1")] + pub error: ::prost::alloc::string::String, +} +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] +#[repr(i32)] +pub enum MetricType { + Unused = 0, + Count = 1, + Rate = 2, + Gauge = 3, + Sketch = 4, +} +impl MetricType { + /// String value of the enum field names used in the ProtoBuf definition. + /// + /// The values are not transformed in any way and thus are considered stable + /// (if the ProtoBuf definition does not change) and safe for programmatic use. + pub fn as_str_name(&self) -> &'static str { + match self { + Self::Unused => "UNUSED", + Self::Count => "Count", + Self::Rate => "Rate", + Self::Gauge => "Gauge", + Self::Sketch => "Sketch", + } + } + /// Creates an enum from field names used in the ProtoBuf definition. + pub fn from_str_name(value: &str) -> ::core::option::Option { + match value { + "UNUSED" => Some(Self::Unused), + "Count" => Some(Self::Count), + "Rate" => Some(Self::Rate), + "Gauge" => Some(Self::Gauge), + "Sketch" => Some(Self::Sketch), + _ => None, + } + } +} +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] +#[repr(i32)] +pub enum ValueType { + /// value is zero, not stored explicitly + Zero = 0, + /// value is stored in valsSint64 + Sint64 = 16, + /// value is stored in valsFloat32 + Float32 = 32, + /// value is stored in valsFloat64 + Float64 = 48, +} +impl ValueType { + /// String value of the enum field names used in the ProtoBuf definition. + /// + /// The values are not transformed in any way and thus are considered stable + /// (if the ProtoBuf definition does not change) and safe for programmatic use. + pub fn as_str_name(&self) -> &'static str { + match self { + Self::Zero => "Zero", + Self::Sint64 => "Sint64", + Self::Float32 => "Float32", + Self::Float64 => "Float64", + } + } + /// Creates an enum from field names used in the ProtoBuf definition. + pub fn from_str_name(value: &str) -> ::core::option::Option { + match value { + "Zero" => Some(Self::Zero), + "Sint64" => Some(Self::Sint64), + "Float32" => Some(Self::Float32), + "Float64" => Some(Self::Float64), + _ => None, + } + } +} +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] +#[repr(i32)] +pub enum MetricFlags { + FlagNone = 0, + /// metric should not be indexed (equivalent to origin metric type == agent_hidden in v2) + FlagNoIndex = 256, + /// timeseries has a unit in the unitRefs column + FlagHasUnit = 512, +} +impl MetricFlags { + /// String value of the enum field names used in the ProtoBuf definition. + /// + /// The values are not transformed in any way and thus are considered stable + /// (if the ProtoBuf definition does not change) and safe for programmatic use. + pub fn as_str_name(&self) -> &'static str { + match self { + Self::FlagNone => "flagNone", + Self::FlagNoIndex => "flagNoIndex", + Self::FlagHasUnit => "flagHasUnit", + } + } + /// Creates an enum from field names used in the ProtoBuf definition. + pub fn from_str_name(value: &str) -> ::core::option::Option { + match value { + "flagNone" => Some(Self::FlagNone), + "flagNoIndex" => Some(Self::FlagNoIndex), + "flagHasUnit" => Some(Self::FlagHasUnit), + _ => None, + } + } +} diff --git a/rustfmt.toml b/rustfmt.toml index 38a6693bd5..1732fefb98 100644 --- a/rustfmt.toml +++ b/rustfmt.toml @@ -3,6 +3,9 @@ unstable_features = true ignore = [ # embedded project waiting for upstream fixes "datadog-ipc/tarpc/", + # @generated by prost-build from a vendored .proto file that must stay byte-for-byte identical + # to its upstream source; wrapping is dictated by that source, not this repo's style + "libdd-metrics-v3/tests/pb/mod.rs", ] format_macro_matchers = true