vector/sources/socket/
mod.rs

1pub mod tcp;
2pub mod udp;
3#[cfg(unix)]
4mod unix;
5
6use vector_lib::{
7    codecs::decoding::DeserializerConfig,
8    config::{LegacyKey, LogNamespace, log_schema},
9    configurable::configurable_component,
10    lookup::{lookup_v2::OptionalValuePath, owned_value_path},
11};
12use vrl::value::{Kind, kind::Collection};
13
14use crate::{
15    codecs::DecodingConfig,
16    config::{GenerateConfig, Resource, SourceConfig, SourceContext, SourceOutput},
17    sources::util::net::TcpSource,
18    tls::MaybeTlsSettings,
19};
20
21/// Configuration for the `socket` source.
22#[configurable_component(source("socket", "Collect logs over a socket."))]
23#[derive(Clone, Debug)]
24pub struct SocketConfig {
25    #[serde(flatten)]
26    pub mode: Mode,
27}
28
29/// Listening mode for the `socket` source.
30#[configurable_component]
31#[derive(Clone, Debug)]
32#[serde(tag = "mode", rename_all = "snake_case")]
33#[configurable(metadata(docs::enum_tag_description = "The type of socket to use."))]
34#[allow(clippy::large_enum_variant)] // just used for configuration
35pub enum Mode {
36    /// Listen on TCP.
37    Tcp(tcp::TcpConfig),
38
39    /// Listen on UDP.
40    Udp(udp::UdpConfig),
41
42    /// Listen on a Unix domain socket (UDS), in datagram mode.
43    #[cfg(unix)]
44    UnixDatagram(unix::UnixConfig),
45
46    /// Listen on a Unix domain socket (UDS), in stream mode.
47    #[cfg(unix)]
48    #[serde(alias = "unix")]
49    UnixStream(unix::UnixConfig),
50}
51
52impl SocketConfig {
53    pub fn new_tcp(tcp_config: tcp::TcpConfig) -> Self {
54        tcp_config.into()
55    }
56
57    pub fn make_basic_tcp_config(addr: std::net::SocketAddr) -> Self {
58        tcp::TcpConfig::from_address(addr.into()).into()
59    }
60
61    fn decoding(&self) -> DeserializerConfig {
62        match &self.mode {
63            Mode::Tcp(config) => config.decoding().clone(),
64            Mode::Udp(config) => config.decoding().clone(),
65            #[cfg(unix)]
66            Mode::UnixDatagram(config) => config.decoding().clone(),
67            #[cfg(unix)]
68            Mode::UnixStream(config) => config.decoding().clone(),
69        }
70    }
71
72    fn log_namespace(&self, global_log_namespace: LogNamespace) -> LogNamespace {
73        match &self.mode {
74            Mode::Tcp(config) => global_log_namespace.merge(config.log_namespace),
75            Mode::Udp(config) => global_log_namespace.merge(config.log_namespace),
76            #[cfg(unix)]
77            Mode::UnixDatagram(config) => global_log_namespace.merge(config.log_namespace),
78            #[cfg(unix)]
79            Mode::UnixStream(config) => global_log_namespace.merge(config.log_namespace),
80        }
81    }
82}
83
84impl From<tcp::TcpConfig> for SocketConfig {
85    fn from(config: tcp::TcpConfig) -> Self {
86        SocketConfig {
87            mode: Mode::Tcp(config),
88        }
89    }
90}
91
92impl From<udp::UdpConfig> for SocketConfig {
93    fn from(config: udp::UdpConfig) -> Self {
94        SocketConfig {
95            mode: Mode::Udp(config),
96        }
97    }
98}
99
100impl GenerateConfig for SocketConfig {
101    fn generate_config() -> toml::Value {
102        toml::from_str(
103            r#"mode = "tcp"
104            address = "0.0.0.0:9000""#,
105        )
106        .unwrap()
107    }
108}
109
110#[async_trait::async_trait]
111#[typetag::serde(name = "socket")]
112impl SourceConfig for SocketConfig {
113    async fn build(&self, cx: SourceContext) -> crate::Result<super::Source> {
114        match self.mode.clone() {
115            Mode::Tcp(config) => {
116                let log_namespace = cx.log_namespace(config.log_namespace);
117
118                let decoding = config.decoding().clone();
119                let decoder = DecodingConfig::new(
120                    config
121                        .framing
122                        .clone()
123                        .unwrap_or_else(|| decoding.default_stream_framing()),
124                    decoding,
125                    log_namespace,
126                )
127                .build()?;
128
129                let tcp = tcp::RawTcpSource::new(config.clone(), decoder, log_namespace);
130                let tls_config = config.tls().as_ref().map(|tls| tls.tls_config.clone());
131                let tls_client_metadata_key = config
132                    .tls()
133                    .as_ref()
134                    .and_then(|tls| tls.client_metadata_key.clone())
135                    .and_then(|k| k.path);
136                let tls = MaybeTlsSettings::from_config(tls_config.as_ref(), true)?;
137                tcp.run(
138                    config.address(),
139                    config.keepalive(),
140                    config.shutdown_timeout_secs(),
141                    tls,
142                    None, // tls_reloader: not wired for this source
143                    tls_client_metadata_key,
144                    config.receive_buffer_bytes(),
145                    config.max_connection_duration_secs(),
146                    cx,
147                    false.into(),
148                    config.connection_limit,
149                    config.permit_origin.map(Into::into),
150                    SocketConfig::NAME,
151                    log_namespace,
152                )
153            }
154            Mode::Udp(config) => {
155                let log_namespace = cx.log_namespace(config.log_namespace);
156                let decoding = config.decoding().clone();
157                let framing = config
158                    .framing()
159                    .clone()
160                    .unwrap_or_else(|| decoding.default_message_based_framing());
161                let decoder = DecodingConfig::new(framing, decoding, log_namespace).build()?;
162                Ok(udp::udp(
163                    config,
164                    decoder,
165                    cx.shutdown,
166                    cx.out,
167                    log_namespace,
168                ))
169            }
170            #[cfg(unix)]
171            Mode::UnixDatagram(config) => {
172                let log_namespace = cx.log_namespace(config.log_namespace);
173                let decoding = config.decoding.clone();
174                let framing = config
175                    .framing
176                    .clone()
177                    .unwrap_or_else(|| decoding.default_message_based_framing());
178                let decoder = DecodingConfig::new(framing, decoding, log_namespace).build()?;
179
180                unix::unix_datagram(config, decoder, cx.shutdown, cx.out, log_namespace)
181            }
182            #[cfg(unix)]
183            Mode::UnixStream(config) => {
184                let log_namespace = cx.log_namespace(config.log_namespace);
185
186                let decoding = config.decoding().clone();
187                let decoder = DecodingConfig::new(
188                    config
189                        .framing
190                        .clone()
191                        .unwrap_or_else(|| decoding.default_stream_framing()),
192                    decoding,
193                    log_namespace,
194                )
195                .build()?;
196
197                unix::unix_stream(config, decoder, cx.shutdown, cx.out, log_namespace)
198            }
199        }
200    }
201
202    fn outputs(&self, global_log_namespace: LogNamespace) -> Vec<SourceOutput> {
203        let log_namespace = self.log_namespace(global_log_namespace);
204
205        let schema_definition = self
206            .decoding()
207            .schema_definition(log_namespace)
208            .with_standard_vector_source_metadata();
209
210        let schema_definition = match &self.mode {
211            Mode::Tcp(config) => {
212                let legacy_host_key = config.host_key().path.map(LegacyKey::InsertIfEmpty);
213
214                let legacy_port_key = config.port_key().clone().path.map(LegacyKey::InsertIfEmpty);
215
216                let tls_client_metadata_path = config
217                    .tls()
218                    .as_ref()
219                    .and_then(|tls| tls.client_metadata_key.as_ref())
220                    .and_then(|k| k.path.clone())
221                    .map(LegacyKey::Overwrite);
222
223                schema_definition
224                    .with_source_metadata(
225                        Self::NAME,
226                        legacy_host_key,
227                        &owned_value_path!("host"),
228                        Kind::bytes(),
229                        Some("host"),
230                    )
231                    .with_source_metadata(
232                        Self::NAME,
233                        legacy_port_key,
234                        &owned_value_path!("port"),
235                        Kind::integer(),
236                        None,
237                    )
238                    .with_source_metadata(
239                        Self::NAME,
240                        tls_client_metadata_path,
241                        &owned_value_path!("tls_client_metadata"),
242                        Kind::object(Collection::empty().with_unknown(Kind::bytes()))
243                            .or_undefined(),
244                        None,
245                    )
246            }
247            Mode::Udp(config) => {
248                let legacy_host_key = config.host_key().path.map(LegacyKey::InsertIfEmpty);
249
250                let legacy_port_key = config.port_key().clone().path.map(LegacyKey::InsertIfEmpty);
251
252                schema_definition
253                    .with_source_metadata(
254                        Self::NAME,
255                        legacy_host_key,
256                        &owned_value_path!("host"),
257                        Kind::bytes(),
258                        None,
259                    )
260                    .with_source_metadata(
261                        Self::NAME,
262                        legacy_port_key,
263                        &owned_value_path!("port"),
264                        Kind::integer(),
265                        None,
266                    )
267            }
268            #[cfg(unix)]
269            Mode::UnixDatagram(config) => {
270                let legacy_host_key = config.host_key().clone().path.map(LegacyKey::InsertIfEmpty);
271
272                schema_definition.with_source_metadata(
273                    Self::NAME,
274                    legacy_host_key,
275                    &owned_value_path!("host"),
276                    Kind::bytes(),
277                    None,
278                )
279            }
280            #[cfg(unix)]
281            Mode::UnixStream(config) => {
282                let legacy_host_key = config.host_key().clone().path.map(LegacyKey::InsertIfEmpty);
283
284                schema_definition.with_source_metadata(
285                    Self::NAME,
286                    legacy_host_key,
287                    &owned_value_path!("host"),
288                    Kind::bytes(),
289                    None,
290                )
291            }
292        };
293
294        vec![SourceOutput::new_maybe_logs(
295            self.decoding().output_type(),
296            schema_definition,
297        )]
298    }
299
300    fn resources(&self) -> Vec<Resource> {
301        match self.mode.clone() {
302            Mode::Tcp(tcp) => vec![tcp.address().as_tcp_resource()],
303            Mode::Udp(udp) => vec![udp.address().as_udp_resource()],
304            #[cfg(unix)]
305            Mode::UnixDatagram(_) => vec![],
306            #[cfg(unix)]
307            Mode::UnixStream(_) => vec![],
308        }
309    }
310
311    fn can_acknowledge(&self) -> bool {
312        false
313    }
314}
315
316pub(crate) fn default_host_key() -> OptionalValuePath {
317    log_schema().host_key().cloned().into()
318}
319
320#[cfg(test)]
321mod test {
322    use std::{
323        collections::HashMap,
324        net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket},
325        sync::{
326            Arc,
327            atomic::{AtomicBool, Ordering},
328        },
329        thread,
330    };
331
332    use approx::assert_relative_eq;
333    use bytes::{BufMut, Bytes, BytesMut};
334    use futures::{StreamExt, stream};
335    use rand::{SeedableRng, rngs::SmallRng, seq::SliceRandom};
336    use serde_json::json;
337    use tokio::{
338        io::AsyncReadExt,
339        net::TcpStream,
340        task::JoinHandle,
341        time::{Duration, Instant, timeout},
342    };
343    #[cfg(unix)]
344    use vector_lib::codecs::{
345        CharacterDelimitedDecoderConfig, decoding::CharacterDelimitedDecoderOptions,
346    };
347    use vector_lib::{
348        codecs::{GelfDeserializerConfig, NewlineDelimitedDecoderConfig},
349        event::EventContainer,
350        lookup::{lookup_v2::OptionalValuePath, owned_value_path, path},
351    };
352    use vrl::{btreemap, value, value::ObjectMap};
353    #[cfg(unix)]
354    use {
355        super::{Mode, unix::UnixConfig},
356        crate::sources::util::unix::UNNAMED_SOCKET_HOST,
357        crate::test_util::wait_for,
358        futures::{SinkExt, Stream},
359        std::future::ready,
360        std::os::unix::fs::PermissionsExt,
361        std::path::PathBuf,
362        tokio::{
363            io::AsyncWriteExt,
364            net::{UnixDatagram, UnixStream},
365            task::yield_now,
366        },
367        tokio_util::codec::{FramedWrite, LinesCodec},
368    };
369
370    use super::{SocketConfig, tcp::TcpConfig, udp::UdpConfig};
371    use crate::{
372        SourceSender,
373        config::{ComponentKey, GlobalOptions, SourceConfig, SourceContext, log_schema},
374        event::{Event, LogEvent},
375        shutdown::{ShutdownSignal, SourceShutdownCoordinator},
376        sinks::util::tcp::TcpSinkConfig,
377        sources::util::net::SocketListenAddr,
378        test_util::{
379            addr::{PortGuard, next_addr, next_addr_any},
380            collect_n, collect_n_limited,
381            components::{
382                COMPONENT_ERROR_TAGS, SOCKET_PUSH_SOURCE_TAGS, assert_source_compliance,
383                assert_source_error,
384            },
385            random_string, send_lines, send_lines_tls, wait_for_tcp,
386        },
387        tls::{self, TlsConfig, TlsEnableableConfig, TlsSourceConfig},
388    };
389
390    async fn wait_for_tcp_and_release(guard: PortGuard, addr: SocketAddr) {
391        wait_for_tcp(addr).await;
392        drop(guard) // Now we're sure the socket was bound by the server and we can release the guard
393    }
394
395    pub fn bind_unused_udp() -> UdpSocket {
396        // Bind to port 0 to let the OS assign an available port
397        UdpSocket::bind((IpAddr::V4(Ipv4Addr::LOCALHOST), 0))
398            .expect("Failed to bind UDP socket to OS-assigned port")
399    }
400
401    /// Bind a UDP socket suitable for sending multicast packets through the loopback interface.
402    /// Sets `IP_MULTICAST_IF` to loopback so packets route through `lo` regardless of the
403    /// system's default multicast route, necessary for reliable local testing on macOS.
404    pub fn bind_unused_udp_multicast() -> UdpSocket {
405        let socket = UdpSocket::bind((IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0))
406            .expect("Failed to bind UDP socket to OS-assigned port");
407        socket2::SockRef::from(&socket)
408            .set_multicast_if_v4(&Ipv4Addr::LOCALHOST)
409            .expect("Failed to set multicast interface to loopback");
410        socket
411    }
412
413    fn get_gelf_payload(message: &str) -> String {
414        serde_json::to_string(&json!({
415            "version": "1.1",
416            "host": "example.org",
417            "short_message": message,
418            "timestamp": 1234567890.123,
419            "level": 6,
420            "_foo": "bar",
421        }))
422        .unwrap()
423    }
424
425    fn create_gelf_chunk(
426        message_id: u64,
427        sequence_number: u8,
428        total_chunks: u8,
429        payload: &[u8],
430    ) -> Bytes {
431        const GELF_MAGIC: [u8; 2] = [0x1e, 0x0f];
432        let mut chunk = BytesMut::new();
433        chunk.put_slice(&GELF_MAGIC);
434        chunk.put_u64(message_id);
435        chunk.put_u8(sequence_number);
436        chunk.put_u8(total_chunks);
437        chunk.put(payload);
438        chunk.freeze()
439    }
440
441    fn get_gelf_chunks(short_message: &str, max_size: usize, rng: &mut SmallRng) -> Vec<Bytes> {
442        let message_id = rand::random();
443        let payload = get_gelf_payload(short_message);
444        let payload_chunks = payload.as_bytes().chunks(max_size).collect::<Vec<_>>();
445        let total_chunks = payload_chunks.len();
446        assert!(total_chunks <= 128, "too many gelf chunks");
447
448        let mut chunks = payload_chunks
449            .into_iter()
450            .enumerate()
451            .map(|(i, payload_chunk)| {
452                create_gelf_chunk(message_id, i as u8, total_chunks as u8, payload_chunk)
453            })
454            .collect::<Vec<_>>();
455        // Shuffle the chunks to simulate out-of-order delivery
456        chunks.shuffle(rng);
457        chunks
458    }
459
460    #[test]
461    fn generate_config() {
462        crate::test_util::test_generate_config::<SocketConfig>();
463    }
464
465    //////// TCP TESTS ////////
466    #[tokio::test]
467    async fn tcp_it_includes_host() {
468        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
469            let (tx, mut rx) = SourceSender::new_test();
470            let (guard, addr) = next_addr();
471
472            let server = SocketConfig::from(TcpConfig::from_address(addr.into()))
473                .build(SourceContext::new_test(tx, None))
474                .await
475                .unwrap();
476            tokio::spawn(server);
477
478            wait_for_tcp_and_release(guard, addr).await;
479
480            let addr = send_lines(addr, vec!["test".to_owned()].into_iter())
481                .await
482                .unwrap();
483
484            let event = rx.next().await.unwrap();
485
486            assert_eq!(event.as_log()["host"], addr.ip().to_string().into());
487            assert_eq!(event.as_log()["port"], addr.port().into());
488        })
489        .await;
490    }
491
492    #[tokio::test]
493    async fn tcp_it_includes_vector_namespaced_fields() {
494        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
495            let (tx, mut rx) = SourceSender::new_test();
496            let (guard, addr) = next_addr();
497            let mut conf = TcpConfig::from_address(addr.into());
498            conf.set_log_namespace(Some(true));
499
500            let server = SocketConfig::from(conf)
501                .build(SourceContext::new_test(tx, None))
502                .await
503                .unwrap();
504            tokio::spawn(server);
505
506            wait_for_tcp_and_release(guard, addr).await;
507
508            let addr = send_lines(addr, vec!["test".to_owned()].into_iter())
509                .await
510                .unwrap();
511
512            let event = rx.next().await.unwrap();
513            let log = event.as_log();
514            let event_meta = log.metadata().value();
515
516            assert_eq!(log.value(), &"test".into());
517            assert_eq!(
518                event_meta.get(path!("vector", "source_type")).unwrap(),
519                &value!(SocketConfig::NAME)
520            );
521            assert_eq!(
522                event_meta.get(path!(SocketConfig::NAME, "host")).unwrap(),
523                &value!(addr.ip().to_string())
524            );
525            assert_eq!(
526                event_meta.get(path!(SocketConfig::NAME, "port")).unwrap(),
527                &value!(addr.port())
528            );
529        })
530        .await;
531    }
532
533    #[tokio::test]
534    async fn tcp_splits_on_newline() {
535        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
536            let (tx, rx) = SourceSender::new_test();
537            let (guard, addr) = next_addr();
538
539            let server = SocketConfig::from(TcpConfig::from_address(addr.into()))
540                .build(SourceContext::new_test(tx, None))
541                .await
542                .unwrap();
543            tokio::spawn(server);
544
545            wait_for_tcp_and_release(guard, addr).await;
546
547            send_lines(addr, vec!["foo\nbar".to_owned()].into_iter())
548                .await
549                .unwrap();
550
551            let events = collect_n(rx, 2).await;
552
553            assert_eq!(events.len(), 2);
554            assert_eq!(
555                events[0].as_log()[log_schema().message_key().unwrap().to_string()],
556                "foo".into()
557            );
558            assert_eq!(
559                events[1].as_log()[log_schema().message_key().unwrap().to_string()],
560                "bar".into()
561            );
562        })
563        .await;
564    }
565
566    #[tokio::test]
567    async fn tcp_it_includes_source_type() {
568        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
569            let (tx, mut rx) = SourceSender::new_test();
570            let (guard, addr) = next_addr();
571
572            let server = SocketConfig::from(TcpConfig::from_address(addr.into()))
573                .build(SourceContext::new_test(tx, None))
574                .await
575                .unwrap();
576            tokio::spawn(server);
577
578            wait_for_tcp_and_release(guard, addr).await;
579            send_lines(addr, vec!["test".to_owned()].into_iter())
580                .await
581                .unwrap();
582
583            let event = rx.next().await.unwrap();
584            assert_eq!(
585                event.as_log()[log_schema().source_type_key().unwrap().to_string()],
586                "socket".into()
587            );
588        })
589        .await;
590    }
591
592    #[tokio::test]
593    async fn tcp_continue_after_long_line() {
594        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
595            let (tx, mut rx) = SourceSender::new_test();
596            let (guard, addr) = next_addr();
597
598            let mut config = TcpConfig::from_address(addr.into());
599            config.set_framing(Some(
600                NewlineDelimitedDecoderConfig::new_with_max_length(10).into(),
601            ));
602
603            let server = SocketConfig::from(config)
604                .build(SourceContext::new_test(tx, None))
605                .await
606                .unwrap();
607            tokio::spawn(server);
608
609            let lines = vec![
610                "short".to_owned(),
611                "this is too long".to_owned(),
612                "more short".to_owned(),
613            ];
614
615            wait_for_tcp_and_release(guard, addr).await;
616            send_lines(addr, lines.into_iter()).await.unwrap();
617
618            let event = rx.next().await.unwrap();
619            assert_eq!(
620                event.as_log()[log_schema().message_key().unwrap().to_string()],
621                "short".into()
622            );
623
624            let event = rx.next().await.unwrap();
625            assert_eq!(
626                event.as_log()[log_schema().message_key().unwrap().to_string()],
627                "more short".into()
628            );
629        })
630        .await;
631    }
632
633    #[tokio::test]
634    async fn tcp_with_tls() {
635        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
636            let (tx, mut rx) = SourceSender::new_test();
637            let (guard, addr) = next_addr();
638
639            let mut config = TcpConfig::from_address(addr.into());
640            config.set_tls(Some(TlsSourceConfig {
641                tls_config: TlsEnableableConfig {
642                    enabled: Some(true),
643                    options: TlsConfig {
644                        verify_certificate: Some(true),
645                        crt_file: Some(tls::TEST_PEM_CRT_PATH.into()),
646                        key_file: Some(tls::TEST_PEM_KEY_PATH.into()),
647                        ca_file: Some(tls::TEST_PEM_CA_PATH.into()),
648                        ..Default::default()
649                    },
650                },
651                client_metadata_key: Some(OptionalValuePath::from(owned_value_path!("tls_peer"))),
652            }));
653
654            let server = SocketConfig::from(config)
655                .build(SourceContext::new_test(tx, None))
656                .await
657                .unwrap();
658            tokio::spawn(server);
659
660            let lines = vec!["one line".to_owned(), "another line".to_owned()];
661
662            wait_for_tcp_and_release(guard, addr).await;
663            send_lines_tls(
664                addr,
665                "localhost".into(),
666                lines.into_iter(),
667                std::path::Path::new(tls::TEST_PEM_CA_PATH),
668                std::path::Path::new(tls::TEST_PEM_CLIENT_CRT_PATH),
669                std::path::Path::new(tls::TEST_PEM_CLIENT_KEY_PATH),
670            )
671            .await
672            .unwrap();
673
674            let event = rx.next().await.unwrap();
675            assert_eq!(
676                event.as_log()[log_schema().message_key().unwrap().to_string()],
677                "one line".into()
678            );
679
680            let tls_meta: ObjectMap = btreemap!(
681                "subject" => "CN=localhost,OU=Vector,O=Datadog,L=New York,ST=New York,C=US"
682            );
683
684            assert_eq!(event.as_log()["tls_peer"], tls_meta.clone().into(),);
685
686            let event = rx.next().await.unwrap();
687            assert_eq!(
688                event.as_log()[log_schema().message_key().unwrap().to_string()],
689                "another line".into()
690            );
691
692            assert_eq!(event.as_log()["tls_peer"], tls_meta.clone().into(),);
693        })
694        .await;
695    }
696
697    #[tokio::test]
698    async fn tcp_with_tls_vector_namespace() {
699        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
700            let (tx, mut rx) = SourceSender::new_test();
701            let (guard, addr) = next_addr();
702
703            let mut config = TcpConfig::from_address(addr.into());
704            config.set_tls(Some(TlsSourceConfig {
705                tls_config: TlsEnableableConfig {
706                    enabled: Some(true),
707                    options: TlsConfig {
708                        verify_certificate: Some(true),
709                        crt_file: Some(tls::TEST_PEM_CRT_PATH.into()),
710                        key_file: Some(tls::TEST_PEM_KEY_PATH.into()),
711                        ca_file: Some(tls::TEST_PEM_CA_PATH.into()),
712                        ..Default::default()
713                    },
714                },
715                client_metadata_key: None,
716            }));
717            config.log_namespace = Some(true);
718
719            let server = SocketConfig::from(config)
720                .build(SourceContext::new_test(tx, None))
721                .await
722                .unwrap();
723            tokio::spawn(server);
724
725            let lines = vec!["one line".to_owned(), "another line".to_owned()];
726
727            wait_for_tcp_and_release(guard, addr).await;
728            send_lines_tls(
729                addr,
730                "localhost".into(),
731                lines.into_iter(),
732                std::path::Path::new(tls::TEST_PEM_CA_PATH),
733                std::path::Path::new(tls::TEST_PEM_CLIENT_CRT_PATH),
734                std::path::Path::new(tls::TEST_PEM_CLIENT_KEY_PATH),
735            )
736            .await
737            .unwrap();
738
739            let event = rx.next().await.unwrap();
740            let log = event.as_log();
741            let event_meta = log.metadata().value();
742
743            assert_eq!(log.value(), &"one line".into());
744
745            let tls_meta: ObjectMap = btreemap!(
746                "subject" => "CN=localhost,OU=Vector,O=Datadog,L=New York,ST=New York,C=US"
747            );
748
749            assert_eq!(
750                event_meta
751                    .get(path!(SocketConfig::NAME, "tls_client_metadata"))
752                    .unwrap(),
753                &value!(tls_meta.clone())
754            );
755
756            let event = rx.next().await.unwrap();
757            let log = event.as_log();
758            let event_meta = log.metadata().value();
759
760            assert_eq!(log.value(), &"another line".into());
761
762            assert_eq!(
763                event_meta
764                    .get(path!(SocketConfig::NAME, "tls_client_metadata"))
765                    .unwrap(),
766                &value!(tls_meta.clone())
767            );
768        })
769        .await;
770    }
771
772    #[tokio::test]
773    async fn tcp_shutdown_simple() {
774        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
775            let source_id = ComponentKey::from("tcp_shutdown_simple");
776            let (tx, mut rx) = SourceSender::new_test();
777            let (guard, addr) = next_addr();
778            let (cx, mut shutdown) = SourceContext::new_shutdown(&source_id, tx);
779
780            // Start TCP Source
781            let server = SocketConfig::from(TcpConfig::from_address(addr.into()))
782                .build(cx)
783                .await
784                .unwrap();
785            let source_handle = tokio::spawn(server);
786
787            // Send data to Source.
788            wait_for_tcp_and_release(guard, addr).await;
789            send_lines(addr, vec!["test".to_owned()].into_iter())
790                .await
791                .unwrap();
792
793            let event = rx.next().await.unwrap();
794            assert_eq!(
795                event.as_log()[log_schema().message_key().unwrap().to_string()],
796                "test".into()
797            );
798
799            // Now signal to the Source to shut down.
800            let deadline = Instant::now() + Duration::from_secs(10);
801            let shutdown_complete = shutdown.shutdown_source(&source_id, deadline);
802            let shutdown_success = shutdown_complete.await;
803            assert!(shutdown_success);
804
805            // Ensure source actually shut down successfully.
806            _ = source_handle.await.unwrap();
807        })
808        .await;
809    }
810
811    // Intentionally not using assert_source_compliance here because this is a round-trip test which
812    // means source and sink will both emit `EventsSent` , triggering multi-emission check.
813    #[tokio::test]
814    async fn tcp_shutdown_infinite_stream() {
815        // We create our TCP source with a larger-than-normal send buffer, which helps ensure that
816        // the source doesn't block on sending the events downstream, otherwise if it was blocked on
817        // doing so, it wouldn't be able to wake up and loop to see that it had been signalled to
818        // shutdown.
819        let (guard, addr) = next_addr();
820
821        let (source_tx, source_rx) = SourceSender::new_test_sender_with_options(10_000, None);
822        let source_key = ComponentKey::from("tcp_shutdown_infinite_stream");
823        let (source_cx, mut shutdown) = SourceContext::new_shutdown(&source_key, source_tx);
824
825        let mut source_config = TcpConfig::from_address(addr.into());
826        source_config.set_shutdown_timeout_secs(1);
827        let source_task = SocketConfig::from(source_config)
828            .build(source_cx)
829            .await
830            .unwrap();
831
832        // Spawn the source task and wait until we're sure it's listening:
833        let source_handle = tokio::spawn(source_task);
834        wait_for_tcp_and_release(guard, addr).await;
835
836        // Now we create a TCP _sink_ which we'll feed with an infinite stream of events to ship to
837        // our TCP source.  This will ensure that our TCP source is fully-loaded as we try to shut
838        // it down, exercising the logic we have to ensure timely shutdown even under load:
839        let message = random_string(512);
840        let message_bytes = Bytes::from(message.clone());
841
842        #[derive(Clone, Debug)]
843        struct Serializer {
844            bytes: Bytes,
845        }
846        impl tokio_util::codec::Encoder<Event> for Serializer {
847            type Error = vector_lib::codecs::encoding::Error;
848
849            fn encode(&mut self, _: Event, buffer: &mut BytesMut) -> Result<(), Self::Error> {
850                buffer.put(self.bytes.as_ref());
851                buffer.put_u8(b'\n');
852                Ok(())
853            }
854        }
855        let sink_config = TcpSinkConfig::from_address(format!("localhost:{}", addr.port()));
856        let encoder = Serializer {
857            bytes: message_bytes,
858        };
859        let (sink, _healthcheck) = sink_config.build(Default::default(), encoder).unwrap();
860
861        tokio::spawn(async move {
862            let input = stream::repeat_with(|| LogEvent::default().into()).boxed();
863            sink.run(input).await.unwrap();
864        });
865
866        // Now with our sink running, feeding events to the source, collect 100 event arrays from
867        // the source and make sure each event within them matches the single message we repeatedly
868        // sent via the sink:
869        let events = collect_n_limited(source_rx, 100)
870            .await
871            .into_iter()
872            .collect::<Vec<_>>();
873        assert_eq!(100, events.len());
874
875        let message_key = log_schema().message_key().unwrap().to_string();
876        let expected_message = message.clone().into();
877        for event in events.into_iter().flat_map(EventContainer::into_events) {
878            assert_eq!(event.as_log()[message_key.as_str()], expected_message);
879        }
880
881        // Now trigger shutdown on the source and ensure that it shuts down before or at the
882        // deadline, and make sure the source task actually finished as well:
883        let shutdown_timeout_limit = Duration::from_secs(10);
884        let deadline = Instant::now() + shutdown_timeout_limit;
885        let shutdown_complete = shutdown.shutdown_source(&source_key, deadline);
886
887        let shutdown_result = timeout(shutdown_timeout_limit, shutdown_complete).await;
888        assert_eq!(shutdown_result, Ok(true));
889
890        let source_result = source_handle.await.expect("source task should not panic");
891        assert_eq!(source_result, Ok(()));
892    }
893
894    #[tokio::test]
895    async fn tcp_connection_close_after_max_duration() {
896        let (tx, _) = SourceSender::new_test();
897        let (guard, addr) = next_addr();
898
899        let mut source_config = TcpConfig::from_address(addr.into());
900        source_config.set_max_connection_duration_secs(Some(1));
901        let source_task = SocketConfig::from(source_config)
902            .build(SourceContext::new_test(tx, None))
903            .await
904            .unwrap();
905
906        // Spawn the source task and wait until we're sure it's listening:
907        drop(tokio::spawn(source_task));
908        wait_for_tcp_and_release(guard, addr).await;
909
910        let mut stream: TcpStream = TcpStream::connect(addr)
911            .await
912            .expect("stream should be able to connect");
913        let start = Instant::now();
914
915        let timeout = tokio::time::sleep(Duration::from_millis(1200));
916        let mut buffer = [0u8; 10];
917
918        tokio::select! {
919             _ = timeout => {
920                 panic!("timed out waiting for stream to close")
921             },
922             read_result = stream.read(&mut buffer) => {
923                 match read_result {
924                    // read resulting with 0 bytes -> the connection was closed
925                    Ok(0) => assert_relative_eq!(start.elapsed().as_secs_f64(), 1.0, epsilon = 0.3),
926                    Ok(_) => panic!("unexpectedly read data from stream"),
927                    Err(e) => panic!("{e:}")
928                 }
929             }
930        }
931    }
932
933    //////// UDP TESTS ////////
934    async fn send_lines_udp(to: SocketAddr, lines: impl IntoIterator<Item = String>) -> UdpSocket {
935        send_lines_udp_from(bind_unused_udp(), to, lines)
936    }
937
938    fn send_lines_udp_from(
939        from: UdpSocket,
940        to: SocketAddr,
941        lines: impl IntoIterator<Item = String>,
942    ) -> UdpSocket {
943        send_packets_udp_from(from, to, lines.into_iter().map(|line| line.into()))
944    }
945
946    async fn send_packets_udp(
947        to: SocketAddr,
948        packets: impl IntoIterator<Item = Bytes>,
949    ) -> UdpSocket {
950        send_packets_udp_from(bind_unused_udp(), to, packets)
951    }
952
953    fn send_packets_udp_from(
954        from: UdpSocket,
955        to: SocketAddr,
956        packets: impl IntoIterator<Item = Bytes>,
957    ) -> UdpSocket {
958        for packet in packets {
959            assert_eq!(
960                from.send_to(&packet, to)
961                    .map_err(|error| panic!("{error:}"))
962                    .ok()
963                    .unwrap(),
964                packet.len()
965            );
966            // Space things out slightly to try to avoid dropped packets
967            thread::sleep(Duration::from_millis(1));
968        }
969
970        // Give packets some time to flow through
971        thread::sleep(Duration::from_millis(10));
972
973        // Done
974        from
975    }
976
977    async fn init_udp_with_shutdown(
978        sender: SourceSender,
979        source_id: &ComponentKey,
980        shutdown: &mut SourceShutdownCoordinator,
981    ) -> (SocketAddr, JoinHandle<Result<(), ()>>) {
982        let (shutdown_signal, _) = shutdown.register_source(source_id, false);
983        init_udp_inner(sender, source_id, shutdown_signal, None, false).await
984    }
985
986    async fn init_udp(sender: SourceSender, use_log_namespace: bool) -> SocketAddr {
987        init_udp_inner(
988            sender,
989            &ComponentKey::from("default"),
990            ShutdownSignal::noop(),
991            None,
992            use_log_namespace,
993        )
994        .await
995        .0
996    }
997
998    async fn init_udp_with_config(sender: SourceSender, config: UdpConfig) -> SocketAddr {
999        init_udp_inner(
1000            sender,
1001            &ComponentKey::from("default"),
1002            ShutdownSignal::noop(),
1003            Some(config),
1004            false,
1005        )
1006        .await
1007        .0
1008    }
1009
1010    async fn init_udp_inner(
1011        sender: SourceSender,
1012        source_key: &ComponentKey,
1013        shutdown_signal: ShutdownSignal,
1014        config: Option<UdpConfig>,
1015        use_vector_namespace: bool,
1016    ) -> (SocketAddr, JoinHandle<Result<(), ()>>) {
1017        let (guard, address, mut config) = match config {
1018            Some(config) => match config.address() {
1019                SocketListenAddr::SocketAddr(addr) => (None, addr, config),
1020                _ => panic!("listen address should not be systemd FD offset in tests"),
1021            },
1022            None => {
1023                let (guard, address) = next_addr();
1024                (
1025                    Some(guard),
1026                    address,
1027                    UdpConfig::from_address(address.into()),
1028                )
1029            }
1030        };
1031
1032        let config = if use_vector_namespace {
1033            config.set_log_namespace(Some(true));
1034            config
1035        } else {
1036            config
1037        };
1038
1039        let server = SocketConfig::from(config)
1040            .build(SourceContext {
1041                key: source_key.clone(),
1042                globals: GlobalOptions::default(),
1043                enrichment_tables: Default::default(),
1044                shutdown: shutdown_signal,
1045                out: sender,
1046                proxy: Default::default(),
1047                acknowledgements: false,
1048                schema: Default::default(),
1049                schema_definitions: HashMap::default(),
1050                extra_context: Default::default(),
1051                metrics_storage: Default::default(),
1052            })
1053            .await
1054            .unwrap();
1055        let source_handle = tokio::spawn(server);
1056
1057        // Wait for UDP to start listening
1058        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
1059
1060        if let Some(guard) = guard {
1061            drop(guard)
1062        }
1063
1064        (address, source_handle)
1065    }
1066
1067    #[tokio::test]
1068    async fn udp_message() {
1069        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1070            let (tx, rx) = SourceSender::new_test();
1071            let address = init_udp(tx, false).await;
1072
1073            send_lines_udp(address, vec!["test".to_string()]).await;
1074            let events = collect_n(rx, 1).await;
1075
1076            assert_eq!(
1077                events[0].as_log()[log_schema().message_key().unwrap().to_string()],
1078                "test".into()
1079            );
1080        })
1081        .await;
1082    }
1083
1084    #[tokio::test]
1085    async fn udp_message_preserves_newline() {
1086        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1087            let (tx, rx) = SourceSender::new_test();
1088            let address = init_udp(tx, false).await;
1089
1090            send_lines_udp(address, vec!["foo\nbar".to_string()]).await;
1091            let events = collect_n(rx, 1).await;
1092
1093            assert_eq!(
1094                events[0].as_log()[log_schema().message_key().unwrap().to_string()],
1095                "foo\nbar".into()
1096            );
1097        })
1098        .await;
1099    }
1100
1101    #[tokio::test]
1102    async fn udp_multiple_packets() {
1103        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1104            let (tx, rx) = SourceSender::new_test();
1105            let address = init_udp(tx, false).await;
1106
1107            send_lines_udp(address, vec!["test".to_string(), "test2".to_string()]).await;
1108            let events = collect_n(rx, 2).await;
1109
1110            assert_eq!(
1111                events[0].as_log()[log_schema().message_key().unwrap().to_string()],
1112                "test".into()
1113            );
1114            assert_eq!(
1115                events[1].as_log()[log_schema().message_key().unwrap().to_string()],
1116                "test2".into()
1117            );
1118        })
1119        .await;
1120    }
1121
1122    #[tokio::test]
1123    async fn udp_max_length() {
1124        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1125            let (tx, rx) = SourceSender::new_test();
1126            let (_, address) = next_addr();
1127            let mut config = UdpConfig::from_address(address.into());
1128            config.max_length = 11;
1129            let address = init_udp_with_config(tx, config).await;
1130
1131            send_lines_udp(
1132                address,
1133                vec![
1134                    "short line".to_string(),
1135                    "test with a long line".to_string(),
1136                    "a short un".to_string(),
1137                ],
1138            )
1139            .await;
1140
1141            let events = collect_n(rx, 2).await;
1142            assert_eq!(
1143                events[0].as_log()[log_schema().message_key().unwrap().to_string()],
1144                "short line".into()
1145            );
1146            assert_eq!(
1147                events[1].as_log()[log_schema().message_key().unwrap().to_string()],
1148                "a short un".into()
1149            );
1150        })
1151        .await;
1152    }
1153
1154    #[cfg(unix)]
1155    #[tokio::test]
1156    /// This test only works on Unix.
1157    /// Unix truncates at max_length giving us the bytes to get the first n delimited messages.
1158    /// Windows will drop the entire packet if we exceed the max_length so we are unable to
1159    /// extract anything.
1160    async fn udp_max_length_delimited() {
1161        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1162            let (tx, rx) = SourceSender::new_test();
1163            let (_, address) = next_addr();
1164            let mut config = UdpConfig::from_address(address.into());
1165            config.max_length = 10;
1166            config.framing = Some(
1167                CharacterDelimitedDecoderConfig {
1168                    character_delimited: CharacterDelimitedDecoderOptions::new(b',', None),
1169                }
1170                .into(),
1171            );
1172            let address = init_udp_with_config(tx, config).await;
1173
1174            send_lines_udp(
1175                address,
1176                vec!["test with, long line".to_string(), "short one".to_string()],
1177            )
1178            .await;
1179
1180            let events = collect_n(rx, 2).await;
1181            assert_eq!(
1182                events[0].as_log()[log_schema().message_key().unwrap().to_string()],
1183                "test with".into()
1184            );
1185            assert_eq!(
1186                events[1].as_log()[log_schema().message_key().unwrap().to_string()],
1187                "short one".into()
1188            );
1189        })
1190        .await;
1191    }
1192
1193    #[tokio::test]
1194    async fn udp_decodes_chunked_gelf_messages() {
1195        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1196            let (tx, rx) = SourceSender::new_test();
1197            let (_, address) = next_addr();
1198            let mut config = UdpConfig::from_address(address.into());
1199            config.decoding = GelfDeserializerConfig::default().into();
1200            let address = init_udp_with_config(tx, config).await;
1201            let seed = 42;
1202            let mut rng = SmallRng::seed_from_u64(seed);
1203            let max_size = 300;
1204            let big_message = "This is a very large message".repeat(500);
1205            let another_big_message = "This is another very large message".repeat(500);
1206            let mut chunks = get_gelf_chunks(big_message.as_str(), max_size, &mut rng);
1207            let mut another_chunks =
1208                get_gelf_chunks(another_big_message.as_str(), max_size, &mut rng);
1209            chunks.append(&mut another_chunks);
1210            chunks.shuffle(&mut rng);
1211
1212            send_packets_udp(address, chunks).await;
1213
1214            let events = collect_n(rx, 2).await;
1215            assert_eq!(
1216                events[1].as_log()[log_schema().message_key().unwrap().to_string()],
1217                big_message.into()
1218            );
1219            assert_eq!(
1220                events[0].as_log()[log_schema().message_key().unwrap().to_string()],
1221                another_big_message.into()
1222            );
1223        })
1224        .await;
1225    }
1226
1227    #[tokio::test]
1228    async fn udp_it_includes_host() {
1229        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1230            let (tx, rx) = SourceSender::new_test();
1231            let address = init_udp(tx, false).await;
1232
1233            let from = send_lines_udp(address, vec!["test".to_string()]).await;
1234            let events = collect_n(rx, 1).await;
1235
1236            assert_eq!(
1237                events[0].as_log()["host"],
1238                from.local_addr().unwrap().ip().to_string().into()
1239            );
1240            assert_eq!(
1241                events[0].as_log()["port"],
1242                from.local_addr().unwrap().port().into()
1243            );
1244        })
1245        .await;
1246    }
1247
1248    #[tokio::test]
1249    async fn udp_it_includes_vector_namespaced_fields() {
1250        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1251            let (tx, rx) = SourceSender::new_test();
1252            let address = init_udp(tx, true).await;
1253
1254            let from = send_lines_udp(address, vec!["test".to_string()]).await;
1255            let events = collect_n(rx, 1).await;
1256            let log = events[0].as_log();
1257            let event_meta = log.metadata().value();
1258
1259            assert_eq!(log.value(), &"test".into());
1260            assert_eq!(
1261                event_meta.get(path!("vector", "source_type")).unwrap(),
1262                &value!(SocketConfig::NAME)
1263            );
1264            assert_eq!(
1265                event_meta.get(path!(SocketConfig::NAME, "host")).unwrap(),
1266                &value!(from.local_addr().unwrap().ip().to_string())
1267            );
1268            assert_eq!(
1269                event_meta.get(path!(SocketConfig::NAME, "port")).unwrap(),
1270                &value!(from.local_addr().unwrap().port())
1271            );
1272        })
1273        .await;
1274    }
1275
1276    #[tokio::test]
1277    async fn udp_it_includes_source_type() {
1278        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1279            let (tx, rx) = SourceSender::new_test();
1280            let address = init_udp(tx, false).await;
1281
1282            _ = send_lines_udp(address, vec!["test".to_string()]).await;
1283            let events = collect_n(rx, 1).await;
1284
1285            assert_eq!(
1286                events[0].as_log()[log_schema().source_type_key().unwrap().to_string()],
1287                "socket".into()
1288            );
1289        })
1290        .await;
1291    }
1292
1293    #[tokio::test]
1294    async fn udp_shutdown_simple() {
1295        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1296            let (tx, rx) = SourceSender::new_test();
1297            let source_id = ComponentKey::from("udp_shutdown_simple");
1298
1299            let mut shutdown = SourceShutdownCoordinator::default();
1300            let (address, source_handle) =
1301                init_udp_with_shutdown(tx, &source_id, &mut shutdown).await;
1302
1303            send_lines_udp(address, vec!["test".to_string()]).await;
1304            let events = collect_n(rx, 1).await;
1305
1306            assert_eq!(
1307                events[0].as_log()[log_schema().message_key().unwrap().to_string()],
1308                "test".into()
1309            );
1310
1311            // Now signal to the Source to shut down.
1312            let deadline = Instant::now() + Duration::from_secs(10);
1313            let shutdown_complete = shutdown.shutdown_source(&source_id, deadline);
1314            let shutdown_success = shutdown_complete.await;
1315            assert!(shutdown_success);
1316
1317            // Ensure source actually shut down successfully.
1318            _ = source_handle.await.unwrap();
1319        })
1320        .await;
1321    }
1322
1323    #[tokio::test]
1324    async fn udp_shutdown_infinite_stream() {
1325        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1326            let (tx, rx) = SourceSender::new_test();
1327            let source_id = ComponentKey::from("udp_shutdown_infinite_stream");
1328
1329            let mut shutdown = SourceShutdownCoordinator::default();
1330            let (address, source_handle) =
1331                init_udp_with_shutdown(tx, &source_id, &mut shutdown).await;
1332
1333            // Stream that keeps sending lines to the UDP source forever.
1334            let run_pump_atomic_sender = Arc::new(AtomicBool::new(true));
1335            let run_pump_atomic_receiver = Arc::clone(&run_pump_atomic_sender);
1336            let pump_handle = tokio::task::spawn_blocking(move || {
1337                let handle = tokio::runtime::Handle::current();
1338                handle.block_on(send_lines_udp(
1339                    address,
1340                    std::iter::repeat("test".to_string())
1341                        .take_while(move |_| run_pump_atomic_receiver.load(Ordering::Relaxed)),
1342                ));
1343            });
1344
1345            // Important that 'rx' doesn't get dropped until the pump has finished sending items to it.
1346            let events = collect_n(rx, 100).await;
1347            assert_eq!(100, events.len());
1348            for event in events {
1349                assert_eq!(
1350                    event.as_log()[log_schema().message_key().unwrap().to_string()],
1351                    "test".into()
1352                );
1353            }
1354
1355            let deadline = Instant::now() + Duration::from_secs(10);
1356            let shutdown_complete = shutdown.shutdown_source(&source_id, deadline);
1357            let shutdown_success = shutdown_complete.await;
1358            assert!(shutdown_success);
1359
1360            // Ensure that the source has actually shut down.
1361            _ = source_handle.await.unwrap();
1362
1363            // Stop the pump from sending lines forever.
1364            run_pump_atomic_sender.store(false, Ordering::Relaxed);
1365            assert!(pump_handle.await.is_ok());
1366        })
1367        .await;
1368    }
1369
1370    #[tokio::test]
1371    async fn multicast_udp_message() {
1372        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1373            let (tx, mut rx) = SourceSender::new_test();
1374            // The socket address must be `IPADDR_ANY` (0.0.0.0) in order to receive multicast packets
1375            let (_guard, socket_address) = next_addr_any();
1376            let multicast_ip_address: Ipv4Addr = "224.0.0.2".parse().unwrap();
1377            let multicast_socket_address =
1378                SocketAddr::new(IpAddr::V4(multicast_ip_address), socket_address.port());
1379            let mut config = UdpConfig::from_address(socket_address.into());
1380            config.multicast_groups = vec![multicast_ip_address];
1381            // Use loopback as the multicast interface so local test packets are delivered
1382            // on all platforms. Without this, macOS joins on the default network interface
1383            // (e.g. en0) instead of loopback, and the sender's packets never arrive.
1384            config.multicast_interface = Some(Ipv4Addr::LOCALHOST);
1385            init_udp_with_config(tx, config).await;
1386
1387            // Bind sender with loopback as the outgoing multicast interface so packets
1388            // route through lo, matching the interface the receiver joined on.
1389            send_lines_udp_from(
1390                bind_unused_udp_multicast(),
1391                multicast_socket_address,
1392                ["test".to_string()],
1393            );
1394
1395            let event = rx.next().await.expect("must receive an event");
1396            assert_eq!(
1397                event.as_log()[log_schema().message_key().unwrap().to_string()],
1398                "test".into()
1399            );
1400        })
1401        .await;
1402    }
1403
1404    #[tokio::test]
1405    async fn multiple_multicast_addresses_udp_message() {
1406        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1407            let (tx, mut rx) = SourceSender::new_test();
1408            let (_guard, socket_address) = next_addr_any();
1409            let multicast_ip_addresses = (2..12)
1410                .map(|i| format!("224.0.0.{i}").parse().unwrap())
1411                .collect::<Vec<Ipv4Addr>>();
1412            let multicast_ip_socket_addresses = multicast_ip_addresses
1413                .iter()
1414                .map(|ip_address| SocketAddr::new(IpAddr::V4(*ip_address), socket_address.port()))
1415                .collect::<Vec<SocketAddr>>();
1416            let mut config = UdpConfig::from_address(socket_address.into());
1417            config.multicast_groups = multicast_ip_addresses;
1418            config.multicast_interface = Some(Ipv4Addr::LOCALHOST);
1419            init_udp_with_config(tx, config).await;
1420
1421            let mut from = bind_unused_udp_multicast();
1422            for multicast_ip_socket_address in multicast_ip_socket_addresses {
1423                from = send_lines_udp_from(
1424                    from,
1425                    multicast_ip_socket_address,
1426                    [multicast_ip_socket_address.to_string()],
1427                );
1428
1429                let event = rx.next().await.expect("must receive an event");
1430                assert_eq!(
1431                    event.as_log()[log_schema().message_key().unwrap().to_string()],
1432                    multicast_ip_socket_address.to_string().into()
1433                );
1434            }
1435        })
1436        .await;
1437    }
1438
1439    #[tokio::test]
1440    async fn multicast_and_unicast_udp_message() {
1441        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1442            let (tx, mut rx) = SourceSender::new_test();
1443            let (_guard, socket_address) = next_addr_any();
1444            let multicast_ip_address: Ipv4Addr = "224.0.0.2".parse().unwrap();
1445            let multicast_socket_address =
1446                SocketAddr::new(IpAddr::V4(multicast_ip_address), socket_address.port());
1447            let mut config = UdpConfig::from_address(socket_address.into());
1448            config.multicast_groups = vec![multicast_ip_address];
1449            config.multicast_interface = Some(Ipv4Addr::LOCALHOST);
1450            init_udp_with_config(tx, config).await;
1451
1452            // Send packet to multicast address using loopback as the outgoing interface
1453            // so it routes through lo, matching the interface the receiver joined on.
1454            let _ = send_lines_udp_from(
1455                bind_unused_udp_multicast(),
1456                multicast_socket_address,
1457                ["test".to_string()],
1458            );
1459            let event = rx.next().await.expect("must receive an event");
1460            assert_eq!(
1461                event.as_log()[log_schema().message_key().unwrap().to_string()],
1462                "test".into()
1463            );
1464
1465            // Windows does not support connecting to `0.0.0.0`,
1466            // therefore we connect to `127.0.0.1` instead (the socket is listening at `0.0.0.0`)
1467            let to = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), socket_address.port());
1468            // Send packet to unicast address
1469            // Use a fresh socket - on macOS, a socket bound to 0.0.0.0 that sends to multicast
1470            // cannot subsequently send unicast packets that the listener receives
1471            send_lines_udp_from(bind_unused_udp(), to, ["test".to_string()]);
1472            let event = rx.next().await.expect("must receive an event");
1473            assert_eq!(
1474                event.as_log()[log_schema().message_key().unwrap().to_string()],
1475                "test".into()
1476            );
1477        })
1478        .await;
1479    }
1480
1481    #[tokio::test]
1482    async fn udp_invalid_multicast_group() {
1483        assert_source_error(&COMPONENT_ERROR_TAGS, async {
1484            let (tx, _rx) = SourceSender::new_test();
1485            let (_, socket_address) = next_addr_any();
1486            let invalid_multicast_ip_address: Ipv4Addr = "192.168.0.3".parse().unwrap();
1487            let mut config = UdpConfig::from_address(socket_address.into());
1488            config.multicast_groups = vec![invalid_multicast_ip_address];
1489            init_udp_with_config(tx, config).await;
1490        })
1491        .await;
1492    }
1493
1494    ////////////// UNIX TEST LIBS //////////////
1495
1496    #[cfg(unix)]
1497    async fn init_unix(sender: SourceSender, stream: bool, use_vector_namespace: bool) -> PathBuf {
1498        init_unix_inner(sender, stream, use_vector_namespace, None).await
1499    }
1500
1501    #[cfg(unix)]
1502    async fn init_unix_with_config(
1503        sender: SourceSender,
1504        stream: bool,
1505        use_vector_namespace: bool,
1506        config: UnixConfig,
1507    ) -> PathBuf {
1508        init_unix_inner(sender, stream, use_vector_namespace, Some(config)).await
1509    }
1510
1511    #[cfg(unix)]
1512    async fn init_unix_inner(
1513        sender: SourceSender,
1514        stream: bool,
1515        use_vector_namespace: bool,
1516        config: Option<UnixConfig>,
1517    ) -> PathBuf {
1518        let mut config = config.unwrap_or_else(|| {
1519            UnixConfig::new(tempfile::tempdir().unwrap().keep().join("unix_test"))
1520        });
1521
1522        let in_path = config.path.clone();
1523
1524        if use_vector_namespace {
1525            config.log_namespace = Some(true);
1526        }
1527
1528        let mode = if stream {
1529            Mode::UnixStream(config)
1530        } else {
1531            Mode::UnixDatagram(config)
1532        };
1533
1534        let server = SocketConfig { mode }
1535            .build(SourceContext::new_test(sender, None))
1536            .await
1537            .unwrap();
1538        tokio::spawn(server);
1539
1540        // Wait for server to accept traffic
1541        while if stream {
1542            std::os::unix::net::UnixStream::connect(&in_path).is_err()
1543        } else {
1544            let socket = std::os::unix::net::UnixDatagram::unbound().unwrap();
1545            socket.connect(&in_path).is_err()
1546        } {
1547            yield_now().await;
1548        }
1549
1550        in_path
1551    }
1552
1553    #[cfg(unix)]
1554    async fn unix_send_lines(stream: bool, path: PathBuf, lines: &[&str]) {
1555        match stream {
1556            false => send_lines_unix_datagram(path, lines).await,
1557            true => send_lines_unix_stream(path, lines).await,
1558        }
1559    }
1560
1561    #[cfg(unix)]
1562    async fn unix_message(
1563        message: &str,
1564        stream: bool,
1565        use_vector_namespace: bool,
1566    ) -> (PathBuf, impl Stream<Item = Event> + use<>) {
1567        let (tx, rx) = SourceSender::new_test();
1568        let path = init_unix(tx, stream, use_vector_namespace).await;
1569        let path_clone = path.clone();
1570
1571        unix_send_lines(stream, path, &[message]).await;
1572
1573        (path_clone, rx)
1574    }
1575
1576    #[cfg(unix)]
1577    async fn unix_multiple_packets(stream: bool) {
1578        let (tx, rx) = SourceSender::new_test();
1579        let path = init_unix(tx, stream, false).await;
1580
1581        unix_send_lines(stream, path, &["test", "test2"]).await;
1582        let events = collect_n(rx, 2).await;
1583
1584        assert_eq!(2, events.len());
1585        assert_eq!(
1586            events[0].as_log()[log_schema().message_key().unwrap().to_string()],
1587            "test".into()
1588        );
1589        assert_eq!(
1590            events[1].as_log()[log_schema().message_key().unwrap().to_string()],
1591            "test2".into()
1592        );
1593    }
1594
1595    #[cfg(unix)]
1596    fn parses_unix_config(mode: &str) -> SocketConfig {
1597        serde_yaml::from_str::<SocketConfig>(&format!(
1598            "mode: \"{mode}\"\npath: \"/does/not/exist\""
1599        ))
1600        .unwrap()
1601    }
1602
1603    #[cfg(unix)]
1604    fn parses_unix_config_file_mode(mode: &str) -> SocketConfig {
1605        serde_yaml::from_str::<SocketConfig>(&format!(
1606            "mode: \"{mode}\"\npath: \"/does/not/exist\"\nsocket_file_mode: 511"
1607        ))
1608        .unwrap()
1609    }
1610
1611    ////////////// UNIX DATAGRAM TESTS //////////////
1612    #[cfg(unix)]
1613    async fn send_lines_unix_datagram(path: PathBuf, lines: &[&str]) {
1614        let packets = lines.iter().map(|line| Bytes::from(line.to_string()));
1615        send_packets_unix_datagram(path, packets).await;
1616    }
1617
1618    #[cfg(unix)]
1619    async fn send_packets_unix_datagram(path: PathBuf, packets: impl IntoIterator<Item = Bytes>) {
1620        let socket = UnixDatagram::unbound().unwrap();
1621        socket.connect(path).unwrap();
1622
1623        for packet in packets {
1624            socket.send(&packet).await.unwrap();
1625        }
1626        socket.shutdown(std::net::Shutdown::Both).unwrap();
1627    }
1628
1629    #[cfg(unix)]
1630    #[tokio::test]
1631    async fn unix_datagram_message() {
1632        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1633            let (_, rx) = unix_message("test", false, false).await;
1634            let events = collect_n(rx, 1).await;
1635
1636            assert_eq!(events.len(), 1);
1637            assert_eq!(
1638                events[0].as_log()[log_schema().message_key().unwrap().to_string()],
1639                "test".into()
1640            );
1641            assert_eq!(
1642                events[0].as_log()[log_schema().source_type_key().unwrap().to_string()],
1643                "socket".into()
1644            );
1645            assert_eq!(events[0].as_log()["host"], UNNAMED_SOCKET_HOST.into());
1646        })
1647        .await;
1648    }
1649
1650    #[ignore]
1651    #[cfg(unix)]
1652    #[tokio::test]
1653    async fn unix_datagram_socket_test() {
1654        // This test is useful for testing the behavior of datagram
1655        // sockets.
1656
1657        use tempfile::tempdir;
1658        use tokio::net::UnixDatagram;
1659
1660        let tmp = tempdir().unwrap();
1661
1662        let tx_path = tmp.path().join("tx");
1663
1664        // Switch this var between "bound" and "unbound" to test
1665        // different types of socket behavior.
1666        let tx_type = "bound";
1667
1668        let tx = if tx_type == "bound" {
1669            UnixDatagram::bind(&tx_path).unwrap()
1670        } else {
1671            UnixDatagram::unbound().unwrap()
1672        };
1673
1674        // The following debug statements showcase some useful info:
1675        // dbg!(tx.local_addr().unwrap());
1676        // dbg!(std::os::unix::prelude::AsRawFd::as_raw_fd(&tx));
1677
1678        // Create another, bound socket
1679        let rx_path = tmp.path().join("rx");
1680        let rx = UnixDatagram::bind(&rx_path).unwrap();
1681
1682        // Connect to the bound socket
1683        tx.connect(&rx_path).unwrap();
1684
1685        // Send to the bound socket
1686        let bytes = b"hello world";
1687        tx.send(bytes).await.unwrap();
1688
1689        let mut buf = vec![0u8; 24];
1690        let (size, _) = rx.recv_from(&mut buf).await.unwrap();
1691
1692        let dgram = &buf[..size];
1693        assert_eq!(dgram, bytes);
1694    }
1695
1696    #[cfg(unix)]
1697    #[tokio::test]
1698    async fn unix_datagram_chunked_gelf_messages() {
1699        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1700            let (tx, rx) = SourceSender::new_test();
1701            let in_path = tempfile::tempdir().unwrap().keep().join("unix_test");
1702            let mut config = UnixConfig::new(in_path.clone());
1703            config.decoding = GelfDeserializerConfig::default().into();
1704            let path = init_unix_with_config(tx, false, false, config).await;
1705            let seed = 42;
1706            let mut rng = SmallRng::seed_from_u64(seed);
1707            let max_size = 20;
1708            let big_message = "This is a very large message".repeat(5);
1709            let another_big_message = "This is another very large message".repeat(5);
1710            let mut chunks = get_gelf_chunks(big_message.as_str(), max_size, &mut rng);
1711            let mut another_chunks =
1712                get_gelf_chunks(another_big_message.as_str(), max_size, &mut rng);
1713            chunks.append(&mut another_chunks);
1714            chunks.shuffle(&mut rng);
1715
1716            send_packets_unix_datagram(path, chunks).await;
1717
1718            let events = collect_n(rx, 2).await;
1719            assert_eq!(
1720                events[0].as_log()[log_schema().message_key().unwrap().to_string()],
1721                big_message.into()
1722            );
1723            assert_eq!(
1724                events[1].as_log()[log_schema().message_key().unwrap().to_string()],
1725                another_big_message.into()
1726            );
1727        })
1728        .await;
1729    }
1730
1731    #[cfg(unix)]
1732    #[tokio::test]
1733    async fn unix_datagram_message_with_vector_namespace() {
1734        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1735            let (_, rx) = unix_message("test", false, true).await;
1736            let events = collect_n(rx, 1).await;
1737            let log = events[0].as_log();
1738            let event_meta = log.metadata().value();
1739
1740            assert_eq!(log.value(), &"test".into());
1741            assert_eq!(events.len(), 1);
1742
1743            assert_eq!(
1744                event_meta.get(path!("vector", "source_type")).unwrap(),
1745                &value!(SocketConfig::NAME)
1746            );
1747
1748            assert_eq!(
1749                event_meta.get(path!(SocketConfig::NAME, "host")).unwrap(),
1750                &value!(UNNAMED_SOCKET_HOST)
1751            );
1752        })
1753        .await;
1754    }
1755
1756    #[cfg(unix)]
1757    #[tokio::test]
1758    async fn unix_datagram_message_preserves_newline() {
1759        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1760            let (_, rx) = unix_message("foo\nbar", false, false).await;
1761            let events = collect_n(rx, 1).await;
1762
1763            assert_eq!(events.len(), 1);
1764            assert_eq!(
1765                events[0].as_log()[log_schema().message_key().unwrap().to_string()],
1766                "foo\nbar".into()
1767            );
1768            assert_eq!(
1769                events[0].as_log()[log_schema().source_type_key().unwrap().to_string()],
1770                "socket".into()
1771            );
1772        })
1773        .await;
1774    }
1775
1776    #[cfg(unix)]
1777    #[tokio::test]
1778    async fn unix_datagram_multiple_packets() {
1779        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1780            unix_multiple_packets(false).await
1781        })
1782        .await;
1783    }
1784
1785    #[cfg(unix)]
1786    #[test]
1787    fn parses_unix_datagram_config() {
1788        let config = parses_unix_config("unix_datagram");
1789        assert!(matches!(config.mode, Mode::UnixDatagram { .. }));
1790    }
1791
1792    #[cfg(unix)]
1793    #[test]
1794    fn parses_unix_datagram_perms() {
1795        let config = parses_unix_config_file_mode("unix_datagram");
1796        assert!(matches!(config.mode, Mode::UnixDatagram { .. }));
1797    }
1798
1799    #[cfg(unix)]
1800    #[tokio::test]
1801    async fn unix_datagram_permissions() {
1802        let in_path = tempfile::tempdir().unwrap().keep().join("unix_test");
1803        let (tx, _) = SourceSender::new_test();
1804
1805        let mut config = UnixConfig::new(in_path.clone());
1806        config.socket_file_mode = Some(0o555);
1807        let mode = Mode::UnixDatagram(config);
1808        let server = SocketConfig { mode }
1809            .build(SourceContext::new_test(tx, None))
1810            .await
1811            .unwrap();
1812        tokio::spawn(server);
1813
1814        wait_for(|| {
1815            match std::fs::metadata(&in_path) {
1816                Ok(meta) => {
1817                    match meta.permissions().mode() {
1818                        // S_IFSOCK   0140000   socket
1819                        0o140555 => ready(true),
1820                        _ => ready(false),
1821                    }
1822                }
1823                Err(_) => ready(false),
1824            }
1825        })
1826        .await;
1827    }
1828
1829    ////////////// UNIX STREAM TESTS //////////////
1830    #[cfg(unix)]
1831    async fn send_lines_unix_stream(path: PathBuf, lines: &[&str]) {
1832        let socket = UnixStream::connect(path).await.unwrap();
1833        let mut sink = FramedWrite::new(socket, LinesCodec::new());
1834
1835        let lines = lines.iter().map(|s| Ok(s.to_string()));
1836        let lines = lines.collect::<Vec<_>>();
1837        sink.send_all(&mut stream::iter(lines)).await.unwrap();
1838
1839        let mut socket = sink.into_inner();
1840        socket.shutdown().await.unwrap();
1841    }
1842
1843    #[cfg(unix)]
1844    #[tokio::test]
1845    async fn unix_stream_message() {
1846        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1847            let (_, rx) = unix_message("test", true, false).await;
1848            let events = collect_n(rx, 1).await;
1849
1850            assert_eq!(1, events.len());
1851            assert_eq!(
1852                events[0].as_log()[log_schema().message_key().unwrap().to_string()],
1853                "test".into()
1854            );
1855            assert_eq!(
1856                events[0].as_log()[log_schema().source_type_key().unwrap().to_string()],
1857                "socket".into()
1858            );
1859        })
1860        .await;
1861    }
1862
1863    #[cfg(unix)]
1864    #[tokio::test]
1865    async fn unix_stream_message_with_vector_namespace() {
1866        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1867            let (_, rx) = unix_message("test", true, true).await;
1868            let events = collect_n(rx, 1).await;
1869            let log = events[0].as_log();
1870            let event_meta = log.metadata().value();
1871
1872            assert_eq!(log.value(), &"test".into());
1873            assert_eq!(1, events.len());
1874            assert_eq!(
1875                event_meta.get(path!("vector", "source_type")).unwrap(),
1876                &value!(SocketConfig::NAME)
1877            );
1878            assert_eq!(
1879                event_meta.get(path!(SocketConfig::NAME, "host")).unwrap(),
1880                &value!(UNNAMED_SOCKET_HOST)
1881            );
1882        })
1883        .await;
1884    }
1885
1886    #[cfg(unix)]
1887    #[tokio::test]
1888    async fn unix_stream_message_splits_on_newline() {
1889        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1890            let (_, rx) = unix_message("foo\nbar", true, false).await;
1891            let events = collect_n(rx, 2).await;
1892
1893            assert_eq!(events.len(), 2);
1894            assert_eq!(
1895                events[0].as_log()[log_schema().message_key().unwrap().to_string()],
1896                "foo".into()
1897            );
1898            assert_eq!(
1899                events[0].as_log()[log_schema().source_type_key().unwrap().to_string()],
1900                "socket".into()
1901            );
1902            assert_eq!(
1903                events[1].as_log()[log_schema().message_key().unwrap().to_string()],
1904                "bar".into()
1905            );
1906            assert_eq!(
1907                events[1].as_log()[log_schema().source_type_key().unwrap().to_string()],
1908                "socket".into()
1909            );
1910        })
1911        .await;
1912    }
1913
1914    #[cfg(unix)]
1915    #[tokio::test]
1916    async fn unix_stream_multiple_packets() {
1917        assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async {
1918            unix_multiple_packets(true).await
1919        })
1920        .await;
1921    }
1922
1923    #[cfg(unix)]
1924    #[test]
1925    fn parses_new_unix_stream_config() {
1926        let config = parses_unix_config("unix_stream");
1927        assert!(matches!(config.mode, Mode::UnixStream { .. }));
1928    }
1929
1930    #[cfg(unix)]
1931    #[test]
1932    fn parses_new_unix_datagram_perms() {
1933        let config = parses_unix_config_file_mode("unix_stream");
1934        assert!(matches!(config.mode, Mode::UnixStream { .. }));
1935    }
1936
1937    #[cfg(unix)]
1938    #[test]
1939    fn parses_old_unix_stream_config() {
1940        let config = parses_unix_config("unix");
1941        assert!(matches!(config.mode, Mode::UnixStream { .. }));
1942    }
1943
1944    #[cfg(unix)]
1945    #[tokio::test]
1946    async fn unix_stream_permissions() {
1947        let in_path = tempfile::tempdir().unwrap().keep().join("unix_test");
1948        let (tx, _) = SourceSender::new_test();
1949
1950        let mut config = UnixConfig::new(in_path.clone());
1951        config.socket_file_mode = Some(0o421);
1952        let mode = Mode::UnixStream(config);
1953        let server = SocketConfig { mode }
1954            .build(SourceContext::new_test(tx, None))
1955            .await
1956            .unwrap();
1957        tokio::spawn(server);
1958
1959        wait_for(|| {
1960            match std::fs::metadata(&in_path) {
1961                Ok(meta) => {
1962                    match meta.permissions().mode() {
1963                        // S_IFSOCK   0140000   socket
1964                        0o140421 => ready(true),
1965                        _ => ready(false),
1966                    }
1967                }
1968                Err(_) => ready(false),
1969            }
1970        })
1971        .await;
1972    }
1973}
vector/sources/socket/mod.rs

vector/sources/socket/
mod.rs
