enhancement(lading_payload): add new service topology-based OTLP traces generator

CHANGELOG.md

@@ -8,6 +8,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## Changed
 - **Breaking Change**: Rename `lost_bytes` to `missed_bytes` in `logrotate_fs`.
   This reflects preexisting terminology.
+- **Breaking Change**: Replace `opentelemetry_traces` payload with a brand-new
+  service topology-based configuration scheme. This allows for defining an 
+  imaginary distributed system -- service type, operations, attributes, etc --
+  which is then simulated to generate realistic-looking traces and spans.
 ## Added
 - Lading now supports histogram approximations in its capture files.
 - HTTP blackhole now tracks distribution of bytes received, both decoded and

ci/fingerprints/otel_traces/fingerprint.txt

@@ -1 +1 @@
-otel_traces_grpc: 7664a22561986337b61f8d807be5ce52688cfdb8b429bac89f32ea9884a17c78 entropy=6.9925
+otel_traces_grpc: 28cd77a717747057a84905d232e08c3010080f98a3f0804274a2d3b2994c1e73 entropy=6.5067

ci/fingerprints/otel_traces/lading.yaml

@@ -1,15 +1,43 @@
 generator:
-  - id: "otel_traces_grpc"
-    grpc:
-      seed: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 6]
-      target_uri: "http://127.0.0.1:4317"
-      variant:
-        opentelemetry_traces: {}
-      bytes_per_second: "100 Mb"
-      maximum_block_size: "1 Mb"
-      maximum_prebuild_cache_size_bytes: "10 Mb"
-      parallel_connections: 1
+    - id: "otel_traces_grpc"
+      grpc:
+          seed: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 6]
+          target_uri: "http://127.0.0.1:4317"
+          variant:
+              opentelemetry_traces:
+                  services:
+                      - name: api-gateway
+                        service_type: http
+                        operations:
+                            - id: get-users
+                              method: GET
+                              route: "/users/{id}"
+                              suboperations:
+                                  - to: user-service/get-user
+                      - name: user-service
+                        service_type: grpc
+                        grpc:
+                            service: UserService
+                        operations:
+                            - id: get-user
+                              method: GetUser
+                              suboperations:
+                                  - to: postgres/select-users
+                      - name: postgres
+                        service_type: database
+                        database:
+                            system: postgresql
+                            name: mydb
+                        operations:
+                            - id: select-users
+                              table: users
+                              query: "SELECT * FROM users WHERE id = $1"
+                              suboperations: []
+          bytes_per_second: "100 Mb"
+          maximum_block_size: "1 Mb"
+          maximum_prebuild_cache_size_bytes: "10 Mb"
+          parallel_connections: 1
 
 blackhole:
-  - http:
-      binding_addr: "127.0.0.1:9999"
+    - http:
+          binding_addr: "127.0.0.1:9999"

integration/sheepdog/src/lib.rs

@@ -461,7 +461,21 @@ generator:
           post:
             maximum_prebuild_cache_size_bytes: "8 MiB"
             variant:
-              opentelemetry_traces: {}
+              opentelemetry_traces:
+                services:
+                  - name: api-gateway
+                    service_type: http
+                    operations:
+                      - id: get-users
+                        method: GET
+                        route: "/users/{id}"
+                        suboperations:
+                          - to: backend/process
+                  - name: backend
+                    operations:
+                      - id: process
+                        span_name: "process"
+                        suboperations: []
         headers:
             Content-Type: "application/x-protobuf"
         "#,

lading_payload/benches/opentelemetry_traces.rs

@@ -1,30 +1,123 @@
 //! Benchmarks for OpenTelemetry trace payload generation.
 
 use criterion::{BenchmarkId, Criterion, Throughput, criterion_group};
-use lading_payload::{OpentelemetryTraces, Serialize, opentelemetry::trace::Config};
+use lading_payload::opentelemetry::trace::{
+    AttributeConfig, AttributeValueConfig, Config, DatabaseServiceConfig, GrpcServiceConfig,
+    OperationConfig, ServiceConfig, ServiceType, SuboperationConfig,
+};
+use lading_payload::{OpentelemetryTraces, Serialize};
 use rand::{SeedableRng, rngs::SmallRng};
 use std::time::Duration;
 
 const MIB: usize = 1_048_576;
 
