OpenTelemetry logs improvement

This commit updates our OTel logs payload to be configurable by the end user. We
allow in a manner similar to metrics for the contexts to be capped, attributes
per level of the message format to be set and seprately from the context
consideration we allow for generation and constraint of total trace-ids.

REF SMPTNG-659

Signed-off-by: Brian L. Troutwine <brian.troutwine@datadoghq.com>

Author: blt

CHANGELOG.md

@@ -5,6 +5,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+## Added
+- Added configuration surface area to the OTel logs payload generator, in a
+  manner similar to OTel metrics.
 
 ## [0.27.0]
 ## Added

CLAUDE.md

@@ -60,6 +60,10 @@ When handling errors:
 This project enforces code style through automated tooling. Use `ci/validate` to
 check style compliance - it will run formatting and linting checks for you.
 
+**Module organization**: Never use `mod.rs` files. Always name modules directly
+(e.g., use `foo.rs` instead of `foo/mod.rs`). This makes the codebase easier to
+navigate and grep.
+
 We do not allow for warnings: all warnings are errors. Deprecation warnings MUST
 be treated as errors. Lading is written in a "naive" style where abstraction is
 not preferred if a duplicated pattern will satisfy. Our reasoning for this is it
@@ -183,3 +187,4 @@ When in doubt, implement rather than import.
 12. Generators must be deterministic - no randomness without explicit seeding
 13. Pre-compute in initialization, not in hot paths
 14. Think about how your code affects the measurement of the target
+15. NEVER use mod.rs files - always name modules directly (e.g., foo.rs not foo/mod.rs)

integration/sheepdog/src/lib.rs

@@ -423,7 +423,8 @@ generator:
         method:
           post:
             maximum_prebuild_cache_size_bytes: "8 MiB"
-            variant: "opentelemetry_logs"
+            variant:
+              opentelemetry_logs: {}
         headers:
             Content-Type: "application/x-protobuf"
         "#,

lading_payload/proptest-regressions/opentelemetry/log/mod.txt

@@ -0,0 +1,7 @@
+# Seeds for failure cases proptest has generated in the past. It is
+# automatically read and these particular cases re-run before any
+# novel cases are generated.
+#
+# It is recommended to check this file in to source control so that
+# everyone who runs the test benefits from these saved cases.
+cc c936baccbc78a4d1a4c4e3f22ba82ca3c8b3c8e74894e7d64167d8ec10c3eb7c # shrinks to seed = 0, total_contexts = 1, steps = 1

lading_payload/proptest-regressions/opentelemetry_log.txt

@@ -0,0 +1,8 @@
+# Seeds for failure cases proptest has generated in the past. It is
+# automatically read and these particular cases re-run before any
+# novel cases are generated.
+#
+# It is recommended to check this file in to source control so that
+# everyone who runs the test benefits from these saved cases.
+cc 48f21c68f6a47b5f04cbc25d9d9907b9db94c3c1352438d61a636a1e819f5aa1 # shrinks to seed = 0, total_contexts = 1, steps = 1
+cc 07dda2b662c8e5b8971ea35f997615743e8be3c2e6604e7fe3ad676ff62a9de4 # shrinks to seed = 6751211249958810060, steps = 2, budget = 1460

lading_payload/src/block.rs

@@ -332,8 +332,15 @@ impl Cache {
                 let _guard = span.enter();
                 construct_block_cache_inner(rng, &mut pyld, maximum_block_bytes, total_bytes.get())?
             }
