diff --git a/lading/Cargo.toml b/lading/Cargo.toml index a2ecd3e6c..2c5f4b883 100644 --- a/lading/Cargo.toml +++ b/lading/Cargo.toml @@ -122,6 +122,10 @@ doctest = false [[bin]] name = "lading" +[[bin]] +name = "decode_lbhc" +path = "src/bin/decode_lbhc.rs" + [package.metadata.cargo-machete] # False positive: used via `#[serde(with = "humantime_serde")]` field attributes. ignored = ["humantime-serde"] diff --git a/lading/build.rs b/lading/build.rs index 0cf4e341b..875cf7127 100644 --- a/lading/build.rs +++ b/lading/build.rs @@ -8,7 +8,13 @@ fn main() -> std::io::Result<()> { prost_build::Config::new() .out_dir("src/proto/") .protoc_arg("--experimental_allow_proto3_optional") - .compile_protos(&["proto/agent_payload.proto"], &includes)?; + // Use `bytes::Bytes` for the capture record's payload field so the + // decoded HTTP body can be moved into the record without a copy. + .bytes([".lading.blackhole.v1.BlackholeCaptureRecord.payload"]) + .compile_protos( + &["proto/agent_payload.proto", "proto/blackhole_capture.proto"], + &includes, + )?; // Compile stateful_encoding.proto with gRPC services tonic_prost_build::configure() diff --git a/lading/proto/blackhole_capture.proto b/lading/proto/blackhole_capture.proto new file mode 100644 index 000000000..0096ac923 --- /dev/null +++ b/lading/proto/blackhole_capture.proto @@ -0,0 +1,44 @@ +// Blackhole HTTP payload capture record. +// +// The wire format on disk is: +// +// File prologue (8 bytes, written once at file create): +// bytes[0..4] magic = "LBHC" // ASCII, "Lading BlackHole Capture" +// bytes[4..6] version = 0x0001 LE // file format version +// bytes[6..8] reserved = 0x0000 LE // reserved, must be zero +// +// Records (repeated): +// A `BlackholeCaptureRecord` serialized as a length-delimited protobuf +// message: a `prost`-style varint containing the length in bytes of the +// following serialized message, followed by the serialized message. +// Compatible with `Message::encode_length_delimited` / +// `Message::decode_length_delimited`. +// +// The file is append-only. A reader observing EOF mid-record (truncated +// varint or short payload read) should treat the run as ending there. + +syntax = "proto3"; + +package lading.blackhole.v1; + +// One HTTP request captured by lading's HTTP blackhole. The record's +// `payload` is the decoded request body; `content_type` and `request_path` +// describe what the target claimed to be sending on this request. +message BlackholeCaptureRecord { + // Milliseconds since the blackhole server started serving. + uint64 relative_ms = 1; + + // Size of the compressed (on-wire) request body in bytes, before the + // blackhole applied Content-Encoding decoding. Informational. + uint64 compressed_bytes = 2; + + // Value of the request's Content-Type header. Empty if absent. + string content_type = 3; + + // Path portion of the request URI (e.g. "/api/v2/logs"). + string request_path = 4; + + // Decoded request body. If the request carried Content-Encoding (gzip, + // deflate, zstd), this field is the decompressed body. + bytes payload = 5; +} diff --git a/lading/src/bin/decode_lbhc.rs b/lading/src/bin/decode_lbhc.rs new file mode 100644 index 000000000..97844cc28 --- /dev/null +++ b/lading/src/bin/decode_lbhc.rs @@ -0,0 +1,277 @@ +//! Decode a `.lbhc` (Lading BlackHole Capture) file to human-readable output. +//! +//! Usage: +//! decode_lbhc [--summary | --full | --json] [--out-dir ] +//! +//! Modes: +//! --summary (default): one line per record with metadata + a short payload preview. +//! --full: full payload printed after each record's metadata. +//! --json: JSON Lines, one object per record with payload as a +//! UTF-8 (lossy) string. +//! +//! Redirection: +//! To capture the mode's output to a file, use shell redirection: +//! decode_lbhc file.lbhc --full > out.txt +//! +//! To dump each record's raw decoded payload to its own file (one file +//! per record, named `NNNN.payload`), pass `--out-dir `. The +//! selected mode still prints to stdout. Combine e.g.: +//! decode_lbhc file.lbhc --summary --out-dir ./payloads > index.txt + +use std::{ + env, + fs::{self, File}, + io::{BufReader, ErrorKind, Read, Write}, + path::PathBuf, + process, +}; + +/// Local copy of the proto struct so this bin doesn't need access to +/// `pub(crate)` proto module in the lading lib. Field tags and types +/// must match `proto/blackhole_capture.proto`. +#[derive(Clone, PartialEq, Eq, prost::Message)] +struct BlackholeCaptureRecord { + #[prost(uint64, tag = "1")] + relative_ms: u64, + #[prost(uint64, tag = "2")] + compressed_bytes: u64, + #[prost(string, tag = "3")] + content_type: String, + #[prost(string, tag = "4")] + request_path: String, + #[prost(bytes = "vec", tag = "5")] + payload: Vec, +} + +#[derive(Copy, Clone)] +enum Mode { + Summary, + Full, + Json, +} + +fn main() { + let mut args: Vec = env::args().skip(1).collect(); + if args.is_empty() { + eprintln!( + "usage: decode_lbhc [--summary|--full|--json] [--out-dir ]" + ); + process::exit(2); + } + let path = args.remove(0); + let mut mode = Mode::Summary; + let mut out_dir: Option = None; + let mut iter = args.into_iter(); + while let Some(arg) = iter.next() { + match arg.as_str() { + "--summary" => mode = Mode::Summary, + "--full" => mode = Mode::Full, + "--json" => mode = Mode::Json, + "--out-dir" => match iter.next() { + Some(d) => out_dir = Some(PathBuf::from(d)), + None => { + eprintln!("--out-dir requires a directory argument"); + process::exit(2); + } + }, + other => { + eprintln!("unknown flag: {other}"); + process::exit(2); + } + } + } + if let Some(ref dir) = out_dir + && let Err(e) = fs::create_dir_all(dir) + { + eprintln!("create --out-dir {}: {e}", dir.display()); + process::exit(1); + } + + let f = File::open(&path).unwrap_or_else(|e| { + eprintln!("open {path}: {e}"); + process::exit(1); + }); + let mut reader = BufReader::new(f); + + // Prologue: 4 bytes magic "LBHC" + u16 LE version + u16 LE reserved. + let mut prologue = [0u8; 8]; + if let Err(e) = reader.read_exact(&mut prologue) { + eprintln!("read prologue: {e}"); + process::exit(1); + } + if &prologue[0..4] != b"LBHC" { + eprintln!("bad magic: {:02x?}", &prologue[0..4]); + process::exit(1); + } + let version = u16::from_le_bytes([prologue[4], prologue[5]]); + if version != 1 { + eprintln!("unsupported file version: {version}"); + process::exit(1); + } + eprintln!("# LBHC v{version}, decoding {path}"); + + let mut seq: u64 = 0; + loop { + let bytes = match read_length_delimited(&mut reader) { + Ok(Some(b)) => b, + Ok(None) => break, + Err(e) => { + eprintln!("record {seq}: framing error: {e}"); + process::exit(1); + } + }; + let record: BlackholeCaptureRecord = match prost::Message::decode(&bytes[..]) { + Ok(r) => r, + Err(e) => { + eprintln!("record {seq}: decode error: {e}"); + process::exit(1); + } + }; + match mode { + Mode::Summary => print_summary(seq, &record), + Mode::Full => print_full(seq, &record), + Mode::Json => print_json(seq, &record), + } + if let Some(ref dir) = out_dir { + if let Err(e) = dump_payload(dir, seq, &record) { + eprintln!("record {seq}: dump failed: {e}"); + process::exit(1); + } + } + seq = seq.saturating_add(1); + } + eprintln!("# {seq} records total"); + if let Some(dir) = out_dir { + eprintln!("# payloads written to {}", dir.display()); + } +} + +/// Write one record's raw decoded payload to `/NNNN.payload`. If the +/// payload parses as JSON, write pretty-printed JSON to `NNNN.json` +/// instead so it's directly `jq`-able / browsable. +fn dump_payload( + dir: &std::path::Path, + seq: u64, + r: &BlackholeCaptureRecord, +) -> std::io::Result<()> { + let (ext, contents): (&str, Vec) = + match serde_json::from_slice::(&r.payload) { + Ok(v) => { + let pretty = serde_json::to_vec_pretty(&v).unwrap_or_else(|_| r.payload.clone()); + ("json", pretty) + } + Err(_) => ("payload", r.payload.clone()), + }; + let name = format!("{seq:04}.{ext}"); + let mut path = dir.to_path_buf(); + path.push(name); + let mut f = File::create(&path)?; + f.write_all(&contents)?; + Ok(()) +} + +/// Read one length-delimited proto message. `Ok(None)` at clean EOF or +/// truncated tail; `Err` on I/O failure or malformed varint. +fn read_length_delimited(reader: &mut R) -> std::io::Result>> { + let mut varint_buf = [0u8; 10]; + let mut varint_len = 0; + loop { + if varint_len >= varint_buf.len() { + return Ok(None); // overlong varint => corrupt tail, stop + } + let mut byte = [0u8; 1]; + match reader.read(&mut byte) { + Ok(0) if varint_len == 0 => return Ok(None), // clean EOF + Ok(0) => return Ok(None), // truncated varint + Ok(_) => { + varint_buf[varint_len] = byte[0]; + varint_len += 1; + if byte[0] & 0x80 == 0 { + break; + } + } + Err(e) if e.kind() == ErrorKind::Interrupted => {} + Err(e) => return Err(e), + } + } + let mut slice = &varint_buf[..varint_len]; + let msg_len = match prost::encoding::decode_varint(&mut slice) { + Ok(v) => v as usize, + Err(e) => return Err(std::io::Error::new(ErrorKind::InvalidData, e)), + }; + let mut buf = vec![0u8; msg_len]; + if !read_full(reader, &mut buf)? { + return Ok(None); // truncated body + } + Ok(Some(buf)) +} + +fn read_full(reader: &mut R, mut buf: &mut [u8]) -> std::io::Result { + while !buf.is_empty() { + match reader.read(buf) { + Ok(0) => return Ok(false), + Ok(n) => { + let tmp = buf; + buf = &mut tmp[n..]; + } + Err(e) if e.kind() == ErrorKind::Interrupted => {} + Err(e) => return Err(e), + } + } + Ok(true) +} + +fn preview(payload: &[u8], max: usize) -> String { + let take = payload.len().min(max); + let s = String::from_utf8_lossy(&payload[..take]).into_owned(); + if payload.len() > max { + format!("{s}...[+{} bytes]", payload.len() - max) + } else { + s + } +} + +fn print_summary(seq: u64, r: &BlackholeCaptureRecord) { + println!( + "#{seq:04} t={:>7}ms compressed={:>7}B decoded={:>7}B path={} ctype={} payload={}", + r.relative_ms, + r.compressed_bytes, + r.payload.len(), + r.request_path, + r.content_type, + preview(&r.payload, 120), + ); +} + +fn print_full(seq: u64, r: &BlackholeCaptureRecord) { + println!( + "#{seq:04} t={}ms compressed_bytes={} decoded_bytes={} path={} content_type={}", + r.relative_ms, + r.compressed_bytes, + r.payload.len(), + r.request_path, + r.content_type, + ); + // If payload parses as JSON, pretty-print. Otherwise lossy UTF-8. + if let Ok(value) = serde_json::from_slice::(&r.payload) { + let pretty = serde_json::to_string_pretty(&value).unwrap_or_default(); + println!("{pretty}"); + } else { + println!("{}", String::from_utf8_lossy(&r.payload)); + } + println!(); +} + +fn print_json(seq: u64, r: &BlackholeCaptureRecord) { + // Preserve the raw payload string as-is (UTF-8 lossy) inside a JSON + // string field so downstream tools can pipe this to jq. + let obj = serde_json::json!({ + "seq": seq, + "relative_ms": r.relative_ms, + "compressed_bytes": r.compressed_bytes, + "content_type": r.content_type, + "request_path": r.request_path, + "payload": String::from_utf8_lossy(&r.payload), + }); + println!("{obj}"); +} diff --git a/lading/src/blackhole/http.rs b/lading/src/blackhole/http.rs index 0e64db010..037b29528 100644 --- a/lading/src/blackhole/http.rs +++ b/lading/src/blackhole/http.rs @@ -16,8 +16,18 @@ use lading_payload::openmetrics; use metrics::counter; use serde::{Deserialize, Deserializer, Serialize, de}; use serde_yaml::with::singleton_map_recursive; -use std::{net::SocketAddr, time::Duration}; -use tracing::error; +use std::{ + io::Write, + net::SocketAddr, + path::PathBuf, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + mpsc::{SyncSender, sync_channel}, + }, + time::{Duration, Instant}, +}; +use tracing::{error, info}; use super::General; use crate::blackhole::common; @@ -162,6 +172,129 @@ pub struct Config { /// delay to add before making a response #[serde(default = "default_response_delay_millis")] pub response_delay_millis: u64, + /// Optional payload capture. When set, decoded HTTP request bodies are + /// written as JSONL to `path`, up to `max_payloads` records. Additional + /// requests are served normally but not captured. + #[serde(default)] + pub capture: Option, +} + +/// Configuration for blackhole HTTP payload capture. +#[derive(Debug, Deserialize, Serialize, Clone, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct CaptureConfig { + /// File path to write JSONL records to. Overwritten on startup. + pub path: PathBuf, + /// Maximum number of payloads to capture. Once reached, subsequent + /// requests skip all capture work with a single relaxed atomic load. + pub max_payloads: usize, + /// Optional cap on the number of payload bytes retained per record. + /// When set, decoded payloads are truncated to this many bytes before + /// being enqueued for the writer. Bounds memory held in the channel. + /// Truncation is zero-copy (`Bytes::slice`). + #[serde(default)] + pub max_payload_bytes: Option, +} + +/// A single captured blackhole payload. Holds `Bytes` (a refcounted view +/// into the decoded body) so cloning into the channel does not memcpy the +/// payload; proto encoding happens on the writer thread, off the request +/// path. +struct CaptureRecord { + /// Milliseconds since the blackhole server started. + relative_ms: u64, + /// Size of the compressed (on-wire) body in bytes. + compressed_bytes: u64, + /// Value of the request's Content-Type header. Empty if absent. + content_type: String, + /// Path portion of the request URI. + request_path: String, + /// Decoded body. + payload: Bytes, +} + +/// Bounded channel capacity between request handlers and the writer thread. +/// Sized to absorb write bursts for full-size (5 MiB) payloads while +/// bounding memory retention to a predictable ceiling. Each slot holds a +/// refcounted `Bytes` view of a decoded payload; with 128 slots the ceiling +/// is `128 * max_payload_size` (~640 MiB at 5 MiB payloads). If +/// `max_payload_bytes` is set the ceiling drops accordingly. +const CAPTURE_CHANNEL_CAPACITY: usize = 128; + +/// File format magic: `LBHC` = Lading blackhole capture. +const CAPTURE_FILE_MAGIC: &[u8; 4] = b"LBHC"; + +/// File format version. Increment on wire-incompatible changes. +const CAPTURE_FILE_VERSION: u16 = 1; + +/// Write the 8-byte file prologue. See `proto/blackhole_capture.proto` for +/// the full on-disk format documentation. +fn write_capture_prologue(writer: &mut W) -> std::io::Result<()> { + writer.write_all(CAPTURE_FILE_MAGIC)?; + writer.write_all(&CAPTURE_FILE_VERSION.to_le_bytes())?; + writer.write_all(&0u16.to_le_bytes())?; // reserved + Ok(()) +} + +/// Spawn a dedicated OS thread that drains capture records and writes a +/// length-delimited proto stream to disk. Deliberately not a tokio task: +/// file I/O is synchronous, and running it on the tokio runtime would pin +/// a worker thread on every flush and back-pressure the request handlers +/// through the runtime. A plain `std::thread` isolates disk stalls from +/// the executor. +/// +/// Flushes after every record so captures survive abrupt shutdown and can +/// be live-tailed. +fn spawn_capture_writer(rx: std::sync::mpsc::Receiver, path: PathBuf) { + std::thread::Builder::new() + .name("blackhole-capture-writer".to_string()) + .spawn(move || { + let file = match std::fs::File::create(&path) { + Ok(f) => f, + Err(e) => { + error!( + "Failed to create blackhole capture file {}: {e}", + path.display() + ); + return; + } + }; + let mut writer = std::io::BufWriter::new(file); + if let Err(e) = write_capture_prologue(&mut writer) { + error!("Failed to write blackhole capture prologue: {e}"); + return; + } + // Reusable encode buffer: sized to grow to the largest record + // seen and then stay there, so per-record cost is a memcpy + // (unavoidable at proto encode time) with no reallocation. + let mut encode_buf: Vec = Vec::with_capacity(64 * 1024); + while let Ok(record) = rx.recv() { + let proto_record = crate::proto::blackhole::v1::BlackholeCaptureRecord { + relative_ms: record.relative_ms, + compressed_bytes: record.compressed_bytes, + content_type: record.content_type, + request_path: record.request_path, + payload: record.payload, + }; + encode_buf.clear(); + if let Err(e) = prost::Message::encode_length_delimited( + &proto_record, + &mut encode_buf, + ) { + error!("Failed to encode blackhole capture record: {e}"); + continue; + } + if let Err(e) = writer + .write_all(&encode_buf) + .and_then(|()| writer.flush()) + { + error!("Blackhole capture write failed: {e}"); + return; + } + } + info!("Blackhole capture writer finished for {}", path.display()); + }) + .expect("failed to spawn blackhole capture writer thread"); } #[derive(Serialize)] @@ -181,6 +314,7 @@ struct KinesisPutRecordBatchResponse { } #[allow(clippy::borrow_interior_mutable_const)] +#[allow(clippy::too_many_arguments)] async fn srv( status: StatusCode, metric_labels: Vec<(String, String)>, @@ -188,6 +322,7 @@ async fn srv( req: Request, headers: HeaderMap, response_delay: Duration, + capture: Option, ) -> Result>, hyper::Error> { counter!("requests_received", &metric_labels).increment(1); @@ -206,6 +341,59 @@ async fn srv( Ok(body) => { counter!("decoded_bytes_received", &metric_labels).increment(body.len() as u64); + // Capture path is deliberately fire-and-forget on a spawned + // task so no capture work — atomic RMWs, timestamp reads, + // channel sends, metric emission — is on the response path. + // + // When capture is disabled the branch is a `None` check. + // When capture is enabled but the payload cap has been reached, + // we pay only a single `Relaxed` load and skip the spawn. + // Under cap we spawn a task that owns the decoded `body` + // (moved, not cloned) and does the rest of the work off-path. + if let Some(cap) = capture.as_ref() + && cap.captured.load(Ordering::Relaxed) < cap.max_payloads + { + let cap = cap.clone(); + let metric_labels = metric_labels.clone(); + // Extract request metadata inline. These are `&str` + // conversions from small headers and are effectively free; + // doing them here avoids moving `parts` into the spawned + // task. + let content_type = parts + .headers + .get(hyper::header::CONTENT_TYPE) + .and_then(|v| v.to_str().ok()) + .unwrap_or("") + .to_string(); + let request_path = parts.uri.path().to_string(); + tokio::spawn(async move { + // Reserve a slot. A second thread may have raced + // past the cap between our load and this RMW, so + // re-check and bail without touching the channel. + let seq = cap.captured.fetch_add(1, Ordering::Relaxed); + if seq >= cap.max_payloads { + return; + } + let payload = match cap.max_payload_bytes { + Some(n) if body.len() > n => body.slice(..n), + _ => body, + }; + let record = CaptureRecord { + relative_ms: u64::try_from(cap.epoch.elapsed().as_millis()) + .unwrap_or(u64::MAX), + compressed_bytes: body_len, + content_type, + request_path, + payload, + }; + // `SyncSender::try_send` returns only `Full` or + // `Disconnected`; both are drops for our purposes. + if cap.tx.try_send(record).is_err() { + counter!("blackhole_capture_dropped", &metric_labels).increment(1); + } + }); + } + tokio::time::sleep(response_delay).await; let mut okay = Response::default(); @@ -217,6 +405,16 @@ async fn srv( } } +/// Shared per-request capture state. +#[derive(Debug, Clone)] +struct CaptureState { + tx: SyncSender, + captured: Arc, + max_payloads: usize, + max_payload_bytes: Option, + epoch: Instant, +} + #[derive(Debug)] /// The HTTP blackhole. pub struct Http { @@ -228,8 +426,10 @@ pub struct Http { status: StatusCode, metric_labels: Vec<(String, String)>, response_delay: Duration, + capture: Option, } + impl Http { /// Create a new [`Http`] server instance /// @@ -274,6 +474,32 @@ impl Http { BodyVariant::OpenMetrics(conf) => openmetrics::OpenMetrics::new(conf)?.into_bytes(), }; + let capture = config.capture.as_ref().map(|cap| { + let (tx, rx) = sync_channel::(CAPTURE_CHANNEL_CAPACITY); + spawn_capture_writer(rx, cap.path.clone()); + if let Some(n) = cap.max_payload_bytes { + info!( + "Blackhole HTTP capture enabled, writing up to {} payloads (truncated to {} bytes) to {}", + cap.max_payloads, + n, + cap.path.display() + ); + } else { + info!( + "Blackhole HTTP capture enabled, writing up to {} payloads to {}", + cap.max_payloads, + cap.path.display() + ); + } + CaptureState { + tx, + captured: Arc::new(AtomicUsize::new(0)), + max_payloads: cap.max_payloads, + max_payload_bytes: cap.max_payload_bytes, + epoch: Instant::now(), + } + }); + Ok(Self { httpd_addr: config.binding_addr, body_bytes, @@ -283,6 +509,7 @@ impl Http { shutdown, metric_labels, response_delay: Duration::from_millis(config.response_delay_millis), + capture, }) } @@ -307,6 +534,7 @@ impl Http { let headers = self.headers.clone(); let status = self.status; let response_delay = self.response_delay; + let capture = self.capture.clone(); hyper::service::service_fn(move |req| { srv( @@ -316,6 +544,7 @@ impl Http { req, headers.clone(), response_delay, + capture.clone(), ) }) }, @@ -351,6 +580,7 @@ body_variant: "nothing" headers: default_headers(), status: default_status_code(), raw_bytes: vec![], + capture: None, }, ); } @@ -375,10 +605,42 @@ raw_bytes: [0x01, 0x02, 0x10] headers: default_headers(), status: default_status_code(), raw_bytes: vec![0x01, 0x02, 0x10], + capture: None, }, ); } + #[test] + fn config_deserializes_capture() { + let contents = r#" +binding_addr: "127.0.0.1:1000" +capture: + path: /tmp/blackhole.jsonl + max_payloads: 100 +"#; + let config: Config = + serde_yaml::from_str(contents).expect("Contents do not match the structure expected"); + let capture = config.capture.expect("capture should be set"); + assert_eq!(capture.path, PathBuf::from("/tmp/blackhole.jsonl")); + assert_eq!(capture.max_payloads, 100); + assert_eq!(capture.max_payload_bytes, None); + } + + #[test] + fn config_deserializes_capture_with_truncation() { + let contents = r#" +binding_addr: "127.0.0.1:1000" +capture: + path: /tmp/blackhole.jsonl + max_payloads: 100 + max_payload_bytes: 4096 +"#; + let config: Config = + serde_yaml::from_str(contents).expect("Contents do not match the structure expected"); + let capture = config.capture.expect("capture should be set"); + assert_eq!(capture.max_payload_bytes, Some(4096)); + } + #[test] fn config_deserializes_tagged_static_variant() { let contents = r#" diff --git a/lading/src/proto.rs b/lading/src/proto.rs index c8bf0a7e4..39682b0fa 100644 --- a/lading/src/proto.rs +++ b/lading/src/proto.rs @@ -1,5 +1,18 @@ //! Module containing structs generated from `proto/` +/// Protobuf definitions for blackhole HTTP payload capture (see +/// `proto/blackhole_capture.proto`). +pub(crate) mod blackhole { + /// v1 schema. + pub(crate) mod v1 { + #![allow(clippy::pedantic)] + #![allow(missing_docs)] + #![allow(unreachable_pub)] + #![allow(dead_code)] + include!("proto/lading.blackhole.v1.rs"); + } +} + /// Protobuf definitions for our `datadog` blackhole pub(crate) mod datadog { /// Related to the [DataDog](https://www.datadoghq.com/) intake API diff --git a/lading/src/proto/lading.blackhole.v1.rs b/lading/src/proto/lading.blackhole.v1.rs new file mode 100644 index 000000000..882ab6ff6 --- /dev/null +++ b/lading/src/proto/lading.blackhole.v1.rs @@ -0,0 +1,24 @@ +// This file is @generated by prost-build. +/// One HTTP request captured by lading's HTTP blackhole. The record's +/// `payload` is the decoded request body; `content_type` and `request_path` +/// describe what the target claimed to be sending on this request. +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct BlackholeCaptureRecord { + /// Milliseconds since the blackhole server started serving. + #[prost(uint64, tag = "1")] + pub relative_ms: u64, + /// Size of the compressed (on-wire) request body in bytes, before the + /// blackhole applied Content-Encoding decoding. Informational. + #[prost(uint64, tag = "2")] + pub compressed_bytes: u64, + /// Value of the request's Content-Type header. Empty if absent. + #[prost(string, tag = "3")] + pub content_type: ::prost::alloc::string::String, + /// Path portion of the request URI (e.g. "/api/v2/logs"). + #[prost(string, tag = "4")] + pub request_path: ::prost::alloc::string::String, + /// Decoded request body. If the request carried Content-Encoding (gzip, + /// deflate, zstd), this field is the decompressed body. + #[prost(bytes = "bytes", tag = "5")] + pub payload: ::prost::bytes::Bytes, +}