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

vector/sources/socket/
mod.rs
