vector/sources/
syslog.rs

1#[cfg(unix)]
2use std::path::PathBuf;
3use std::{net::SocketAddr, num::NonZeroU64, time::Duration};
4
5use bytes::Bytes;
6use chrono::Utc;
7use futures::StreamExt;
8use listenfd::ListenFd;
9use smallvec::SmallVec;
10use tokio_util::udp::UdpFramed;
11use vector_lib::{
12    EstimatedJsonEncodedSizeOf,
13    codecs::{
14        BytesDecoder, OctetCountingDecoder, SyslogDeserializerConfig,
15        decoding::{Deserializer, Framer},
16    },
17    config::{LegacyKey, LogNamespace},
18    configurable::configurable_component,
19    internal_event::{ByteSize, BytesReceived, InternalEventHandle as _, Protocol},
20    ipallowlist::IpAllowlistConfig,
21    lookup::{OwnedValuePath, lookup_v2::OptionalValuePath, path},
22};
23use vrl::event_path;
24
25#[cfg(unix)]
26use crate::sources::util::build_unix_stream_source;
27use crate::{
28    SourceSender,
29    codecs::Decoder,
30    config::{
31        DataType, GenerateConfig, Resource, SourceConfig, SourceContext, SourceOutput, log_schema,
32    },
33    event::Event,
34    internal_events::{
35        SocketBindError, SocketEventsReceived, SocketMode, SocketReceiveError, StreamClosedError,
36    },
37    net,
38    shutdown::ShutdownSignal,
39    sources::util::net::{SocketListenAddr, TcpNullAcker, TcpSource, try_bind_udp_socket},
40    tcp::TcpKeepaliveConfig,
41    tls::{MaybeTlsSettings, TlsSourceConfig},
42};
43
44/// Configuration for the `syslog` source.
45#[configurable_component(source("syslog", "Collect logs sent via Syslog."))]
46#[derive(Clone, Debug)]
47pub struct SyslogConfig {
48    #[serde(flatten)]
49    mode: Mode,
50
51    /// The maximum buffer size of incoming messages, in bytes.
52    ///
53    /// Messages larger than this are truncated.
54    #[serde(default = "crate::serde::default_max_length")]
55    #[configurable(metadata(docs::type_unit = "bytes"))]
56    max_length: usize,
57
58    /// Overrides the name of the log field used to add the peer host to each event.
59    ///
60    /// If using TCP or UDP, the value is the peer host's address, including the port. For example, `1.2.3.4:9000`. If using
61    /// UDS, the value is the socket path itself.
62    ///
63    /// By default, the [global `log_schema.host_key` option][global_host_key] is used.
64    ///
65    /// [global_host_key]: https://vector.dev/docs/reference/configuration/global-options/#log_schema.host_key
66    host_key: Option<OptionalValuePath>,
67
68    /// The namespace to use for logs. This overrides the global setting.
69    #[configurable(metadata(docs::hidden))]
70    #[serde(default)]
71    pub log_namespace: Option<bool>,
72}
73
74/// Listener mode for the `syslog` source.
75#[configurable_component]
76#[derive(Clone, Debug)]
77#[serde(tag = "mode", rename_all = "snake_case")]
78#[configurable(metadata(docs::enum_tag_description = "The type of socket to use."))]
79#[allow(clippy::large_enum_variant)]
80pub enum Mode {
81    /// Listen on TCP.
82    Tcp {
83        #[configurable(derived)]
84        address: SocketListenAddr,
85
86        #[configurable(derived)]
87        keepalive: Option<TcpKeepaliveConfig>,
88
89        #[configurable(derived)]
90        permit_origin: Option<IpAllowlistConfig>,
91
92        #[configurable(derived)]
93        tls: Option<TlsSourceConfig>,
94
95        /// The size of the receive buffer used for each connection.
96        ///
97        /// This should not typically needed to be changed.
98        #[configurable(metadata(docs::type_unit = "bytes"))]
99        receive_buffer_bytes: Option<usize>,
100
101        /// The maximum number of TCP connections that are allowed at any given time.
102        connection_limit: Option<u32>,
103
104        /// The timeout, in seconds, before a TLS handshake is aborted if it has not completed.
105        ///
106        /// This bounds how long a connection can hold its slot against `connection_limit`
107        /// before the TLS handshake finishes, protecting against clients that open a
108        /// connection but never complete (or never start) a handshake.
109        #[configurable(metadata(docs::type_unit = "seconds"))]
110        tls_handshake_timeout_secs: Option<NonZeroU64>,
111    },
112
113    /// Listen on UDP.
114    Udp {
115        #[configurable(derived)]
116        address: SocketListenAddr,
117
118        /// The size of the receive buffer used for the listening socket.
119        ///
120        /// This should not typically needed to be changed.
121        #[configurable(metadata(docs::type_unit = "bytes"))]
122        receive_buffer_bytes: Option<usize>,
123    },
124
125    /// Listen on UDS (Unix domain socket). This only supports Unix stream sockets.
126    ///
127    /// For Unix datagram sockets, use the `socket` source instead.
128    #[cfg(unix)]
129    Unix {
130        /// The Unix socket path.
131        ///
132        /// This should be an absolute path.
133        #[configurable(metadata(docs::examples = "/path/to/socket"))]
134        path: PathBuf,
135
136        /// Unix file mode bits to be applied to the unix socket file as its designated file permissions.
137        ///
138        /// The file mode value can be specified in any numeric format supported by your configuration
139        /// language, but it is most intuitive to use an octal number.
140        socket_file_mode: Option<u32>,
141    },
142}
143
144impl SyslogConfig {
145    #[cfg(test)]
146    pub fn from_mode(mode: Mode) -> Self {
147        Self {
148            mode,
149            host_key: None,
150            max_length: crate::serde::default_max_length(),
151            log_namespace: None,
152        }
153    }
154}
155
156impl Default for SyslogConfig {
157    fn default() -> Self {
158        Self {
159            mode: Mode::Tcp {
160                address: SocketListenAddr::SocketAddr("0.0.0.0:514".parse().unwrap()),
161                keepalive: None,
162                permit_origin: None,
163                tls: None,
164                receive_buffer_bytes: None,
165                connection_limit: None,
166                tls_handshake_timeout_secs: None,
167            },
168            host_key: None,
169            max_length: crate::serde::default_max_length(),
170            log_namespace: None,
171        }
172    }
173}
174
175impl GenerateConfig for SyslogConfig {
176    fn generate_config() -> serde_json::Value {
177        serde_json::to_value(SyslogConfig::default()).unwrap()
178    }
179}
180
181#[async_trait::async_trait]
182#[typetag::serde(name = "syslog")]
183impl SourceConfig for SyslogConfig {
184    async fn build(&self, cx: SourceContext) -> crate::Result<super::Source> {
185        let log_namespace = cx.log_namespace(self.log_namespace);
186        let host_key = self
187            .host_key
188            .clone()
189            .and_then(|k| k.path)
190            .or(log_schema().host_key().cloned());
191
192        match self.mode.clone() {
193            Mode::Tcp {
194                address,
195                keepalive,
196                permit_origin,
197                tls,
198                receive_buffer_bytes,
199                connection_limit,
200                tls_handshake_timeout_secs,
201            } => {
202                let source = SyslogTcpSource {
203                    max_length: self.max_length,
204                    host_key,
205                    log_namespace,
206                };
207                let shutdown_secs = Duration::from_secs(30);
208                let tls_config = tls.as_ref().map(|tls| tls.tls_config.clone());
209                let tls_client_metadata_key = tls
210                    .as_ref()
211                    .and_then(|tls| tls.client_metadata_key.clone())
212                    .and_then(|k| k.path);
213                let tls = MaybeTlsSettings::from_config(tls_config.as_ref(), true)?;
214                source.run(
215                    address,
216                    keepalive,
217                    shutdown_secs,
218                    tls,
219                    None, // tls_reloader: not wired for this source
220                    tls_client_metadata_key,
221                    receive_buffer_bytes,
222                    None,
223                    tls_handshake_timeout_secs,
224                    cx,
225                    false.into(),
226                    connection_limit,
227                    permit_origin.map(Into::into),
228                    SyslogConfig::NAME,
229                    log_namespace,
230                )
231            }
232            Mode::Udp {
233                address,
234                receive_buffer_bytes,
235            } => Ok(udp(
236                address,
237                self.max_length,
238                host_key,
239                receive_buffer_bytes,
240                cx.shutdown,
241                log_namespace,
242                cx.out,
243            )),
244            #[cfg(unix)]
245            Mode::Unix {
246                path,
247                socket_file_mode,
248            } => {
249                let decoder = Decoder::new(
250                    Framer::OctetCounting(OctetCountingDecoder::new_with_max_length(
251                        self.max_length,
252                    )),
253                    Deserializer::Syslog(
254                        SyslogDeserializerConfig::from_source(SyslogConfig::NAME).build(),
255                    ),
256                );
257
258                build_unix_stream_source(
259                    path,
260                    socket_file_mode,
261                    decoder,
262                    move |events, host| handle_events(events, &host_key, host, log_namespace),
263                    cx.shutdown,
264                    cx.out,
265                )
266            }
267        }
268    }
269
270    fn outputs(&self, global_log_namespace: LogNamespace) -> Vec<SourceOutput> {
271        let log_namespace = global_log_namespace.merge(self.log_namespace);
272        let schema_definition = SyslogDeserializerConfig::from_source(SyslogConfig::NAME)
273            .schema_definition(log_namespace)
274            .with_standard_vector_source_metadata();
275
276        vec![SourceOutput::new_maybe_logs(
277            DataType::Log,
278            schema_definition,
279        )]
280    }
281
282    fn resources(&self) -> Vec<Resource> {
283        match self.mode.clone() {
284            Mode::Tcp { address, .. } => vec![address.as_tcp_resource()],
285            Mode::Udp { address, .. } => vec![address.as_udp_resource()],
286            #[cfg(unix)]
287            Mode::Unix { .. } => vec![],
288        }
289    }
290
291    fn can_acknowledge(&self) -> bool {
292        false
293    }
294}
295
296#[derive(Debug, Clone)]
297struct SyslogTcpSource {
298    max_length: usize,
299    host_key: Option<OwnedValuePath>,
300    log_namespace: LogNamespace,
301}
302
303impl TcpSource for SyslogTcpSource {
304    type Error = vector_lib::codecs::decoding::Error;
305    type Item = SmallVec<[Event; 1]>;
306    type Decoder = Decoder;
307    type Acker = TcpNullAcker;
308
309    fn decoder(&self) -> Self::Decoder {
310        Decoder::new(
311            Framer::OctetCounting(OctetCountingDecoder::new_with_max_length(self.max_length)),
312            Deserializer::Syslog(SyslogDeserializerConfig::from_source(SyslogConfig::NAME).build()),
313        )
314    }
315
316    fn handle_events(&self, events: &mut [Event], host: SocketAddr) {
317        handle_events(
318            events,
319            &self.host_key,
320            Some(host.ip().to_string().into()),
321            self.log_namespace,
322        );
323    }
324
325    fn build_acker(&self, _: &[Self::Item]) -> Self::Acker {
326        TcpNullAcker
327    }
328}
329
330pub fn udp(
331    addr: SocketListenAddr,
332    _max_length: usize,
333    host_key: Option<OwnedValuePath>,
334    receive_buffer_bytes: Option<usize>,
335    shutdown: ShutdownSignal,
336    log_namespace: LogNamespace,
337    mut out: SourceSender,
338) -> super::Source {
339    Box::pin(async move {
340        let listenfd = ListenFd::from_env();
341        let socket = try_bind_udp_socket(addr, listenfd).await.map_err(|error| {
342            emit!(SocketBindError {
343                mode: SocketMode::Udp,
344                error: &error,
345            })
346        })?;
347
348        if let Some(receive_buffer_bytes) = receive_buffer_bytes
349            && let Err(error) = net::set_receive_buffer_size(&socket, receive_buffer_bytes)
350        {
351            warn!(message = "Failed configuring receive buffer size on UDP socket.", %error);
352        }
353
354        info!(
355            message = "Listening.",
356            addr = %addr,
357            r#type = "udp"
358        );
359
360        let bytes_received = register!(BytesReceived::from(Protocol::UDP));
361
362        let mut stream = UdpFramed::new(
363            socket,
364            Decoder::new(
365                Framer::Bytes(BytesDecoder::new()),
366                Deserializer::Syslog(
367                    SyslogDeserializerConfig::from_source(SyslogConfig::NAME).build(),
368                ),
369            ),
370        )
371        .take_until(shutdown)
372        .filter_map(|frame| {
373            let host_key = host_key.clone();
374            let bytes_received = bytes_received.clone();
375            async move {
376                match frame {
377                    Ok(((mut events, byte_size), received_from)) => {
378                        let count = events.len();
379                        bytes_received.emit(ByteSize(byte_size));
380                        emit!(SocketEventsReceived {
381                            mode: SocketMode::Udp,
382                            byte_size: events.estimated_json_encoded_size_of(),
383                            count,
384                        });
385                        let received_from = received_from.ip().to_string().into();
386                        handle_events(&mut events, &host_key, Some(received_from), log_namespace);
387                        Some(events.remove(0))
388                    }
389                    Err(error) => {
390                        emit!(SocketReceiveError {
391                            mode: SocketMode::Udp,
392                            error: &error,
393                        });
394                        None
395                    }
396                }
397            }
398        })
399        .boxed();
400
401        match out.send_event_stream(&mut stream).await {
402            Ok(()) => {
403                debug!("Finished sending.");
404                Ok(())
405            }
406            Err(_) => {
407                let (count, _) = stream.size_hint();
408                emit!(StreamClosedError { count });
409                Err(())
410            }
411        }
412    })
413}
414
415fn handle_events(
416    events: &mut [Event],
417    host_key: &Option<OwnedValuePath>,
418    default_host: Option<Bytes>,
419    log_namespace: LogNamespace,
420) {
421    for event in events {
422        enrich_syslog_event(event, host_key, default_host.clone(), log_namespace);
423    }
424}
425
426fn enrich_syslog_event(
427    event: &mut Event,
428    host_key: &Option<OwnedValuePath>,
429    default_host: Option<Bytes>,
430    log_namespace: LogNamespace,
431) {
432    let log = event.as_mut_log();
433
434    if let Some(default_host) = &default_host {
435        log_namespace.insert_source_metadata(
436            SyslogConfig::NAME,
437            log,
438            Some(LegacyKey::Overwrite(path!("source_ip"))),
439            path!("source_ip"),
440            default_host.clone(),
441        );
442    }
443
444    let parsed_hostname = log
445        .get(event_path!("hostname"))
446        .map(|hostname| hostname.coerce_to_bytes());
447
448    if let Some(parsed_host) = parsed_hostname.or(default_host) {
449        let legacy_host_key = host_key.as_ref().map(LegacyKey::Overwrite);
450
451        log_namespace.insert_source_metadata(
452            SyslogConfig::NAME,
453            log,
454            legacy_host_key,
455            path!("host"),
456            parsed_host,
457        );
458    }
459
460    log_namespace.insert_standard_vector_source_metadata(log, SyslogConfig::NAME, Utc::now());
461
462    if log_namespace == LogNamespace::Legacy {
463        let timestamp = log
464            .get(event_path!("timestamp"))
465            .and_then(|timestamp| timestamp.as_timestamp().cloned())
466            .unwrap_or_else(Utc::now);
467        log.maybe_insert(log_schema().timestamp_key_target_path(), timestamp);
468    }
469
470    trace!(
471        message = "Processing one event.",
472        event = ?event
473    );
474}
475
476#[cfg(test)]
477mod test {
478    use std::{collections::HashMap, fmt, str::FromStr};
479
480    use chrono::prelude::*;
481    use indoc::indoc;
482    use rand::{RngExt, rng};
483    use serde::Deserialize;
484    use tokio::time::{Duration, Instant, sleep};
485    use tokio_util::codec::BytesCodec;
486    use vector_lib::{
487        assert_event_data_eq,
488        codecs::decoding::format::Deserializer,
489        config::ComponentKey,
490        lookup::{OwnedTargetPath, PathPrefix, event_path, owned_value_path},
491        schema::Definition,
492    };
493    use vrl::value::{Kind, ObjectMap, Value, kind::Collection};
494
495    use super::*;
496    use crate::{
497        config::log_schema,
498        event::{Event, LogEvent},
499        test_util::{
500            CountReceiver,
501            addr::next_addr,
502            components::{SOCKET_PUSH_SOURCE_TAGS, assert_source_compliance},
503            random_maps, random_string, send_encodable, send_lines, wait_for_tcp,
504        },
505    };
506
507    fn event_from_bytes(
508        host_key: &str,
509        default_host: Option<Bytes>,
510        bytes: Bytes,
511        log_namespace: LogNamespace,
512    ) -> Option<Event> {
513        let parser = SyslogDeserializerConfig::from_source(SyslogConfig::NAME).build();
514        let mut events = parser.parse(bytes, LogNamespace::Legacy).ok()?;
515        handle_events(
516            &mut events,
517            &Some(owned_value_path!(host_key)),
518            default_host,
519            log_namespace,
520        );
521        Some(events.remove(0))
522    }
523
524    #[test]
525    fn generate_config() {
526        crate::test_util::test_generate_config::<SyslogConfig>();
527    }
528
529    #[test]
530    fn output_schema_definition_vector_namespace() {
531        let config = SyslogConfig {
532            log_namespace: Some(true),
533            ..Default::default()
534        };
535
536        let definitions = config
537            .outputs(LogNamespace::Vector)
538            .remove(0)
539            .schema_definition(true);
540
541        let expected_definition =
542            Definition::new_with_default_metadata(Kind::bytes(), [LogNamespace::Vector])
543                .with_meaning(OwnedTargetPath::event_root(), "message")
544                .with_metadata_field(
545                    &owned_value_path!("vector", "source_type"),
546                    Kind::bytes(),
547                    None,
548                )
549                .with_metadata_field(
550                    &owned_value_path!("vector", "ingest_timestamp"),
551                    Kind::timestamp(),
552                    None,
553                )
554                .with_metadata_field(
555                    &owned_value_path!("syslog", "timestamp"),
556                    Kind::timestamp(),
557                    Some("timestamp"),
558                )
559                .with_metadata_field(
560                    &owned_value_path!("syslog", "hostname"),
561                    Kind::bytes().or_undefined(),
562                    Some("host"),
563                )
564                .with_metadata_field(
565                    &owned_value_path!("syslog", "source_ip"),
566                    Kind::bytes().or_undefined(),
567                    None,
568                )
569                .with_metadata_field(
570                    &owned_value_path!("syslog", "severity"),
571                    Kind::bytes().or_undefined(),
572                    Some("severity"),
573                )
574                .with_metadata_field(
575                    &owned_value_path!("syslog", "facility"),
576                    Kind::bytes().or_undefined(),
577                    None,
578                )
579                .with_metadata_field(
580                    &owned_value_path!("syslog", "version"),
581                    Kind::integer().or_undefined(),
582                    None,
583                )
584                .with_metadata_field(
585                    &owned_value_path!("syslog", "appname"),
586                    Kind::bytes().or_undefined(),
587                    Some("service"),
588                )
589                .with_metadata_field(
590                    &owned_value_path!("syslog", "msgid"),
591                    Kind::bytes().or_undefined(),
592                    None,
593                )
594                .with_metadata_field(
595                    &owned_value_path!("syslog", "procid"),
596                    Kind::integer().or_bytes().or_undefined(),
597                    None,
598                )
599                .with_metadata_field(
600                    &owned_value_path!("syslog", "structured_data"),
601                    Kind::object(Collection::from_unknown(Kind::object(
602                        Collection::from_unknown(Kind::bytes()),
603                    ))),
604                    None,
605                )
606                .with_metadata_field(
607                    &owned_value_path!("syslog", "tls_client_metadata"),
608                    Kind::object(Collection::empty().with_unknown(Kind::bytes())).or_undefined(),
609                    None,
610                );
611
612        assert_eq!(definitions, Some(expected_definition));
613    }
614
615    #[test]
616    fn output_schema_definition_legacy_namespace() {
617        let config = SyslogConfig::default();
618
619        let definitions = config
620            .outputs(LogNamespace::Legacy)
621            .remove(0)
622            .schema_definition(true);
623
624        let expected_definition = Definition::new_with_default_metadata(
625            Kind::object(Collection::empty()),
626            [LogNamespace::Legacy],
627        )
628        .with_event_field(
629            &owned_value_path!("message"),
630            Kind::bytes(),
631            Some("message"),
632        )
633        .with_event_field(
634            &owned_value_path!("timestamp"),
635            Kind::timestamp(),
636            Some("timestamp"),
637        )
638        .with_event_field(
639            &owned_value_path!("hostname"),
640            Kind::bytes().or_undefined(),
641            Some("host"),
642        )
643        .with_event_field(
644            &owned_value_path!("source_ip"),
645            Kind::bytes().or_undefined(),
646            None,
647        )
648        .with_event_field(
649            &owned_value_path!("severity"),
650            Kind::bytes().or_undefined(),
651            Some("severity"),
652        )
653        .with_event_field(
654            &owned_value_path!("facility"),
655            Kind::bytes().or_undefined(),
656            None,
657        )
658        .with_event_field(
659            &owned_value_path!("version"),
660            Kind::integer().or_undefined(),
661            None,
662        )
663        .with_event_field(
664            &owned_value_path!("appname"),
665            Kind::bytes().or_undefined(),
666            Some("service"),
667        )
668        .with_event_field(
669            &owned_value_path!("msgid"),
670            Kind::bytes().or_undefined(),
671            None,
672        )
673        .with_event_field(
674            &owned_value_path!("procid"),
675            Kind::integer().or_bytes().or_undefined(),
676            None,
677        )
678        .unknown_fields(Kind::object(Collection::from_unknown(Kind::bytes())))
679        .with_standard_vector_source_metadata();
680
681        assert_eq!(definitions, Some(expected_definition));
682    }
683
684    #[test]
685    fn config_tcp() {
686        let config: SyslogConfig = serde_yaml::from_str(indoc! {
687            r#"
688            mode: tcp
689            address: "127.0.0.1:1235"
690            "#,
691        })
692        .unwrap();
693        assert!(matches!(config.mode, Mode::Tcp { .. }));
694    }
695
696    #[test]
697    fn config_tcp_with_receive_buffer_size() {
698        let config: SyslogConfig = serde_yaml::from_str(indoc! {
699            r#"
700            mode: tcp
701            address: "127.0.0.1:1235"
702            receive_buffer_bytes: 256
703            "#,
704        })
705        .unwrap();
706
707        let receive_buffer_bytes = match config.mode {
708            Mode::Tcp {
709                receive_buffer_bytes,
710                ..
711            } => receive_buffer_bytes,
712            _ => panic!("expected Mode::Tcp"),
713        };
714
715        assert_eq!(receive_buffer_bytes, Some(256));
716    }
717
718    #[test]
719    fn config_tcp_keepalive_empty() {
720        let config: SyslogConfig = serde_yaml::from_str(indoc! {
721            r#"
722            mode: tcp
723            address: "127.0.0.1:1235"
724            "#,
725        })
726        .unwrap();
727
728        let keepalive = match config.mode {
729            Mode::Tcp { keepalive, .. } => keepalive,
730            _ => panic!("expected Mode::Tcp"),
731        };
732
733        assert_eq!(keepalive, None);
734    }
735
736    #[test]
737    fn config_tcp_keepalive_full() {
738        let config: SyslogConfig = serde_yaml::from_str(indoc! {
739            r#"
740            mode: tcp
741            address: "127.0.0.1:1235"
742            keepalive:
743              time_secs: 7200
744            "#,
745        })
746        .unwrap();
747
748        let keepalive = match config.mode {
749            Mode::Tcp { keepalive, .. } => keepalive,
750            _ => panic!("expected Mode::Tcp"),
751        };
752
753        let keepalive = keepalive.expect("keepalive config not set");
754
755        assert_eq!(keepalive.time_secs, Some(7200));
756    }
757
758    #[test]
759    fn config_udp() {
760        let config: SyslogConfig = serde_yaml::from_str(indoc! {
761            r#"
762            mode: udp
763            address: "127.0.0.1:1235"
764            max_length: 32187
765            "#,
766        })
767        .unwrap();
768        assert!(matches!(config.mode, Mode::Udp { .. }));
769    }
770
771    #[test]
772    fn config_udp_with_receive_buffer_size() {
773        let config: SyslogConfig = serde_yaml::from_str(indoc! {
774            r#"
775            mode: udp
776            address: "127.0.0.1:1235"
777            max_length: 32187
778            receive_buffer_bytes: 256
779            "#,
780        })
781        .unwrap();
782
783        let receive_buffer_bytes = match config.mode {
784            Mode::Udp {
785                receive_buffer_bytes,
786                ..
787            } => receive_buffer_bytes,
788            _ => panic!("expected Mode::Udp"),
789        };
790
791        assert_eq!(receive_buffer_bytes, Some(256));
792    }
793
794    #[cfg(unix)]
795    #[test]
796    fn config_unix() {
797        let config: SyslogConfig = serde_yaml::from_str(indoc! {
798            r#"
799            mode: unix
800            path: "127.0.0.1:1235"
801            "#,
802        })
803        .unwrap();
804        assert!(matches!(config.mode, Mode::Unix { .. }));
805    }
806
807    #[cfg(unix)]
808    #[test]
809    fn config_unix_permissions() {
810        let config: SyslogConfig = serde_yaml::from_str(indoc! {
811            r#"
812            mode: unix
813            path: "127.0.0.1:1235"
814            socket_file_mode: 511
815            "#,
816        })
817        .unwrap();
818        let socket_file_mode = match config.mode {
819            Mode::Unix {
820                path: _,
821                socket_file_mode,
822            } => socket_file_mode,
823            _ => panic!("expected Mode::Unix"),
824        };
825
826        assert_eq!(socket_file_mode, Some(0o777));
827    }
828
829    #[test]
830    fn syslog_ng_network_syslog_protocol() {
831        // this should also match rsyslog omfwd with template=RSYSLOG_SyslogProtocol23Format
832        let msg = "i am foobar";
833        let raw = format!(
834            r#"<13>1 2019-02-13T19:48:34+00:00 74794bfb6795 root 8449 - {}{} {}"#,
835            r#"[meta sequenceId="1" sysUpTime="37" language="EN"]"#,
836            r#"[origin ip="192.168.0.1" software="test"]"#,
837            msg
838        );
839
840        let mut expected = Event::Log(LogEvent::from(msg));
841
842        {
843            let expected = expected.as_mut_log();
844            expected.insert(
845                (PathPrefix::Event, log_schema().timestamp_key().unwrap()),
846                Utc.with_ymd_and_hms(2019, 2, 13, 19, 48, 34)
847                    .single()
848                    .expect("invalid timestamp"),
849            );
850            expected.insert(
851                log_schema().source_type_key_target_path().unwrap(),
852                "syslog",
853            );
854            expected.insert(event_path!("host"), "74794bfb6795");
855            expected.insert(event_path!("hostname"), "74794bfb6795");
856
857            expected.insert(event_path!("meta", "sequenceId"), "1");
858            expected.insert(event_path!("meta", "sysUpTime"), "37");
859            expected.insert(event_path!("meta", "language"), "EN");
860            expected.insert(event_path!("origin", "software"), "test");
861            expected.insert(event_path!("origin", "ip"), "192.168.0.1");
862
863            expected.insert(event_path!("severity"), "notice");
864            expected.insert(event_path!("facility"), "user");
865            expected.insert(event_path!("version"), 1);
866            expected.insert(event_path!("appname"), "root");
867            expected.insert(event_path!("procid"), 8449);
868            expected.insert(event_path!("source_ip"), "192.168.0.254");
869        }
870
871        assert_event_data_eq!(
872            event_from_bytes(
873                "host",
874                Some(Bytes::from("192.168.0.254")),
875                raw.into(),
876                LogNamespace::Legacy
877            )
878            .unwrap(),
879            expected
880        );
881    }
882
883    #[test]
884    fn handles_incorrect_sd_element() {
885        let msg = "qwerty";
886        let raw = format!(
887            r#"<13>1 2019-02-13T19:48:34+00:00 74794bfb6795 root 8449 - {} {}"#,
888            r"[incorrect x]", msg
889        );
890
891        let mut expected = Event::Log(LogEvent::from(msg));
892        {
893            let expected = expected.as_mut_log();
894            expected.insert(
895                (PathPrefix::Event, log_schema().timestamp_key().unwrap()),
896                Utc.with_ymd_and_hms(2019, 2, 13, 19, 48, 34)
897                    .single()
898                    .expect("invalid timestamp"),
899            );
900            expected.insert(
901                (PathPrefix::Event, log_schema().host_key().unwrap()),
902                "74794bfb6795",
903            );
904            expected.insert(event_path!("hostname"), "74794bfb6795");
905            expected.insert(
906                log_schema().source_type_key_target_path().unwrap(),
907                "syslog",
908            );
909            expected.insert(event_path!("severity"), "notice");
910            expected.insert(event_path!("facility"), "user");
911            expected.insert(event_path!("version"), 1);
912            expected.insert(event_path!("appname"), "root");
913            expected.insert(event_path!("procid"), 8449);
914            expected.insert(event_path!("source_ip"), "192.168.0.254");
915        }
916
917        let event = event_from_bytes(
918            "host",
919            Some(Bytes::from("192.168.0.254")),
920            raw.into(),
921            LogNamespace::Legacy,
922        )
923        .unwrap();
924        assert_event_data_eq!(event, expected);
925
926        let raw = format!(
927            r#"<13>1 2019-02-13T19:48:34+00:00 74794bfb6795 root 8449 - {} {}"#,
928            r"[incorrect x=]", msg
929        );
930
931        let event = event_from_bytes(
932            "host",
933            Some(Bytes::from("192.168.0.254")),
934            raw.into(),
935            LogNamespace::Legacy,
936        )
937        .unwrap();
938        assert_event_data_eq!(event, expected);
939    }
940
941    #[test]
942    fn handles_empty_sd_element() {
943        fn there_is_map_called_empty(event: Event) -> bool {
944            event
945                .as_log()
946                .get(event_path!("empty"))
947                .expect("empty exists")
948                .is_object()
949        }
950
951        let msg = format!(
952            r#"<13>1 2019-02-13T19:48:34+00:00 74794bfb6795 root 8449 - {} qwerty"#,
953            r"[empty]"
954        );
955
956        let event = event_from_bytes("host", None, msg.into(), LogNamespace::Legacy).unwrap();
957        assert!(there_is_map_called_empty(event));
958
959        let msg = format!(
960            r#"<13>1 2019-02-13T19:48:34+00:00 74794bfb6795 root 8449 - {} qwerty"#,
961            r#"[non_empty x="1"][empty]"#
962        );
963
964        let event = event_from_bytes("host", None, msg.into(), LogNamespace::Legacy).unwrap();
965        assert!(there_is_map_called_empty(event));
966
967        let msg = format!(
968            r#"<13>1 2019-02-13T19:48:34+00:00 74794bfb6795 root 8449 - {} qwerty"#,
969            r#"[empty][non_empty x="1"]"#
970        );
971
972        let event = event_from_bytes("host", None, msg.into(), LogNamespace::Legacy).unwrap();
973        assert!(there_is_map_called_empty(event));
974
975        let msg = format!(
976            r#"<13>1 2019-02-13T19:48:34+00:00 74794bfb6795 root 8449 - {} qwerty"#,
977            r#"[empty not_really="testing the test"]"#
978        );
979
980        let event = event_from_bytes("host", None, msg.into(), LogNamespace::Legacy).unwrap();
981        assert!(there_is_map_called_empty(event));
982    }
983
984    #[test]
985    fn handles_weird_whitespace() {
986        // this should also match rsyslog omfwd with template=RSYSLOG_SyslogProtocol23Format
987        let raw = r#"
988            <13>1 2019-02-13T19:48:34+00:00 74794bfb6795 root 8449 - [meta sequenceId="1"] i am foobar
989            "#;
990        let cleaned = r#"<13>1 2019-02-13T19:48:34+00:00 74794bfb6795 root 8449 - [meta sequenceId="1"] i am foobar"#;
991
992        assert_event_data_eq!(
993            event_from_bytes("host", None, raw.to_owned().into(), LogNamespace::Legacy).unwrap(),
994            event_from_bytes(
995                "host",
996                None,
997                cleaned.to_owned().into(),
998                LogNamespace::Legacy
999            )
1000            .unwrap()
1001        );
1002    }
1003
1004    #[test]
1005    fn handles_dots_in_sdata() {
1006        let raw =
1007            r#"<190>Feb 13 21:31:56 74794bfb6795 liblogging-stdlog:  [origin foo.bar="baz"] hello"#;
1008        let event =
1009            event_from_bytes("host", None, raw.to_owned().into(), LogNamespace::Legacy).unwrap();
1010        assert_eq!(
1011            event.as_log().get(event_path!("origin", "foo.bar")),
1012            Some(&Value::from("baz"))
1013        );
1014    }
1015
1016    #[test]
1017    fn syslog_ng_default_network() {
1018        let msg = "i am foobar";
1019        let raw = format!(r#"<13>Feb 13 20:07:26 74794bfb6795 root[8539]: {msg}"#);
1020        let event = event_from_bytes(
1021            "host",
1022            Some(Bytes::from("192.168.0.254")),
1023            raw.into(),
1024            LogNamespace::Legacy,
1025        )
1026        .unwrap();
1027
1028        let mut expected = Event::Log(LogEvent::from(msg));
1029        {
1030            let value = event.as_log().get(event_path!("timestamp")).unwrap();
1031            let year = value.as_timestamp().unwrap().naive_local().year();
1032
1033            let expected = expected.as_mut_log();
1034            let expected_date: DateTime<Utc> = Local
1035                .with_ymd_and_hms(year, 2, 13, 20, 7, 26)
1036                .single()
1037                .expect("invalid timestamp")
1038                .into();
1039
1040            expected.insert(
1041                (PathPrefix::Event, log_schema().timestamp_key().unwrap()),
1042                expected_date,
1043            );
1044            expected.insert(
1045                (PathPrefix::Event, log_schema().host_key().unwrap()),
1046                "74794bfb6795",
1047            );
1048            expected.insert(
1049                log_schema().source_type_key_target_path().unwrap(),
1050                "syslog",
1051            );
1052            expected.insert(event_path!("hostname"), "74794bfb6795");
1053            expected.insert(event_path!("severity"), "notice");
1054            expected.insert(event_path!("facility"), "user");
1055            expected.insert(event_path!("appname"), "root");
1056            expected.insert(event_path!("procid"), 8539);
1057            expected.insert(event_path!("source_ip"), "192.168.0.254");
1058        }
1059
1060        assert_event_data_eq!(event, expected);
1061    }
1062
1063    #[test]
1064    fn rsyslog_omfwd_tcp_default() {
1065        let msg = "start";
1066        let raw = format!(
1067            r#"<190>Feb 13 21:31:56 74794bfb6795 liblogging-stdlog:  [origin software="rsyslogd" swVersion="8.24.0" x-pid="8979" x-info="http://www.rsyslog.com"] {msg}"#
1068        );
1069        let event = event_from_bytes(
1070            "host",
1071            Some(Bytes::from("192.168.0.254")),
1072            raw.into(),
1073            LogNamespace::Legacy,
1074        )
1075        .unwrap();
1076
1077        let mut expected = Event::Log(LogEvent::from(msg));
1078        {
1079            let value = event.as_log().get(event_path!("timestamp")).unwrap();
1080            let year = value.as_timestamp().unwrap().naive_local().year();
1081
1082            let expected = expected.as_mut_log();
1083            let expected_date: DateTime<Utc> = Local
1084                .with_ymd_and_hms(year, 2, 13, 21, 31, 56)
1085                .single()
1086                .expect("invalid timestamp")
1087                .into();
1088            expected.insert(
1089                (PathPrefix::Event, log_schema().timestamp_key().unwrap()),
1090                expected_date,
1091            );
1092            expected.insert(
1093                log_schema().source_type_key_target_path().unwrap(),
1094                "syslog",
1095            );
1096            expected.insert(event_path!("host"), "74794bfb6795");
1097            expected.insert(event_path!("hostname"), "74794bfb6795");
1098            expected.insert(event_path!("severity"), "info");
1099            expected.insert(event_path!("facility"), "local7");
1100            expected.insert(event_path!("appname"), "liblogging-stdlog");
1101            expected.insert(event_path!("origin", "software"), "rsyslogd");
1102            expected.insert(event_path!("origin", "swVersion"), "8.24.0");
1103            expected.insert(event_path!("source_ip"), "192.168.0.254");
1104            expected.insert(event_path!("origin", "x-pid"), "8979");
1105            expected.insert(event_path!("origin", "x-info"), "http://www.rsyslog.com");
1106        }
1107
1108        assert_event_data_eq!(event, expected);
1109    }
1110
1111    #[test]
1112    fn rsyslog_omfwd_tcp_forward_format() {
1113        let msg = "start";
1114        let raw = format!(
1115            r#"<190>2019-02-13T21:53:30.605850+00:00 74794bfb6795 liblogging-stdlog:  [origin software="rsyslogd" swVersion="8.24.0" x-pid="9043" x-info="http://www.rsyslog.com"] {msg}"#
1116        );
1117
1118        let mut expected = Event::Log(LogEvent::from(msg));
1119        {
1120            let expected = expected.as_mut_log();
1121            expected.insert(
1122                (PathPrefix::Event, log_schema().timestamp_key().unwrap()),
1123                Utc.with_ymd_and_hms(2019, 2, 13, 21, 53, 30)
1124                    .single()
1125                    .and_then(|t| t.with_nanosecond(605_850 * 1000))
1126                    .expect("invalid timestamp"),
1127            );
1128            expected.insert(
1129                log_schema().source_type_key_target_path().unwrap(),
1130                "syslog",
1131            );
1132            expected.insert(event_path!("host"), "74794bfb6795");
1133            expected.insert(event_path!("hostname"), "74794bfb6795");
1134            expected.insert(event_path!("severity"), "info");
1135            expected.insert(event_path!("facility"), "local7");
1136            expected.insert(event_path!("appname"), "liblogging-stdlog");
1137            expected.insert(event_path!("origin", "software"), "rsyslogd");
1138            expected.insert(event_path!("origin", "swVersion"), "8.24.0");
1139            expected.insert(event_path!("origin", "x-pid"), "9043");
1140            expected.insert(event_path!("origin", "x-info"), "http://www.rsyslog.com");
1141        }
1142
1143        assert_event_data_eq!(
1144            event_from_bytes("host", None, raw.into(), LogNamespace::Legacy).unwrap(),
1145            expected
1146        );
1147    }
1148
1149    #[tokio::test]
1150    async fn test_tcp_syslog() {
1151        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1152            let num_messages: usize = 10000;
1153            let (_guard, in_addr) = next_addr();
1154
1155            // Create and spawn the source.
1156            let config = SyslogConfig::from_mode(Mode::Tcp {
1157                address: in_addr.into(),
1158                permit_origin: None,
1159                keepalive: None,
1160                tls: None,
1161                receive_buffer_bytes: None,
1162                connection_limit: None,
1163                tls_handshake_timeout_secs: None,
1164            });
1165
1166            let key = ComponentKey::from("in");
1167            let (tx, rx) = SourceSender::new_test();
1168            let (context, shutdown) = SourceContext::new_shutdown(&key, tx);
1169            let shutdown_complete = shutdown.shutdown_tripwire();
1170
1171            let source = config
1172                .build(context)
1173                .await
1174                .expect("source should not fail to build");
1175            tokio::spawn(source);
1176
1177            // Wait for source to become ready to accept traffic.
1178            wait_for_tcp(in_addr).await;
1179
1180            let output_events = CountReceiver::receive_events(rx);
1181
1182            // Now craft and send syslog messages to the source, and collect them on the other side.
1183            let input_messages: Vec<SyslogMessageRfc5424> = (0..num_messages)
1184                .map(|i| SyslogMessageRfc5424::random(i, 30, 4, 3, 3))
1185                .collect();
1186
1187            let input_lines: Vec<String> =
1188                input_messages.iter().map(|msg| msg.to_string()).collect();
1189
1190            send_lines(in_addr, input_lines).await.unwrap();
1191
1192            // Wait a short period of time to ensure the messages get sent.
1193            sleep(Duration::from_secs(2)).await;
1194
1195            // Shutdown the source, and make sure we've got all the messages we sent in.
1196            shutdown
1197                .shutdown_all(Some(Instant::now() + Duration::from_millis(100)))
1198                .await;
1199            shutdown_complete.await;
1200
1201            let output_events = output_events.await;
1202            assert_eq!(output_events.len(), num_messages);
1203
1204            let output_messages: Vec<SyslogMessageRfc5424> = output_events
1205                .into_iter()
1206                .map(|mut e| {
1207                    e.as_mut_log().remove(event_path!("hostname")); // Vector adds this field which will cause a parse error.
1208                    e.as_mut_log().remove(event_path!("source_ip")); // Vector adds this field which will cause a parse error.
1209                    e.into()
1210                })
1211                .collect();
1212            assert_eq!(output_messages, input_messages);
1213        })
1214        .await;
1215    }
1216
1217    #[tokio::test]
1218    async fn test_udp_syslog() {
1219        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1220            let num_messages: usize = 1000;
1221            let (_guard, in_addr) = next_addr();
1222
1223            // Create and spawn the source.
1224            let config = SyslogConfig::from_mode(Mode::Udp {
1225                address: in_addr.into(),
1226                receive_buffer_bytes: Some(4 * 1024 * 1024),
1227            });
1228
1229            let key = ComponentKey::from("in");
1230            let (tx, rx) = SourceSender::new_test();
1231            let (context, shutdown) = SourceContext::new_shutdown(&key, tx);
1232            let shutdown_complete = shutdown.shutdown_tripwire();
1233
1234            let source = config
1235                .build(context)
1236                .await
1237                .expect("source should not fail to build");
1238            tokio::spawn(source);
1239
1240            // Give UDP a brief moment to start listening.
1241            sleep(Duration::from_millis(150)).await;
1242
1243            let output_events = CountReceiver::receive_events(rx);
1244
1245            // Craft and send syslog messages as individual UDP datagrams.
1246            let input_messages: Vec<SyslogMessageRfc5424> = (0..num_messages)
1247                .map(|i| SyslogMessageRfc5424::random(i, 30, 4, 3, 3))
1248                .collect();
1249
1250            let input_lines: Vec<String> =
1251                input_messages.iter().map(|msg| msg.to_string()).collect();
1252
1253            let socket = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap();
1254            for line in input_lines {
1255                socket.send_to(line.as_bytes(), in_addr).await.unwrap();
1256            }
1257
1258            // Wait a short period of time to ensure the messages get sent.
1259            sleep(Duration::from_secs(2)).await;
1260
1261            // Shutdown the source, and make sure we've got all the messages we sent in.
1262            shutdown
1263                .shutdown_all(Some(Instant::now() + Duration::from_millis(100)))
1264                .await;
1265            shutdown_complete.await;
1266
1267            let output_events = output_events.await;
1268            assert_eq!(output_events.len(), num_messages);
1269
1270            let output_messages: Vec<SyslogMessageRfc5424> = output_events
1271                .into_iter()
1272                .map(|mut e| {
1273                    e.as_mut_log().remove(event_path!("hostname")); // Vector adds this field which will cause a parse error.
1274                    e.as_mut_log().remove(event_path!("source_ip")); // Vector adds this field which will cause a parse error.
1275                    e.into()
1276                })
1277                .collect();
1278            assert_eq!(output_messages, input_messages);
1279        })
1280        .await;
1281    }
1282
1283    #[cfg(unix)]
1284    #[tokio::test]
1285    async fn test_unix_stream_syslog() {
1286        use std::os::unix::net::UnixStream as StdUnixStream;
1287
1288        use futures_util::{SinkExt, stream};
1289        use tokio::{io::AsyncWriteExt, net::UnixStream};
1290        use tokio_util::codec::{FramedWrite, LinesCodec};
1291
1292        use crate::test_util::components::SOCKET_PUSH_SOURCE_TAGS;
1293
1294        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1295            let num_messages: usize = 1;
1296            let in_path = tempfile::tempdir().unwrap().keep().join("stream_test");
1297
1298            // Create and spawn the source.
1299            let config = SyslogConfig::from_mode(Mode::Unix {
1300                path: in_path.clone(),
1301                socket_file_mode: None,
1302            });
1303
1304            let key = ComponentKey::from("in");
1305            let (tx, rx) = SourceSender::new_test();
1306            let (context, shutdown) = SourceContext::new_shutdown(&key, tx);
1307            let shutdown_complete = shutdown.shutdown_tripwire();
1308
1309            let source = config
1310                .build(context)
1311                .await
1312                .expect("source should not fail to build");
1313            tokio::spawn(source);
1314
1315            // Wait for source to become ready to accept traffic.
1316            while StdUnixStream::connect(&in_path).is_err() {
1317                tokio::task::yield_now().await;
1318            }
1319
1320            let output_events = CountReceiver::receive_events(rx);
1321
1322            // Now craft and send syslog messages to the source, and collect them on the other side.
1323            let input_messages: Vec<SyslogMessageRfc5424> = (0..num_messages)
1324                .map(|i| SyslogMessageRfc5424::random(i, 30, 4, 3, 3))
1325                .collect();
1326
1327            let stream = UnixStream::connect(&in_path).await.unwrap();
1328            let mut sink = FramedWrite::new(stream, LinesCodec::new());
1329
1330            let lines: Vec<String> = input_messages.iter().map(|msg| msg.to_string()).collect();
1331            let mut lines = stream::iter(lines).map(Ok);
1332            sink.send_all(&mut lines).await.unwrap();
1333
1334            let stream = sink.get_mut();
1335            stream.shutdown().await.unwrap();
1336
1337            // Wait a short period of time to ensure the messages get sent.
1338            sleep(Duration::from_secs(1)).await;
1339
1340            shutdown
1341                .shutdown_all(Some(Instant::now() + Duration::from_millis(100)))
1342                .await;
1343            shutdown_complete.await;
1344
1345            let output_events = output_events.await;
1346            assert_eq!(output_events.len(), num_messages);
1347
1348            let output_messages: Vec<SyslogMessageRfc5424> = output_events
1349                .into_iter()
1350                .map(|mut e| {
1351                    e.as_mut_log().remove(event_path!("hostname")); // Vector adds this field which will cause a parse error.
1352                    e.as_mut_log().remove(event_path!("source_ip")); // Vector adds this field which will cause a parse error.
1353                    e.into()
1354                })
1355                .collect();
1356            assert_eq!(output_messages, input_messages);
1357        })
1358        .await;
1359    }
1360
1361    #[tokio::test]
1362    async fn test_octet_counting_syslog() {
1363        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1364            let num_messages: usize = 10000;
1365            let (_guard, in_addr) = next_addr();
1366
1367            // Create and spawn the source.
1368            let config = SyslogConfig::from_mode(Mode::Tcp {
1369                address: in_addr.into(),
1370                permit_origin: None,
1371                keepalive: None,
1372                tls: None,
1373                receive_buffer_bytes: None,
1374                connection_limit: None,
1375                tls_handshake_timeout_secs: None,
1376            });
1377
1378            let key = ComponentKey::from("in");
1379            let (tx, rx) = SourceSender::new_test();
1380            let (context, shutdown) = SourceContext::new_shutdown(&key, tx);
1381            let shutdown_complete = shutdown.shutdown_tripwire();
1382
1383            let source = config
1384                .build(context)
1385                .await
1386                .expect("source should not fail to build");
1387            tokio::spawn(source);
1388
1389            // Wait for source to become ready to accept traffic.
1390            wait_for_tcp(in_addr).await;
1391
1392            let output_events = CountReceiver::receive_events(rx);
1393
1394            // Now craft and send syslog messages to the source, and collect them on the other side.
1395            let input_messages: Vec<SyslogMessageRfc5424> = (0..num_messages)
1396                .map(|i| {
1397                    let mut msg = SyslogMessageRfc5424::random(i, 30, 4, 3, 3);
1398                    msg.message.push('\n');
1399                    msg.message.push_str(&random_string(30));
1400                    msg
1401                })
1402                .collect();
1403
1404            let codec = BytesCodec::new();
1405            let input_lines: Vec<Bytes> = input_messages
1406                .iter()
1407                .map(|msg| {
1408                    let s = msg.to_string();
1409                    format!("{} {}", s.len(), s).into()
1410                })
1411                .collect();
1412
1413            send_encodable(in_addr, codec, input_lines).await.unwrap();
1414
1415            // Wait a short period of time to ensure the messages get sent.
1416            sleep(Duration::from_secs(2)).await;
1417
1418            // Shutdown the source, and make sure we've got all the messages we sent in.
1419            shutdown
1420                .shutdown_all(Some(Instant::now() + Duration::from_millis(100)))
1421                .await;
1422            shutdown_complete.await;
1423
1424            let output_events = output_events.await;
1425            assert_eq!(output_events.len(), num_messages);
1426
1427            let output_messages: Vec<SyslogMessageRfc5424> = output_events
1428                .into_iter()
1429                .map(|mut e| {
1430                    e.as_mut_log().remove(event_path!("hostname")); // Vector adds this field which will cause a parse error.
1431                    e.as_mut_log().remove(event_path!("source_ip")); // Vector adds this field which will cause a parse error.
1432                    e.into()
1433                })
1434                .collect();
1435            assert_eq!(output_messages, input_messages);
1436        })
1437        .await;
1438    }
1439
1440    #[derive(Deserialize, PartialEq, Clone, Debug)]
1441    struct SyslogMessageRfc5424 {
1442        msgid: String,
1443        severity: Severity,
1444        facility: Facility,
1445        version: u8,
1446        timestamp: String,
1447        host: String,
1448        source_type: String,
1449        appname: String,
1450        procid: usize,
1451        message: String,
1452        #[serde(flatten)]
1453        structured_data: StructuredData,
1454    }
1455
1456    impl SyslogMessageRfc5424 {
1457        fn random(
1458            id: usize,
1459            msg_len: usize,
1460            field_len: usize,
1461            max_map_size: usize,
1462            max_children: usize,
1463        ) -> Self {
1464            let msg = random_string(msg_len);
1465            let structured_data = random_structured_data(max_map_size, max_children, field_len);
1466
1467            let timestamp = Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true);
1468            //"secfrac" can contain up to 6 digits, but TCP sinks uses `AutoSi`
1469
1470            Self {
1471                msgid: format!("test{id}"),
1472                severity: Severity::LOG_INFO,
1473                facility: Facility::LOG_USER,
1474                version: 1,
1475                timestamp,
1476                host: "hogwarts".to_owned(),
1477                source_type: "syslog".to_owned(),
1478                appname: "harry".to_owned(),
1479                procid: rng().random_range(0..32768),
1480                structured_data,
1481                message: msg,
1482            }
1483        }
1484    }
1485
1486    impl fmt::Display for SyslogMessageRfc5424 {
1487        fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
1488            write!(
1489                f,
1490                "<{}>{} {} {} {} {} {} {} {}",
1491                encode_priority(self.severity, self.facility),
1492                self.version,
1493                self.timestamp,
1494                self.host,
1495                self.appname,
1496                self.procid,
1497                self.msgid,
1498                format_structured_data_rfc5424(&self.structured_data),
1499                self.message
1500            )
1501        }
1502    }
1503
1504    impl From<Event> for SyslogMessageRfc5424 {
1505        fn from(e: Event) -> Self {
1506            let (value, _) = e.into_log().into_parts();
1507            let mut fields = value.into_object().unwrap();
1508
1509            Self {
1510                msgid: fields.remove("msgid").map(value_to_string).unwrap(),
1511                severity: fields
1512                    .remove("severity")
1513                    .map(value_to_string)
1514                    .and_then(|s| Severity::from_str(s.as_str()))
1515                    .unwrap(),
1516                facility: fields
1517                    .remove("facility")
1518                    .map(value_to_string)
1519                    .and_then(|s| Facility::from_str(s.as_str()))
1520                    .unwrap(),
1521                version: fields
1522                    .remove("version")
1523                    .map(value_to_string)
1524                    .map(|s| u8::from_str(s.as_str()).unwrap())
1525                    .unwrap(),
1526                timestamp: fields.remove("timestamp").map(value_to_string).unwrap(),
1527                host: fields.remove("host").map(value_to_string).unwrap(),
1528                source_type: fields.remove("source_type").map(value_to_string).unwrap(),
1529                appname: fields.remove("appname").map(value_to_string).unwrap(),
1530                procid: fields
1531                    .remove("procid")
1532                    .map(value_to_string)
1533                    .map(|s| usize::from_str(s.as_str()).unwrap())
1534                    .unwrap(),
1535                message: fields.remove("message").map(value_to_string).unwrap(),
1536                structured_data: structured_data_from_fields(fields),
1537            }
1538        }
1539    }
1540
1541    fn structured_data_from_fields(fields: ObjectMap) -> StructuredData {
1542        let mut structured_data = StructuredData::default();
1543
1544        for (key, value) in fields.into_iter() {
1545            let subfields = value
1546                .into_object()
1547                .unwrap()
1548                .into_iter()
1549                .map(|(k, v)| (k.into(), value_to_string(v)))
1550                .collect();
1551
1552            structured_data.insert(key.into(), subfields);
1553        }
1554
1555        structured_data
1556    }
1557
1558    #[allow(non_camel_case_types, clippy::upper_case_acronyms)]
1559    #[derive(Copy, Clone, Deserialize, PartialEq, Eq, Debug)]
1560    pub enum Severity {
1561        #[serde(rename(deserialize = "emergency"))]
1562        LOG_EMERG,
1563        #[serde(rename(deserialize = "alert"))]
1564        LOG_ALERT,
1565        #[serde(rename(deserialize = "critical"))]
1566        LOG_CRIT,
1567        #[serde(rename(deserialize = "error"))]
1568        LOG_ERR,
1569        #[serde(rename(deserialize = "warn"))]
1570        LOG_WARNING,
1571        #[serde(rename(deserialize = "notice"))]
1572        LOG_NOTICE,
1573        #[serde(rename(deserialize = "info"))]
1574        LOG_INFO,
1575        #[serde(rename(deserialize = "debug"))]
1576        LOG_DEBUG,
1577    }
1578
1579    impl Severity {
1580        fn from_str(s: &str) -> Option<Self> {
1581            match s {
1582                "emergency" => Some(Self::LOG_EMERG),
1583                "alert" => Some(Self::LOG_ALERT),
1584                "critical" => Some(Self::LOG_CRIT),
1585                "error" => Some(Self::LOG_ERR),
1586                "warn" => Some(Self::LOG_WARNING),
1587                "notice" => Some(Self::LOG_NOTICE),
1588                "info" => Some(Self::LOG_INFO),
1589                "debug" => Some(Self::LOG_DEBUG),
1590
1591                x => {
1592                    #[allow(clippy::print_stdout)]
1593                    {
1594                        println!("converting severity str, got {x}");
1595                    }
1596                    None
1597                }
1598            }
1599        }
1600    }
1601
1602    #[allow(non_camel_case_types, clippy::upper_case_acronyms)]
1603    #[derive(Copy, Clone, PartialEq, Eq, Deserialize, Debug)]
1604    pub enum Facility {
1605        #[serde(rename(deserialize = "kernel"))]
1606        LOG_KERN = 0 << 3,
1607        #[serde(rename(deserialize = "user"))]
1608        LOG_USER = 1 << 3,
1609        #[serde(rename(deserialize = "mail"))]
1610        LOG_MAIL = 2 << 3,
1611        #[serde(rename(deserialize = "daemon"))]
1612        LOG_DAEMON = 3 << 3,
1613        #[serde(rename(deserialize = "auth"))]
1614        LOG_AUTH = 4 << 3,
1615        #[serde(rename(deserialize = "syslog"))]
1616        LOG_SYSLOG = 5 << 3,
1617    }
1618
1619    impl Facility {
1620        fn from_str(s: &str) -> Option<Self> {
1621            match s {
1622                "kernel" => Some(Self::LOG_KERN),
1623                "user" => Some(Self::LOG_USER),
1624                "mail" => Some(Self::LOG_MAIL),
1625                "daemon" => Some(Self::LOG_DAEMON),
1626                "auth" => Some(Self::LOG_AUTH),
1627                "syslog" => Some(Self::LOG_SYSLOG),
1628                _ => None,
1629            }
1630        }
1631    }
1632
1633    type StructuredData = HashMap<String, HashMap<String, String>>;
1634
1635    fn random_structured_data(
1636        max_map_size: usize,
1637        max_children: usize,
1638        field_len: usize,
1639    ) -> StructuredData {
1640        let amount = rng().random_range(0..max_children);
1641
1642        random_maps(max_map_size, field_len)
1643            .filter(|m| !m.is_empty()) //syslog_rfc5424 ignores empty maps, tested separately
1644            .take(amount)
1645            .enumerate()
1646            .map(|(i, map)| (format!("id{i}"), map))
1647            .collect()
1648    }
1649
1650    fn format_structured_data_rfc5424(data: &StructuredData) -> String {
1651        if data.is_empty() {
1652            "-".to_string()
1653        } else {
1654            let mut res = String::new();
1655            for (id, params) in data {
1656                res = res + "[" + id;
1657                for (name, value) in params {
1658                    res = res + " " + name + "=\"" + value + "\"";
1659                }
1660                res += "]";
1661            }
1662
1663            res
1664        }
1665    }
1666
1667    const fn encode_priority(severity: Severity, facility: Facility) -> u8 {
1668        facility as u8 | severity as u8
1669    }
1670
1671    fn value_to_string(v: Value) -> String {
1672        if v.is_bytes() {
1673            let buf = v.as_bytes().unwrap();
1674            String::from_utf8_lossy(buf).to_string()
1675        } else if v.is_timestamp() {
1676            let ts = v.as_timestamp().unwrap();
1677            ts.to_rfc3339_opts(SecondsFormat::AutoSi, true)
1678        } else {
1679            v.to_string()
1680        }
1681    }
1682}
vector/sources/syslog.rs

vector/sources/
syslog.rs
