netget/server/gelf/
mod.rs

1//! One-way GELF collector with bounded stream parsing and owned tasks.
2pub mod actions;
3pub mod codec;
4use crate::protocol::{Event, SpawnContext};
5use crate::server::{
6    accept_bounded::{accept_bounded, ConnectionLimiter, DEFAULT_MAX_CONNECTIONS},
7    connection::ConnectionId,
8};
9use crate::state::{
10    server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo},
11    AccessLogOwner,
12};
13use crate::{console_error, console_info};
14use anyhow::{Context, Result};
15use codec::{Message, Reassembler, Transport};
16use serde_json::json;
17use std::{
18    net::SocketAddr,
19    time::{Duration, SystemTime, UNIX_EPOCH},
20};
21use tokio::{
22    io::AsyncReadExt,
23    net::{TcpListener, TcpStream},
24};
25
26pub const READ_TIMEOUT: Duration = Duration::from_secs(30);
27pub struct GelfServer;
28impl GelfServer {
29    pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
30        let llm_fallback = ctx
31            .startup_params
32            .as_ref()
33            .map(|p| p.get_optional_bool("llm_fallback"))
34            .transpose()?
35            .flatten()
36            .unwrap_or(codec::DEFAULT_LLM_FALLBACK);
37        let transport = Transport::parse(
38            &ctx.startup_params
39                .as_ref()
40                .map(|p| p.get_optional_string("transport"))
41                .transpose()?
42                .flatten()
43                .unwrap_or_else(|| codec::DEFAULT_TRANSPORT.into()),
44        )?;
45        if transport == Transport::Udp {
46            return udp(ctx, llm_fallback).await;
47        }
48        let listener = TcpListener::bind(ctx.legacy_listen_addr()).await?;
49        let local = listener.local_addr()?;
50        console_info!(
51            ctx.status_tx,
52            "GELF TCP listening on {} (llm_fallback={})",
53            local,
54            llm_fallback
55        );
56        let registrar = ctx.state.clone();
57        let server_id = ctx.server_id;
58        let task = tokio::spawn(async move {
59            let limiter = ConnectionLimiter::new(DEFAULT_MAX_CONNECTIONS);
60            loop {
61                // GELF has no response grammar: refuse excess peers by closing.
62                let (socket, peer, permit) =
63                    match accept_bounded(&listener, &limiter, b"", "GELF", Some(&ctx.status_tx))
64                        .await
65                    {
66                        Ok(accepted) => accepted,
67                        Err(error) => {
68                            console_error!(ctx.status_tx, "GELF accept failed: {}", error);
69                            break;
70                        }
71                    };
72                let id = ConnectionId::new(ctx.state.get_next_unified_id().await);
73                let now = crate::utils::clock::Instant::now();
74                ctx.state
75                    .add_connection_to_server(
76                        server_id,
77                        ConnectionState {
78                            id,
79                            remote_addr: peer,
80                            local_addr: local,
81                            bytes_sent: 0,
82                            bytes_received: 0,
83                            packets_sent: 0,
84                            packets_received: 0,
85                            last_activity: now,
86                            status: ConnectionStatus::Active,
87                            status_changed_at: now,
88                            protocol_info: ProtocolConnectionInfo::empty(),
89                        },
90                    )
91                    .await;
92                let child_ctx = ctx.clone();
93                let child = tokio::spawn(async move {
94                    let _permit = permit;
95                    if let Err(error) = session(socket, peer, id, &child_ctx, llm_fallback).await {
96                        console_error!(child_ctx.status_tx, "GELF peer {} closed: {}", peer, error);
97                        child_ctx
98                            .state
99                            .record_access_log(
100                                AccessLogOwner::Server(server_id.as_u32()),
101                                "GELF",
102                                Some(id.as_u32()),
103                                "gelf_invalid_stream",
104                                json!({"source_addr":peer.to_string()}),
105                                vec![json!({"error":error.to_string()})],
106                            )
107                            .await;
108                    }
109                    child_ctx
110                        .state
111                        .remove_connection_from_server(server_id, id)
112                        .await;
113                    let _ = child_ctx.status_tx.send("__UPDATE_UI__".into());
114                });
115                ctx.state.register_server_task(server_id, child).await;
116            }
117        });
118        registrar.register_server_task(server_id, task).await;
119        Ok(local)
120    }
121}
122async fn session(
123    mut socket: TcpStream,
124    peer: SocketAddr,
125    id: ConnectionId,
126    ctx: &SpawnContext,
127    llm_fallback: bool,
128) -> Result<()> {
129    let mut decoder = codec::TcpDecoder::default();
130    let mut buffer = [0u8; 8192];
131    loop {
132        let deadline = tokio::time::Instant::now() + READ_TIMEOUT;
133        let message = loop {
134            if let Some(message) = decoder.next_message()? {
135                break message;
136            }
137            let n = tokio::time::timeout_at(deadline, socket.read(&mut buffer))
138                .await
139                .context("GELF frame read deadline exceeded")??;
140            if n == 0 {
141                return decoder.finish();
142            }
143            ctx.state
144                .update_connection_stats(ctx.server_id, id, Some(n as u64), None, Some(1), None)
145                .await;
146            decoder.feed(&buffer[..n])?;
147        };
148        dispatch(message, peer, id, ctx, Transport::Tcp, llm_fallback).await?;
149    }
150}
151async fn dispatch(
152    mut message: Message,
153    peer: SocketAddr,
154    id: ConnectionId,
155    ctx: &SpawnContext,
156    transport: Transport,
157    llm_fallback: bool,
158) -> Result<()> {
159    if message.timestamp.is_none() {
160        message.timestamp = Some(SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs_f64());
161    }
162    if message.level.is_none() {
163        message.level = Some(1);
164    }
165    let event = Event::new(
166        &actions::GELF_MESSAGE_EVENT,
167        json!({"message":message,"source_addr":peer.to_string(),"transport":transport.as_str()}),
168    );
169    let configured = ctx
170        .state
171        .get_event_handler_config(ctx.server_id)
172        .await
173        .is_some_and(|c| c.find_handler("gelf_message").is_some());
174    if llm_fallback || configured {
175        let result = crate::llm::action_helper::call_llm(
176            &ctx.llm_client,
177            &ctx.state,
178            ctx.server_id,
179            Some(id),
180            &event,
181            &actions::GelfProtocol::new(),
182        )
183        .await?;
184        if !result.failures.is_empty() {
185            console_error!(
186                ctx.status_tx,
187                "GELF decision=fail_closed_handler_action_error failed_actions={}",
188                result.failures.len()
189            );
190            ctx.state.record_access_log(
191                AccessLogOwner::Server(ctx.server_id.as_u32()), "GELF", Some(id.as_u32()),
192                "gelf_handler_failed", event.data,
193                vec![json!({"decision":"fail_closed_handler_action_error","failed_action_count":result.failures.len()})],
194            ).await;
195            anyhow::bail!("GELF handler action failed");
196        }
197        for message in result.messages {
198            console_info!(ctx.status_tx, "{}", message);
199        }
200    } else {
201        ctx.state
202            .record_access_log(
203                AccessLogOwner::Server(ctx.server_id.as_u32()),
204                "GELF",
205                Some(id.as_u32()),
206                "gelf_message",
207                event.data,
208                vec![json!({"type":"collect_gelf_message"})],
209            )
210            .await;
211    }
212    let _ = ctx.status_tx.send("__UPDATE_UI__".into());
213    Ok(())
214}
215async fn udp(ctx: SpawnContext, llm_fallback: bool) -> Result<SocketAddr> {
216    let socket = tokio::net::UdpSocket::bind(ctx.legacy_listen_addr()).await?;
217    let local = socket.local_addr()?;
218    console_info!(
219        ctx.status_tx,
220        "GELF UDP listening on {} (llm_fallback={})",
221        local,
222        llm_fallback
223    );
224    let registrar = ctx.state.clone();
225    let server_id = ctx.server_id;
226    let task = tokio::spawn(async move {
227        let mut buffer = vec![0; codec::MAX_DATAGRAM_BYTES + 1];
228        let mut reassembler = Reassembler::default();
229        let mut expiry = tokio::time::interval(Duration::from_secs(1));
230        loop {
231            let (n, peer) = tokio::select! {
232                _=expiry.tick()=>{reassembler.expire(tokio::time::Instant::now());continue;},
233                result=socket.recv_from(&mut buffer)=>match result {Ok(r)=>r,Err(error)=>{console_error!(ctx.status_tx,"GELF receive failed: {}",error);break;}}
234            };
235            let message = match reassembler.push(peer, &buffer[..n], tokio::time::Instant::now()) {
236                Ok(Some(m)) => m,
237                Ok(None) => continue,
238                Err(error) => {
239                    ctx.state
240                        .record_access_log(
241                            AccessLogOwner::Server(server_id.as_u32()),
242                            "GELF",
243                            None,
244                            "gelf_invalid_datagram",
245                            json!({"source_addr":peer.to_string(),"received_bytes":n}),
246                            vec![json!({"error":error.to_string()})],
247                        )
248                        .await;
249                    continue;
250                }
251            };
252            let id = ConnectionId::new(ctx.state.get_next_unified_id().await);
253            let now = crate::utils::clock::Instant::now();
254            ctx.state
255                .add_connection_to_server(
256                    server_id,
257                    ConnectionState {
258                        id,
259                        remote_addr: peer,
260                        local_addr: local,
261                        bytes_sent: 0,
262                        bytes_received: n as u64,
263                        packets_sent: 0,
264                        packets_received: 1,
265                        last_activity: now,
266                        status: ConnectionStatus::Active,
267                        status_changed_at: now,
268                        protocol_info: ProtocolConnectionInfo::empty(),
269                    },
270                )
271                .await;
272            if let Err(error) =
273                dispatch(message, peer, id, &ctx, Transport::Udp, llm_fallback).await
274            {
275                console_error!(
276                    ctx.status_tx,
277                    "GELF decision=fail_closed_handler_error: {}",
278                    error
279                );
280            }
281            ctx.state.remove_connection_from_server(server_id, id).await;
282        }
283    });
284    registrar.register_server_task(server_id, task).await;
285    Ok(local)
286}
netget/server/gelf/mod.rs

netget/server/gelf/
mod.rs
