wp-core-connectors 0.4.1

//! Syslog source factory implementation
//!
//! This module provides the factory for creating syslog data sources
//! with support for both UDP and TCP protocols.

use super::config::{Protocol, SyslogSourceSpec};
use super::tcp_source::TcpSyslogSource;
use super::udp_source::UdpSyslogSource;
use crate::sources::tcp::{FramingMode, TcpAcceptor, TcpSource, tcp_reader_batch_channel_cap};
use orion_conf::{ErrorWith, ToStructError};
use std::collections::HashSet;
use std::sync::{Arc, Mutex};
use tokio::sync::mpsc;
use wp_connector_api::{
    AcceptorHandle, ConnectorDef, SourceBuildCtx, SourceDefProvider, SourceFactory, SourceHandle,
    SourceMeta, SourceReason, SourceResult, SourceSvcIns, Tags,
};

/// Syslog source factory that creates both UDP and TCP syslog sources
pub struct SyslogSourceFactory {}

impl SyslogSourceFactory {
    /// Create a new syslog source factory
    pub fn new() -> Self {
        Self {}
    }
}

impl Default for SyslogSourceFactory {
    fn default() -> Self {
        Self::new()
    }
}

#[async_trait::async_trait]
impl SourceFactory for SyslogSourceFactory {
    fn kind(&self) -> &'static str {
        "syslog"
    }

    async fn build(
        &self,
        spec: &wp_connector_api::SourceSpec,
        _ctx: &SourceBuildCtx,
    ) -> SourceResult<SourceSvcIns> {
        let fut = async {
            let config = SyslogSourceSpec::from_params(&spec.params)?;
            let mut base_tags = {
                let mut tags = Tags::new();
                for item in &spec.tags {
                    if let Some((k, v)) = item.split_once("=").or_else(|| item.split_once(":")) {
                        tags.set(k, v);
                    }
                }
                tags
            };
            base_tags.set("access_source", "syslog".to_string());
            base_tags.set("syslog_protocol", format!("{:?}", config.protocol));

            let meta_builder = |name: &str, tagset: &Tags| -> SourceMeta {
                let mut meta = SourceMeta::new(name.to_string(), spec.kind.clone());
                for (k, v) in tagset.iter() {
                    meta.tags.set(k, v);
                }
                meta
            };

            let svc = match config.protocol {
                Protocol::Udp => {
                    info_ctrl!(
                        "syslog UDP factory build: strip_header={}, attach_meta_tags={}, fast_strip={}, udp_recv_buffer={}",
                        config.strip_header,
                        config.attach_meta_tags,
                        config.fast_strip,
                        config.udp_recv_buffer
                    );
                    let tagset = base_tags.clone();
                    let source = UdpSyslogSource::new(
                        spec.name.clone(),
                        config.address(),
                        tagset.clone(),
                        config.strip_header,
                        config.attach_meta_tags,
                        config.fast_strip,
                        config.udp_recv_buffer,
                    )
                    .await?;
                    let meta = meta_builder(&spec.name, &tagset);
                    SourceSvcIns::new()
                        .with_sources(vec![SourceHandle::new(Box::new(source), meta)])
                }
                Protocol::Tcp => {
                    let tags = base_tags.clone();
                    let connection_registry = Arc::new(Mutex::new(HashSet::<u64>::new()));
                    let mut instance_reg_txs = Vec::with_capacity(config.instances);
                    let mut source_handles = Vec::with_capacity(config.instances);
                    let framing = FramingMode::Auto;

                    for idx in 0..config.instances {
                        let (reader_reg_tx, reader_reg_rx) =
                            mpsc::channel(tcp_reader_batch_channel_cap());
                        instance_reg_txs.push(reader_reg_tx);

                        let key = if config.instances == 1 {
                            spec.name.clone()
                        } else {
                            format!("{}#{}", spec.name, idx + 1)
                        };
                        let inner = TcpSource::new(
                            key.clone(),
                            tags.clone(),
                            config.address(),
                            config.tcp_recv_bytes,
                            framing,
                            connection_registry.clone(),
                            reader_reg_rx,
                        )
                        .map_err(|e| anyhow::anyhow!("{}", e))?;
                        let syslog = TcpSyslogSource::new(
                            key.clone(),
                            tags.clone(),
                            config.strip_header,
                            config.attach_meta_tags,
                            config.fast_strip,
                            inner,
                        )
                        .await
                        .map_err(|e| anyhow::anyhow!("{}", e))?;

                        let mut meta = SourceMeta::new(key.clone(), spec.kind.clone());
                        for (k, v) in tags.iter() {
                            meta.tags.set(k, v);
                        }
                        if config.instances > 1 {
                            meta.tags.set("instance".to_string(), (idx + 1).to_string());
                        }
                        source_handles.push(SourceHandle::new(Box::new(syslog), meta));
                    }

                    let acceptor = TcpAcceptor::new(
                        spec.name.clone(),
                        config.address(),
                        1000,
                        connection_registry,
                        instance_reg_txs,
                    );

                    SourceSvcIns::new()
                        .with_sources(source_handles)
                        .with_acceptor(AcceptorHandle::new(spec.name.clone(), Box::new(acceptor)))
                }
            };

            Ok(svc)
        };

        let fut: anyhow::Result<SourceSvcIns> = fut.await;
        fut.map_err(|e| {
            SourceReason::core_conf()
                .to_err()
                .with_detail(e.to_string())
        })
        .with_context(spec.name.as_str())
        .doing("build syslog source service")
    }
}

impl SourceDefProvider for SyslogSourceFactory {
    fn source_def(&self) -> ConnectorDef {
        crate::builtin::source_def("syslog_src").expect("builtin source def missing: syslog_src")
    }
}