1use super::codec::{DEFAULT_LLM_FALLBACK, DEFAULT_TRANSPORT, 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#[derive(Default)]
16pub struct GelfProtocol;
17impl GelfProtocol {
18 pub fn new() -> Self {
19 Self
20 }
21}
22pub fn parameter(name: &str, type_hint: &str, description: &str, required: bool) -> Parameter {
23 Parameter {
24 name: name.into(),
25 type_hint: type_hint.into(),
26 description: description.into(),
27 required,
28 }
29}
30pub fn transport_parameter() -> ParameterDefinition {
31 ParameterDefinition {name:"transport".into(),type_hint:"string".into(),description:"udp (default) or tcp. TCP uses uncompressed NUL-delimited JSON; UDP accepts JSON, gzip/zlib and chunks.".into(),required:false,example:json!("tcp"),default:Some(json!(DEFAULT_TRANSPORT))}
32}
33fn collect_action() -> ActionDefinition {
34 ActionDefinition {name:"collect_gelf_message".into(),description:"Observe one validated GELF message in the bounded access log. No persistence or reply traffic.".into(),parameters:vec![],example:json!({"type":"collect_gelf_message"}),log_template: Some(LogTemplate::new().with_info("GELF message observed"))}
35}
36pub static GELF_MESSAGE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
37 EventType::new("gelf_message","One validated GELF 1.1 message after TCP framing or UDP reassembly/decompression. Missing timestamp resolves to receiver time; missing level resolves to ALERT (1). Unmatched events collect without model calls by default.",collect_action().example)
38.with_parameters(vec![parameter("message","object","Structured host, short_message, optional full_message, timestamp, level, deprecated facility/file/line, and additional_fields with unprefixed names and string/number values.",true),parameter("source_addr","string","Remote IP and port of the GELF emitter",true),parameter("transport","string","Selected GELF transport: udp or tcp",true)])
39.with_actions(vec![collect_action()])
40});
41impl Protocol for GelfProtocol {
42 fn protocol_name(&self) -> &'static str {
43 "GELF"
44 }
45 fn stack_name(&self) -> &'static str {
46 "ETH>IP>UDP|TCP>GELF"
47 }
48 fn keywords(&self) -> Vec<&'static str> {
49 vec!["gelf", "graylog"]
50 }
51 fn description(&self) -> &'static str {
52 "GELF 1.1 UDP/TCP structured log collector; no model calls by default"
53 }
54 fn example_prompt(&self) -> &'static str {
55 "Collect GELF logs on UDP port 12201"
56 }
57 fn group_name(&self) -> &'static str {
58 "Network Services"
59 }
60 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
61 vec![]
62 }
63 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
64 vec![collect_action()]
65 }
66 fn get_event_types(&self) -> Vec<EventType> {
67 vec![GELF_MESSAGE_EVENT.clone()]
68 }
69 fn metadata(&self) -> ProtocolMetadataV2 {
70 ProtocolMetadataV2::builder().request_only("GELF is a one-way log stream; collecting a received message has no reply or unsolicited server-message operation.").deliberately_silent().state(DevelopmentState::Experimental).well_known_udp_port(12201).max_inbound_bytes(MAX_MESSAGE_BYTES)
71 .implementation("Native bounded GELF 1.1 JSON, UDP chunks/gzip/zlib and NUL-delimited TCP; flate2 compression")
72 .llm_control("Explicit static/script/manual/model handlers; llm_fallback=false collects unmatched messages without model calls")
73 .e2e_testing("Codec bounds/negative cases, both transports, lifecycle and independent pygelf emitter / official Graylog go-gelf readers")
74 .notes("256KiB JSON/compressed message, 8192-byte datagrams, 128 chunks within 5s, 128 pending messages/4MiB payload plus bounded metadata/recent IDs, 256 TCP peers, 30s absolute frame deadline. No TLS, HTTP input, authentication, persistence, acknowledgment or retry. No Stable/fuzz/pcap or complete Graylog platform claim. Metadata is not connectionless because TCP sessions and UDP reassembly must remain live.").build()
75 }
76 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
77 vec![
78 transport_parameter(),
79 ParameterDefinition {
80 name: "llm_fallback".into(),
81 type_hint: "boolean".into(),
82 description:
83 "Opt unmatched messages into the model; configured handlers always run.".into(),
84 required: false,
85 example: json!(true),
86 default: Some(json!(DEFAULT_LLM_FALLBACK)),
87 },
88 ]
89 }
90 fn get_startup_examples(&self) -> StartupExamples {
91 StartupExamples::new(
92 json!({"type":"open_server","base_stack":"gelf","port":12201,"startup_params":{"llm_fallback":true},"instruction":"Summarize suspicious messages"}),
93 json!({"type":"open_server","base_stack":"gelf","port":12201,"event_handlers":[{"event_pattern":"gelf_message","handler":{"type":"script","language":"python","code":"import json,sys\njson.load(sys.stdin)\nprint(json.dumps({'actions':[{'type':'collect_gelf_message'}]}))"}}]}),
94 json!({"type":"open_server","base_stack":"gelf","port":12201,"startup_params":{"transport":"tcp"},"event_handlers":[{"event_pattern":"gelf_message","handler":{"type":"static","actions":[{"type":"collect_gelf_message"}]}}]}),
95 )
96 }
97}
98impl Server for GelfProtocol {
99 fn spawn(
100 &self,
101 ctx: SpawnContext,
102 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
103 {
104 Box::pin(super::GelfServer::spawn(ctx))
105 }
106 fn execute_action(&self, action: Value) -> Result<ActionResult> {
107 match action["type"].as_str() {
108 Some("collect_gelf_message") => Ok(ActionResult::NoAction),
109 _ => bail!("unknown GELF collector action"),
110 }
111 }
112}