Optimization passes for Otel metrics payload generation

Cargo.toml

@@ -47,6 +47,7 @@ lto = true        # Optimize our binary at link stage.
 codegen-units = 1 # Increases compile time but improves optimization alternatives.
 opt-level = 3     # Optimize with 'all' optimization flipped on. May produce larger binaries than 's' or 'z'.
 panic = "abort"
+debug = false
 
 [profile.dev]
 panic = "abort"

examples/lading-otel-metrics.yaml

@@ -0,0 +1,63 @@
+generator:
+  - http:
+      seed: [2, 3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 37, 41, 43, 47, 53,
+             59, 61, 67, 71, 73, 79, 83, 89, 97, 101, 103, 107, 109, 113, 127, 131]
+      headers:
+        content-type: "application/x-protobuf"
+      target_uri: "http://127.0.0.1:4318/v1/metrics"
+      bytes_per_second: "50 MiB"
+      parallel_connections: 5
+      method:
+        post:
+          maximum_prebuild_cache_size_bytes: "1 GiB"
+          variant:
+            # all numbers arbitraray
+            opentelemetry_metrics:
+              metric_weights:
+                gauge: 50
+                sum: 50
+              contexts:
+                total_contexts:
+                  constant: 100
+                # host., service. etc
+                attributes_per_resource:
+                  inclusive:
+                    min: 1
+                    max: 64
+                # auto-instrumentation in client libraries, DB connection etc
+                scopes_per_resource:
+                  inclusive:
+                    min: 1
+                    max: 32
+                # rare? build info possibly
+                attributes_per_scope:
+                  inclusive:
+                    min: 0
+                    max: 4
+                # exported instruments / telemetry by libraries and custom code
+                metrics_per_scope:
+                  inclusive:
+                    min: 1
+                    max: 128
+                # stuff like exit code, user id, cgroup id
+                attributes_per_metric:
+                  inclusive:
+                    min: 0
+                    max: 255
+
+blackhole:
+  - http:
+      binding_addr: "127.0.0.1:9091"
+  - http:
+      binding_addr: "127.0.0.1:9092"
+  - http:
+      binding_addr: "127.0.0.1:4217"  # OTLP HTTP endpoint
+      body_variant: "nothing"
+  # - grpc:
+  #     binding_addr: "127.0.0.1:4318"  # OTLP gRPC endpoint
+
+target_metrics:
+  - prometheus:
+      uri: "http://127.0.0.1:8888/metrics"  # OTel collector metrics endpoint
+      tags:
+        component: "otel-collector"

lading/Cargo.toml

@@ -43,6 +43,7 @@ http-serde = "2.1"
 hyper = { workspace = true, features = ["client", "http1", "http2", "server"] }
 hyper-util = { workspace = true, features = ["default", "client", "client-legacy"] }
 is_executable = "1.0.4"
+jemallocator = { version = "0.5" }
 metrics = { workspace = true }
 metrics-exporter-prometheus = { workspace = true }
 metrics-util = { workspace = true }

lading/src/bin/captool.rs

@@ -7,13 +7,17 @@ use async_compression::tokio::bufread::ZstdDecoder;
 use average::{Estimate, Max, Min, Variance, concatenate};
 use clap::Parser;
 use futures::io;
+use jemallocator::Jemalloc;
 use lading_capture::json::{Line, MetricKind};
 use tokio::io::{AsyncBufReadExt, BufReader};
 use tokio_stream::StreamExt;
 use tokio_stream::wrappers::LinesStream;
 use tracing::{error, info};
 use tracing_subscriber::{fmt::format::FmtSpan, util::SubscriberInitExt};
 
