diff --git a/.github/workflows/changes.yml b/.github/workflows/changes.yml index 4cd9045342e97..972bac2b3cf37 100644 --- a/.github/workflows/changes.yml +++ b/.github/workflows/changes.yml @@ -144,6 +144,8 @@ on: value: ${{ jobs.int_tests.outputs.redis }} splunk: value: ${{ jobs.int_tests.outputs.splunk }} + syslog: + value: ${{ jobs.int_tests.outputs.syslog }} webhdfs: value: ${{ jobs.int_tests.outputs.webhdfs }} # e2e tests @@ -404,6 +406,7 @@ jobs: pulsar: ${{ steps.filter.outputs.pulsar }} redis: ${{ steps.filter.outputs.redis }} splunk: ${{ steps.filter.outputs.splunk }} + syslog: ${{ steps.filter.outputs.syslog }} webhdfs: ${{ steps.filter.outputs.webhdfs }} any: ${{ steps.detect-changes.outputs.any }} steps: @@ -470,6 +473,7 @@ jobs: "pulsar": ${{ steps.filter.outputs.pulsar }}, "redis": ${{ steps.filter.outputs.redis }}, "splunk": ${{ steps.filter.outputs.splunk }}, + "syslog": ${{ steps.filter.outputs.syslog }}, "webhdfs": ${{ steps.filter.outputs.webhdfs }} } EOF diff --git a/.github/workflows/ci-integration-review.yml b/.github/workflows/ci-integration-review.yml index ee58ad0d7fdf5..589b237426522 100644 --- a/.github/workflows/ci-integration-review.yml +++ b/.github/workflows/ci-integration-review.yml @@ -141,6 +141,7 @@ jobs: "pulsar", "redis", "splunk", + "syslog", "webhdfs" ] steps: diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml index 9794672002cd0..62cc5ec7159ea 100644 --- a/.github/workflows/integration.yml +++ b/.github/workflows/integration.yml @@ -133,6 +133,7 @@ jobs: "pulsar", "redis", "splunk", + "syslog", "webhdfs" ] timeout-minutes: 90 diff --git a/Cargo.toml b/Cargo.toml index a23e0c6df3293..cfc8090c27e83 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -938,6 +938,7 @@ sinks-logs = [ "sinks-sematext", "sinks-socket", "sinks-splunk_hec", + "sinks-syslog", "sinks-vector", "sinks-webhdfs", "sinks-websocket", @@ -1011,6 +1012,7 @@ sinks-sematext = ["sinks-elasticsearch", "sinks-influxdb"] sinks-socket = ["sinks-utils-udp"] sinks-splunk_hec = [] sinks-statsd = ["sinks-utils-udp", "tokio-util/net"] +sinks-syslog = ["codecs-syslog", "sinks-utils-udp"] sinks-utils-udp = [] sinks-vector = ["sinks-utils-udp", "dep:tonic", "protobuf-build", "dep:prost"] sinks-websocket = ["dep:tokio-tungstenite"] @@ -1060,6 +1062,7 @@ all-integration-tests = [ "pulsar-integration-tests", "redis-integration-tests", "splunk-integration-tests", + "syslog-integration-tests", "vector-api-tests", "webhdfs-integration-tests", "windows-event-log-integration-tests", @@ -1129,6 +1132,7 @@ prometheus-integration-tests = ["sinks-prometheus", "sources-prometheus", "sinks pulsar-integration-tests = ["sinks-pulsar", "sources-pulsar"] redis-integration-tests = ["sinks-redis", "sources-redis"] splunk-integration-tests = ["sinks-splunk_hec"] +syslog-integration-tests = ["sinks-syslog"] dnstap-integration-tests = ["sources-dnstap", "dep:bollard"] webhdfs-integration-tests = ["sinks-webhdfs"] windows-event-log-integration-tests = ["sources-windows_event_log-integration-tests"] diff --git a/changelog.d/6863_syslog_sink.feature.md b/changelog.d/6863_syslog_sink.feature.md new file mode 100644 index 0000000000000..c228e94bc0d98 --- /dev/null +++ b/changelog.d/6863_syslog_sink.feature.md @@ -0,0 +1,6 @@ +Added a new `syslog` sink for sending log events in syslog format (RFC 5424 or RFC 3164) over TCP, +UDP, or Unix stream sockets, with configurable stream framing, TLS support, configurable facility, +severity, app_name, proc_id, and msg_id field mappings, and improved syslog encoder compatibility +for header fields, structured-data names, and common facility/severity aliases. + +authors: tot19 diff --git a/lib/codecs/src/encoding/format/mod.rs b/lib/codecs/src/encoding/format/mod.rs index f760ea5c8dd83..ed4bc3edcac14 100644 --- a/lib/codecs/src/encoding/format/mod.rs +++ b/lib/codecs/src/encoding/format/mod.rs @@ -48,7 +48,7 @@ pub use otlp::{OtlpSerializer, OtlpSerializerConfig}; pub use protobuf::{ProtobufSerializer, ProtobufSerializerConfig, ProtobufSerializerOptions}; pub use raw_message::{RawMessageSerializer, RawMessageSerializerConfig}; #[cfg(feature = "syslog")] -pub use syslog::{SyslogSerializer, SyslogSerializerConfig}; +pub use syslog::{SyslogSerializer, SyslogSerializerConfig, SyslogSerializerOptions}; pub use text::{TextSerializer, TextSerializerConfig}; use vector_core::event::Event; diff --git a/lib/codecs/src/encoding/format/syslog.rs b/lib/codecs/src/encoding/format/syslog.rs index 7cf7147fcdb5d..f8c1806b8fc7d 100644 --- a/lib/codecs/src/encoding/format/syslog.rs +++ b/lib/codecs/src/encoding/format/syslog.rs @@ -5,8 +5,7 @@ use serde_json; use std::borrow::Cow; use std::collections::BTreeMap; use std::fmt::Write; -use std::str::FromStr; -use strum::{EnumString, FromRepr, VariantNames}; +use strum::FromRepr; use tokio_util::codec::Encoder; use tracing::debug; use vector_config::configurable_component; @@ -89,11 +88,13 @@ impl Encoder for SyslogSerializer { type Error = vector_common::Error; fn encode(&mut self, event: Event, buffer: &mut BytesMut) -> Result<(), Self::Error> { - if let Event::Log(log_event) = event { - let syslog_message = ConfigDecanter::new(&log_event).decant_config(&self.config.syslog); - let encoded = syslog_message.encode(&self.config.syslog.rfc); - buffer.put_slice(encoded.as_bytes()); - } + let Event::Log(log_event) = event else { + return Err("Syslog serializer only supports log events.".into()); + }; + + let syslog_message = ConfigDecanter::new(&log_event).decant_config(&self.config.syslog); + let encoded = syslog_message.encode(&self.config.syslog.rfc); + buffer.put_slice(encoded.as_bytes()); Ok(()) } @@ -109,37 +110,39 @@ impl<'a> ConfigDecanter<'a> { } fn decant_config(&self, config: &SyslogSerializerOptions) -> SyslogMessage { - let mut app_name = self - .get_value(config.app_name.as_ref()) // P1: Configured path - .unwrap_or_else(|| { - // P2: Semantic fallback to the field designated as "service" in the schema. - // P3: Fall back to "vector" when that field is absent. - self.log - .get_by_meaning("service") - .map_or_else(|| "vector".to_owned(), |v| v.to_string_lossy().to_string()) - }); + let mut app_name = self.get_value(config.app_name.as_ref()).unwrap_or_else(|| { + self.log + .get_by_meaning("service") + .map_or_else(|| "vector".to_owned(), |v| v.to_string_lossy().into_owned()) + }); let mut proc_id = self.get_value(config.proc_id.as_ref()); let mut msg_id = self.get_value(config.msg_id.as_ref()); + let hostname = self + .log + .get_host() + .map(|v| v.to_string_lossy().into_owned()); match config.rfc { SyslogRFC::Rfc3164 => { - // RFC 3164: TAG field (app_name and proc_id) must be ASCII printable - app_name = sanitize_to_ascii(&app_name).into_owned(); - if let Some(pid) = &mut proc_id { - *pid = sanitize_to_ascii(pid).into_owned(); - } + // RFC 3164: the TAG (app_name and proc_id) must be printable + // US-ASCII without spaces, since a space terminates the TAG on + // most receivers. RFC 3164 has no NILVALUE, so an app_name + // that is empty after sanitization falls back to the default. + app_name = sanitize_printusascii_field(app_name, RFC3164_TAG_MAX_LENGTH) + .unwrap_or_else(|| "vector".to_owned()); + proc_id = proc_id + .and_then(|pid| sanitize_printusascii_field(pid, RFC3164_TAG_MAX_LENGTH)); } SyslogRFC::Rfc5424 => { - // Truncate to character limits (not byte limits to avoid UTF-8 panics) - truncate_chars(&mut app_name, 48); - if let Some(pid) = &mut proc_id { - truncate_chars(pid, 128); - } - if let Some(mid) = &mut msg_id { - truncate_chars(mid, 32); - } + app_name = sanitize_printusascii_field(app_name, APP_NAME_MAX_LENGTH) + .unwrap_or_else(|| NIL_VALUE.to_owned()); + proc_id = + proc_id.and_then(|pid| sanitize_printusascii_field(pid, PROC_ID_MAX_LENGTH)); + msg_id = msg_id.and_then(|mid| sanitize_printusascii_field(mid, MSG_ID_MAX_LENGTH)); } } + let hostname = + hostname.and_then(|host| sanitize_printusascii_field(host, HOSTNAME_MAX_LENGTH)); SyslogMessage { pri: Pri { @@ -147,7 +150,7 @@ impl<'a> ConfigDecanter<'a> { severity: self.get_severity(config), }, timestamp: self.get_timestamp(), - hostname: self.log.get_host().map(|v| v.to_string_lossy().to_string()), + hostname, tag: Tag { app_name, proc_id, @@ -159,14 +162,14 @@ impl<'a> ConfigDecanter<'a> { } fn get_value(&self, path: Option<&ConfigTargetPath>) -> Option { - path.and_then(|p| self.log.get(p).cloned()) - .map(|v| v.to_string_lossy().to_string()) + path.and_then(|p| self.log.get(p)) + .map(|v| v.to_string_lossy().into_owned()) } fn get_structured_data(&self) -> Option { self.log .get(event_path!("structured_data")) - .and_then(|v| v.clone().into_object()) + .and_then(|v| v.as_object().cloned()) .map(StructuredData::from) } @@ -180,13 +183,16 @@ impl<'a> ConfigDecanter<'a> { fn get_payload(&self) -> String { self.log .get_message() - .map(|v| v.to_string_lossy().to_string()) + .map(|v| v.to_string_lossy().into_owned()) .unwrap_or_default() } fn get_facility(&self, config: &SyslogSerializerOptions) -> Facility { config.facility.as_ref().map_or(Facility::User, |path| { - self.get_syslog_code(path, Facility::from_repr, Facility::User) + self.log + .get(path) + .and_then(parse_facility) + .unwrap_or(Facility::User) }) } @@ -195,51 +201,26 @@ impl<'a> ConfigDecanter<'a> { .severity .as_ref() .map_or(Severity::Informational, |path| { - self.get_syslog_code(path, Severity::from_repr, Severity::Informational) + self.log + .get(path) + .and_then(parse_severity) + .unwrap_or(Severity::Informational) }) } - - // https://github.com/vectordotdev/vector/issues/23659 - #[allow( - clippy::cast_possible_truncation, - reason = "Preserve the existing numeric conversion and wire-format behavior." - )] - #[allow( - clippy::cast_sign_loss, - reason = "Preserve the existing numeric fallback behavior." - )] - fn get_syslog_code( - &self, - path: &ConfigTargetPath, - from_repr_fn: fn(usize) -> Option, - default_value: T, - ) -> T - where - T: Copy + FromStr, - { - if let Some(value) = self.log.get(path).cloned() { - let s = value.to_string_lossy(); - if let Ok(val_from_name) = s.to_ascii_lowercase().parse::() { - return val_from_name; - } - if let Value::Integer(n) = value - && let Some(val_from_num) = from_repr_fn(n as usize) - { - return val_from_num; - } - } - default_value - } } const NIL_VALUE: &str = "-"; const SYSLOG_V1: &str = "1"; const RFC3164_TAG_MAX_LENGTH: usize = 32; -const SD_ID_MAX_LENGTH: usize = 32; +const HOSTNAME_MAX_LENGTH: usize = 255; +const APP_NAME_MAX_LENGTH: usize = 48; +const PROC_ID_MAX_LENGTH: usize = 128; +const MSG_ID_MAX_LENGTH: usize = 32; +const SD_NAME_MAX_LENGTH: usize = 32; -/// Replaces invalid characters with '_' +/// Replaces invalid characters with `replacement` #[inline] -fn sanitize_with(s: &str, is_valid: F) -> Cow<'_, str> +fn sanitize_with(s: &str, is_valid: F, replacement: char) -> Cow<'_, str> where F: Fn(char) -> bool, { @@ -250,7 +231,7 @@ where let (valid_prefix, remainder) = s.split_at(first_invalid_idx); result.push_str(valid_prefix); for c in remainder.chars() { - result.push(if is_valid(c) { c } else { '_' }); + result.push(if is_valid(c) { c } else { replacement }); } Cow::Owned(result) @@ -258,12 +239,21 @@ where } } -/// Sanitize a string to ASCII printable characters (space to tilde, ASCII 32-126) -/// Used for RFC 3164 TAG field (`app_name` and `proc_id`) -/// Invalid characters are replaced with '_' +/// Sanitize a PRINTUSASCII header field, replacing spaces and control +/// characters with underscores and returning `None` for empty fields. #[inline] -fn sanitize_to_ascii(s: &str) -> Cow<'_, str> { - sanitize_with(s, |c| (' '..='~').contains(&c)) +fn sanitize_printusascii_field(s: String, max_chars: usize) -> Option { + let mut sanitized = match sanitize_with(&s, |c| c.is_ascii_graphic(), '_') { + Cow::Borrowed(_) => s, // Already clean, reuse the caller's allocation + Cow::Owned(owned) => owned, + }; + truncate_chars(&mut sanitized, max_chars); + + if sanitized.is_empty() { + None + } else { + Some(sanitized) + } } /// Sanitize SD-ID or PARAM-NAME according to RFC 5424 @@ -272,9 +262,11 @@ fn sanitize_to_ascii(s: &str) -> Cow<'_, str> { /// Invalid characters are replaced with '_' #[inline] fn sanitize_name(name: &str) -> Cow<'_, str> { - sanitize_with(name, |c| { - c.is_ascii_graphic() && !matches!(c, '=' | ']' | '"') - }) + sanitize_with( + name, + |c| c.is_ascii_graphic() && !matches!(c, '=' | ']' | '"'), + '_', + ) } /// Escape PARAM-VALUE according to RFC 5424 @@ -307,6 +299,156 @@ fn truncate_chars(s: &mut String, max_chars: usize) { } } +fn truncate_to_string(s: Cow<'_, str>, max_chars: usize) -> String { + let mut s = s.into_owned(); + truncate_chars(&mut s, max_chars); + s +} + +fn sanitize_sd_name(name: &str) -> Option { + let sanitized = truncate_to_string(sanitize_name(name), SD_NAME_MAX_LENGTH); + + if sanitized.is_empty() { + None + } else { + Some(sanitized) + } +} + +/// Joins an already-sanitized leaf name onto a (possibly empty) dotted prefix, +/// keeping the result within `SD_NAME_MAX_LENGTH` characters. The prefix is +/// truncated rather than the leaf so that sibling keys under a long path stay +/// distinguishable. +fn join_sd_name(prefix: &str, leaf: String) -> String { + if prefix.is_empty() { + return leaf; + } + + let prefix_budget = SD_NAME_MAX_LENGTH.saturating_sub(leaf.chars().count() + 1); + if prefix_budget == 0 { + return leaf; + } + + let mut joined = prefix.to_owned(); + truncate_chars(&mut joined, prefix_budget); + joined.push('.'); + joined.push_str(&leaf); + joined +} + +fn deduped_key(map: &BTreeMap, base: String) -> String { + if !map.contains_key(&base) { + return base; + } + + for index in 1usize.. { + let suffix = format!("_{index}"); + let mut candidate = base.clone(); + truncate_chars( + &mut candidate, + SD_NAME_MAX_LENGTH.saturating_sub(suffix.len()), + ); + candidate.push_str(&suffix); + + if !map.contains_key(&candidate) { + return candidate; + } + } + + unreachable!("usize iteration should produce a unique structured-data name") +} + +fn insert_structured_param(result: &mut BTreeMap, key: String, value: String) { + let key = deduped_key(result, key); + result.insert(key, value); +} + +/// Resolves a facility or severity from an event value. +/// +/// Accepts integers, numeric strings, and case-insensitive names, converting the +/// value to a string at most once and without allocating for valid UTF-8 bytes. +fn parse_syslog_code( + value: &Value, + from_repr: fn(usize) -> Option, + names: &[(&str, T)], +) -> Option { + if let Value::Integer(n) = value { + return usize::try_from(*n).ok().and_then(from_repr); + } + + let raw = value.to_string_lossy(); + let name = raw.trim(); + if let Ok(n) = name.parse::() { + return from_repr(n); + } + + names + .iter() + .find(|(candidate, _)| candidate.eq_ignore_ascii_case(name)) + .map(|(_, code)| *code) +} + +const FACILITY_NAMES: &[(&str, Facility)] = &[ + ("kern", Facility::Kern), + ("kernel", Facility::Kern), + ("user", Facility::User), + ("mail", Facility::Mail), + ("daemon", Facility::Daemon), + ("auth", Facility::Auth), + ("syslog", Facility::Syslog), + ("lpr", Facility::Lpr), + ("news", Facility::News), + ("uucp", Facility::Uucp), + ("cron", Facility::Cron), + ("authpriv", Facility::Authpriv), + ("auth-priv", Facility::Authpriv), + ("auth_priv", Facility::Authpriv), + ("ftp", Facility::Ftp), + ("ntp", Facility::Ntp), + ("security", Facility::Security), + ("audit", Facility::Security), + ("log-alert", Facility::LogAlert), + ("log_alert", Facility::LogAlert), + ("alert", Facility::LogAlert), + ("console", Facility::LogAlert), + ("solaris-cron", Facility::SolarisCron), + ("solaris_cron", Facility::SolarisCron), + ("clockd", Facility::SolarisCron), + ("local0", Facility::Local0), + ("local1", Facility::Local1), + ("local2", Facility::Local2), + ("local3", Facility::Local3), + ("local4", Facility::Local4), + ("local5", Facility::Local5), + ("local6", Facility::Local6), + ("local7", Facility::Local7), +]; + +const SEVERITY_NAMES: &[(&str, Severity)] = &[ + ("emergency", Severity::Emergency), + ("emerg", Severity::Emergency), + ("panic", Severity::Emergency), + ("alert", Severity::Alert), + ("critical", Severity::Critical), + ("crit", Severity::Critical), + ("error", Severity::Error), + ("err", Severity::Error), + ("warning", Severity::Warning), + ("warn", Severity::Warning), + ("notice", Severity::Notice), + ("informational", Severity::Informational), + ("info", Severity::Informational), + ("debug", Severity::Debug), +]; + +fn parse_facility(value: &Value) -> Option { + parse_syslog_code(value, Facility::from_repr, FACILITY_NAMES) +} + +fn parse_severity(value: &Value) -> Option { + parse_syslog_code(value, Severity::from_repr, SEVERITY_NAMES) +} + /// The syslog RFC standard to use for formatting. #[configurable_component] #[derive(PartialEq, Clone, Debug, Default)] @@ -333,7 +475,7 @@ impl SyslogMessage { fn encode(&self, rfc: &SyslogRFC) -> String { let mut result = String::with_capacity(256); - result.push_str(&self.pri.encode()); + write!(result, "<{}>", self.pri.prival()).expect("write to String never fails"); if *rfc == SyslogRFC::Rfc5424 { result.push_str(SYSLOG_V1); @@ -343,7 +485,7 @@ impl SyslogMessage { match rfc { SyslogRFC::Rfc3164 => { write!(result, "{} ", self.timestamp.format("%b %e %H:%M:%S")) - .expect("writing to a String cannot fail"); + .expect("write to String never fails"); } SyslogRFC::Rfc5424 => { result.push_str( @@ -360,47 +502,46 @@ impl SyslogMessage { result.push(' '); match rfc { - SyslogRFC::Rfc3164 => result.push_str(&self.tag.encode_rfc_3164()), - SyslogRFC::Rfc5424 => result.push_str(&self.tag.encode_rfc_5424()), - } - result.push(' '); + SyslogRFC::Rfc3164 => { + result.push_str(&self.tag.encode_rfc_3164()); + + // RFC 3164 does not support structured data + if self + .structured_data + .as_ref() + .is_some_and(|sd| !sd.elements.is_empty()) + { + debug!( + "Structured data present but ignored - RFC 3164 does not support structured data. Consider using RFC 5424 instead." + ); + } - if *rfc == SyslogRFC::Rfc3164 { - // RFC 3164 does not support structured data - if let Some(sd) = &self.structured_data - && !sd.elements.is_empty() - { - debug!( - "Structured data present but ignored - RFC 3164 does not support structured data. Consider using RFC 5424 instead." - ); - } - } else { - if let Some(sd) = &self.structured_data { - result.push_str(&sd.encode()); - } else { - result.push_str(NIL_VALUE); + if !self.message.is_empty() { + result.push(' '); + result.push_str(&Self::sanitize_rfc3164_message(&self.message)); + } } - if !self.message.is_empty() { + SyslogRFC::Rfc5424 => { + result.push_str(&self.tag.encode_rfc_5424()); result.push(' '); - } - } - if !self.message.is_empty() { - if *rfc == SyslogRFC::Rfc3164 { - result.push_str(&Self::sanitize_rfc3164_message(&self.message)); - } else { - result.push_str(&self.message); + match &self.structured_data { + Some(sd) => result.push_str(&sd.encode()), + None => result.push_str(NIL_VALUE), + } + + if !self.message.is_empty() { + result.push(' '); + result.push_str(&self.message); + } } } result } - fn sanitize_rfc3164_message(message: &str) -> String { - message - .chars() - .map(|ch| if (' '..='~').contains(&ch) { ch } else { ' ' }) - .collect() + fn sanitize_rfc3164_message(message: &str) -> Cow<'_, str> { + sanitize_with(message, |ch| (' '..='~').contains(&ch), ' ') } } @@ -413,19 +554,28 @@ struct Tag { impl Tag { fn encode_rfc_3164(&self) -> String { - let mut tag = if let Some(proc_id) = self.proc_id.as_deref() { - format!("{}[{proc_id}]:", self.app_name) - } else { - format!("{}:", self.app_name) - }; - if tag.chars().count() > RFC3164_TAG_MAX_LENGTH { - truncate_chars(&mut tag, RFC3164_TAG_MAX_LENGTH); - if !tag.ends_with(':') { - tag.pop(); - tag.push(':'); + // The TAG is limited to `RFC3164_TAG_MAX_LENGTH` characters, including + // the terminating ':'. When the full tag does not fit, truncate the + // app_name and keep "[proc_id]" intact; if even that is impossible, + // drop the proc_id entirely rather than emit a corrupted, unbalanced + // "app[123:" tag. + if let Some(proc_id) = self.proc_id.as_deref() { + let tag = format!("{}[{proc_id}]:", self.app_name); + if tag.chars().count() <= RFC3164_TAG_MAX_LENGTH { + return tag; + } + + let bracket_chars = proc_id.chars().count() + "[]:".len(); + if bracket_chars < RFC3164_TAG_MAX_LENGTH { + let mut app_name = self.app_name.clone(); + truncate_chars(&mut app_name, RFC3164_TAG_MAX_LENGTH - bracket_chars); + return format!("{app_name}[{proc_id}]:"); } } - tag + + let mut app_name = self.app_name.clone(); + truncate_chars(&mut app_name, RFC3164_TAG_MAX_LENGTH - 1); + format!("{app_name}:") } fn encode_rfc_5424(&self) -> String { @@ -449,11 +599,14 @@ impl StructuredData { self.elements .iter() .fold(String::new(), |mut acc, (sd_id, sd_params)| { - write!(acc, "[{sd_id}").expect("writing to a String cannot fail"); + acc.push('['); + acc.push_str(sd_id); for (key, value) in sd_params { - let esc_val = escape_sd_value(value); - write!(acc, " {key}=\"{esc_val}\"") - .expect("writing to a String cannot fail"); + acc.push(' '); + acc.push_str(key); + acc.push_str("=\""); + acc.push_str(&escape_sd_value(value)); + acc.push('"'); } acc.push(']'); acc @@ -464,33 +617,31 @@ impl StructuredData { impl From for StructuredData { fn from(fields: ObjectMap) -> Self { - let elements = fields - .into_iter() - .map(|(sd_id, value)| { - let sd_id_str: String = sd_id.into(); - let sanitized_id = sanitize_name(&sd_id_str); + let mut elements = BTreeMap::new(); + + for (sd_id, value) in fields { + let sd_id_str: String = sd_id.into(); + let Some(final_id) = sanitize_sd_name(&sd_id_str) else { + debug!("Dropping structured-data element whose SD-ID is empty after sanitization."); + continue; + }; + let final_id = deduped_key(&elements, final_id); + + let sd_params = match value { + Value::Object(obj) => { + let mut map = BTreeMap::new(); + flatten_object(obj, String::new(), &mut map); + map + } + scalar => { + let mut map = BTreeMap::new(); + map.insert("value".to_string(), scalar.to_string_lossy().to_string()); + map + } + }; + elements.insert(final_id, sd_params); + } - let final_id = if sanitized_id.chars().count() > SD_ID_MAX_LENGTH { - sanitized_id.chars().take(SD_ID_MAX_LENGTH).collect() - } else { - sanitized_id.into_owned() - }; - - let sd_params = match value { - Value::Object(obj) => { - let mut map = BTreeMap::new(); - flatten_object(obj, String::new(), &mut map); - map - } - scalar => { - let mut map = BTreeMap::new(); - map.insert("value".to_string(), scalar.to_string_lossy().to_string()); - map - } - }; - (final_id, sd_params) - }) - .collect(); Self { elements } } } @@ -504,14 +655,12 @@ impl From for StructuredData { fn flatten_object(obj: ObjectMap, prefix: String, result: &mut BTreeMap) { for (key, value) in obj { let key_str: String = key.into(); + let Some(sanitized_key) = sanitize_sd_name(&key_str) else { + debug!("Dropping structured-data param whose name is empty after sanitization."); + continue; + }; - let sanitized_key = sanitize_name(&key_str); - - let mut full_key = prefix.clone(); - if !full_key.is_empty() { - full_key.push('.'); - } - full_key.push_str(&sanitized_key); + let full_key = join_sd_name(&prefix, sanitized_key); match value { Value::Object(nested) => { @@ -519,13 +668,13 @@ fn flatten_object(obj: ObjectMap, prefix: String, result: &mut BTreeMap { if let Ok(json) = serde_json::to_string(&arr) { - result.insert(full_key, json); + insert_structured_param(result, full_key, json); } else { - result.insert(full_key, format!("{arr:?}")); + insert_structured_param(result, full_key, format!("{arr:?}")); } } scalar => { - result.insert(full_key, scalar.to_string_lossy().to_string()); + insert_structured_param(result, full_key, scalar.to_string_lossy().to_string()); } } } @@ -540,15 +689,17 @@ struct Pri { impl Pri { // The last paragraph describes how to compose the enums into `PRIVAL`: // https://datatracker.ietf.org/doc/html/rfc5424#section-6.2.1 - fn encode(&self) -> String { - let pri_val = (self.facility as u8 * 8) + self.severity as u8; - format!("<{pri_val}>") + fn prival(&self) -> u8 { + (self.facility as u8 * 8) + self.severity as u8 } } /// Syslog facility -#[derive(Default, Debug, EnumString, FromRepr, VariantNames, Copy, Clone, PartialEq, Eq)] -#[strum(serialize_all = "kebab-case")] +/// +/// Facility and severity names (including aliases) are resolved at runtime by +/// `parse_facility`/`parse_severity`, which are the single source of truth for +/// recognized spellings. +#[derive(Default, Debug, FromRepr, Copy, Clone, PartialEq, Eq)] #[configurable_component] pub enum Facility { /// Kern @@ -580,8 +731,8 @@ pub enum Facility { Ntp = 12, /// Security Security = 13, - /// Console - Console = 14, + /// Log alert + LogAlert = 14, /// `SolarisCron` SolarisCron = 15, /// Local0 @@ -603,8 +754,7 @@ pub enum Facility { } /// Syslog severity -#[derive(Default, Debug, EnumString, FromRepr, VariantNames, Copy, Clone, PartialEq, Eq)] -#[strum(serialize_all = "kebab-case")] +#[derive(Default, Debug, FromRepr, Copy, Clone, PartialEq, Eq)] #[configurable_component] pub enum Severity { /// Emergency @@ -657,7 +807,15 @@ mod tests { } fn create_simple_log() -> LogEvent { - let mut log = LogEvent::from("original message"); + create_simple_log_with_message("original message") + } + + // Sets the message at construction time instead of overwriting the + // literal "message" field afterwards: the decoder tests in this crate + // mutate the process-global log schema's message key, so path-based + // message overrides race with them under parallel test execution. + fn create_simple_log_with_message(message: &str) -> LogEvent { + let mut log = LogEvent::from(message); log.insert( event_path!("timestamp"), NaiveDate::from_ymd_opt(2025, 8, 28) @@ -747,7 +905,7 @@ mod tests { toml::from_str::(r#"facility = ".syslog_facility""#).unwrap(); let config_sev = toml::from_str::(r#"severity = ".syslog_severity""#).unwrap(); - //check lowercase and digit + log.insert(event_path!("syslog_facility"), "daemon"); log.insert(event_path!("syslog_severity"), "critical"); let decanter = ConfigDecanter::new(&log); @@ -765,7 +923,6 @@ mod tests { assert_eq!(facility, Facility::Daemon); assert_eq!(severity, Severity::Critical); - //check digit log.insert(event_path!("syslog_facility"), Value::from(3u8)); log.insert(event_path!("syslog_severity"), Value::from(2u8)); let decanter = ConfigDecanter::new(&log); @@ -774,49 +931,89 @@ mod tests { assert_eq!(facility, Facility::Daemon); assert_eq!(severity, Severity::Critical); - //check short-form severity aliases - log.insert(event_path!("syslog_severity"), "crit"); - let decanter = ConfigDecanter::new(&log); - assert_eq!(decanter.get_severity(&config_sev), Severity::Critical); - - log.insert(event_path!("syslog_severity"), "emerg"); - let decanter = ConfigDecanter::new(&log); - assert_eq!(decanter.get_severity(&config_sev), Severity::Emergency); - - log.insert(event_path!("syslog_severity"), "err"); - let decanter = ConfigDecanter::new(&log); - assert_eq!(decanter.get_severity(&config_sev), Severity::Error); - - log.insert(event_path!("syslog_severity"), "info"); - let decanter = ConfigDecanter::new(&log); - assert_eq!(decanter.get_severity(&config_sev), Severity::Informational); - - log.insert(event_path!("syslog_severity"), "warn"); + log.insert(event_path!("syslog_facility"), "3"); + log.insert(event_path!("syslog_severity"), "2"); let decanter = ConfigDecanter::new(&log); - assert_eq!(decanter.get_severity(&config_sev), Severity::Warning); - - log.insert(event_path!("syslog_severity"), "panic"); - let decanter = ConfigDecanter::new(&log); - assert_eq!(decanter.get_severity(&config_sev), Severity::Emergency); + let facility = decanter.get_facility(&config_fac); + let severity = decanter.get_severity(&config_sev); + assert_eq!(facility, Facility::Daemon); + assert_eq!(severity, Severity::Critical); - //check uppercase short-form aliases - log.insert(event_path!("syslog_severity"), "CRIT"); - let decanter = ConfigDecanter::new(&log); - assert_eq!(decanter.get_severity(&config_sev), Severity::Critical); + for (name, expected) in [ + ("kern", Facility::Kern), + ("kernel", Facility::Kern), + ("audit", Facility::Security), + ("alert", Facility::LogAlert), + ("log-alert", Facility::LogAlert), + ("clockd", Facility::SolarisCron), + ("local7", Facility::Local7), + ] { + log.insert(event_path!("syslog_facility"), name); + let decanter = ConfigDecanter::new(&log); + assert_eq!(decanter.get_facility(&config_fac), expected); + } - log.insert(event_path!("syslog_severity"), "EMERG"); - let decanter = ConfigDecanter::new(&log); - assert_eq!(decanter.get_severity(&config_sev), Severity::Emergency); + for (name, expected) in [ + ("emerg", Severity::Emergency), + ("emergency", Severity::Emergency), + ("panic", Severity::Emergency), + ("EMERG", Severity::Emergency), + ("CRIT", Severity::Critical), + ("crit", Severity::Critical), + ("critical", Severity::Critical), + ("err", Severity::Error), + ("error", Severity::Error), + ("warn", Severity::Warning), + ("warning", Severity::Warning), + ("info", Severity::Informational), + ("informational", Severity::Informational), + ] { + log.insert(event_path!("syslog_severity"), name); + let decanter = ConfigDecanter::new(&log); + assert_eq!(decanter.get_severity(&config_sev), expected); + } //check defaults with empty config let empty_config = toml::from_str::(r#"facility = ".missing_field""#).unwrap(); + let decanter = ConfigDecanter::new(&log); let default_facility = decanter.get_facility(&empty_config); let default_severity = decanter.get_severity(&empty_config); assert_eq!(default_facility, Facility::User); assert_eq!(default_severity, Severity::Informational); } + #[test] + fn test_pri_aliases_from_syslog_decoder_names() { + let config = toml::from_str::( + r#" + [syslog] + rfc = "rfc5424" + facility = ".facility" + severity = ".severity" + "#, + ) + .unwrap(); + + let mut log = create_simple_log(); + log.insert(event_path!("facility"), "audit"); + log.insert(event_path!("severity"), "info"); + let output = run_encode(config.clone(), Event::Log(log)); + assert!(output.starts_with("<110>1"), "unexpected PRI: {output}"); + + let mut log = create_simple_log(); + log.insert(event_path!("facility"), "clockd"); + log.insert(event_path!("severity"), "err"); + let output = run_encode(config.clone(), Event::Log(log)); + assert!(output.starts_with("<123>1"), "unexpected PRI: {output}"); + + let mut log = create_simple_log(); + log.insert(event_path!("facility"), "3"); + log.insert(event_path!("severity"), "2"); + let output = run_encode(config, Event::Log(log)); + assert!(output.starts_with("<26>1"), "unexpected PRI: {output}"); + } + #[test] fn test_rfc3164_sanitization() { let config = toml::from_str::( @@ -827,11 +1024,8 @@ mod tests { ) .unwrap(); - let mut log = create_simple_log(); - log.insert( - event_path!("message"), - "A\nB\tC, Привіт D, E\u{0007}F", //newline, tab, unicode - ); + //newline, tab, unicode + let log = create_simple_log_with_message("A\nB\tC, Привіт D, E\u{0007}F"); let output = run_encode(config, Event::Log(log)); let expected_message = "A B C, D, E F"; @@ -866,6 +1060,72 @@ mod tests { assert_eq!(message.tag.msg_id.unwrap().len(), 32); } + #[test] + fn test_rfc5424_header_field_sanitization() { + let config = toml::from_str::( + r#" + [syslog] + rfc = "rfc5424" + app_name = ".app" + proc_id = ".proc" + msg_id = ".msg" + "#, + ) + .unwrap(); + + let mut log = create_simple_log(); + log.insert(event_path!("host"), "bad host\n😀"); + log.insert(event_path!("app"), "my app\n"); + log.insert(event_path!("proc"), "proc\tid"); + log.insert(event_path!("msg"), "msg id\u{0007}"); + + let output = run_encode(config, Event::Log(log)); + let expected = "<14>1 2025-08-28T18:30:00.123456Z bad_host__ my_app_ proc_id msg_id_ - original message"; + assert_eq!(output, expected); + } + + #[test] + fn test_rfc5424_empty_header_fields_use_nilvalue() { + let config = toml::from_str::( + r#" + [syslog] + rfc = "rfc5424" + app_name = ".app" + proc_id = ".proc" + msg_id = ".msg" + "#, + ) + .unwrap(); + + let mut log = create_simple_log(); + log.insert(event_path!("host"), ""); + log.insert(event_path!("app"), ""); + log.insert(event_path!("proc"), ""); + log.insert(event_path!("msg"), ""); + + let output = run_encode(config, Event::Log(log)); + let expected = "<14>1 2025-08-28T18:30:00.123456Z - - - - - original message"; + assert_eq!(output, expected); + } + + #[test] + fn test_rfc3164_hostname_sanitization() { + let config = toml::from_str::( + r#" + [syslog] + rfc = "rfc3164" + "#, + ) + .unwrap(); + + let mut log = create_simple_log(); + log.insert(event_path!("host"), "bad host\n😀"); + + let output = run_encode(config, Event::Log(log)); + let expected = "<14>Aug 28 18:30:00 bad_host__ vector: original message"; + assert_eq!(output, expected); + } + #[test] fn test_rfc3164_tag_truncation() { let config = toml::from_str::( @@ -887,9 +1147,109 @@ mod tests { ); log.insert(event_path!("proc_id"), "1234567890"); + let output = run_encode(config.clone(), Event::Log(log)); + // The app_name is truncated so that "[proc_id]:" stays intact and the + // whole tag fits in 32 characters. + let expected_tag = " this-is-a-very-very[1234567890]: "; + assert!(output.contains(expected_tag), "got: {output}"); + + // Without a proc_id the app_name itself is truncated to fit. + let mut log = create_simple_log(); + log.insert( + event_path!("app_name"), + "this-is-a-very-very-long-application-name", + ); let output = run_encode(config, Event::Log(log)); - let expected_tag = "this-is-a-very-very-long-applic:"; - assert!(output.contains(expected_tag)); + let expected_tag = " this-is-a-very-very-long-applic: "; + assert!(output.contains(expected_tag), "got: {output}"); + } + + #[test] + fn test_rfc3164_tag_oversized_proc_id_dropped() { + let config = toml::from_str::( + r#" + [syslog] + rfc = "rfc3164" + app_name = ".app_name" + proc_id = ".proc_id" + "#, + ) + .unwrap(); + + // "[proc_id]:" alone cannot fit in the 32-char tag, so the proc_id is + // dropped entirely instead of being truncated into a wrong value. + let mut log = create_simple_log(); + log.insert(event_path!("app_name"), "myapp"); + log.insert(event_path!("proc_id"), "a".repeat(30)); + + let output = run_encode(config.clone(), Event::Log(log)); + assert!(output.contains(" myapp: "), "got: {output}"); + assert!(!output.contains('['), "got: {output}"); + + // An empty proc_id is treated as absent rather than emitting "[]". + let mut log = create_simple_log(); + log.insert(event_path!("app_name"), "myapp"); + log.insert(event_path!("proc_id"), ""); + + let output = run_encode(config, Event::Log(log)); + assert!(output.contains(" myapp: "), "got: {output}"); + assert!(!output.contains('['), "got: {output}"); + } + + #[test] + fn test_rfc3164_tag_space_sanitization() { + let config = toml::from_str::( + r#" + [syslog] + rfc = "rfc3164" + app_name = ".app" + "#, + ) + .unwrap(); + + // A space would terminate the TAG early on most receivers. + let mut log = create_simple_log(); + log.insert(event_path!("app"), "my app"); + + let output = run_encode(config, Event::Log(log)); + assert!(output.contains(" my_app: "), "got: {output}"); + } + + #[test] + fn test_rfc3164_empty_app_name_falls_back_to_default() { + let config = toml::from_str::( + r#" + [syslog] + rfc = "rfc3164" + app_name = ".app" + "#, + ) + .unwrap(); + + // RFC 3164 has no NILVALUE; an empty app_name must not produce a bare + // ":" tag. + let mut log = create_simple_log(); + log.insert(event_path!("app"), ""); + + let output = run_encode(config, Event::Log(log)); + assert!(output.contains(" vector: "), "got: {output}"); + } + + #[test] + fn test_rfc3164_empty_message_no_trailing_space() { + let config = toml::from_str::( + r#" + [syslog] + rfc = "rfc3164" + "#, + ) + .unwrap(); + + let log = create_simple_log_with_message(""); + + let output = run_encode(config, Event::Log(log)); + let expected = "<14>Aug 28 18:30:00 test-host.com vector:"; + assert_eq!(output, expected); } #[test] @@ -952,8 +1312,7 @@ mod tests { ) .unwrap(); - let mut log = create_simple_log(); - log.insert(event_path!("message"), ""); + let mut log = create_simple_log_with_message(""); log.insert(event_path!("structured_data"), value!({})); let output = run_encode(config, Event::Log(log)); @@ -962,7 +1321,7 @@ mod tests { } #[test] - fn test_non_log_event_filtering() { + fn test_non_log_event_rejected() { let config = toml::from_str::( r#" [syslog] @@ -985,7 +1344,7 @@ mod tests { let result = serializer.encode(metric_event, &mut buffer); - assert!(result.is_ok()); + assert!(result.is_err()); assert!(buffer.is_empty()); } @@ -1168,7 +1527,7 @@ mod tests { let output = run_encode(config, Event::Log(log)); // All invalid characters should be replaced with _ - assert!(output.contains(r"[my_id")); + assert!(output.contains("[my_id")); assert!(output.contains(r#"foo_bar="value1""#)); assert!(output.contains(r#"has_quote="value2""#)); assert!(output.contains(r#"user_name="alice""#)); @@ -1200,6 +1559,140 @@ mod tests { assert!(!output.contains(&format!("[{}", "a".repeat(50)))); } + #[test] + fn test_structured_data_param_name_length_limit_and_collision() { + let config = toml::from_str::( + r#" + [syslog] + rfc = "rfc5424" + "#, + ) + .unwrap(); + + let mut log = create_simple_log(); + log.insert( + event_path!("structured_data"), + value!({ + "metrics": { + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa": "first", + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaab": "second", + } + }), + ); + + let output = run_encode(config, Event::Log(log)); + let truncated_key = "a".repeat(32); + let deduped_key = format!("{}_1", "a".repeat(30)); + assert!(output.contains(&format!(r#"{truncated_key}="first""#))); + assert!(output.contains(&format!(r#"{deduped_key}="second""#))); + assert!(!output.contains(&"a".repeat(50))); + } + + #[test] + fn test_structured_data_long_prefix_preserves_leaf_names() { + let config = toml::from_str::( + r#" + [syslog] + rfc = "rfc5424" + "#, + ) + .unwrap(); + + // A nested prefix at the 32-char PARAM-NAME limit: the prefix is + // truncated instead of the leaf, so sibling keys stay distinguishable. + let long_prefix = "a".repeat(32); + let leaf: ObjectMap = [("x".into(), Value::from(1)), ("y".into(), Value::from(2))] + .into_iter() + .collect(); + let nested: ObjectMap = [(long_prefix.into(), Value::Object(leaf))] + .into_iter() + .collect(); + let sd: ObjectMap = [("el".into(), Value::Object(nested))].into_iter().collect(); + + let mut log = create_simple_log(); + log.insert(event_path!("structured_data"), Value::Object(sd)); + + let output = run_encode(config, Event::Log(log)); + let truncated_prefix = "a".repeat(30); + assert!( + output.contains(&format!(r#"{truncated_prefix}.x="1""#)), + "got: {output}" + ); + assert!( + output.contains(&format!(r#"{truncated_prefix}.y="2""#)), + "got: {output}" + ); + } + + #[test] + fn test_structured_data_prefix_truncation_collision_deduped() { + let config = toml::from_str::( + r#" + [syslog] + rfc = "rfc5424" + "#, + ) + .unwrap(); + + // Two distinct prefixes that only differ past the truncation point + // collide after joining; the dedup suffix must keep both params. + let prefix_a = "a".repeat(32); + let prefix_b = format!("{}b", "a".repeat(31)); + let leaf_a: ObjectMap = [("v".into(), Value::from(1))].into_iter().collect(); + let leaf_b: ObjectMap = [("v".into(), Value::from(2))].into_iter().collect(); + let nested: ObjectMap = [ + (prefix_a.into(), Value::Object(leaf_a)), + (prefix_b.into(), Value::Object(leaf_b)), + ] + .into_iter() + .collect(); + let sd: ObjectMap = [("el".into(), Value::Object(nested))].into_iter().collect(); + + let mut log = create_simple_log(); + log.insert(event_path!("structured_data"), Value::Object(sd)); + + let output = run_encode(config, Event::Log(log)); + let truncated = "a".repeat(30); + assert!( + output.contains(&format!(r#"{truncated}.v="1""#)), + "got: {output}" + ); + assert!( + output.contains(&format!(r#"{truncated}_1="2""#)), + "got: {output}" + ); + } + + #[test] + fn test_structured_data_empty_names_are_skipped() { + let config = toml::from_str::( + r#" + [syslog] + rfc = "rfc5424" + "#, + ) + .unwrap(); + + let mut log = create_simple_log(); + log.insert( + event_path!("structured_data"), + value!({ + "": { + "ignored": "value" + }, + "valid": { + "": "also ignored", + "kept": "yes" + } + }), + ); + + let output = run_encode(config, Event::Log(log)); + assert!(output.contains(r#"[valid kept="yes"]"#)); + assert!(!output.contains("ignored")); + assert!(!output.contains("also ignored")); + } + #[test] fn test_utf8_safe_truncation() { let config = toml::from_str::( diff --git a/lib/codecs/src/encoding/mod.rs b/lib/codecs/src/encoding/mod.rs index 361a62a301533..3346d2af1e2b0 100644 --- a/lib/codecs/src/encoding/mod.rs +++ b/lib/codecs/src/encoding/mod.rs @@ -34,7 +34,7 @@ pub use format::{ ParquetCompression, ParquetSchemaMode, ParquetSerializer, ParquetSerializerConfig, }; #[cfg(feature = "syslog")] -pub use format::{SyslogSerializer, SyslogSerializerConfig}; +pub use format::{SyslogSerializer, SyslogSerializerConfig, SyslogSerializerOptions}; pub use framing::{ BoxedFramer, BoxedFramingError, BytesEncoder, BytesEncoderConfig, CharacterDelimitedEncoder, CharacterDelimitedEncoderConfig, CharacterDelimitedEncoderOptions, Framer, FramingConfig, diff --git a/lib/codecs/src/lib.rs b/lib/codecs/src/lib.rs index 61bfea1122286..dbef0ace8fd99 100644 --- a/lib/codecs/src/lib.rs +++ b/lib/codecs/src/lib.rs @@ -41,6 +41,8 @@ pub use encoding::{ RawMessageSerializer, RawMessageSerializerConfig, SinkType, TextSerializer, TextSerializerConfig, TimestampFormat, Transformer, }; +#[cfg(feature = "syslog")] +pub use encoding::{SyslogSerializer, SyslogSerializerConfig, SyslogSerializerOptions}; pub use gelf::{VALID_FIELD_REGEX, gelf_fields}; pub use ready_frames::ReadyFrames; use vector_config_macros::configurable_component; diff --git a/src/sinks/mod.rs b/src/sinks/mod.rs index 85825449d6ed3..0cea929e33a77 100644 --- a/src/sinks/mod.rs +++ b/src/sinks/mod.rs @@ -110,6 +110,8 @@ pub mod socket; pub mod splunk_hec; #[cfg(feature = "sinks-statsd")] pub mod statsd; +#[cfg(feature = "sinks-syslog")] +pub mod syslog; #[cfg(feature = "sinks-vector")] pub mod vector; #[cfg(feature = "sinks-webhdfs")] diff --git a/src/sinks/syslog.rs b/src/sinks/syslog.rs new file mode 100644 index 0000000000000..e4995284e489a --- /dev/null +++ b/src/sinks/syslog.rs @@ -0,0 +1,997 @@ +use bytes::BytesMut; +use vector_lib::{ + codecs::{ + NewlineDelimitedEncoder, SyslogSerializerConfig, SyslogSerializerOptions, encoding::Framer, + }, + configurable::configurable_component, +}; +use vrl::value::Kind; + +#[cfg(not(windows))] +use crate::sinks::util::unix::UnixSinkConfig; +use crate::{ + codecs::{Encoder, Transformer, encoding::BoxedFramingError}, + config::{ + AcknowledgementsConfig, DataType, GenerateConfig, Input, SinkConfig, SinkContext, + ValidatedSink, + }, + schema, + sinks::util::{tcp::TcpSinkConfig, udp::UdpSinkConfig}, +}; + +/// Configuration for the `syslog` sink. +/// +/// Sends log events as syslog messages over TCP, UDP, or Unix stream sockets. +/// For syslog over TLS (RFC 5425), configure TCP with TLS, RFC 5424, and +/// octet-counting framing. +#[configurable_component(sink( + "syslog", + "Deliver log events to a remote syslog endpoint in RFC 5424 or RFC 3164 format." +))] +#[derive(Clone, Debug)] +pub struct SyslogSinkConfig { + #[serde(flatten)] + pub mode: Mode, + + /// Syslog encoding options. + /// + /// Controls the RFC format, facility, severity, and field mappings for the syslog output. + #[serde(default)] + #[configurable(metadata(docs::examples = "syslog_options_example()"))] + pub syslog: SyslogSerializerOptions, + + #[serde( + default, + deserialize_with = "crate::serde::bool_or_struct", + skip_serializing_if = "crate::serde::is_default" + )] + pub acknowledgements: AcknowledgementsConfig, +} + +/// Syslog sink mode. +#[configurable_component] +#[derive(Clone, Debug)] +#[serde(tag = "mode", rename_all = "snake_case")] +#[configurable(metadata(docs::enum_tag_description = "The type of socket to use."))] +pub enum Mode { + /// Send over TCP. + Tcp(TcpMode), + + /// Send over UDP. + Udp(UdpMode), + + /// Send over a Unix domain socket (UDS), in stream mode. + #[serde(alias = "unix")] + UnixStream(UnixMode), +} + +/// TCP configuration. +#[configurable_component] +#[derive(Clone, Debug)] +pub struct TcpMode { + #[serde(flatten)] + pub config: TcpSinkConfig, + + #[serde(default)] + #[configurable(metadata(docs::examples = "framing_example()"))] + pub framing: SyslogFramingConfig, +} + +/// UDP configuration. +#[configurable_component] +#[derive(Clone, Debug)] +pub struct UdpMode { + #[serde(flatten)] + pub config: UdpSinkConfig, +} + +/// Unix Domain Socket configuration. +#[configurable_component] +#[derive(Clone, Debug)] +pub struct UnixMode { + #[serde(flatten)] + pub config: UnixSinkConfig, + + #[serde(default)] + #[configurable(metadata(docs::examples = "framing_example()"))] + pub framing: SyslogFramingConfig, +} + +fn syslog_options_example() -> SyslogSerializerOptions { + serde_json::from_value(serde_json::json!({ + "rfc": "rfc5424", + "facility": ".facility", + "severity": ".severity", + "app_name": ".app_name", + "proc_id": ".proc_id", + "msg_id": ".msg_id", + })) + .expect("syslog options example should deserialize") +} + +const fn framing_example() -> SyslogFramingConfig { + SyslogFramingConfig { + method: SyslogFramingMethod::OctetCounting, + } +} + +/// Stream framing configuration. +/// +/// Applies only to stream-oriented transports: TCP and Unix stream sockets. UDP +/// sends exactly one syslog message per datagram and does not use framing. +#[configurable_component] +#[derive(Clone, Debug, Default)] +#[serde(default, deny_unknown_fields)] +pub struct SyslogFramingConfig { + /// The framing method used to separate syslog messages in stream transports. + pub method: SyslogFramingMethod, +} + +impl SyslogFramingConfig { + fn build(&self) -> Framer { + match self.method { + SyslogFramingMethod::NewlineDelimited => NewlineDelimitedEncoder::default().into(), + SyslogFramingMethod::OctetCounting => Framer::Boxed(Box::new(OctetCountingEncoder)), + } + } +} + +/// Stream framing method. +#[configurable_component] +#[derive(Clone, Debug, Default)] +#[serde(rename_all = "snake_case")] +pub enum SyslogFramingMethod { + /// Terminates each syslog message with a newline (LF) character. + /// + /// This is RFC 6587 non-transparent framing. Use octet-counting if + /// messages can contain embedded newlines. + #[default] + NewlineDelimited, + + /// Prefixes each syslog message with its byte length and a space. + /// + /// This is RFC 6587 octet-counting framing. When used with TCP, TLS, and + /// RFC 5424 messages, this is the framing required by RFC 5425. + OctetCounting, +} + +/// RFC 6587/RFC 5425 octet-counting framing. +#[derive(Clone, Debug, Default)] +struct OctetCountingEncoder; + +impl tokio_util::codec::Encoder<()> for OctetCountingEncoder { + type Error = BoxedFramingError; + + fn encode(&mut self, _: (), buffer: &mut BytesMut) -> Result<(), Self::Error> { + use std::fmt::Write; + + // The buffer holds exactly one serialized message; prepend its byte + // length and a space. Writing the digits via `fmt::Write` avoids a + // `to_string` allocation per frame. + let payload = buffer.split(); + buffer.reserve(21 + payload.len()); + write!(buffer, "{} ", payload.len()).expect("writing to BytesMut should never fail"); + buffer.extend_from_slice(&payload); + + Ok(()) + } +} + +#[cfg(all(test, feature = "syslog-integration-tests"))] +mod integration_tests; + +// Workaround for https://github.com/vectordotdev/vector/issues/22198. +#[cfg(windows)] +/// A Unix Domain Socket sink. +#[configurable_component] +#[derive(Clone, Debug)] +pub struct UnixSinkConfig { + /// The Unix socket path. + /// + /// This should be an absolute path. + #[configurable(metadata(docs::examples = "/path/to/socket"))] + pub path: std::path::PathBuf, +} + +impl GenerateConfig for SyslogSinkConfig { + fn generate_config() -> serde_json::Value { + serde_yaml::from_str(indoc::indoc! { + r#"address: "127.0.0.1:514" + mode: tcp + syslog: + rfc: rfc5424"#, + }) + .unwrap() + } +} + +impl SyslogSinkConfig { + fn build_serializer(&self) -> vector_lib::codecs::SyslogSerializer { + SyslogSerializerConfig { + syslog: self.syslog.clone(), + } + .build() + } +} + +#[async_trait::async_trait] +#[typetag::serde(name = "syslog")] +impl SinkConfig for SyslogSinkConfig { + fn input(&self) -> Input { + let requirement = schema::Requirement::empty() + .optional_meaning("host", Kind::bytes()) + .optional_meaning("service", Kind::bytes()); + + Input::new(DataType::Log).with_schema_requirement(requirement) + } + + fn acknowledgements(&self) -> &AcknowledgementsConfig { + &self.acknowledgements + } +} + +#[derive(Clone, Debug)] +pub enum ValidatedSyslog { + Tcp { + host: String, + port: u16, + }, + Udp, + #[cfg(unix)] + UnixStream, +} + +#[async_trait::async_trait] +impl ValidatedSink for SyslogSinkConfig { + type Validated = ValidatedSyslog; + + fn validate(&self) -> crate::Result { + match &self.mode { + Mode::Tcp(TcpMode { config, .. }) => { + let (host, port) = config.parse_address()?; + Ok(ValidatedSyslog::Tcp { host, port }) + } + Mode::Udp(UdpMode { config }) => { + config.parse_address()?; + Ok(ValidatedSyslog::Udp) + } + #[cfg(unix)] + Mode::UnixStream(_) => Ok(ValidatedSyslog::UnixStream), + #[cfg(not(unix))] + Mode::UnixStream(_) => { + Err("Unix stream mode is supported only on Unix platforms.".into()) + } + } + } + + async fn build( + &self, + validated: &ValidatedSyslog, + _cx: SinkContext, + ) -> crate::Result<(super::VectorSink, super::Healthcheck)> { + let serializer = self.build_serializer(); + // No user-configurable transformer: the syslog serializer handles all + // field extraction internally. Users who need field filtering should + // use the `socket` sink with `encoding.codec = "syslog"` instead. + let transformer = Transformer::default(); + + match (validated, &self.mode) { + (ValidatedSyslog::Tcp { host, port }, Mode::Tcp(TcpMode { config, framing })) => { + let encoder = Encoder::::new(framing.build(), serializer.into()); + config.build_with_address(host.clone(), *port, transformer, encoder) + } + (ValidatedSyslog::Udp, Mode::Udp(UdpMode { config })) => { + let encoder = Encoder::<()>::new(serializer.into()); + config.build(transformer, encoder, None) + } + #[cfg(unix)] + (ValidatedSyslog::UnixStream, Mode::UnixStream(UnixMode { config, framing })) => { + let encoder = Encoder::::new(framing.build(), serializer.into()); + config.build( + transformer, + encoder, + super::util::service::net::UnixMode::Stream, + ) + } + #[cfg(not(unix))] + (_, Mode::UnixStream(_)) => { + Err("Unix stream mode is supported only on Unix platforms.".into()) + } + _ => unreachable!("validated state does not match config mode"), + } + } +} + +#[cfg(test)] +mod tests { + use std::{future::ready, net::SocketAddr, time::Duration}; + + use futures::{StreamExt, stream}; + use tokio::{ + io::AsyncReadExt, + net::TcpListener, + task::JoinHandle, + time::{sleep, timeout}, + }; + use tokio_stream::wrappers::TcpListenerStream; + use tokio_util::codec::{FramedRead, LinesCodec}; + use vector_lib::event::{BatchNotifier, BatchStatus, Event, LogEvent}; + use vrl::event_path; + + use super::*; + use crate::{ + config::SinkContext, + test_util::{ + CountReceiver, + addr::next_addr, + components::{SINK_TAGS, assert_sink_compliance, run_and_assert_sink_compliance}, + random_lines_with_stream, trace_init, + }, + tls::{self, MaybeTlsSettings, TlsConfig, TlsEnableableConfig}, + }; + + #[test] + fn generate_config() { + crate::test_util::test_generate_config::(); + } + + #[test] + fn validate_rejects_invalid_addresses() { + for mode in ["tcp", "udp"] { + let config: SyslogSinkConfig = toml::from_str(&format!( + r#" + mode = "{mode}" + address = "missing-port" + "#, + )) + .expect("config should parse"); + + assert!( + config.validate().is_err(), + "{mode} address without a port should fail validation" + ); + } + } + + #[tokio::test] + async fn component_spec_compliance() { + trace_init(); + + let (_guard, addr) = next_addr(); + let _receiver = CountReceiver::receive_lines(addr); + + let mut config: SyslogSinkConfig = + serde_json::from_value(SyslogSinkConfig::generate_config()) + .expect("config should be valid"); + // Point to our local test listener instead of the default address. + config.mode = Mode::Tcp(TcpMode { + config: TcpSinkConfig::from_address(addr.to_string()), + framing: Default::default(), + }); + + let context = SinkContext::default(); + let (sink, _healthcheck) = SinkConfig::build(&config, context).await.unwrap(); + + let event = Event::Log(LogEvent::from("spec compliance")); + run_and_assert_sink_compliance(sink, stream::once(ready(event)), &SINK_TAGS).await; + } + + #[tokio::test] + async fn tcp_syslog_rfc5424() { + trace_init(); + + let (_guard, addr) = next_addr(); + let mut receiver = CountReceiver::receive_lines(addr); + + let config = SyslogSinkConfig { + mode: Mode::Tcp(TcpMode { + config: TcpSinkConfig::from_address(addr.to_string()), + framing: Default::default(), + }), + syslog: Default::default(), + acknowledgements: Default::default(), + }; + + let mut event = Event::Log(LogEvent::from("test syslog message")); + event.as_mut_log().insert(event_path!("host"), "test-host"); + + assert_sink_compliance(&SINK_TAGS, async move { + let context = SinkContext::default(); + let (sink, _healthcheck) = SinkConfig::build(&config, context).await.unwrap(); + sink.run(stream::once(ready(event.into()))).await + }) + .await + .expect("Running sink failed"); + + receiver.connected().await; + let output = receiver.await; + assert!(!output.is_empty(), "No messages received"); + let line = &output[0]; + // RFC 5424: VERSION TIMESTAMP HOSTNAME APP-NAME ... + // The version is always "1", so after the closing ">" we expect "1 ". + assert!(line.starts_with('<'), "Expected syslog format, got: {line}"); + assert!( + line.contains(">1 "), + "Expected RFC 5424 version marker, got: {line}" + ); + assert!( + line.contains("test-host"), + "Host not found in output: {line}" + ); + assert!( + line.contains("test syslog message"), + "Message not found in output: {line}" + ); + } + + #[tokio::test] + async fn tcp_syslog_rfc3164() { + trace_init(); + + let (_guard, addr) = next_addr(); + let mut receiver = CountReceiver::receive_lines(addr); + + let config: SyslogSinkConfig = toml::from_str(&format!( + r#" + mode = "tcp" + address = "{addr}" + syslog.rfc = "rfc3164" + "#, + )) + .expect("config should parse"); + + let mut event = Event::Log(LogEvent::from("rfc3164 test message")); + event.as_mut_log().insert(event_path!("host"), "myhost"); + + assert_sink_compliance(&SINK_TAGS, async move { + let context = SinkContext::default(); + let (sink, _healthcheck) = SinkConfig::build(&config, context).await.unwrap(); + sink.run(stream::once(ready(event.into()))).await + }) + .await + .expect("Running sink failed"); + + receiver.connected().await; + let output = receiver.await; + assert!(!output.is_empty(), "No messages received"); + let line = &output[0]; + // RFC 3164: TIMESTAMP HOSTNAME TAG: MSG + // Should NOT contain ">1 " (that's RFC 5424's version field). + assert!(line.starts_with('<'), "Expected syslog format, got: {line}"); + assert!( + !line.contains(">1 "), + "RFC 3164 should not have version field, got: {line}" + ); + assert!(line.contains("myhost"), "Host not found in output: {line}"); + assert!( + line.contains("rfc3164 test message"), + "Message not found in output: {line}" + ); + } + + /// Verifies that custom syslog field paths (app_name, facility) are + /// correctly wired through to the serializer output. + #[tokio::test] + async fn tcp_syslog_custom_fields() { + trace_init(); + + let (_guard, addr) = next_addr(); + let mut receiver = CountReceiver::receive_lines(addr); + + let config: SyslogSinkConfig = toml::from_str(&format!( + r#" + mode = "tcp" + address = "{addr}" + syslog.rfc = "rfc5424" + syslog.app_name = ".my_app" + syslog.facility = ".syslog_facility" + syslog.severity = ".syslog_severity" + "#, + )) + .expect("config should parse"); + + let mut event = Event::Log(LogEvent::from("custom fields test")); + event + .as_mut_log() + .insert(event_path!("my_app"), "myservice"); + event + .as_mut_log() + .insert(event_path!("syslog_facility"), "local0"); + event + .as_mut_log() + .insert(event_path!("syslog_severity"), "error"); + + assert_sink_compliance(&SINK_TAGS, async move { + let context = SinkContext::default(); + let (sink, _healthcheck) = SinkConfig::build(&config, context).await.unwrap(); + sink.run(stream::once(ready(event.into()))).await + }) + .await + .expect("Running sink failed"); + + receiver.connected().await; + let output = receiver.await; + assert!(!output.is_empty(), "No messages received"); + let line = &output[0]; + assert!( + line.starts_with("<131>"), + "Custom facility and severity did not produce expected PRI: {line}" + ); + assert!( + line.contains("myservice"), + "Custom app_name not found in output: {line}" + ); + assert!( + line.contains("custom fields test"), + "Message not found in output: {line}" + ); + } + + /// Sends multiple events over TCP and verifies that newline-delimited + /// framing produces the correct number of distinct messages. + #[tokio::test] + async fn tcp_multiple_events() { + trace_init(); + + let (_guard, addr) = next_addr(); + let mut receiver = CountReceiver::receive_lines(addr); + + let config = SyslogSinkConfig { + mode: Mode::Tcp(TcpMode { + config: TcpSinkConfig::from_address(addr.to_string()), + framing: Default::default(), + }), + syslog: Default::default(), + acknowledgements: Default::default(), + }; + + let (lines, events) = random_lines_with_stream(10, 100, None); + + assert_sink_compliance(&SINK_TAGS, async move { + let context = SinkContext::default(); + let (sink, _healthcheck) = SinkConfig::build(&config, context).await.unwrap(); + sink.run(events).await + }) + .await + .expect("Running sink failed"); + + receiver.connected().await; + let output = receiver.await; + assert_eq!( + lines.len(), + output.len(), + "Expected {} messages but got {}", + lines.len(), + output.len() + ); + for (source, received) in lines.iter().zip(output.iter()) { + assert!( + received.starts_with('<'), + "Expected syslog format, got: {received}" + ); + assert!( + received.contains(source), + "Source line not found in output.\nExpected to contain: {source}\nGot: {received}" + ); + } + } + + #[tokio::test] + async fn tcp_syslog_octet_counting() { + trace_init(); + + let (_guard, addr) = next_addr(); + let receiver = spawn_tcp_receiver(addr).await; + + let config: SyslogSinkConfig = toml::from_str(&format!( + r#" + mode = "tcp" + address = "{addr}" + framing.method = "octet_counting" + syslog.rfc = "rfc5424" + "#, + )) + .expect("config should parse"); + + let event = Event::Log(LogEvent::from("line one\nline two")); + + assert_sink_compliance(&SINK_TAGS, async move { + let context = SinkContext::default(); + let (sink, _healthcheck) = SinkConfig::build(&config, context).await.unwrap(); + sink.run(stream::once(ready(event.into()))).await + }) + .await + .expect("Running sink failed"); + + let output = receiver.await.expect("receiver task should complete"); + let frames = decode_octet_counted_frames(&output); + assert_eq!(frames.len(), 1, "Expected one octet-counted frame"); + assert!( + frames[0].contains("line one\nline two"), + "Multiline message was not preserved in one frame: {:?}", + frames[0] + ); + assert!( + !output.ends_with(b"\n"), + "Octet-counting should not add newline framing: {:?}", + String::from_utf8_lossy(&output) + ); + } + + #[tokio::test] + async fn tcp_tls_syslog_octet_counting_rfc5425() { + trace_init(); + + let (_guard, addr) = next_addr(); + let receiver = spawn_tls_tcp_receiver(addr).await; + + let config = SyslogSinkConfig { + mode: Mode::Tcp(TcpMode { + config: TcpSinkConfig::new( + addr.to_string(), + None, + Some(TlsEnableableConfig { + enabled: Some(true), + options: TlsConfig { + verify_certificate: Some(false), + verify_hostname: Some(false), + ca_file: Some(tls::TEST_PEM_CRT_PATH.into()), + ..Default::default() + }, + }), + None, + ), + framing: SyslogFramingConfig { + method: SyslogFramingMethod::OctetCounting, + }, + }), + syslog: Default::default(), + acknowledgements: Default::default(), + }; + + let event = Event::Log(LogEvent::from("tls syslog test")); + + assert_sink_compliance(&SINK_TAGS, async move { + let context = SinkContext::default(); + let (sink, _healthcheck) = SinkConfig::build(&config, context).await.unwrap(); + sink.run(stream::once(ready(event.into()))).await + }) + .await + .expect("Running sink failed"); + + let output = receiver.await.expect("receiver task should complete"); + let frames = decode_octet_counted_frames(&output); + assert_eq!(frames.len(), 1, "Expected one RFC5425 frame"); + assert!( + frames[0].contains(">1 "), + "Expected RFC5424 message inside RFC5425 frame: {:?}", + frames[0] + ); + assert!( + frames[0].contains("tls syslog test"), + "Message not found in RFC5425 frame: {:?}", + frames[0] + ); + } + + #[tokio::test] + async fn udp_syslog() { + trace_init(); + + let (_guard, addr) = next_addr(); + let receiver = std::net::UdpSocket::bind(addr).expect("Failed to bind UDP socket"); + // Fail instead of hanging the runtime if the datagram never arrives. + receiver + .set_read_timeout(Some(Duration::from_secs(10))) + .expect("Failed to set read timeout"); + + let config = SyslogSinkConfig { + mode: Mode::Udp(UdpMode { + config: UdpSinkConfig::from_address(addr.to_string()), + }), + syslog: Default::default(), + acknowledgements: Default::default(), + }; + + let context = SinkContext::default(); + assert_sink_compliance(&SINK_TAGS, async move { + let (sink, _healthcheck) = SinkConfig::build(&config, context).await.unwrap(); + let event = Event::Log(LogEvent::from("udp syslog test")); + sink.run(stream::once(ready(event.into()))).await + }) + .await + .expect("Running sink failed"); + + let mut buf = [0; 1024]; + let size = receiver + .recv_from(&mut buf) + .expect("Did not receive message") + .0; + + let packet = String::from_utf8(buf[..size].to_vec()).expect("Invalid data received"); + assert!( + packet.starts_with('<'), + "Expected syslog format, got: {packet}" + ); + assert!( + packet.contains(">1 "), + "Expected RFC 5424 version marker, got: {packet}" + ); + assert!( + packet.contains("udp syslog test"), + "Message not found in output: {packet}" + ); + assert!( + !packet.ends_with('\n'), + "UDP syslog packets should contain exactly one syslog message without stream framing: {packet:?}" + ); + } + + #[cfg(unix)] + #[tokio::test] + async fn unix_stream_syslog() { + trace_init(); + + let out_path = temp_uds_path("syslog_unix_stream_test"); + let mut receiver = CountReceiver::receive_lines_unix(out_path.clone()); + + let config = SyslogSinkConfig { + mode: Mode::UnixStream(UnixMode { + config: UnixSinkConfig::new(out_path), + framing: Default::default(), + }), + syslog: Default::default(), + acknowledgements: Default::default(), + }; + + let event = Event::Log(LogEvent::from("unix syslog test")); + + assert_sink_compliance(&SINK_TAGS, async move { + let context = SinkContext::default(); + let (sink, _healthcheck) = SinkConfig::build(&config, context).await.unwrap(); + sink.run(stream::once(ready(event.into()))).await + }) + .await + .expect("Running sink failed"); + + receiver.connected().await; + let output = receiver.await; + assert!(!output.is_empty(), "No messages received"); + let line = &output[0]; + assert!(line.starts_with('<'), "Expected syslog format, got: {line}"); + assert!( + line.contains(">1 "), + "Expected RFC 5424 version marker, got: {line}" + ); + assert!( + line.contains("unix syslog test"), + "Message not found in output: {line}" + ); + } + + /// Verifies that batch finalizers are marked `Delivered` after a successful + /// TCP send so downstream acknowledgement chains complete correctly. + #[tokio::test] + async fn tcp_finalizers_delivered_on_success() { + trace_init(); + + let (_guard, addr) = next_addr(); + let _receiver = CountReceiver::receive_lines(addr); + + let config = SyslogSinkConfig { + mode: Mode::Tcp(TcpMode { + config: TcpSinkConfig::from_address(addr.to_string()), + framing: Default::default(), + }), + syslog: Default::default(), + acknowledgements: Default::default(), + }; + + let (batch, mut receiver) = BatchNotifier::new_with_receiver(); + let event = Event::Log(LogEvent::from("ack-tcp-test").with_batch_notifier(&batch)); + drop(batch); + + assert_sink_compliance(&SINK_TAGS, async move { + let context = SinkContext::default(); + let (sink, _healthcheck) = SinkConfig::build(&config, context).await.unwrap(); + sink.run(stream::once(ready(event.into()))).await + }) + .await + .expect("Running sink failed"); + + assert_eq!( + receiver.try_recv(), + Ok(BatchStatus::Delivered), + "TCP sink should mark batch finalizer Delivered" + ); + } + + /// Verifies that batch finalizers are marked `Delivered` after a successful + /// UDP send. This guards the connectionless datagram path in + /// `src/sinks/util/datagram.rs::send_and_emit`. + #[tokio::test] + async fn udp_finalizers_delivered_on_success() { + trace_init(); + + let (_guard, addr) = next_addr(); + let receiver = std::net::UdpSocket::bind(addr).expect("Failed to bind UDP socket"); + // Fail instead of hanging the runtime if the datagram never arrives. + receiver + .set_read_timeout(Some(Duration::from_secs(10))) + .expect("Failed to set read timeout"); + + let config = SyslogSinkConfig { + mode: Mode::Udp(UdpMode { + config: UdpSinkConfig::from_address(addr.to_string()), + }), + syslog: Default::default(), + acknowledgements: Default::default(), + }; + + let (batch, mut ack_receiver) = BatchNotifier::new_with_receiver(); + let event = Event::Log(LogEvent::from("ack-udp-test").with_batch_notifier(&batch)); + drop(batch); + + let context = SinkContext::default(); + assert_sink_compliance(&SINK_TAGS, async move { + let (sink, _healthcheck) = SinkConfig::build(&config, context).await.unwrap(); + sink.run(stream::once(ready(event.into()))).await + }) + .await + .expect("Running sink failed"); + + // Drain the datagram so the receiver socket is properly cleaned up. + let mut buf = [0; 1024]; + receiver + .recv_from(&mut buf) + .expect("Did not receive message"); + + assert_eq!( + ack_receiver.try_recv(), + Ok(BatchStatus::Delivered), + "UDP sink should mark batch finalizer Delivered" + ); + } + + /// Forces a hard TCP disconnect mid-stream and verifies the sink reconnects + /// to a freshly-bound listener and continues delivering events. Mirrors + /// `src/sinks/socket.rs::tests::reconnect`. + #[tokio::test] + async fn tcp_reconnect_after_server_close() { + trace_init(); + + let (_guard, addr) = next_addr(); + let config = SyslogSinkConfig { + mode: Mode::Tcp(TcpMode { + config: TcpSinkConfig::from_address(addr.to_string()), + framing: Default::default(), + }), + syslog: Default::default(), + acknowledgements: Default::default(), + }; + + let context = SinkContext::default(); + let (sink, _healthcheck) = SinkConfig::build(&config, context).await.unwrap(); + + let (_, events) = random_lines_with_stream(1000, 10000, None); + let sink_handle = tokio::spawn(run_and_assert_sink_compliance(sink, events, &SINK_TAGS)); + + // First listener: drain a handful of events then drop the connection. + let mut count = 20usize; + TcpListenerStream::new(TcpListener::bind(addr).await.unwrap()) + .next() + .await + .unwrap() + .map(|socket| FramedRead::new(socket, LinesCodec::new())) + .unwrap() + .map(|x| x.unwrap()) + .take_while(|_| { + ready(if count > 0 { + count -= 1; + true + } else { + false + }) + }) + .collect::>() + .await; + + if cfg!(windows) { + sleep(Duration::from_secs(1)).await; + } + + // Second listener: if the sink failed to reconnect this never connects + // and the test hangs out to its tokio timeout. + assert!( + timeout( + Duration::from_secs(5), + CountReceiver::receive_lines(addr).connected() + ) + .await + .is_ok(), + "syslog sink did not reconnect after the TCP peer closed" + ); + + sink_handle.await.unwrap(); + } + + #[cfg(unix)] + fn temp_uds_path(name: &str) -> std::path::PathBuf { + tempfile::tempdir().unwrap().keep().join(name) + } + + async fn spawn_tcp_receiver(addr: SocketAddr) -> JoinHandle> { + let listener = TcpListener::bind(addr) + .await + .expect("TCP receiver should bind"); + + tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.expect("TCP should accept"); + let mut output = Vec::new(); + timeout(Duration::from_secs(5), socket.read_to_end(&mut output)) + .await + .expect("TCP read should complete") + .expect("TCP read should succeed"); + output + }) + } + + async fn spawn_tls_tcp_receiver(addr: SocketAddr) -> JoinHandle> { + let config = Some(TlsEnableableConfig::test_config()); + let tls = MaybeTlsSettings::from_config(config.as_ref(), true) + .expect("TLS settings should build"); + let mut listener = tls.bind(&addr).await.expect("TLS receiver should bind"); + + tokio::spawn(async move { + let mut socket = listener.accept().await.expect("TLS should accept"); + let mut output = Vec::new(); + timeout(Duration::from_secs(5), socket.read_to_end(&mut output)) + .await + .expect("TLS read should complete") + .expect("TLS read should succeed"); + output + }) + } + + fn decode_octet_counted_frames(output: &[u8]) -> Vec { + let mut frames = Vec::new(); + let mut cursor = 0; + + while cursor < output.len() { + let length_start = cursor; + while cursor < output.len() && output[cursor].is_ascii_digit() { + cursor += 1; + } + + assert!( + cursor > length_start, + "Octet-counted frame must start with a non-empty length: {:?}", + String::from_utf8_lossy(output) + ); + assert!( + cursor < output.len() && output[cursor] == b' ', + "Octet-counted frame length must be followed by a space: {:?}", + String::from_utf8_lossy(output) + ); + + let length = std::str::from_utf8(&output[length_start..cursor]) + .expect("Length should be UTF-8") + .parse::() + .expect("Length should be numeric"); + cursor += 1; + + let frame_end = cursor + length; + assert!( + frame_end <= output.len(), + "Octet-counted frame length exceeds available bytes" + ); + frames.push( + String::from_utf8(output[cursor..frame_end].to_vec()) + .expect("Syslog frame should be UTF-8"), + ); + cursor = frame_end; + } + + frames + } +} diff --git a/src/sinks/syslog/integration_tests.rs b/src/sinks/syslog/integration_tests.rs new file mode 100644 index 0000000000000..d6ebba4e8bc3c --- /dev/null +++ b/src/sinks/syslog/integration_tests.rs @@ -0,0 +1,509 @@ +use std::{ + future::ready, + path::PathBuf, + time::{Duration, Instant}, +}; + +use futures::stream; +use tokio::time::sleep; +use vector_lib::event::{Event, LogEvent, ObjectMap, Value}; +use vrl::event_path; + +use super::SyslogSinkConfig; +use crate::{ + config::{SinkConfig, SinkContext}, + test_util::{ + components::{SINK_TAGS, assert_sink_compliance}, + random_string, trace_init, wait_for_tcp, + }, + tls::{self, TlsConfig, TlsEnableableConfig}, +}; + +// These tests exercise receiver-visible interoperability with a real rsyslog +// instance. Unit tests cover exact wire bytes; integration tests stay focused on +// common deployment shapes that operators are likely to run. +const TCP_LINE_DEFAULT: &str = "rsyslog:5515"; + +fn syslog_udp_address() -> String { + std::env::var("SYSLOG_UDP_ADDRESS").unwrap_or_else(|_| "rsyslog:5514".to_owned()) +} + +fn syslog_tcp_line_address() -> String { + std::env::var("SYSLOG_TCP_LINE_ADDRESS").unwrap_or_else(|_| TCP_LINE_DEFAULT.to_owned()) +} + +fn syslog_tcp_octet_address() -> String { + std::env::var("SYSLOG_TCP_OCTET_ADDRESS").unwrap_or_else(|_| "rsyslog:5516".to_owned()) +} + +fn syslog_ng_tcp_tls_address() -> String { + std::env::var("SYSLOG_NG_TCP_TLS_ADDRESS").unwrap_or_else(|_| "syslog-ng:5517".to_owned()) +} + +fn syslog_ng_udp_address() -> String { + std::env::var("SYSLOG_NG_UDP_ADDRESS").unwrap_or_else(|_| "syslog-ng:5514".to_owned()) +} + +fn syslog_ng_tcp_line_address() -> String { + std::env::var("SYSLOG_NG_TCP_LINE_ADDRESS").unwrap_or_else(|_| "syslog-ng:5515".to_owned()) +} + +fn syslog_ng_tcp_octet_address() -> String { + std::env::var("SYSLOG_NG_TCP_OCTET_ADDRESS").unwrap_or_else(|_| "syslog-ng:5516".to_owned()) +} + +fn syslog_log_dir() -> PathBuf { + std::env::var_os("SYSLOG_LOG_DIR") + .map(PathBuf::from) + .unwrap_or_else(|| PathBuf::from("/syslog_logs")) +} + +fn log_event(message: &str, facility: &str, severity: &str) -> Event { + let mut event = Event::Log(LogEvent::from(message.to_owned())); + let log = event.as_mut_log(); + log.insert(event_path!("host"), "vector-integration-host"); + log.insert(event_path!("app"), "vector-integration-app"); + log.insert(event_path!("facility"), facility); + log.insert(event_path!("severity"), severity); + event +} + +async fn run_sink(config: SyslogSinkConfig, event: Event) { + // Build the sink inside `assert_sink_compliance` so the registered-event + // names from `register!(BytesSent::from(Protocol::UDP))` are captured + // after `init_test` clears the event recorder. + assert_sink_compliance(&SINK_TAGS, async move { + let context = SinkContext::default(); + let (sink, healthcheck) = SinkConfig::build(&config, context) + .await + .expect("sink should build"); + healthcheck.await.expect("healthcheck should pass"); + sink.run(stream::once(ready(event.into()))) + .await + .expect("sink should run"); + }) + .await; +} + +/// Fields a receiver parsed out of one syslog message, as written by the JSON +/// templates in `tests/integration/syslog/data`. +#[derive(Debug)] +struct ParsedRecord(serde_json::Value); + +impl ParsedRecord { + /// Returns a field as a string. syslog-ng may emit numeric macros as JSON + /// numbers, so numbers are rendered without quotes; missing fields are empty. + fn field(&self, name: &str) -> String { + match self.0.get(name) { + Some(serde_json::Value::String(s)) => s.clone(), + Some(serde_json::Value::Null) | None => String::new(), + Some(other) => other.to_string(), + } + } + + /// The parsed MSG part. rsyslog keeps the space that follows an RFC 3164 + /// TAG in `msg`, so leading whitespace is not significant here. + fn message(&self) -> String { + self.field("message").trim_start().to_owned() + } +} + +/// Waits for a receiver to log a message whose parsed MSG contains `needle`. +/// +/// The receiver log files are shared between tests (and across local runs), +/// so every test uses a unique message and matches only its own record. +async fn wait_for_record(file_name: &str, needle: &str) -> ParsedRecord { + let path = syslog_log_dir().join(file_name); + let started = Instant::now(); + let mut contents = String::new(); + + while started.elapsed() <= Duration::from_secs(15) { + contents = std::fs::read_to_string(&path).unwrap_or_default(); + let record = contents + .lines() + .filter_map(|line| serde_json::from_str(line).ok()) + .map(ParsedRecord) + .find(|record| record.message().contains(needle)); + if let Some(record) = record { + return record; + } + + sleep(Duration::from_millis(100)).await; + } + + panic!( + "timed out waiting for {needle:?} in {}; contents:\n{contents}", + path.display() + ); +} + +/// Asserts that the receiver parsed the PRI, HOSTNAME, APP-NAME, and MSG that +/// the sink sent. +fn assert_received_syslog(record: &ParsedRecord, pri: u8, message: &str) { + assert_received_header(record, pri); + assert_eq!(record.message(), message, "message mismatch: {record:?}"); +} + +/// Asserts that the receiver parsed the PRI, HOSTNAME, and APP-NAME that the +/// sink sent. +fn assert_received_header(record: &ParsedRecord, pri: u8) { + assert_eq!( + record.field("facility"), + (pri / 8).to_string(), + "facility mismatch: {record:?}" + ); + assert_eq!( + record.field("severity"), + (pri % 8).to_string(), + "severity mismatch: {record:?}" + ); + assert_eq!( + record.field("hostname"), + "vector-integration-host", + "hostname mismatch: {record:?}" + ); + assert_eq!( + record.field("app_name"), + "vector-integration-app", + "app name mismatch: {record:?}" + ); +} + +async fn wait_for_rsyslog() { + wait_for_tcp(syslog_tcp_line_address()).await; +} + +async fn wait_for_syslog_ng() { + wait_for_tcp(syslog_ng_tcp_line_address()).await; +} + +async fn wait_for_syslog_ng_tls() { + wait_for_tcp(syslog_ng_tcp_tls_address()).await; +} + +#[tokio::test] +async fn udp_rfc3164_reaches_rsyslog() { + trace_init(); + wait_for_rsyslog().await; + + let message = format!("udp-rfc3164-{}", random_string(12)); + let config: SyslogSinkConfig = toml::from_str(&format!( + r#" + mode = "udp" + address = "{}" + syslog.rfc = "rfc3164" + syslog.app_name = ".app" + syslog.facility = ".facility" + syslog.severity = ".severity" + "#, + syslog_udp_address(), + )) + .expect("config should parse"); + + run_sink(config, log_event(&message, "local0", "notice")).await; + + let record = wait_for_record("udp.log", &message).await; + assert_received_syslog(&record, 133, &message); +} + +#[tokio::test] +async fn udp_rfc5424_reaches_rsyslog() { + trace_init(); + wait_for_rsyslog().await; + + let message = format!("udp-rfc5424-{}", random_string(12)); + let config: SyslogSinkConfig = toml::from_str(&format!( + r#" + mode = "udp" + address = "{}" + syslog.rfc = "rfc5424" + syslog.app_name = ".app" + syslog.facility = ".facility" + syslog.severity = ".severity" + "#, + syslog_udp_address(), + )) + .expect("config should parse"); + + run_sink(config, log_event(&message, "local0", "info")).await; + + let record = wait_for_record("udp.log", &message).await; + assert_received_syslog(&record, 134, &message); +} + +#[tokio::test] +async fn tcp_newline_rfc3164_reaches_rsyslog() { + trace_init(); + wait_for_rsyslog().await; + + let message = format!("tcp-newline-rfc3164-{}", random_string(12)); + let config: SyslogSinkConfig = toml::from_str(&format!( + r#" + mode = "tcp" + address = "{}" + syslog.rfc = "rfc3164" + syslog.app_name = ".app" + syslog.facility = ".facility" + syslog.severity = ".severity" + "#, + syslog_tcp_line_address(), + )) + .expect("config should parse"); + + run_sink(config, log_event(&message, "local1", "warning")).await; + + let record = wait_for_record("tcp-line.log", &message).await; + assert_received_syslog(&record, 140, &message); +} + +#[tokio::test] +async fn tcp_newline_rfc5424_reaches_rsyslog() { + trace_init(); + wait_for_rsyslog().await; + + let message = format!("tcp-newline-rfc5424-{}", random_string(12)); + let config: SyslogSinkConfig = toml::from_str(&format!( + r#" + mode = "tcp" + address = "{}" + syslog.rfc = "rfc5424" + syslog.app_name = ".app" + syslog.facility = ".facility" + syslog.severity = ".severity" + "#, + syslog_tcp_line_address(), + )) + .expect("config should parse"); + + run_sink(config, log_event(&message, "local1", "err")).await; + + let record = wait_for_record("tcp-line.log", &message).await; + assert_received_syslog(&record, 139, &message); +} + +#[tokio::test] +async fn tcp_octet_counting_rfc5424_reaches_rsyslog() { + trace_init(); + wait_for_rsyslog().await; + + let message = format!("tcp-octet-rfc5424-{}", random_string(12)); + let config: SyslogSinkConfig = toml::from_str(&format!( + r#" + mode = "tcp" + address = "{}" + framing.method = "octet_counting" + syslog.rfc = "rfc5424" + syslog.app_name = ".app" + syslog.facility = ".facility" + syslog.severity = ".severity" + "#, + syslog_tcp_octet_address(), + )) + .expect("config should parse"); + + run_sink(config, log_event(&message, "local2", "crit")).await; + + let record = wait_for_record("tcp-octet.log", &message).await; + assert_received_syslog(&record, 146, &message); +} + +/// Verifies that `proc_id`, `msg_id`, and `structured_data` configured at the +/// sink level are routed through the encoder and arrive intact at rsyslog. +/// This catches regressions in the field-path plumbing inside `decant_config` +/// or its inputs, which the codec-only unit tests can't see. +#[tokio::test] +async fn tcp_octet_counting_rfc5424_with_proc_id_msg_id_structured_data_reaches_rsyslog() { + trace_init(); + wait_for_rsyslog().await; + + let id = random_string(12); + let message = format!("tcp-octet-fields-{id}"); + let proc_id = format!("pid-{id}"); + let msg_id = format!("msg-{id}"); + let sd_param_value = format!("retry-{id}"); + + let config: SyslogSinkConfig = toml::from_str(&format!( + r#" + mode = "tcp" + address = "{}" + framing.method = "octet_counting" + syslog.rfc = "rfc5424" + syslog.app_name = ".app" + syslog.proc_id = ".pid" + syslog.msg_id = ".mid" + syslog.facility = ".facility" + syslog.severity = ".severity" + "#, + syslog_tcp_octet_address(), + )) + .expect("config should parse"); + + let mut event = log_event(&message, "local2", "info"); + let log = event.as_mut_log(); + log.insert(event_path!("pid"), proc_id.as_str()); + log.insert(event_path!("mid"), msg_id.as_str()); + let mut sd_params = ObjectMap::new(); + sd_params.insert("retry".into(), Value::from(sd_param_value.as_str())); + let mut sd_root = ObjectMap::new(); + sd_root.insert("metrics@1234".into(), Value::from(sd_params)); + log.insert(event_path!("structured_data"), Value::from(sd_root)); + + run_sink(config, event).await; + + let record = wait_for_record("tcp-octet.log", &message).await; + assert_received_syslog(&record, 150, &message); + assert_eq!( + record.field("proc_id"), + proc_id, + "proc_id mismatch: {record:?}" + ); + assert_eq!( + record.field("msg_id"), + msg_id, + "msg_id mismatch: {record:?}" + ); + assert_eq!( + record.field("structured_data"), + format!("[metrics@1234 retry=\"{sd_param_value}\"]"), + "structured data mismatch: {record:?}" + ); +} + +#[tokio::test] +async fn tcp_octet_counting_rfc5424_multiline_reaches_rsyslog() { + trace_init(); + wait_for_rsyslog().await; + + let id = random_string(12); + let first_line = format!("tcp-octet-multiline-{id}-first"); + let second_line = format!("tcp-octet-multiline-{id}-second"); + let message = format!("{first_line}\n{second_line}"); + let config: SyslogSinkConfig = toml::from_str(&format!( + r#" + mode = "tcp" + address = "{}" + framing.method = "octet_counting" + syslog.rfc = "rfc5424" + syslog.app_name = ".app" + syslog.facility = ".facility" + syslog.severity = ".severity" + "#, + syslog_tcp_octet_address(), + )) + .expect("config should parse"); + + run_sink(config, log_event(&message, "local2", "crit")).await; + + let record = wait_for_record("tcp-octet.log", &first_line).await; + assert_received_header(&record, 146); + // rsyslog escapes control characters on receipt by default (LF becomes + // `#012`), so assert that both lines were parsed as a single message + // rather than comparing the separator byte. + let received = record.message(); + assert!( + received.starts_with(&first_line) && received.ends_with(&second_line), + "expected both lines in a single parsed message: {record:?}" + ); +} + +/// RFC 5425 (syslog over TLS) interop with syslog-ng's TLS network +/// transport using octet-counted framing. Catches regressions in the +/// TLS handshake or framing paths that the in-process TLS unit test +/// can miss because it doesn't speak to a real syslog daemon. +#[tokio::test] +async fn tcp_tls_octet_counting_rfc5424_to_syslog_ng() { + trace_init(); + wait_for_syslog_ng_tls().await; + + let message = format!("tcp-tls-rfc5424-{}", random_string(12)); + let mut config: SyslogSinkConfig = toml::from_str(&format!( + r#" + mode = "tcp" + address = "{}" + framing.method = "octet_counting" + syslog.rfc = "rfc5424" + syslog.app_name = ".app" + syslog.facility = ".facility" + syslog.severity = ".severity" + "#, + syslog_ng_tcp_tls_address(), + )) + .expect("config should parse"); + + if let super::Mode::Tcp(tcp_mode) = &mut config.mode { + // Trust the project test CA but skip hostname verification, since + // the syslog-ng cert is for `localhost` not `syslog-ng`. + tcp_mode.config = crate::sinks::util::tcp::TcpSinkConfig::new( + syslog_ng_tcp_tls_address(), + None, + Some(TlsEnableableConfig { + enabled: Some(true), + options: TlsConfig { + verify_certificate: Some(true), + verify_hostname: Some(false), + ca_file: Some(tls::TEST_PEM_INTERMEDIATE_CA_PATH.into()), + ..Default::default() + }, + }), + None, + ); + } else { + panic!("expected TCP mode after config parse"); + } + + run_sink(config, log_event(&message, "local3", "warning")).await; + + let record = wait_for_record("syslog-ng-tcp-tls.log", &message).await; + assert_received_syslog(&record, 156, &message); +} + +#[tokio::test] +async fn udp_rfc5424_reaches_syslog_ng() { + trace_init(); + wait_for_syslog_ng().await; + + let message = format!("syslog-ng-udp-{}", random_string(12)); + let config: SyslogSinkConfig = toml::from_str(&format!( + r#" + mode = "udp" + address = "{}" + syslog.rfc = "rfc5424" + syslog.app_name = ".app" + syslog.facility = ".facility" + syslog.severity = ".severity" + "#, + syslog_ng_udp_address(), + )) + .expect("config should parse"); + + run_sink(config, log_event(&message, "local4", "info")).await; + + let record = wait_for_record("syslog-ng-udp.log", &message).await; + assert_received_syslog(&record, 166, &message); +} + +#[tokio::test] +async fn tcp_octet_counting_rfc5424_reaches_syslog_ng() { + trace_init(); + wait_for_syslog_ng().await; + + let message = format!("syslog-ng-tcp-octet-{}", random_string(12)); + let config: SyslogSinkConfig = toml::from_str(&format!( + r#" + mode = "tcp" + address = "{}" + framing.method = "octet_counting" + syslog.rfc = "rfc5424" + syslog.app_name = ".app" + syslog.facility = ".facility" + syslog.severity = ".severity" + "#, + syslog_ng_tcp_octet_address(), + )) + .expect("config should parse"); + + run_sink(config, log_event(&message, "local5", "notice")).await; + + let record = wait_for_record("syslog-ng-tcp-octet.log", &message).await; + assert_received_syslog(&record, 173, &message); +} diff --git a/tests/integration/syslog/config/compose.yaml b/tests/integration/syslog/config/compose.yaml new file mode 100644 index 0000000000000..604f4bbd48430 --- /dev/null +++ b/tests/integration/syslog/config/compose.yaml @@ -0,0 +1,32 @@ +version: "3" + +services: + rsyslog: + image: docker.io/rsyslog/rsyslog:${CONFIG_VERSION} + user: root + environment: + - RSYSLOG_HOSTNAME=vector-rsyslog + volumes: + - ../data/rsyslog.conf:/etc/rsyslog.d/vector-test.conf:ro + - logs:/var/log/vector-test + + syslog-ng: + image: docker.io/balabit/syslog-ng:4.8.3 + command: ["--no-caps", "--foreground", "-f", "/etc/syslog-ng/vector-test.conf"] + volumes: + - ../data/syslog-ng.conf:/etc/syslog-ng/vector-test.conf:ro + - ../../../../tests/data/ca/intermediate_server/certs/localhost.cert.pem:/etc/syslog-ng/tls/localhost.cert.pem:ro + - ../../../../tests/data/ca/intermediate_server/private/localhost.key.pem:/etc/syslog-ng/tls/localhost.key.pem:ro + - logs:/var/log/vector-test + +volumes: + # Use external volume 'syslog_logs' that's shared with the test runner. + # The runner creates this volume and mounts it at /syslog_logs. + logs: + name: syslog_logs + external: true + +networks: + default: + name: ${VECTOR_NETWORK} + external: true diff --git a/tests/integration/syslog/config/test.yaml b/tests/integration/syslog/config/test.yaml new file mode 100644 index 0000000000000..ec883a97c420a --- /dev/null +++ b/tests/integration/syslog/config/test.yaml @@ -0,0 +1,30 @@ +features: + - syslog-integration-tests + +test_filter: "::syslog::integration_tests::" + +env: + SYSLOG_UDP_ADDRESS: rsyslog:5514 + SYSLOG_TCP_LINE_ADDRESS: rsyslog:5515 + SYSLOG_TCP_OCTET_ADDRESS: rsyslog:5516 + SYSLOG_NG_UDP_ADDRESS: syslog-ng:5514 + SYSLOG_NG_TCP_LINE_ADDRESS: syslog-ng:5515 + SYSLOG_NG_TCP_OCTET_ADDRESS: syslog-ng:5516 + SYSLOG_NG_TCP_TLS_ADDRESS: syslog-ng:5517 + SYSLOG_LOG_DIR: /syslog_logs + +runner: + volumes: + syslog_logs: /syslog_logs + +matrix: + version: ["2026-02"] + +paths: + - "lib/codecs/src/encoding/format/syslog.rs" + - "lib/codecs/src/encoding/encoder.rs" + - "lib/codecs/src/encoding/framing/**" + - "src/sinks/syslog.rs" + - "src/sinks/syslog/**" + - "src/sinks/util/**" + - "tests/integration/syslog/**" diff --git a/tests/integration/syslog/data/rsyslog.conf b/tests/integration/syslog/data/rsyslog.conf new file mode 100644 index 0000000000000..a67cd15314040 --- /dev/null +++ b/tests/integration/syslog/data/rsyslog.conf @@ -0,0 +1,44 @@ +module(load="imudp") +module(load="imtcp") + +# Write the fields rsyslog parsed out of each message as one JSON object per +# line, so tests verify that the receiver understood the message rather than +# just that the bytes arrived. For RFC 3164 input, rsyslog derives app-name and +# procid from the TAG. +template(name="ParsedSyslogMessage" type="list") { + constant(value="{\"facility\":\"") + property(name="syslogfacility") + constant(value="\",\"severity\":\"") + property(name="syslogseverity") + constant(value="\",\"hostname\":\"") + property(name="hostname" format="json") + constant(value="\",\"app_name\":\"") + property(name="app-name" format="json") + constant(value="\",\"proc_id\":\"") + property(name="procid" format="json") + constant(value="\",\"msg_id\":\"") + property(name="msgid" format="json") + constant(value="\",\"structured_data\":\"") + property(name="structured-data" format="json") + constant(value="\",\"message\":\"") + property(name="msg" format="json") + constant(value="\",\"raw\":\"") + property(name="rawmsg" format="json") + constant(value="\"}\n") +} + +input(type="imudp" port="5514" ruleset="udp") +input(type="imtcp" port="5515" ruleset="tcp_line" supportOctetCountedFraming="off") +input(type="imtcp" port="5516" ruleset="tcp_octet" supportOctetCountedFraming="on") + +ruleset(name="udp") { + action(type="omfile" file="/var/log/vector-test/udp.log" template="ParsedSyslogMessage") +} + +ruleset(name="tcp_line") { + action(type="omfile" file="/var/log/vector-test/tcp-line.log" template="ParsedSyslogMessage") +} + +ruleset(name="tcp_octet") { + action(type="omfile" file="/var/log/vector-test/tcp-octet.log" template="ParsedSyslogMessage") +} diff --git a/tests/integration/syslog/data/syslog-ng.conf b/tests/integration/syslog/data/syslog-ng.conf new file mode 100644 index 0000000000000..25ae2e0fe24a9 --- /dev/null +++ b/tests/integration/syslog/data/syslog-ng.conf @@ -0,0 +1,79 @@ +@version: current + +options { + chain_hostnames(off); + use_dns(no); + keep_hostname(yes); + create_dirs(yes); +}; + +# Write the fields syslog-ng parsed out of each message as one JSON object per +# line, so tests verify that the receiver understood the message rather than +# just that the bytes arrived. +template parsed { + template("$(format-json --scope none facility=$FACILITY_NUM severity=$LEVEL_NUM hostname=$HOST app_name=$PROGRAM proc_id=$PID msg_id=$MSGID structured_data=$SDATA message=$MSG raw=$RAWMSG)\n"); +}; + +source s_udp { + network( + ip("0.0.0.0") + port(5514) + transport("udp") + flags(syslog-protocol, store-raw-message) + ); +}; + +source s_tcp_line { + network( + ip("0.0.0.0") + port(5515) + transport("tcp") + flags(syslog-protocol, store-raw-message) + ); +}; + +# The syslog() driver (unlike network()) parses RFC 6587 octet-counted +# framing, so these sources genuinely validate the frames Vector sends. +source s_tcp_octet { + syslog( + ip("0.0.0.0") + port(5516) + transport("tcp") + flags(store-raw-message) + ); +}; + +source s_tcp_tls { + syslog( + ip("0.0.0.0") + port(5517) + transport("tls") + flags(store-raw-message) + tls( + key-file("/etc/syslog-ng/tls/localhost.key.pem") + cert-file("/etc/syslog-ng/tls/localhost.cert.pem") + peer-verify("optional-untrusted") + ) + ); +}; + +destination d_udp { + file("/var/log/vector-test/syslog-ng-udp.log" template(parsed)); +}; + +destination d_tcp_line { + file("/var/log/vector-test/syslog-ng-tcp-line.log" template(parsed)); +}; + +destination d_tcp_octet { + file("/var/log/vector-test/syslog-ng-tcp-octet.log" template(parsed)); +}; + +destination d_tcp_tls { + file("/var/log/vector-test/syslog-ng-tcp-tls.log" template(parsed)); +}; + +log { source(s_udp); destination(d_udp); }; +log { source(s_tcp_line); destination(d_tcp_line); }; +log { source(s_tcp_octet); destination(d_tcp_octet); }; +log { source(s_tcp_tls); destination(d_tcp_tls); }; diff --git a/website/content/en/docs/reference/configuration/sinks/syslog.md b/website/content/en/docs/reference/configuration/sinks/syslog.md new file mode 100644 index 0000000000000..5c1dbde1b029b --- /dev/null +++ b/website/content/en/docs/reference/configuration/sinks/syslog.md @@ -0,0 +1,14 @@ +--- +title: Syslog +description: Deliver logs to a syslog endpoint using RFC 5424 or RFC 3164 formatting +component_kind: sink +layout: component +tags: ["syslog", "remote", "component", "sink", "logs"] +--- + +{{/* +This doc is generated using: + +1. The template in layouts/docs/component.html +2. The relevant CUE data in cue/reference/components/... +*/}} diff --git a/website/cue/reference/components/sinks/generated/syslog.cue b/website/cue/reference/components/sinks/generated/syslog.cue new file mode 100644 index 0000000000000..afe25fc202198 --- /dev/null +++ b/website/cue/reference/components/sinks/generated/syslog.cue @@ -0,0 +1,170 @@ +package metadata + +generated: components: sinks: syslog: configuration: { + acknowledgements: { + description: """ + Controls how acknowledgements are handled for this sink. + + See [End-to-end Acknowledgements][e2e_acks] for more information on how event acknowledgement is handled. + + [e2e_acks]: https://vector.dev/docs/architecture/end-to-end-acknowledgements/ + """ + required: false + type: _schemaDefinitions["vector_core::config::AcknowledgementsConfig"] + } + address: { + description: """ + The address to connect to. + + Both IP address and hostname are accepted formats. + + The address _must_ include a port. + """ + relevant_when: "mode = \"tcp\" or mode = \"udp\"" + required: true + type: string: examples: ["92.12.333.224:5000", "https://somehost:5000"] + } + framing: { + description: """ + Stream framing configuration. + + Applies only to stream-oriented transports: TCP and Unix stream sockets. UDP + sends exactly one syslog message per datagram and does not use framing. + """ + relevant_when: "mode = \"tcp\" or mode = \"unix_stream\"" + required: false + type: object: { + examples: [{ + method: "octet_counting" + }] + options: method: { + description: "The framing method used to separate syslog messages in stream transports." + required: false + type: string: { + default: "newline_delimited" + enum: { + newline_delimited: """ + Terminates each syslog message with a newline (LF) character. + + This is RFC 6587 non-transparent framing. Use octet-counting if + messages can contain embedded newlines. + """ + octet_counting: """ + Prefixes each syslog message with its byte length and a space. + + This is RFC 6587 octet-counting framing. When used with TCP, TLS, and + RFC 5424 messages, this is the framing required by RFC 5425. + """ + } + } + } + } + } + keepalive: { + description: "TCP keepalive settings for socket-based components." + relevant_when: "mode = \"tcp\"" + required: false + type: _schemaDefinitions["core::option::Option"] + } + mode: { + description: "The type of socket to use." + required: true + type: string: enum: { + tcp: "Send over TCP." + udp: "Send over UDP." + unix_stream: "Send over a Unix domain socket (UDS), in stream mode." + } + } + path: { + description: """ + The Unix socket path. + + This should be an absolute path. + """ + relevant_when: "mode = \"unix_stream\"" + required: true + type: string: examples: ["/path/to/socket"] + } + send_buffer_bytes: { + description: """ + The size of the socket's send buffer. + + If set, the value of the setting is passed via the `SO_SNDBUF` option. + """ + relevant_when: "mode = \"tcp\" or mode = \"udp\"" + required: false + type: uint: { + examples: [ + 65536 + ] + unit: "bytes" + } + } + syslog: { + description: """ + Syslog encoding options. + + Controls the RFC format, facility, severity, and field mappings for the syslog output. + """ + required: false + type: object: { + examples: [{ + app_name: ".app_name" + facility: ".facility" + msg_id: ".msg_id" + proc_id: ".proc_id" + rfc: "rfc5424" + severity: ".severity" + }] + options: { + app_name: { + description: """ + Path to a field in the event to use for the app name. + + If not provided, the encoder checks for a semantic "service" field. + If that is also missing, it defaults to "vector". + """ + required: false + type: string: {} + } + facility: { + description: "Path to a field in the event to use for the facility. Defaults to \"user\"." + required: false + type: string: {} + } + msg_id: { + description: "Path to a field in the event to use for the msg ID." + required: false + type: string: {} + } + proc_id: { + description: "Path to a field in the event to use for the proc ID." + required: false + type: string: {} + } + rfc: { + description: "RFC to use for formatting." + required: false + type: string: { + default: "rfc5424" + enum: { + rfc3164: "The legacy RFC3164 syslog format." + rfc5424: "The modern RFC5424 syslog format." + } + } + } + severity: { + description: "Path to a field in the event to use for the severity. Defaults to \"informational\"." + required: false + type: string: {} + } + } + } + } + tls: { + description: "Configures the TLS options for incoming/outgoing connections." + relevant_when: "mode = \"tcp\"" + required: false + type: _schemaDefinitions["core::option::Option"] + } +} diff --git a/website/cue/reference/components/sinks/syslog.cue b/website/cue/reference/components/sinks/syslog.cue new file mode 100644 index 0000000000000..70370596a594e --- /dev/null +++ b/website/cue/reference/components/sinks/syslog.cue @@ -0,0 +1,124 @@ +package metadata + +components: sinks: syslog: { + title: "Syslog" + + classes: { + delivery: "best_effort" + development: "beta" + egress_method: "stream" + service_providers: [] + stateful: false + } + + features: { + acknowledgements: true + auto_generated: true + healthcheck: enabled: true + send: { + compression: enabled: false + encoding: enabled: false + send_buffer_bytes: { + enabled: true + relevant_when: "mode = `tcp` or mode = `udp`" + } + keepalive: enabled: true + request: enabled: false + tls: { + enabled: true + can_verify_certificate: true + can_verify_hostname: true + enabled_default: false + enabled_by_scheme: false + } + to: { + service: services.syslog + + interface: { + socket: { + api: { + title: "Syslog" + url: urls.syslog + } + direction: "outgoing" + protocols: ["tcp", "udp", "unix"] + ssl: "optional" + } + } + } + } + } + + support: { + requirements: [] + warnings: [ + """ + For RFC 5425 syslog over TLS compliance, configure `mode = "tcp"`, + `tls.enabled = true`, `syslog.rfc = "rfc5424"`, and + `framing.method = "octet_counting"`. The default + `newline_delimited` framing is not RFC 5425 compliant. + """, + """ + `newline_delimited` stream framing can split messages that contain + embedded newlines, such as stack traces or multiline JSON. Use + `framing.method = "octet_counting"` for multiline syslog messages. + """, + """ + Unix datagram sockets, including typical Linux `/dev/log` sockets, + are not supported by this sink. Use the `socket` sink with + `mode = "unix_datagram"` for local syslog daemons that require + datagram sockets. + """, + """ + RFC 3164 timestamps are emitted in UTC. + """, + """ + This sink does not expose `except_fields`, `only_fields`, or + `timestamp_format` transformation options. Use the `socket` sink + with `encoding.codec = "syslog"` if you need field filtering or + timestamp formatting before serialization. + """, + ] + } + + configuration: generated.components.sinks.syslog.configuration + + input: { + logs: true + metrics: null + traces: false + } + + configuration_examples: [ + { + title: "UDP RFC 5424" + configuration: { + type: "syslog" + mode: "udp" + address: "syslog.example.com:514" + syslog: rfc: "rfc5424" + } + }, + { + title: "TCP RFC 5424 with octet-counting" + configuration: { + type: "syslog" + mode: "tcp" + address: "syslog.example.com:514" + framing: method: "octet_counting" + syslog: rfc: "rfc5424" + } + }, + { + title: "RFC 5425 over TLS" + configuration: { + type: "syslog" + mode: "tcp" + address: "syslog.example.com:6514" + framing: method: "octet_counting" + syslog: rfc: "rfc5424" + tls: enabled: true + } + }, + ] +} diff --git a/website/generated/example-configs/sinks/syslog/advanced.yaml b/website/generated/example-configs/sinks/syslog/advanced.yaml new file mode 100644 index 0000000000000..4ae9d892c0cc7 --- /dev/null +++ b/website/generated/example-configs/sinks/syslog/advanced.yaml @@ -0,0 +1,17 @@ +sinks: + my_sink_id: + type: syslog + inputs: + - my-source-or-transform-id + address: 92.12.333.224:5000 + framing: + method: octet_counting + mode: tcp + send_buffer_bytes: 65536 + syslog: + app_name: .app_name + facility: .facility + msg_id: .msg_id + proc_id: .proc_id + rfc: rfc5424 + severity: .severity diff --git a/website/generated/example-configs/sinks/syslog/minimal.yaml b/website/generated/example-configs/sinks/syslog/minimal.yaml new file mode 100644 index 0000000000000..fb4d923a5f499 --- /dev/null +++ b/website/generated/example-configs/sinks/syslog/minimal.yaml @@ -0,0 +1,7 @@ +sinks: + my_sink_id: + type: syslog + inputs: + - my-source-or-transform-id + address: 92.12.333.224:5000 + mode: tcp