{"record":{"id":"d6aae89aaa7217e1","repo":"vectordotdev/vector","slug":"docker-logs-stream-must-exist-in-the-metadata","errorCode":null,"errorMessage":"`docker_logs.stream` must exist in the metadata","messagePattern":"`docker_logs\\.stream` must exist in the metadata","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/sources/docker_logs/mod.rs","lineNumber":1353,"sourceCode":"    log_namespace: LogNamespace,\n) -> impl Stream<Item = LogEvent> {\n    let line_agg_in = inner.map(move |mut log| {\n        let message_value = match log_namespace {\n            LogNamespace::Vector => log\n                .remove(&vrl::path::OwnedTargetPath::event_root())\n                .expect(\"`.` must exist in the event\"),\n            LogNamespace::Legacy => log\n                .remove(\n                    log_schema()\n                        .message_key_target_path()\n                        .expect(\"global log_schema.message_key to be valid path\"),\n                )\n                .expect(\"`message` must exist in the event\"),\n        };\n        let stream_value = match log_namespace {\n            LogNamespace::Vector => log\n                .get(metadata_path!(DockerLogsConfig::NAME, STREAM))\n                .expect(\"`docker_logs.stream` must exist in the metadata\"),\n            LogNamespace::Legacy => log\n                .get(event_path!(STREAM))\n                .expect(\"stream must exist in the event\"),\n        };\n\n        let stream = stream_value.coerce_to_bytes();\n        let message = message_value.coerce_to_bytes();\n        (stream, message, log)\n    });\n    let line_agg_out = LineAgg::<_, Bytes, LogEvent>::new(line_agg_in, logic);\n    line_agg_out.map(move |(_, message, mut log, _)| {\n        match log_namespace {\n            LogNamespace::Vector => log.insert(&vrl::path::OwnedTargetPath::event_root(), message),\n            LogNamespace::Legacy => log.insert(\n                log_schema()\n                    .message_key_target_path()\n                    .expect(\"global log_schema.message_key to be valid path\"),\n                message,","sourceCodeStart":1335,"sourceCodeEnd":1371,"githubUrl":"https://github.com/vectordotdev/vector/blob/3708c39b12a93212ed8b8d7510b4cc7769cb5864/src/sources/docker_logs/mod.rs#L1335-L1371","documentation":"In line_agg_adapter's Vector-namespace branch, the stream tag used to group multiline messages is read with log.get(metadata_path!(DockerLogsConfig::NAME, STREAM)).expect(\"`docker_logs.stream` must exist in the metadata\"). The docker_logs source inserts this metadata (stdout/stderr) when it creates each event, so the expect asserts that the event passed through that insertion stage unchanged.","triggerScenarios":"An event reaching the aggregator without docker_logs source metadata - a regression or a custom stage between the source and line_agg that drops or rebuilds metadata, while multiline is enabled with log_namespace = true.","commonSituations":"Vector upgrades changing source-metadata insertion, forks inserting transforms before aggregation.","solutions":["Verify events still carry the docker_logs.stream metadata (inspect with a console sink) with the same config minus multiline","Remove any custom stages between the source and the aggregator","Patch: fall back to Bytes::new() for the stream value with a warning","Report upstream and upgrade"],"exampleFix":"// before\nlet stream_value = log\n    .get(metadata_path!(DockerLogsConfig::NAME, STREAM))\n    .expect(\"`docker_logs.stream` must exist in the metadata\");\n\n// after\nlet stream_value = log\n    .get(metadata_path!(DockerLogsConfig::NAME, STREAM))\n    .unwrap_or_else(|| {\n        warn!(message = \"event missing docker_logs.stream metadata\");\n        &Value::Bytes(Bytes::new())\n    });","handlingStrategy":"validation","validationCode":"let stream_value = log\n    .get(metadata_path!(DockerLogsConfig::NAME, STREAM))\n    .cloned()\n    .unwrap_or_else(|| {\n        warn!(message = \"missing docker_logs.stream metadata\");\n        Value::Bytes(Bytes::new())\n    });","typeGuard":"fn has_stream_metadata(log: &LogEvent) -> bool {\n    log.get(metadata_path!(DockerLogsConfig::NAME, STREAM)).is_some()\n}","tryCatchPattern":"match log.get(metadata_path!(DockerLogsConfig::NAME, STREAM)) {\n    Some(v) => v.coerce_to_bytes(),\n    None => { warn!(\"missing stream metadata\"); Bytes::new() }\n}","preventionTips":["Insert source metadata and consume it through shared constants so they cannot drift","Test Vector-namespace events end to end whenever adding stages before aggregation","Default missing grouping keys to an empty value and log, instead of expecting"],"tags":["rust","panic","invariant","log-metadata","multiline","docker-logs"],"backgroundTag":"missing-event-field","analyzedSha":"3708c39b12a93212ed8b8d7510b4cc7769cb5864","analyzedAt":"2026-08-20T07:02:18.786Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T15:18:49.778Z"}