+/// A minimal but representative topology for benchmarking.
+fn bench_config() -> Config {
+    Config {
+        services: vec![
+            ServiceConfig {
+                name: "api-gateway".into(),
+                service_type: Some(ServiceType::Http),
+                grpc: None,
+                database: None,
+                operations: vec![OperationConfig {
+                    id: "get-users".into(),
+                    span_name: None,
+                    method: Some("GET".into()),
+                    route: Some("/users/{id}".into()),
+                    table: None,
+                    query: None,
+                    suboperations: vec![SuboperationConfig {
+                        to: "user-service/get-user".into(),
+                        rate: 1.0,
+                        max_repeat: 1,
+                        attributes: vec![],
+                    }],
+                    attributes: vec![],
+                    root: None,
+                }],
+                resource_attributes: vec![AttributeConfig::new(
+                    "deployment.environment",
+                    AttributeValueConfig::Static("production".into()),
+                )],
+                scope_name: Some("com.example.gateway".into()),
+                scope_version: "1.0.0".into(),
+            },
+            ServiceConfig {
+                name: "user-service".into(),
+                service_type: Some(ServiceType::Grpc),
+                grpc: Some(GrpcServiceConfig {
+                    service: "UserService".into(),
+                }),
+                database: None,
+                operations: vec![OperationConfig {
+                    id: "get-user".into(),
+                    span_name: None,
+                    method: Some("GetUser".into()),
+                    route: None,
+                    table: None,
+                    query: None,
+                    suboperations: vec![SuboperationConfig {
+                        to: "postgres/select-users".into(),
+                        rate: 1.0,
+                        max_repeat: 1,
+                        attributes: vec![],
+                    }],
+                    attributes: vec![],
+                    root: None,
+                }],
+                resource_attributes: vec![],
+                scope_name: Some("com.example.users".into()),
+                scope_version: "1.0.0".into(),
+            },
+            ServiceConfig {
+                name: "postgres".into(),
+                service_type: Some(ServiceType::Database),
+                grpc: None,
+                database: Some(DatabaseServiceConfig {
+                    system: "postgresql".into(),
+                    name: Some("mydb".into()),
+                }),
+                operations: vec![OperationConfig {
+                    id: "select-users".into(),
+                    span_name: None,
+                    method: None,
+                    route: None,
+                    table: Some("users".into()),
+                    query: Some("SELECT * FROM users WHERE id = $1".into()),
+                    suboperations: vec![],
+                    attributes: vec![],
+                    root: None,
+                }],
+                resource_attributes: vec![],
+                scope_name: None,
+                scope_version: "1.0.0".into(),
+            },
+        ],
+        error_rate: 0.01,
+    }
+}
+
 fn opentelemetry_traces_setup(c: &mut Criterion) {
+    let config = bench_config();
     c.bench_function("opentelemetry_traces_setup", |b| {
         b.iter(|| {
             let mut rng = SmallRng::seed_from_u64(19_690_716);
-            let _ot = OpentelemetryTraces::with_config(Config::default(), &mut rng)
+            let _ot = OpentelemetryTraces::with_config(&config, &mut rng)
                 .expect("failed to create trace generator");
         })
     });
 }
 
 fn opentelemetry_traces_all(c: &mut Criterion) {
+    let config = bench_config();
     let mut group = c.benchmark_group("opentelemetry_traces_all");
     for size in &[MIB, 10 * MIB, 100 * MIB, 1_000 * MIB] {
         group.throughput(Throughput::Bytes(*size as u64));
         group.bench_with_input(BenchmarkId::from_parameter(size), size, |b, &size| {
             b.iter(|| {
                 let mut rng = SmallRng::seed_from_u64(19_690_716);
-                let mut ot = OpentelemetryTraces::with_config(Config::default(), &mut rng)
+                let mut ot = OpentelemetryTraces::with_config(&config, &mut rng)
                     .expect("failed to create trace generator");
                 let mut writer = Vec::with_capacity(size);
 

lading_payload/src/block.rs

@@ -355,7 +355,7 @@ impl Cache {
                 )?
             }
             crate::Config::OpentelemetryTraces(config) => {
-                let mut pyld = crate::OpentelemetryTraces::with_config(*config, &mut rng)?;
+                let mut pyld = crate::OpentelemetryTraces::with_config(config, &mut rng)?;
                 let span = span!(Level::INFO, "fixed", payload = "otel-traces");
                 let _guard = span.enter();
                 construct_block_cache_inner(rng, &mut pyld, maximum_block_bytes, total_bytes.get())?

lading_payload/src/lib.rs

@@ -145,7 +145,7 @@ pub enum Config {
     Json,
     /// Generates a Apache Common log lines
     ApacheCommon,
-    /// Generates OpenTelemetry traces
+    /// Generates OpenTelemetry traces from a service topology graph
     OpentelemetryTraces(crate::opentelemetry::trace::Config),
     /// Generates OpenTelemetry logs
     OpentelemetryLogs(crate::opentelemetry::log::Config),