+#[global_allocator]
+static GLOBAL: Jemalloc = Jemalloc;
+
 #[derive(Parser, Debug)]
 #[clap(author, version, about, long_about = None)]
 struct Args {

lading/src/bin/lading.rs

@@ -8,6 +8,7 @@ use std::{
 };
 
 use clap::{ArgGroup, Parser, Subcommand};
+use jemallocator::Jemalloc;
 use lading::{
     blackhole,
     captures::CaptureManager,
@@ -32,6 +33,9 @@ use tokio::{
 use tracing::{Instrument, debug, error, info, info_span, warn};
 use tracing_subscriber::{EnvFilter, util::SubscriberInitExt};
 
+#[global_allocator]
+static GLOBAL: Jemalloc = Jemalloc;
+
 #[derive(thiserror::Error, Debug)]
 enum Error {
     #[error("Target related error: {0}")]

lading/src/bin/payloadtool.rs

@@ -4,12 +4,16 @@ use std::{io, num::NonZeroU32};
 
 use byte_unit::{Unit, UnitType};
 use clap::Parser;
+use jemallocator::Jemalloc;
 use lading::generator::http::Method;
 use lading_payload::block;
 use rand::{SeedableRng, rngs::StdRng};
-use tracing::{debug, error, info, warn};
+use tracing::{error, info, trace, warn};
 use tracing_subscriber::{fmt::format::FmtSpan, util::SubscriberInitExt};
 
+#[global_allocator]
+static GLOBAL: Jemalloc = Jemalloc;
+
 const UDP_PACKET_LIMIT_BYTES: byte_unit::Byte =
     byte_unit::Byte::from_u64_with_unit(65_507, Unit::B).expect("valid bytes");
 
@@ -51,26 +55,28 @@ fn generate_and_check(
         block::Cache::Fixed { blocks, .. } => blocks,
     };
     info!("Payload generation took {:?}", start.elapsed());
-    debug!("Payload: {:#?}", blocks);
+    trace!("Payload: {:#?}", blocks);
 
     let mut total_generated_bytes: u32 = 0;
     for block in blocks.iter() {
         total_generated_bytes += block.total_bytes.get();
     }
+    let total_requested_bytes =
+        byte_unit::Byte::from_u128(total_bytes.get().into()).expect("total_bytes must be non-zero");
+    let total_requested_bytes_str = total_requested_bytes
+        .get_appropriate_unit(UnitType::Binary)
+        .to_string();
     if total_bytes.get().abs_diff(total_generated_bytes) > 1_000_000 {
-        let total_requested_bytes = byte_unit::Byte::from_u128(total_bytes.get().into())
-            .expect("total_bytes must be non-zero");
-        let total_requested_bytes_str = total_requested_bytes
-            .get_appropriate_unit(UnitType::Binary)
-            .to_string();
         let total_generated_bytes = byte_unit::Byte::from_u128(total_generated_bytes.into())
             .expect("total_generated_bytes must be non-zero");
         let total_generated_bytes_str = total_generated_bytes
             .get_appropriate_unit(UnitType::Binary)
             .to_string();
         warn!(
-            "Generator failed to generate {total_requested_bytes_str}, instead only found {total_generated_bytes_str} of data"
+            "Generator failed to generate {total_requested_bytes_str}, producing {total_generated_bytes_str} of data"
         )
+    } else {
+        info!("Generator succeeded in generating {total_requested_bytes_str} of data")
     }
 
     Ok(())

lading/src/bin/simple_target.rs

@@ -1,6 +1,9 @@
 //! A simple target for lading that runs forever
 
+use jemallocator::Jemalloc;
 use std::{thread, time};
+#[global_allocator]
+static GLOBAL: Jemalloc = Jemalloc;
 
 pub fn main() {
     loop {

lading_payload/benches/opentelemetry_metric.rs

@@ -1,14 +1,28 @@
 use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
 
-use lading_payload::{OpentelemetryMetrics, Serialize, opentelemetry_metric::Config};
+use lading_payload::common::config::ConfRange;
+use lading_payload::{
+    OpentelemetryMetrics, Serialize,
+    opentelemetry_metric::{Config, Contexts, MetricWeights},
+};
 use rand::{SeedableRng, rngs::SmallRng};
 use std::time::Duration;
 
 fn opentelemetry_metric_setup(c: &mut Criterion) {
     c.bench_function("opentelemetry_metric_setup", |b| {
         b.iter(|| {
             let mut rng = SmallRng::seed_from_u64(19690716);
-            let config = Config::default();
+            let config = Config {
+                metric_weights: MetricWeights { gauge: 50, sum: 50 },
+                contexts: Contexts {
+                    total_contexts: ConfRange::Constant(100),
+                    attributes_per_resource: ConfRange::Inclusive { min: 1, max: 64 },
+                    scopes_per_resource: ConfRange::Inclusive { min: 1, max: 32 },
+                    attributes_per_scope: ConfRange::Inclusive { min: 0, max: 4 },
+                    metrics_per_scope: ConfRange::Inclusive { min: 1, max: 128 },
+                    attributes_per_metric: ConfRange::Inclusive { min: 0, max: 255 },
+                },
+            };
             let _ot = OpentelemetryMetrics::new(config, &mut rng)
                 .expect("failed to create metrics generator");
         })
@@ -24,7 +38,17 @@ fn opentelemetry_metric_all(c: &mut Criterion) {
         group.bench_with_input(BenchmarkId::from_parameter(size), size, |b, &size| {
             b.iter(|| {
                 let mut rng = SmallRng::seed_from_u64(19690716);
-                let config = Config::default();
+                let config = Config {
+                    metric_weights: MetricWeights { gauge: 50, sum: 50 },
+                    contexts: Contexts {
+                        total_contexts: ConfRange::Constant(100),
+                        attributes_per_resource: ConfRange::Inclusive { min: 1, max: 64 },
+                        scopes_per_resource: ConfRange::Inclusive { min: 1, max: 32 },
+                        attributes_per_scope: ConfRange::Inclusive { min: 0, max: 4 },
+                        metrics_per_scope: ConfRange::Inclusive { min: 1, max: 128 },
+                        attributes_per_metric: ConfRange::Inclusive { min: 0, max: 255 },
+                    },
+                };
                 let ot = OpentelemetryMetrics::new(config, &mut rng)
                     .expect("failed to create metrics generator");
                 let mut writer = Vec::with_capacity(size);

lading_payload/src/block.rs

@@ -489,17 +489,18 @@ where
     S: crate::Serialize,
     R: Rng + ?Sized,
 {
-    let mut block_cache: Vec<Block> = Vec::with_capacity(128);
-    let mut bytes_remaining = total_bytes;
     let mut min_block_size = 0;
     let mut max_actual_block_size = 0;
     let mut rejected_block_sizes = 0;
     let mut success_block_sizes = 0;
+
     info!(
         ?max_block_size,
         ?total_bytes,
         "Constructing requested block cache"
     );
+    let mut block_cache: Vec<Block> = Vec::with_capacity(128);
+    let mut bytes_remaining = total_bytes;
 
     let start = Instant::now();
     let mut next_minute = 1;

lading_payload/src/common.rs

@@ -1,3 +1,5 @@
-pub(crate) mod config;
+//! Common utilities for all lading payloads
+
+pub mod config;
 pub(crate) mod strings;
 pub(crate) mod tags;

lading_payload/src/common/config.rs

@@ -1,3 +1,5 @@
+//! Common configuration for all lading payloads
+
 use rand::distr::uniform::SampleUniform;
 use serde::{Deserialize, Serialize as SerdeSerialize};
 use std::cmp;

lading_payload/src/lib.rs

@@ -46,7 +46,7 @@ pub use trace_agent::TraceAgent;
 
 pub mod apache_common;
 pub mod ascii;
-pub(crate) mod common;
+pub mod common;
 pub mod datadog_logs;
 pub mod dogstatsd;
 pub mod fluent;
@@ -84,6 +84,9 @@ pub enum Error {
     /// See [`unit::Error`]
     #[error(transparent)]
     Unit(#[from] opentelemetry_metric::unit::Error),
+    /// See [`prost::EncodeError`]
+    #[error(transparent)]
+    ProstEncode(#[from] prost::EncodeError),
 }
 
 /// To serialize into bytes