-            crate::Config::OpentelemetryLogs => {
-                let mut pyld = crate::OpentelemetryLogs::new(&mut rng);
+            crate::Config::OpentelemetryLogs(config) => {
+                match config.valid() {
+                    Ok(()) => (),
+                    Err(e) => {
+                        warn!("Invalid OpentelemetryLogs configuration: {}", e);
+                        return Err(Error::InvalidConfig(e));
+                    }
+                }
+                let mut pyld = crate::OpentelemetryLogs::new(*config, &mut rng)?;
                 let span = span!(Level::INFO, "fixed", payload = "otel-logs");
                 let _guard = span.enter();
                 construct_block_cache_inner(rng, &mut pyld, maximum_block_bytes, total_bytes.get())?

lading_payload/src/lib.rs

@@ -96,9 +96,14 @@ pub enum Error {
     /// See [`prost::EncodeError`]
     #[error(transparent)]
     ProstEncode(#[from] prost::EncodeError),
-    /// See [`opentelemetry_metric::templates::PoolError`]
+    /// See [`opentelemetry::common::PoolError`]
     #[error("Unable to choose from pool: {0}")]
-    Pool(#[from] opentelemetry::metric::templates::PoolError),
+    Pool(
+        #[from] opentelemetry::common::templates::PoolError<opentelemetry::common::GeneratorError>,
+    ),
+    /// Validation error
+    #[error("Validation error: {0}")]
+    Validation(String),
 }
 
 /// To serialize into bytes
@@ -172,7 +177,7 @@ pub enum Config {
     /// Generates OpenTelemetry traces
     OpentelemetryTraces,
     /// Generates OpenTelemetry logs
-    OpentelemetryLogs,
+    OpentelemetryLogs(crate::opentelemetry::log::Config),
     /// Generates OpenTelemetry metrics
     OpentelemetryMetrics(crate::opentelemetry::metric::Config),
     /// Generates `DogStatsD`

lading_payload/src/opentelemetry.rs

@@ -2,6 +2,7 @@
 //!
 //! This module contains payload generators for OpenTelemetry formats.
 
+pub mod common;
 pub mod log;
 pub mod metric;
 pub mod trace;

…payload/src/opentelemetry/metric/tags.rs …ding_payload/src/opentelemetry/common.rslading_payload/src/opentelemetry/metric/tags.rs renamed to lading_payload/src/opentelemetry/common.rs lading_payload/src/opentelemetry/metric/tags.rs renamed to lading_payload/src/opentelemetry/common.rs

@@ -1,20 +1,37 @@
-//! Tag generation for OpenTelemetry metric payloads
-use std::{cmp, rc::Rc};
+//! Common utilities and types for OpenTelemetry payload generation
+//!
+//! This module contains shared code used by both metrics and logs implementations.
+
+pub(crate) mod templates;
 
-use super::templates::GeneratorError;
-use crate::{Error, Generator, common::config::ConfRange, common::strings::Pool};
+use crate::{Error, Generator, SizedGenerator, common::config::ConfRange, common::strings::Pool};
 use opentelemetry_proto::tonic::common::v1::{AnyValue, KeyValue, any_value};
 use prost::Message;
+use std::{cmp, rc::Rc};
+
+/// Errors that can occur during generation
+#[derive(thiserror::Error, Debug, Clone, Copy)]
+pub enum GeneratorError {
+    /// Generator exhausted bytes budget prematurely
+    #[error("Generator exhausted bytes budget prematurely")]
+    SizeExhausted,
+    /// Failed to generate string
+    #[error("Failed to generate string")]
+    StringGenerate,
+}
+
+/// Ratio of unique tags to use in tag generation
+pub(crate) const UNIQUE_TAG_RATIO: f32 = 0.75;
 
+/// Smallest useful `KeyValue` protobuf, determined by experimentation and enforced in tests
+pub(crate) const SMALLEST_KV_PROTOBUF: usize = 10;
+
+/// Tag generator for OpenTelemetry attributes
 #[derive(Debug, Clone)]
 pub(crate) struct TagGenerator {
     inner: crate::common::tags::Generator,
 }
 
-// smallest useful protobuf, determined by experimentation and enforced in
-// smallest_kv_protobuf test
-const SMALLEST_KV_PROTOBUF: usize = 10;
-
 impl TagGenerator {
     /// Creates a new tag generator
     ///
@@ -43,7 +60,8 @@ impl TagGenerator {
     }
 }
 
-fn varint_len(v: usize) -> usize {
+/// Calculate the length of a varint encoding
+pub(crate) fn varint_len(v: usize) -> usize {
     let mut v = v;
     let mut n = 1;
     while v > 0x7f {
@@ -53,14 +71,15 @@ fn varint_len(v: usize) -> usize {
     n
 }
 
-fn overhead(v: usize) -> usize {
+/// Calculate the overhead for a `KeyValue` in a repeated field
+pub(crate) fn overhead(v: usize) -> usize {
     // overhead in a repeated field is per-item, so:
     //
     // [tag-byte] [varint-length] [kv-bytes…]
     varint_len(v) + 1 + v
 }
 
-impl<'a> crate::SizedGenerator<'a> for TagGenerator {
+impl<'a> SizedGenerator<'a> for TagGenerator {
     type Output = Vec<KeyValue>;
     type Error = GeneratorError;
 
@@ -143,11 +162,11 @@ mod test {
             }),
         };
 
-        let encoded_size = overhead(kv.encoded_len());
+        let sz = overhead(kv.encoded_len());
 
-        assert!(
-            encoded_size == SMALLEST_KV_PROTOBUF,
-            "Minimal useful request size ({encoded_size}) should be == SMALLEST_KV_PROTOBUF ({SMALLEST_KV_PROTOBUF})"
+        assert_eq!(
+            sz, SMALLEST_KV_PROTOBUF,
+            "Minimal useful key/value pair should have size {SMALLEST_KV_PROTOBUF}, was {sz}"
         );
     }
 }

lading_payload/src/opentelemetry/common/templates.rs

@@ -0,0 +1,91 @@
+//! Template utilities for OpenTelemetry payload generation
+
+use prost::Message;
+use rand::{Rng, prelude::*, seq::IteratorRandom};
+use std::collections::BTreeMap;
+
+/// Errors related to pool operations
+#[derive(thiserror::Error, Debug, Clone, Copy)]
+pub enum PoolError<E> {
+    /// Choice could not be made on empty container
+    #[error("Choice could not be made on empty container.")]
+    EmptyChoice,
+    /// Generation error
+    #[error("Generation error: {0}")]
+    Generator(E),
+}
+
+/// A pool that stores pre-generated instances indexed by their encoded size
+#[derive(Debug, Clone)]
+pub(crate) struct Pool<T, G> {
+    context_cap: u32,
+    /// key: encoded size; val: templates with that size
+    by_size: BTreeMap<usize, Vec<T>>,
+    generator: G,
+    len: u32,
+}
+
+impl<T, G> Pool<T, G>
+where
+    T: Message,
+{
+    /// Build an empty pool that can hold at most `context_cap` templates.
+    pub(crate) fn new(context_cap: u32, generator: G) -> Self {
+        Self {
+            context_cap,
+            by_size: BTreeMap::new(),
+            generator,
+            len: 0,
+        }
+    }
+
+    /// Return a reference to an item from the pool.
+    ///
+    /// Instances returned by this function are guaranteed to be of an encoded
+    /// size no greater than budget. No greater than `context_cap` instances
+    /// will ever be stored in this structure.
+    pub(crate) fn fetch<'a, R>(
+        &'a mut self,
+        rng: &mut R,
+        budget: &mut usize,
+    ) -> Result<&'a T, PoolError<G::Error>>
+    where
+        R: Rng + ?Sized,
+        G: crate::SizedGenerator<'a, Output = T>,
+        G::Error: 'a,
+    {
+        // If we are at context cap, search by_size for templates <= budget and
+        // return a random choice. If we are not at context cap, call
+        // generator with the budget and then store the result
+        // for future use in `by_size`.
+        //
+        // Size search is in the interval (0, budget].
+
+        let upper = *budget;
+
+        // Generate new instances until either context_cap is hit or the
+        // remaining space drops below our lookup interval.
+        if self.len < self.context_cap {
+            let mut limit = *budget;
+            match self.generator.generate(rng, &mut limit) {
+                Ok(item) => {
+                    let sz = item.encoded_len();
+                    self.by_size.entry(sz).or_default().push(item);
+                    self.len += 1;
+                }
+                Err(e) => return Err(PoolError::Generator(e)),
+            }
+        }
+
+        let (choice_sz, choices) = self
+            .by_size
+            .range(..=upper)
+            .choose(rng)
+            .ok_or(PoolError::EmptyChoice)?;
+
+        let choice = choices.choose(rng).ok_or(PoolError::EmptyChoice)?;
+        *budget = budget.saturating_sub(*choice_sz);
+
+        Ok(choice)
+    }
+}