1pub mod actions;
3pub mod codec;
4use crate::protocol::{Event, SpawnContext};
5use crate::server::connection::ConnectionId;
6use crate::state::{
7 server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo},
8 AccessLogOwner,
9};
10use crate::{console_error, console_info};
11use anyhow::{ensure, Result};
12use serde_json::json;
13use std::{net::SocketAddr, time::Duration};
14pub const MAX_QUEUED_EVENTS: usize = 32;
15pub struct SflowServer;
16pub fn duration(value: u64) -> Result<Duration> {
17 ensure!(
18 (1..=86400).contains(&value),
19 "sFlow duration must be1..86400seconds"
20 );
21 Ok(Duration::from_secs(value))
22}
23impl SflowServer {
24 pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
25 let params = ctx.startup_params.as_ref();
26 let idle = duration(
27 params
28 .map(|p| p.get_optional_u64("session_idle_seconds"))
29 .transpose()?
30 .flatten()
31 .unwrap_or(codec::DEFAULT_SESSION_IDLE),
32 )?;
33 let llm = params
34 .map(|p| p.get_optional_bool("llm_fallback"))
35 .transpose()?
36 .flatten()
37 .unwrap_or(codec::DEFAULT_LLM_FALLBACK);
38 let socket = tokio::net::UdpSocket::bind(ctx.legacy_listen_addr()).await?;
39 let local = socket.local_addr()?;
40 console_info!(
41 ctx.status_tx,
42 "sFlow UDP listening on {} (llm_fallback={})",
43 local,
44 llm
45 );
46 let (tx, mut rx) =
47 tokio::sync::mpsc::channel::<(SocketAddr, usize, codec::Message)>(MAX_QUEUED_EVENTS);
48 let dispatcher = ctx.clone();
49 let owner = ctx.state.clone();
50 let id = ctx.server_id;
51 owner
52 .spawn_server_task(id, async move {
53 while let Some((peer, n, message)) = rx.recv().await {
54 let cid = ConnectionId::new(dispatcher.state.get_next_unified_id().await);
55 let now = crate::utils::clock::Instant::now();
56 dispatcher
57 .state
58 .add_connection_to_server(
59 id,
60 ConnectionState {
61 id: cid,
62 remote_addr: peer,
63 local_addr: local,
64 bytes_sent: 0,
65 bytes_received: n as u64,
66 packets_sent: 0,
67 packets_received: 1,
68 last_activity: now,
69 status: ConnectionStatus::Active,
70 status_changed_at: now,
71 protocol_info: ProtocolConnectionInfo::empty(),
72 },
73 )
74 .await;
75 let event = Event::new(
76 &actions::SFLOW_MESSAGE_EVENT,
77 json!({"message":message, "source_addr":peer.to_string()}),
78 );
79 let configured = dispatcher
80 .state
81 .get_event_handler_config(id)
82 .await
83 .is_some_and(|c| c.find_handler("sflow_message").is_some());
84 if llm || configured {
85 match crate::llm::action_helper::call_llm(
86 &dispatcher.llm_client,
87 &dispatcher.state,
88 id,
89 Some(cid),
90 &event,
91 &actions::SflowProtocol::new(),
92 )
93 .await
94 {
95 Ok(result) if result.failures.is_empty() => {
96 let decision = if result
97 .protocol_result_actions
98 .iter()
99 .any(|action| action == "collect_sflow_samples")
100 {
101 "handler_collect"
102 } else {
103 "handler_silent"
104 };
105 for m in result.messages {
106 console_info!(dispatcher.status_tx, "{}", m);
107 }
108 console_info!(
109 dispatcher.status_tx,
110 "sFlow source={} decision={} udp_silent=true",
111 peer,
112 decision
113 );
114 dispatcher
115 .state
116 .record_access_log(
117 AccessLogOwner::Server(id.as_u32()),
118 "sFlow",
119 Some(cid.as_u32()),
120 "sflow_handler_decision",
121 event.data,
122 vec![json!({"decision":decision,"udp_silent":true})],
123 )
124 .await;
125 }
126 result => {
127 let (decision, reason) = match result {
128 Ok(r) => (
129 "fail_closed_action_error",
130 format!("{} failed actions", r.failures.len()),
131 ),
132 Err(e) => ("fail_closed_dispatch_error", e.to_string()),
133 };
134 console_error!(
135 dispatcher.status_tx,
136 "sFlow source={} decision={} udp_silent=true error={}",
137 peer,
138 decision,
139 reason
140 );
141 dispatcher
142 .state
143 .record_access_log(
144 AccessLogOwner::Server(id.as_u32()),
145 "sFlow",
146 Some(cid.as_u32()),
147 "sflow_handler_failed",
148 event.data,
149 vec![json!({"decision":decision, "error":reason,"terminal_decision":decision,"udp_silent":true})],
150 )
151 .await;
152 }
153 }
154 } else {
155 console_info!(
156 dispatcher.status_tx,
157 "sFlow source={} decision=default_collect udp_silent=true",
158 peer
159 );
160 dispatcher
161 .state
162 .record_access_log(
163 AccessLogOwner::Server(id.as_u32()),
164 "sFlow",
165 Some(cid.as_u32()),
166 "sflow_message",
167 event.data,
168 vec![json!({"type":"collect_sflow_samples","decision":"default_collect","udp_silent":true})],
169 )
170 .await;
171 }
172 dispatcher
173 .state
174 .remove_connection_from_server(id, cid)
175 .await;
176 let _ = dispatcher.status_tx.send("__UPDATE_UI__".into());
177 }
178 })
179 .await;
180 owner
181 .spawn_server_task(id, async move {
182 let mut cache = codec::SequenceCache::new(idle);
183 let mut buffer = vec![0; codec::MAX_MESSAGE_BYTES + 1];
184 let mut expiry = tokio::time::interval(Duration::from_secs(1));
185 expiry.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
186 loop {
187 let (n, peer) = tokio::select! {
188 _ = expiry.tick() => {
189 cache.expire(tokio::time::Instant::now());
190 continue;
191 },
192 r = socket.recv_from(&mut buffer) => match r {
193 Ok(v) => v,
194 Err(e) => {
195 console_error!(ctx.status_tx, "sFlow decision=fail_closed_receive_error udp_silent=true error={}", e);
196 break;
197 }
198 }
199 };
200 match cache.ingest(peer, &buffer[..n], tokio::time::Instant::now()) {
201 Ok(message) => {
202 if let Err(error) = tx.try_send((peer, n, message)) {
203 let closed = matches!(
204 error,
205 tokio::sync::mpsc::error::TrySendError::Closed(_)
206 );
207 let decision = if closed {
208 "fail_closed_dispatch_closed"
209 } else {
210 "fail_closed_event_capacity"
211 };
212 console_error!(
213 ctx.status_tx,
214 "sFlow source={} decision={} udp_silent=true event_cap={}",
215 peer,
216 decision,
217 MAX_QUEUED_EVENTS
218 );
219 ctx.state
220 .record_access_log(
221 AccessLogOwner::Server(id.as_u32()),
222 "sFlow",
223 None,
224 "sflow_event_capacity",
225 json!({"source_addr":peer.to_string()}),
226 vec![json!({"decision":"fail_closed_event_capacity", "event_cap":MAX_QUEUED_EVENTS,"terminal_decision":decision,"udp_silent":true})],
227 )
228 .await;
229 if closed {
230 break;
231 }
232 }
233 }
234 Err(error) => {
235 console_error!(
236 ctx.status_tx,
237 "sFlow source={} decision=fail_closed_invalid_datagram udp_silent=true received_bytes={} error={}",
238 peer,
239 n,
240 error
241 );
242 ctx.state
243 .record_access_log(
244 AccessLogOwner::Server(id.as_u32()),
245 "sFlow",
246 None,
247 "sflow_invalid_datagram",
248 json!({"source_addr":peer.to_string(), "received_bytes":n}),
249 vec![json!({"decision":"fail_closed_invalid_datagram", "error":error.to_string()})],
250 )
251 .await;
252 }
253 }
254 }
255 })
256 .await;
257 Ok(local)
258 }
259}