netget/server/sflow/
actions.rs

1use super::codec::{DEFAULT_LLM_FALLBACK, DEFAULT_SESSION_IDLE, MAX_MESSAGE_BYTES};
2use crate::llm::actions::{
3    protocol_trait::{ActionResult, Protocol, Server},
4    ActionDefinition, Parameter, ParameterDefinition, StartupExamples,
5};
6use crate::protocol::log_template::LogTemplate;
7use crate::protocol::{
8    metadata::{DevelopmentState, ProtocolMetadataV2},
9    EventType, SpawnContext,
10};
11use crate::state::app_state::AppState;
12use anyhow::{bail, Result};
13use serde_json::{json, Value};
14use std::sync::LazyLock;
15
16#[derive(Default)]
17pub struct SflowProtocol;
18impl SflowProtocol {
19    pub fn new() -> Self {
20        Self
21    }
22}
23pub fn parameter(name: &str, type_hint: &str, description: &str, required: bool) -> Parameter {
24    Parameter {
25        name: name.into(),
26        type_hint: type_hint.into(),
27        description: description.into(),
28        required,
29    }
30}
31fn collect() -> ActionDefinition {
32    ActionDefinition { name:"collect_sflow_samples".into(),
33        description:"Observe typed sFlow samples through the bounded common access log. UDP is silent; no protocol flow store, persistence or aggregation.".into(),
34        parameters:vec![],example:json!({"type":"collect_sflow_samples"}),log_template:Some(LogTemplate::new().with_info("sFlow samples collected; UDP remains silent")) }
35}
36pub static SFLOW_MESSAGE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
37    EventType::new("sflow_message","A validated sFlow v5 UDP datagram. Selected flow/counter records are typed; header payload and unsupported record bytes are discarded. Datagram sequence gaps and uptime decreases are observations, not reliable delivery/reboot identification.",collect().example)
38        .with_parameters(vec![parameter("message","object","Agent address,sub-agent,datagram sequence,uptime,compact/expanded samples,selected typed records,unknown enterprise/format/byte counts and bounded sequence diagnostics",true),
39            parameter("source_addr","string","UDP exporter IP and port, scoped to this collector",true)])
40        .with_actions(vec![collect()])
41});
42impl Protocol for SflowProtocol {
43    fn protocol_name(&self) -> &'static str {
44        "sFlow"
45    }
46    fn stack_name(&self) -> &'static str {
47        "ETH>IP>UDP>SFLOW"
48    }
49    fn keywords(&self) -> Vec<&'static str> {
50        vec!["sflow", "sflow-v5"]
51    }
52    fn description(&self) -> &'static str {
53        "Typed bounded sFlow v5 UDP flow/counter collector; no model calls by default"
54    }
55    fn example_prompt(&self) -> &'static str {
56        "Collect sFlow samples on UDP6343"
57    }
58    fn group_name(&self) -> &'static str {
59        "Network Management"
60    }
61    fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
62        vec![]
63    }
64    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
65        vec![collect()]
66    }
67    fn get_event_types(&self) -> Vec<EventType> {
68        vec![SFLOW_MESSAGE_EVENT.clone()]
69    }
70    fn metadata(&self) -> ProtocolMetadataV2 {
71        ProtocolMetadataV2::builder().deliberately_silent().state(DevelopmentState::Experimental)
72            .well_known_udp_port(6343).max_inbound_bytes(MAX_MESSAGE_BYTES)
73            .implementation("Native bounded sFlow v5 XDR collector; compact/expanded flow/counter samples, typed selected records and transient sequence diagnostics")
74            .llm_control("Common event handlers/shared memory/access log; llm_fallback=false collects unmatched samples without a model call; all UDP outcomes are silent and decision-tagged")
75            .e2e_testing("Pinned unmodified BSD Cistern public exporter/decoder with explicit known source-ID workaround; actual GoFlow2 collector, literal dual wire/typed golden, malformed/bounds/sequence/owned lifecycle checks")
76            .notes("Experimental v5 telemetry subset, not a full sFlow collector/agent: no SNMP configuration, aggregation/domain store, persistence, authentication, ACK, reliable delivery, sampler or packet capture. Four sample formats; flow records1(header)/2(Ethernet)/3(IPv4)/4(IPv6)/1001(switch), counters1(interface)/2(Ethernet)/5(VLAN). Header summaries cover Ethernet/up to2VLANs,IPv4/IPv6,ports/TCP flags,up to8IPv6 extensions; truncated/unknown headers remain metadata-only and payloads never reach model data. No checksum/capture-integrity claim.8192bytes/datagram,32samples,64records/sample,256records,256headerbytes,4096unknownbytes,128sessions,32queued events+oneactive. Cache key exporterIP/port+declared agent+subagent;session_idle_seconds(default1800,1..86400) expires on ingest and owned1s tick. Datagram sequence modulo2^32; late/duplicate does not regress, loweruptime diagnostic does not prove reboot; expiry resets expectation. Caller-owned sample sequence per sFlow instance is passed through, not inferred per datasource. Failed handler actions do not roll back prior wire state/common actions. No fuzz/pcap/full compliance claim.")
77            .build()
78    }
79    fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
80        vec![
81            ParameterDefinition {
82                name: "session_idle_seconds".into(),
83                type_hint: "number".into(),
84                description:
85                    "UDP exporter/agent/sub-agent diagnostic state idle timeout,1..86400seconds"
86                        .into(),
87                required: false,
88                example: json!(DEFAULT_SESSION_IDLE),
89                default: Some(json!(DEFAULT_SESSION_IDLE)),
90            },
91            ParameterDefinition {
92                name: "llm_fallback".into(),
93                type_hint: "boolean".into(),
94                description:
95                    "Enable unmatched model calls; explicit matching handlers always dispatch"
96                        .into(),
97                required: false,
98                example: json!(true),
99                default: Some(json!(DEFAULT_LLM_FALLBACK)),
100            },
101        ]
102    }
103    fn get_startup_examples(&self) -> StartupExamples {
104        StartupExamples::new(
105            json!({"type":"open_server","base_stack":"sflow","port":6343,"startup_params":{"llm_fallback":true},"instruction":"Review telemetry"}),
106            json!({"type":"open_server","base_stack":"sflow","port":6343,"event_handlers":[{"event_pattern":"sflow_message","handler":{"type":"script","language":"python","code":"import json,sys\njson.load(sys.stdin)\nprint(json.dumps({'actions':[{'type':'collect_sflow_samples'}]}))"}}]}),
107            json!({"type":"open_server","base_stack":"sflow","port":6343,"event_handlers":[{"event_pattern":"sflow_message","handler":{"type":"static","actions":[collect().example]}}]}),
108        )
109    }
110}
111impl Server for SflowProtocol {
112    fn spawn(
113        &self,
114        ctx: SpawnContext,
115    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
116    {
117        Box::pin(super::SflowServer::spawn(ctx))
118    }
119    fn execute_action(&self, action: Value) -> Result<ActionResult> {
120        match action["type"].as_str() {
121            Some("collect_sflow_samples") => Ok(ActionResult::NoAction),
122            _ => bail!("unknown sFlow collector action"),
123        }
124    }
125}
netget/server/sflow/actions.rs

netget/server/sflow/
actions.rs
