1use crate::llm::actions::{
2 client_trait::{Client, ClientActionResult},
3 protocol_trait::Protocol,
4 ActionDefinition, StartupExamples,
5};
6use crate::protocol::log_template::LogTemplate;
7use crate::protocol::{
8 metadata::{DevelopmentState, ProtocolMetadataV2},
9 ConnectContext, EventType,
10};
11use crate::server::sflow::{
12 actions::parameter,
13 codec::{self, Batch},
14};
15use crate::state::app_state::AppState;
16use anyhow::{bail, Context, Result};
17use serde_json::{json, Value};
18use std::sync::LazyLock;
19
20#[derive(Default)]
21pub struct SflowClientProtocol;
22impl SflowClientProtocol {
23 pub fn new() -> Self {
24 Self
25 }
26}
27pub fn example_batch() -> Value {
28 json!({"agent_address":"192.0.2.1","sub_agent_id":0,"samples":[{
29 "kind":"flow","sequence_number":1,"source":{"class":0,"index":3},
30 "sampling_rate":1000,"sample_pool":1000,"drops":0,
31 "input":{"format":0,"value":3},"output":{"format":0,"value":4},
32 "records":[{"kind":"sampled_ipv4","packet_length":64,"protocol":17,
33 "source_ip":"192.0.2.2","destination_ip":"198.51.100.2","source_port":53,
34 "destination_port":123,"tcp_flags":0,"traffic_class":0}]}]})
35}
36fn send() -> ActionDefinition {
37 ActionDefinition {
38 name:"export_sflow_samples".into(),
39 description:"Validate a complete typed sFlow v5 batch and send one UDP datagram. Sequence counts locally sent datagrams; transport acceptance does not acknowledge collector reception. Sample sequence/pool/drop counters are supplied by the caller.".into(),
40 parameters:vec![parameter("batch","object","agent_address,sub_agent_id,optional uptime_ms,1..32 samples. kind flow/counters; expanded optional(defaultfalse);sequence_number,source(class0..2,index). Flow requires sampling_rate,sample_pool,drops,input/output(format0..2,value),records. Flow records: sampled_ipv4/ipv6 with addresses,packet_length,protocol,ports,tcp_flags,traffic_class; synthesized_header with packet plus frame_length/stripped; ethernet with MACs/ether_type;extended_switch. Counter records: interface(all19fields),ethernet(all13fields),vlan(all6fields).8192bytes,64records/sample,256total; no opaque packet bytes.",true)],
41 example:json!({"type":"export_sflow_samples","batch":example_batch()}),
42 log_template:Some(LogTemplate::new().with_info("sFlow samples queued for local UDP export")),
43 }
44}
45fn disconnect() -> ActionDefinition {
46 ActionDefinition {
47 name: "disconnect".into(),
48 description: "Cancel the exporter, handlers and command channel".into(),
49 parameters: vec![],
50 example: json!({"type":"disconnect"}),
51 log_template: Some(LogTemplate::new().with_info("sFlow exporter disconnected")),
52 }
53}
54pub static SFLOW_CONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
55 EventType::new(
56 "sflow_connected",
57 "sFlow v5 UDP exporter is locally ready; reception is unconfirmed.",
58 send().example,
59 )
60 .with_parameters(vec![
61 parameter(
62 "remote_addr",
63 "string",
64 "Target collector UDP address and port",
65 true,
66 ),
67 parameter("local_addr", "string", "UDP exporter socket", true),
68 ])
69 .with_actions(vec![send(), disconnect()])
70});
71pub static SFLOW_EXPORTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
72 EventType::new("sflow_exported","One datagram accepted by local UDP transport, without a receipt or persistence acknowledgment.",send().example)
73 .with_parameters(vec![parameter("agent_address","string","Declared monitored agent IPv4 or IPv6 address",true),parameter("sub_agent_id","number","Unsigned identifier of this agent exporter instance",true),
74 parameter("sequence_number","number","Datagram sequence before increment",true),parameter("uptime_ms","number","Unsigned32 reported uptime",true),
75 parameter("sample_count","number","Number of flow and counter samples in the datagram",true),parameter("record_count","number","Total telemetry records across all samples",true),
76 parameter("byte_count","number","Number of encoded bytes accepted by local UDP transport",true),parameter("local_transport_only","boolean","True reports local UDP send acceptance, without collector acknowledgment",true)])
77 .with_actions(vec![send(),disconnect()])
78});
79impl Protocol for SflowClientProtocol {
80 fn protocol_name(&self) -> &'static str {
81 "sFlow"
82 }
83 fn stack_name(&self) -> &'static str {
84 "ETH>IP>UDP>SFLOW"
85 }
86 fn keywords(&self) -> Vec<&'static str> {
87 vec!["sflow", "sflow-v5"]
88 }
89 fn description(&self) -> &'static str {
90 "Typed bounded sFlow v5 UDP sample exporter"
91 }
92 fn example_prompt(&self) -> &'static str {
93 "Export sFlow samples to localhost:6343"
94 }
95 fn group_name(&self) -> &'static str {
96 "Network Management"
97 }
98 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
99 vec![send(), disconnect()]
100 }
101 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
102 vec![]
103 }
104 fn get_event_types(&self) -> Vec<EventType> {
105 vec![SFLOW_CONNECTED_EVENT.clone(), SFLOW_EXPORTED_EVENT.clone()]
106 }
107 fn metadata(&self) -> ProtocolMetadataV2 {
108 ProtocolMetadataV2::builder().state(DevelopmentState::Experimental).well_known_udp_port(6343)
109 .implementation("Native bounded sFlow v5 XDR encoder with compact/expanded flow and counter samples; no Cargo protocol dependency")
110 .llm_control("Typed export actions and transport-only events through common handlers/shared memory; command injection independent of parked handlers")
111 .e2e_testing("Required pinned unmodified BSD Cistern exporter/decoder, actual GoFlow2 collector, literal wire/typed goldens and native lifecycle/bounds checks")
112 .notes("Experimental manually supplied telemetry subset, not a full sFlow agent: no packet capture, statistical sampler, automatic counter polling, SNMP configuration, aggregation/store, ACK, reliability or authentication. v5 only; four sample formats; selected flow records1/2/3/4/1001 and counters1/2/5. Synthesized raw headers use typed IPv4/IPv6 TCP/UDP summaries; no payload, transport-checksum or capture-integrity claim. 8192bytes/datagram,32samples,64records/sample,256records;32declared agent/sub-agent sequence states, modulo2^32 datagram count starting0. Explicit sample sequence numbers describe caller-owned sFlow instances, not inferred from source alone; uptime optional caller value or elapsed logical client time modulo2^32.10s resolve/write deadlines,oneinflight,32queued events/actions,followup depth8. Required Cistern peer source-ID encoder defect compensated through public arguments; VLAN encoder excluded, evidence uses literal and GoFlow2. No fuzz/pcap/full compliance claim.")
113 .build()
114 }
115 fn get_startup_examples(&self) -> StartupExamples {
116 StartupExamples::new(
117 json!({"type":"open_client","base_stack":"sflow","remote_addr":"localhost:6343"}),
118 json!({"type":"open_client","base_stack":"sflow","remote_addr":"localhost:6343","event_handlers":[{"event_pattern":"sflow_connected","handler":{"type":"script","language":"python","code":format!("import json,sys\njson.load(sys.stdin)\nprint({:?})",json!({"actions":[send().example]}).to_string())}}]}),
119 json!({"type":"open_client","base_stack":"sflow","remote_addr":"localhost:6343","event_handlers":[{"event_pattern":"sflow_connected","handler":{"type":"static","actions":[send().example]}}]}),
120 )
121 }
122}
123impl Client for SflowClientProtocol {
124 fn connect(
125 &self,
126 ctx: ConnectContext,
127 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
128 {
129 Box::pin(super::SflowClient::connect(ctx))
130 }
131 fn execute_action(&self, action: Value) -> Result<ClientActionResult> {
132 match action["type"].as_str() {
133 Some("export_sflow_samples") => {
134 let batch: Batch =
135 serde_json::from_value(action.get("batch").context("missing batch")?.clone())?;
136 codec::encode(&batch, 0, batch.uptime_ms.unwrap_or(0))?;
137 Ok(ClientActionResult::Custom {
138 name: "export_sflow_samples".into(),
139 data: serde_json::to_value(batch)?,
140 })
141 }
142 Some("disconnect") => Ok(ClientActionResult::Disconnect),
143 _ => bail!("unknown sFlow exporter action"),
144 }
145 }
146}