feat(vrl): add parse_dnstap function

.github/actions/spelling/patterns.txt

@@ -226,7 +226,7 @@ user:P@ssw0rd
 /.+@base64/.+
 
 # Ignore base64 encoded values in VRL examples (requires padding to avoid false positives)
-"[A-Za-z0-9]*=="
+"[A-Za-z0-9+\/]*=="
 
 # ignore uuid_from_friendly_id argument
 uuid_from_friendly_id!\(".*"\)

Cargo.toml

@@ -103,6 +103,7 @@ members = [
   ".",
   "lib/codecs",
   "lib/dnsmsg-parser",
+  "lib/dnstap-parser",
   "lib/docs-renderer",
   "lib/enrichment",
   "lib/fakedata",
@@ -133,13 +134,15 @@ members = [
 ]
 
 [workspace.dependencies]
+anyhow = "1.0.94"
 cfg-if = { version = "1.0.0", default-features = false }
 chrono = { version = "0.4.38", default-features = false, features = ["clock", "serde"] }
 chrono-tz = { version = "0.10.0", default-features = false, features = ["serde"] }
 clap = { version = "4.5.22", default-features = false, features = ["derive", "error-context", "env", "help", "std", "string", "usage", "wrap_help"] }
 flate2 = { version = "1.0.35", default-features = false, features = ["default"] }
 futures = { version = "0.3.31", default-features = false, features = ["compat", "io-compat", "std"], package = "futures" }
 glob = { version = "0.3.1", default-features = false }
+hickory-proto = { version = "0.24.1", default-features = false, features = ["dnssec"] }
 indexmap = { version = "2.7.0", default-features = false, features = ["serde", "std"] }
 metrics = "0.24.1"
 metrics-tracing-context = { version = "0.17.0", default-features = false }
@@ -177,6 +180,7 @@ snafu.workspace = true
 
 # Internal libs
 dnsmsg-parser = { path = "lib/dnsmsg-parser", optional = true }
+dnstap-parser = { path = "lib/dnstap-parser", optional = true }
 fakedata = { path = "lib/fakedata", optional = true }
 portpicker = { path = "lib/portpicker" }
 tracing-limit = { path = "lib/tracing-limit" }
@@ -368,7 +372,7 @@ tokio-postgres = { version = "0.7.12", default-features = false, features = ["ru
 tokio-tungstenite = { version = "0.20.1", default-features = false, features = ["connect"], optional = true }
 toml.workspace = true
 tonic = { workspace = true, optional = true }
-hickory-proto = { version = "0.24.1", default-features = false, features = ["dnssec"], optional = true }
+hickory-proto = { workspace = true, optional = true }
 typetag = { version = "0.2.18", default-features = false }
 url = { version = "2.5.4", default-features = false, features = ["serde"] }
 warp = { version = "0.3.7", default-features = false }
@@ -584,7 +588,7 @@ sources-aws_s3 = ["aws-core", "dep:aws-sdk-sqs", "dep:aws-sdk-s3", "dep:semver",
 sources-aws_sqs = ["aws-core", "dep:aws-sdk-sqs"]
 sources-datadog_agent = ["sources-utils-http-error", "protobuf-build", "dep:prost"]
 sources-demo_logs = ["dep:fakedata"]
-sources-dnstap = ["sources-utils-net-tcp", "dep:base64", "dep:hickory-proto", "dep:dnsmsg-parser", "protobuf-build", "dep:prost"]
+sources-dnstap = ["sources-utils-net-tcp", "dep:base64", "dep:hickory-proto", "dep:dnsmsg-parser", "dep:dnstap-parser", "protobuf-build", "dep:prost"]
 sources-docker_logs = ["docker"]
 sources-eventstoredb_metrics = []
 sources-exec = []

buf.yaml

@@ -3,6 +3,7 @@ modules:
   - path: lib/vector-core/proto
   - path: proto/third-party
   - path: proto/vector
+  - path: lib/dnstap-parser/proto
 lint:
   use:
     - DEFAULT

build.rs

@@ -116,7 +116,6 @@ fn main() {
 
     #[cfg(feature = "protobuf-build")]
     {
-        println!("cargo:rerun-if-changed=proto/third-party/dnstap.proto");
         println!("cargo:rerun-if-changed=proto/third-party/google/pubsub/v1/pubsub.proto");
         println!("cargo:rerun-if-changed=proto/third-party/google/rpc/status.proto");
         println!("cargo:rerun-if-changed=proto/vector/dd_metric.proto");
@@ -144,7 +143,6 @@ fn main() {
                 prost_build,
                 &[
                     "lib/vector-core/proto/event.proto",
-                    "proto/third-party/dnstap.proto",
                     "proto/vector/ddsketch_full.proto",
                     "proto/vector/dd_metric.proto",
                     "proto/vector/dd_trace.proto",

changelog.d/21985_parse_dnstap.feature.md

@@ -0,0 +1,3 @@
+Add VRL function `parse_dnstap` that can parse dnstap data and produce output in the same format as `dnstap` source.
+
+authors: esensar

lib/dnsmsg-parser/Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT"
 
 [dependencies]
 data-encoding = "2.6"
-hickory-proto = { version = "0.24", features = ["dnssec"] }
+hickory-proto.workspace = true
 snafu.workspace = true
 
 [dev-dependencies]

lib/dnstap-parser/Cargo.toml

@@ -0,0 +1,26 @@
+[package]
+name = "dnstap-parser"
+version = "0.1.0"
+authors = ["Vector Contributors <[email protected]>"]
+edition = "2021"
+publish = false
+license = "MIT"
+
+[dependencies]
+base64 = { version = "0.22.1", default-features = false }
+bytes = { version = "1.9.0", default-features = false, features = ["serde"] }
+chrono.workspace = true
+dnsmsg-parser = { path = "../dnsmsg-parser" }
+hickory-proto.workspace = true
+prost.workspace = true
+snafu.workspace = true
+tracing = { version = "0.1.34", default-features = false }
+vector-lib = { path = "../vector-lib" }
+vrl.workspace = true
+
+[build-dependencies]
+prost-build.workspace = true
+
+[dev-dependencies]
+anyhow.workspace = true
+chrono-tz.workspace = true

lib/dnstap-parser/build.rs

@@ -0,0 +1,8 @@
+fn main() {
+    println!("cargo:rerun-if-changed=proto/dnstap.proto");
+    let mut prost_build = prost_build::Config::new();
+    prost_build.btree_map(["."]);
+    prost_build
+        .compile_protos(&["proto/dnstap.proto"], &["proto"])
+        .expect("Failed to compile proto files");
+}

lib/dnstap-parser/src/internal_events.rs

@@ -0,0 +1,20 @@
+use tracing::warn;
+use vector_lib::internal_event::InternalEvent;
+use vector_lib::internal_event::{error_stage, error_type};
+
+#[derive(Debug)]
+pub(crate) struct DnstapParseWarning<E> {
+    pub error: E,
+}
+
+impl<E: std::fmt::Display> InternalEvent for DnstapParseWarning<E> {
+    fn emit(self) {
+        warn!(
+            message = "Recoverable error occurred while parsing dnstap data.",
+            error = %self.error,
+            stage = error_stage::PROCESSING,
+            error_type = error_type::PARSER_FAILED,
+            internal_log_rate_limit = true,
+        );
+    }
+}

lib/dnstap-parser/src/lib.rs

@@ -0,0 +1,12 @@
+#![deny(warnings)]
+
+use vrl::compiler::Function;
+
+mod internal_events;
+pub mod parser;
+pub mod schema;
+mod vrl_functions;
+
+pub fn vrl_functions() -> Vec<Box<dyn Function>> {
+    vrl_functions::all()
+}

src/sources/dnstap/parser.rs → lib/dnstap-parser/src/parser.rs

@@ -5,6 +5,7 @@ use std::{
     net::{IpAddr, Ipv4Addr, Ipv6Addr},
     sync::LazyLock,
 };
+use vector_lib::emit;
 
 use base64::prelude::{Engine as _, BASE64_STANDARD};
 use bytes::Bytes;
@@ -18,9 +19,8 @@ use prost::Message;
 use snafu::Snafu;
 use vrl::{owned_value_path, path};
 
-use crate::{
+use vector_lib::{
     event::{LogEvent, Value},
-    internal_events::DnstapParseWarning,
     Error, Result,
 };
 
@@ -29,7 +29,7 @@ mod dnstap_proto {
     include!(concat!(env!("OUT_DIR"), "/dnstap.rs"));
 }
 
-use crate::sources::dnstap::schema::DNSTAP_VALUE_PATHS;
+use crate::{internal_events::DnstapParseWarning, schema::DNSTAP_VALUE_PATHS};
 use dnstap_proto::{
     message::Type as DnstapMessageType, Dnstap, Message as DnstapMessage, SocketFamily,
     SocketProtocol,
@@ -38,7 +38,7 @@ use vector_lib::config::log_schema;
 use vector_lib::lookup::lookup_v2::ValuePath;
 use vector_lib::lookup::PathPrefix;
 
-use super::{
+use dnsmsg_parser::{
     dns_message::{
         DnsRecord, EdnsOptionEntry, OptPseudoSection, QueryHeader, QueryQuestion, UpdateHeader,
         ZoneInfo,
@@ -459,14 +459,15 @@ impl DnstapParser {
             );
         }
 
-        Ok(if let Some(response_port) = dnstap_message.response_port {
+        if let Some(response_port) = dnstap_message.response_port {
             DnstapParser::insert(
                 event,
                 prefix.clone(),
                 &DNSTAP_VALUE_PATHS.response_port,
                 response_port,
             );
-        })
+        };
+        Ok(())
     }
 
     fn log_time<'a>(
@@ -1015,7 +1016,6 @@ fn to_dnstap_message_type(type_id: i32) -> String {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::event::Value;
     use chrono::DateTime;
     use dnsmsg_parser::dns_message_parser::DnsParserOptions;
     use std::collections::BTreeMap;

src/sources/dnstap/schema.rs → lib/dnstap-parser/src/schema.rs

@@ -12,7 +12,7 @@ pub struct DnstapEventSchema;
 
 impl DnstapEventSchema {
     /// The message schema for the request and response message fields
-    fn request_message_schema_definition(&self) -> Collection<Field> {
+    pub(crate) fn request_message_schema_definition(&self) -> Collection<Field> {
         let mut result: BTreeMap<Field, Kind> = BTreeMap::new();
         result.insert(
             DNSTAP_VALUE_PATHS.time.to_string().into(),
@@ -288,7 +288,7 @@ pub struct DnstapPaths {
 }
 
 /// Lazily initialized singleton.
-pub(crate) static DNSTAP_VALUE_PATHS: LazyLock<DnstapPaths> = LazyLock::new(|| DnstapPaths {
+pub static DNSTAP_VALUE_PATHS: LazyLock<DnstapPaths> = LazyLock::new(|| DnstapPaths {
     server_identity: owned_value_path!("serverId"),
     server_version: owned_value_path!("serverVersion"),
     extra: owned_value_path!("extraInfo"),

lib/dnstap-parser/src/vrl_functions/mod.rs

@@ -0,0 +1,7 @@
+use vrl::compiler::Function;
+
+pub mod parse_dnstap;
+
+pub fn all() -> Vec<Box<dyn Function>> {
+    vec![Box::new(parse_dnstap::ParseDnstap) as _]
+}