wp-core-connectors 0.5.2

Core connector registry and sink runtimes for WarpParse
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126

//! Configuration structures for syslog sources

use super::constants::{DEFAULT_TCP_RECV_BYTES, DEFAULT_UDP_RECV_BUFFER};
use anyhow::ensure;

/// Configuration for syslog sources
#[derive(Debug, Clone)]
pub struct SyslogSourceSpec {
    pub addr: String,
    pub port: u16,
    pub protocol: Protocol,
    pub tcp_recv_bytes: usize,
    /// UDP socket receive buffer size (bytes)
    pub udp_recv_buffer: usize,
    pub strip_header: bool,
    pub attach_meta_tags: bool,
    /// Fast strip mode (works for both UDP and TCP when header_mode=skip)
    /// Enables fast path that skips full syslog parsing when only stripping header
    pub fast_strip: bool,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Protocol {
    Udp,
    Tcp,
}

impl SyslogSourceSpec {
    /// Parse configuration directly from params table (Factory path)
    pub fn from_params(params: &wp_connector_api::ParamMap) -> anyhow::Result<Self> {
        if let Some(v) = params.get("protocol").and_then(|v| v.as_str()) {
            let p = v.to_ascii_lowercase();
            ensure!(
                p == "udp" || p == "tcp",
                "invalid protocol: {} (must be 'udp' or 'tcp')",
                v
            );
        }
        if let Some(v) = params.get("tcp_recv_bytes").and_then(|v| v.as_i64()) {
            ensure!(v > 0, "tcp_recv_bytes must be > 0 (got {})", v);
        }
        if let Some(v) = params.get("udp_recv_buffer").and_then(|v| v.as_i64()) {
            ensure!(v > 0, "udp_recv_buffer must be > 0 (got {})", v);
        }
        if let Some(v) = params.get("port").and_then(|v| v.as_i64()) {
            ensure!(
                (0..=65535).contains(&v),
                "port out of range: {} (allow 0 or 1..=65535)",
                v
            );
        }

        let addr = params
            .get("addr")
            .and_then(|v| v.as_str())
            .unwrap_or("0.0.0.0")
            .to_string();
        let port = params.get("port").and_then(|v| v.as_i64()).unwrap_or(514) as u16;
        let protocol = params
            .get("protocol")
            .and_then(|v| v.as_str())
            .unwrap_or("UDP");
        let protocol = match protocol.to_uppercase().as_str() {
            "TCP" => Protocol::Tcp,
            _ => Protocol::Udp,
        };
        let tcp_recv_bytes = params
            .get("tcp_recv_bytes")
            .and_then(|v| v.as_i64())
            .filter(|&v| v > 0)
            .unwrap_or(DEFAULT_TCP_RECV_BYTES as i64) as usize;
        let udp_recv_buffer = params
            .get("udp_recv_buffer")
            .and_then(|v| v.as_i64())
            .filter(|&v| v > 0)
            .unwrap_or(DEFAULT_UDP_RECV_BUFFER as i64) as usize;
        // header_mode: controls how syslog header is handled
        //   New names (preferred):
        //     raw  => keep original message untouched
        //     skip => strip header, keep message body only
        //     tag  => strip header + extract syslog.pri/facility/severity tags
        //   Legacy aliases (backward-compatible):
        //     keep  => raw
        //     strip => skip
        //     parse => tag
        let header_mode = params
            .get("header_mode")
            .and_then(|v| v.as_str())
            .unwrap_or("skip")
            .to_ascii_lowercase();
        let (strip_header, attach_meta_tags) = match header_mode.as_str() {
            "raw" | "keep" => (false, false),
            "skip" | "strip" => (true, false),
            "tag" | "parse" => (true, true),
            other => {
                log::warn!("syslog.header_mode invalid: '{}', fallback to 'tag'", other);
                (true, true)
            }
        };
        log::info!(
            "syslog config: header_mode='{}' => strip_header={}, attach_meta_tags={}",
            header_mode,
            strip_header,
            attach_meta_tags
        );
        let fast_strip = params
            .get("fast_strip")
            .and_then(|v| v.as_bool())
            .unwrap_or(false);
        Ok(Self {
            addr,
            port,
            protocol,
            tcp_recv_bytes,
            udp_recv_buffer,
            strip_header,
            attach_meta_tags,
            fast_strip,
        })
    }

    /// Get the full address string
    pub fn address(&self) -> String {
        format!("{}:{}", self.addr, self.port)
    }
}