test: shared fixture corpus for tests and benches

.gitignore

@@ -27,4 +27,6 @@ bottlecap/proptest-regressions
 integration-tests/cdk.context.json
 
 .gitlab/pipeline*
-/CLAUDE_PERSONAL.md
+/CLAUDE_PERSONAL.md
+# Generated benchmark fixture corpus (see bottlecap/tests/fixtures/README.md)
+bottlecap/tests/fixtures/generated

bottlecap/src/extension/telemetry/mod.rs

@@ -5,7 +5,8 @@ use crate::extension::{EXTENSION_ID_HEADER, base_url};
 
 pub mod events;
 pub mod listener;
-mod stitch;
+#[doc(hidden)]
+pub mod stitch;
 
 pub const TELEMETRY_SUBSCRIPTION_ROUTE: &str = "2022-07-01/telemetry";
 

bottlecap/src/extension/telemetry/stitch.rs

@@ -67,7 +67,7 @@ const MAX_SEAMS: usize = 64;
 
 /// What came of pairing an unparseable body with a held fragment.
 #[derive(Debug)]
-pub(crate) enum Stitch {
+pub enum Stitch {
     /// The body completed a split payload.
     Complete(Vec<TelemetryEvent>),
     /// The body opens a split payload, and is held for its continuation.
@@ -78,13 +78,13 @@ pub(crate) enum Stitch {
 
 /// Holds the leading half of a split payload until its continuation arrives.
 #[derive(Clone, Default)]
-pub(crate) struct FragmentBuffer {
+pub struct FragmentBuffer {
     held: Arc<Mutex<Option<Fragment>>>,
 }
 
 impl FragmentBuffer {
     /// Joins `body` onto the held fragment, or holds `body` if it opens a split payload.
-    pub(crate) fn stitch(&self, body: &[u8]) -> Stitch {
+    pub fn stitch(&self, body: &[u8]) -> Stitch {
         let mut slot = self.held.lock().unwrap_or_else(PoisonError::into_inner);
 
         if let Some(stale) = slot.take_if(|held| held.received.elapsed() > FRAGMENT_TTL) {

bottlecap/tests/fixtures/README.md

@@ -0,0 +1,70 @@
+# Fixture corpus
+
+Shared inputs for the criterion benches (`bottlecap/benches/`), the fault-injection harness and
+the Lambda simulator. Everything is generated by the Rust module in this directory, so no large
+blob is checked in.
+
+## Use from Rust
+
+```rust
+#[path = "../tests/fixtures/mod.rs"] // from benches/; integration tests use `mod fixtures;`
+mod fixtures;
+use fixtures::{dogstatsd, events, logs, telemetry, traces};
+```
+
+| Module | Generator | What it returns |
+|---|---|---|
+| `telemetry` | `platform_batch(request_id) -> Vec<u8>` | One cold-start invocation: `initStart`, `initRuntimeDone`, `initReport`, `start`, `runtimeDone`, `report` |
+| | `log_batch(LogKind, n) -> Vec<u8>` | One POST body of `n` function log events (`LogKind::{Text, JsonObject, JsonString}`) |
+| | `log_batches(LogKind, n, per_post) -> Vec<Vec<u8>>` | The same, split into POSTs of `per_post` events (Lambda sends at most 1000) |
+| | `invocation_batch(LogKind, n) -> Vec<u8>` | `platform.start`, `n` log lines, `runtimeDone`, `report` |
+| | `split_record_posts(total, piece, per_post) -> Vec<Vec<u8>>` | A record of `total` bytes cut into `piece`-byte pieces, spread over POSTs; the last carries `runtimeDone`. Feed to `FragmentBuffer::stitch` in order |
+| | `split_record_single_post(total, piece) -> Vec<u8>` | All pieces of one split record in one body (seams only) |
+| | `runtime_done_with(id, status, with_metrics)`, `logs_dropped(n)`, `unknown_event()` | A `runtimeDone` of any status with or without `metrics` (Lambda omits it on a timeout), a `platform.logsDropped`, an event of an unknown type |
+| | `event(millis, type, record_json)`, `init_start()`, `platform_start(id)`, `runtime_done(id)`, `report(id)`, `function_event(kind, i)`, `extension_event(i)`, `batch(&[String])` | Building blocks |
+| | `REQUEST_ID`, `FUNCTION_ARN` | Constants used everywhere |
+| `logs` | `text_line(i)`, `json_object(i) -> Value`, `json_line(i)`, `text_line_of_len(i, len)`, `OOM_LINES` | Raw log lines: Lambda text format (~200 B), Powertools-like JSON (~2 KB), one OOM line per runtime |
+| `traces` | `make_traces(n_traces, spans_per_trace, with_sql) -> Vec<Vec<pb::Span>>` | Synthetic traces: `aws.lambda` root, SQL and HTTP clients, internal spans |
+| | `trace_2000_spans()`, `v04_2000_spans() -> Vec<u8>` | The reference payload: 100 x 20 spans, ~1 MB v0.4 msgpack |
+| | `v04_msgpack(&traces)`, `header_tags()`, `TRACE_HEADERS` | Encoding and the matching tracer headers |
+| `dogstatsd` | `line(Kind, i)`, `packet(Kind, first, n)`, `packets(Kind, total, per_packet)`, `unique_tag_line(i)` | Datagrams: 50 metric names x 100 endpoints; `Kind::{Count, Gauge, Distribution}` |
+| | `ENRICH_TAGS`, `DIST_8_TAGS`, `COUNT_2_TAGS`, `SKIPPED_LINES` | Constants |
+| `events` | `payload(name)`, `payload_names()`, `all_payloads()` | The 38 trigger events in `tests/payloads/` |
+| | `sqs_event(records, body_bytes)`, `sqs_220kb()` | SQS batch with `_datadog` trace context; reference 10 x 22 KB |
+| | `dynamodb_event(records, attrs)`, `dynamodb_100_records()` | DynamoDB stream batch; reference 100 x 30 attributes |
+| | `lwa_next_headers(request_id)` | Headers of the `/invocation/next` response LWA intercepts |
+| root | `write_corpus(&dir) -> Vec<(PathBuf, usize)>`, `default_corpus_dir()` | Writes the files below |
+
+## Files on disk
+
+Generated on demand. Run from `bottlecap/`:
+
+```sh
+cargo test --test fixtures_corpus write_fixture_corpus -- --ignored
+```
+
+Output goes to `bottlecap/tests/fixtures/generated/` (gitignored), or to `$BOTTLECAP_FIXTURES_DIR`.
+`MANIFEST.txt` lists every file with its size.
+
+```
+telemetry/platform_batch.json                 platform events of one cold start
+telemetry/function_text_1000.json             1000 text log events, one POST
+telemetry/function_json_object_600.json       600 JSON-object log events, one POST under 1 MiB
+telemetry/function_json_string_600.json       600 JSON-in-string log events, one POST under 1 MiB
+telemetry/invocation_batch_text_200.json      start + 200 lines + runtimeDone + report
+telemetry/split_record_1mib_post_{0..1}.json  a 1 MiB record in 256 KiB pieces, 3 per POST
+telemetry/split_record_1mib_single_post.json  the same pieces in one body
+traces/v04_2000_spans.msgpack                 100 traces x 20 spans
+traces/v04_100_spans.msgpack                  5 traces x 20 spans
+logs/text_lines_1000.txt                      raw text lines
+logs/json_lines_1000.jsonl                    raw JSON lines
+logs/oom_lines.txt                            one OOM line per runtime
+dogstatsd/count_packet_{0..9}.txt             20 count lines per datagram
+dogstatsd/distribution_packet_{0..9}.txt      20 distribution lines per datagram
+dogstatsd/enrich_tags.txt                     the tags bottlecap adds to every metric
+events/sqs_220kb.json                         10 SQS records x 22 KB
+events/dynamodb_100_records.json              100 DynamoDB records x 30 attributes
+events/payloads/<name>.json                   a copy of the 38 trigger events in tests/payloads/
+```
+
+The checked-in trigger events stay in `bottlecap/tests/payloads/`.

bottlecap/tests/fixtures/dogstatsd.rs

@@ -0,0 +1,69 @@
+//! DogStatsD datagrams as a client library sends them to UDP 8125.
+
+/// The tags bottlecap adds to every metric (`SortedTags::parse(ENRICH_TAGS)`).
+pub const ENRICH_TAGS: &str = "function_arn:arn:aws:lambda:us-east-1:123456789012:function:my-checkout-fn,region:us-east-1,account_id:123456789012,aws_account:123456789012,functionname:my-checkout-fn,resource:my-checkout-fn,executedversion:$LATEST,memorysize:128,architecture:arm64,runtime:nodejs22.x,_dd.runtime_family:nodejs,env:prod,service:checkout,version:1.2.3,dd_extension_version:90,_dd.compute_stats:1";
+
+/// A distribution with 8 tags, the heaviest common line.
+pub const DIST_8_TAGS: &str = "app.checkout.latency:12.5|d|#env:prod,service:checkout,version:1.2.3,region:us-east-1,team:payments,endpoint:/api/v1/pay,status:200,method:POST";
+
+/// A count with 2 tags, the lightest common line.
+pub const COUNT_2_TAGS: &str = "my.metric:1|c|#tag1:a,tag2:b";
+
+/// Lines bottlecap drops before parsing: service checks, events, and its own enhanced metric.
+pub const SKIPPED_LINES: [&str; 3] = [
+    "_sc|my.check|0|#env:prod",
+    "_e{5,4}:title|text",
+    "aws.lambda.enhanced.invocations:1|c",
+];
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Kind {
+    Count,
+    Gauge,
+    Distribution,
+}
+
+/// One metric line. The `endpoint` tag cycles over 100 values and the name moves to the next of
+/// 50 values every 100 lines, so a stream of `n` lines has `min(n, 5000)` contexts.
+pub fn line(kind: Kind, i: usize) -> String {
+    match kind {
+        Kind::Count => format!(
+            "app.request.count{}:1|c|#endpoint:/api/v{},status:200,method:GET",
+            (i / 100) % 50,
+            i % 100
+        ),
+        Kind::Gauge => format!(
+            "app.queue.depth{}:{}|g|#endpoint:/api/v{},status:200,method:GET",
+            (i / 100) % 50,
+            i % 900,
+            i % 100
+        ),
+        Kind::Distribution => format!(
+            "app.request.latency{}:{}|d|#endpoint:/api/v{},status:200,method:GET",
+            (i / 100) % 50,
+            (i % 900) + 1,
+            i % 100
+        ),
+    }
+}
+
+/// A line whose tag value is unique per `i` (one new interned string per line).
+pub fn unique_tag_line(i: usize) -> String {
+    format!("my.metric:1|c|#env:prod,service:api,request_id:{i:08x}-0000-4000-8000-{i:012x}")
+}
+
+/// `n` lines joined by `\n`, one UDP datagram (keep it under 8 KiB).
+pub fn packet(kind: Kind, first: usize, n: usize) -> String {
+    (first..first + n)
+        .map(|i| line(kind, i))
+        .collect::<Vec<_>>()
+        .join("\n")
+}
+
+/// `total` lines split into datagrams of `per_packet` lines.
+pub fn packets(kind: Kind, total: usize, per_packet: usize) -> Vec<String> {
+    (0..total)
+        .step_by(per_packet.max(1))
+        .map(|first| packet(kind, first, per_packet.min(total - first)))
+        .collect()
+}

bottlecap/tests/fixtures/events.rs

@@ -0,0 +1,155 @@
+//! Invocation payloads: the 38 trigger events in `tests/payloads/`, plus large generated
+//! SQS and DynamoDB events.
+
+use std::path::{Path, PathBuf};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use serde_json::{Value, json};
+
+fn payloads_dir() -> PathBuf {
+    Path::new(env!("CARGO_MANIFEST_DIR"))
+        .join("tests")
+        .join("payloads")
+}
+
+/// Reads `tests/payloads/<name>.json`.
+pub fn payload(name: &str) -> String {
+    let path = payloads_dir().join(format!("{name}.json"));
+    std::fs::read_to_string(&path).unwrap_or_else(|e| panic!("read {}: {e}", path.display()))
+}
+
+/// Names of every file in `tests/payloads/`, sorted, without the `.json` suffix.
+pub fn payload_names() -> Vec<String> {
+    let mut names: Vec<String> = std::fs::read_dir(payloads_dir())
+        .expect("list tests/payloads")
+        .filter_map(Result::ok)
+        .filter_map(|e| {
+            let name = e.file_name().to_str()?.to_string();
+            name.strip_suffix(".json").map(str::to_string)
+        })
+        .collect();
+    names.sort();
+    names
+}
+
+/// Every payload, parsed, with its name.
+pub fn all_payloads() -> Vec<(String, Value)> {
+    payload_names()
+        .into_iter()
+        .map(|name| {
+            let value: Value = serde_json::from_str(&payload(&name))
+                .unwrap_or_else(|e| panic!("parse payload {name}: {e}"));
+            (name, value)
+        })
+        .collect()
+}
+
+/// An SQS batch of `records` messages, each with a JSON array body of exactly `body_bytes`
+/// bytes (at least 2) and a `_datadog` message attribute carrying trace context.
+pub fn sqs_event(records: usize, body_bytes: usize) -> Value {
+    let mut body = String::from("[");
+    for i in 0.. {
+        let item = format!("{{\"k{i}\":\"v{i}xxxxxxxxxxxxxxxxxxxxxx\"}},");
+        // Keep one byte for the closing `]`.
+        if body.len() + item.len() >= body_bytes {
+            break;
+        }
+        body.push_str(&item);
+    }
+    // Pad with spaces before the `]` to reach the exact size.
+    let body = format!(
+        "{:<width$}]",
+        body.trim_end_matches(','),
+        width = body_bytes.saturating_sub(1)
+    );
+    let recs: Vec<Value> = (0..records)
+        .map(|i| {
+            json!({
+                "messageId": format!("059f36b4-87a3-44ab-83d2-66197{i:07}"),
+                "receiptHandle": "AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a...",
+                "body": body,
+                "attributes": {
+                    "ApproximateReceiveCount": "1",
+                    "SentTimestamp": "1545082649183",
+                    "SenderId": "AIDAIENQZJOLO23YVJ4VO",
+                    "ApproximateFirstReceiveTimestamp": "1545082649185"
+                },
+                "messageAttributes": {
+                    "_datadog": {
+                        "stringValue": "{\"x-datadog-trace-id\":\"2684756524522091840\",\"x-datadog-parent-id\":\"7431398482019833808\",\"x-datadog-sampling-priority\":\"1\"}",
+                        "stringListValues": [],
+                        "binaryListValues": [],
+                        "dataType": "String"
+                    }
+                },
+                "md5OfBody": "e4e68fb7bd0e697a0ae8f1bb342846b3",
+                "eventSource": "aws:sqs",
+                "eventSourceARN": "arn:aws:sqs:us-east-1:123456789012:my-queue",
+                "awsRegion": "us-east-1"
+            })
+        })
+        .collect();
+    json!({ "Records": recs })
+}
+
+/// The reference SQS payload: 10 records of 22 KB, about 250 KB serialized.
+pub fn sqs_220kb() -> Value {
+    sqs_event(10, 22_000)
+}
+
+/// A DynamoDB stream batch of `records` INSERT records with `attrs` attributes each.
+pub fn dynamodb_event(records: usize, attrs: usize) -> Value {
+    let recs: Vec<Value> = (0..records)
+        .map(|i| {
+            let mut img = serde_json::Map::new();
+            for a in 0..attrs {
+                img.insert(format!("attr{a}"), json!({"S": format!("value-{i}-{a}")}));
+            }
+            json!({
+                "eventID": format!("{i}"),
+                "eventName": "INSERT",
+                "eventVersion": "1.1",
+                "eventSource": "aws:dynamodb",
+                "awsRegion": "us-east-1",
+                "dynamodb": {
+                    "ApproximateCreationDateTime": 1_428_537_600,
+                    "Keys": {"Id": {"N": "101"}},
+                    "NewImage": Value::Object(img),
+                    "SequenceNumber": format!("{i}00000000000000000"),
+                    "SizeBytes": 26,
+                    "StreamViewType": "NEW_AND_OLD_IMAGES"
+                },
+                "eventSourceARN": "arn:aws:dynamodb:us-east-1:123456789012:table/ExampleTableWithStream/stream/2015-06-27T00:48:05.899"
+            })
+        })
+        .collect();
+    json!({ "Records": recs })
+}
+
+/// The reference DynamoDB payload: 100 records of 30 attributes, about 100 KB serialized.
+pub fn dynamodb_100_records() -> Value {
+    dynamodb_event(100, 30)
+}
+
+/// Headers of the `GET /runtime/invocation/next` response that LWA intercepts. The deadline is
+/// 60 s from now, so a runtime client that honors it does not time out at once.
+pub fn lwa_next_headers(request_id: &str) -> Vec<(&'static str, String)> {
+    let deadline_ms = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .expect("clock after epoch")
+        .as_millis()
+        + 60_000;
+    vec![
+        ("content-type", "application/json".to_string()),
+        ("lambda-runtime-aws-request-id", request_id.to_string()),
+        ("lambda-runtime-deadline-ms", deadline_ms.to_string()),
+        (
+            "lambda-runtime-invoked-function-arn",
+            crate::fixtures::telemetry::FUNCTION_ARN.to_string(),
+        ),
+        (
+            "lambda-runtime-trace-id",
+            "Root=1-67d448e8-3ae320be0e2cdf2f53ccdaba;Lineage=1:73f724a8:0".to_string(),
+        ),
+    ]
+}