netget/client/gelf/
actions.rs

1use crate::llm::actions::{
2    client_trait::{Client, ClientActionResult},
3    protocol_trait::Protocol,
4    ActionDefinition, ParameterDefinition, StartupExamples,
5};
6use crate::protocol::log_template::LogTemplate;
7use crate::protocol::{
8    metadata::{DevelopmentState, ProtocolMetadataV2},
9    ConnectContext, EventType,
10};
11use crate::server::gelf::{
12    actions::{parameter, transport_parameter},
13    codec::{self, Message, DEFAULT_CHUNK_SIZE, DEFAULT_COMPRESSION},
14};
15use crate::state::app_state::AppState;
16use anyhow::{bail, Context, Result};
17use serde_json::{json, Value};
18use std::sync::LazyLock;
19#[derive(Default)]
20pub struct GelfClientProtocol;
21impl GelfClientProtocol {
22    pub fn new() -> Self {
23        Self
24    }
25}
26pub fn send_action() -> ActionDefinition {
27    ActionDefinition {name:"send_gelf_message".into(),description:"Emit one structured GELF 1.1 log. Entire message/compression/chunk plan is validated before writing. Local acceptance does not confirm collector persistence; no automatic retries.".into(),parameters:vec![parameter("message","object","host and short_message required; optional full_message, finite nonnegative timestamp, level 0..7, facility/file/line, additional_fields object with unprefixed names and string/number values. id prohibited. Encoded JSON <=256KiB; UDP requires <=128 chunks.",true)],example:json!({"type":"send_gelf_message","message":{"host":"demo","short_message":"Started","timestamp":1700000000.25,"level":6,"additional_fields":{"service":"api"}}}),log_template: Some(LogTemplate::new().with_info("-> GELF host={message.host} level={message.level} short_message_bytes={message.short_message_len}"))}
28}
29fn disconnect_action() -> ActionDefinition {
30    ActionDefinition {
31        name: "disconnect".into(),
32        description: "Close the GELF transport and command handle".into(),
33        parameters: vec![],
34        example: json!({"type":"disconnect"}),
35        log_template: Some(LogTemplate::new().with_info("-> GELF disconnect")),
36    }
37}
38pub static GELF_CONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
39    EventType::new(
40        "gelf_connected",
41        "GELF emitter transport ready; neither UDP nor TCP has per-message acknowledgment.",
42        send_action().example,
43    )
44    .with_parameters(vec![
45        parameter(
46            "remote_addr",
47            "string",
48            "Remote IP and port of the GELF collector",
49            true,
50        ),
51        parameter(
52            "local_addr",
53            "string",
54            "Local IP and port of this GELF emitter",
55            true,
56        ),
57        parameter(
58            "transport",
59            "string",
60            "Selected GELF transport: udp or tcp",
61            true,
62        ),
63    ])
64    .with_actions(vec![send_action(), disconnect_action()])
65});
66impl Protocol for GelfClientProtocol {
67    fn protocol_name(&self) -> &'static str {
68        "GELF"
69    }
70    fn stack_name(&self) -> &'static str {
71        "ETH>IP>UDP|TCP>GELF"
72    }
73    fn keywords(&self) -> Vec<&'static str> {
74        vec!["gelf", "graylog"]
75    }
76    fn description(&self) -> &'static str {
77        "Emit GELF 1.1 structured logs over UDP or TCP"
78    }
79    fn example_prompt(&self) -> &'static str {
80        "Send a GELF log to localhost:12201"
81    }
82    fn group_name(&self) -> &'static str {
83        "Network Services"
84    }
85    fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
86        vec![send_action(), disconnect_action()]
87    }
88    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
89        vec![]
90    }
91    fn get_event_types(&self) -> Vec<EventType> {
92        vec![GELF_CONNECTED_EVENT.clone()]
93    }
94    fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
95        vec![transport_parameter(),ParameterDefinition{name:"compression".into(),type_hint:"string".into(),description:"auto chooses UDP gzip/TCP none; none, gzip or zlib. Explicit TCP compression is rejected.".into(),required:false,example:json!("zlib"),default:Some(json!(DEFAULT_COMPRESSION))},ParameterDefinition{name:"chunk_size".into(),type_hint:"number".into(),description:"UDP datagram size, including 12-byte chunk header; 13..8192. TCP ignores this validated setting.".into(),required:false,example:json!(1420),default:Some(json!(DEFAULT_CHUNK_SIZE))}]
96    }
97    fn metadata(&self) -> ProtocolMetadataV2 {
98        ProtocolMetadataV2::builder().state(DevelopmentState::Experimental).well_known_udp_port(12201).implementation("Native GELF JSON encoder, bounded UDP compression/chunks and TCP NUL frames").llm_control("Connected event with standard memory updates, typed send action and live command injection").e2e_testing("Both transports and independent official Graylog go-gelf UDP/TCP readers; pygelf emitters in server tests").notes("256KiB message, 128 UDP chunks, 10s connect/write deadline. No TLS, HTTP, acknowledgments, storage or retry; transport success is not end-to-end delivery.").build()
99    }
100    fn get_startup_examples(&self) -> StartupExamples {
101        StartupExamples::new(
102            json!({"type":"open_client","base_stack":"gelf","remote_addr":"localhost:12201","instruction":"Send an informational startup log"}),
103            json!({"type":"open_client","base_stack":"gelf","remote_addr":"localhost:12201","event_handlers":[{"event_pattern":"gelf_connected","handler":{"type":"script","language":"python","code":"import json,sys\njson.load(sys.stdin)\nprint(json.dumps({'actions':[{'type':'send_gelf_message','message':{'host':'demo','short_message':'Started'}}]}))"}}]}),
104            json!({"type":"open_client","base_stack":"gelf","remote_addr":"localhost:12201","startup_params":{"transport":"tcp"},"event_handlers":[{"event_pattern":"gelf_connected","handler":{"type":"static","actions":[send_action().example]}}]}),
105        )
106    }
107}
108impl Client for GelfClientProtocol {
109    fn connect(
110        &self,
111        ctx: ConnectContext,
112    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
113    {
114        Box::pin(super::GelfClient::connect(ctx))
115    }
116    fn execute_action(&self, action: Value) -> Result<ClientActionResult> {
117        match action["type"].as_str() {
118            Some("send_gelf_message") => {
119                let message: Message = serde_json::from_value(
120                    action.get("message").context("missing message")?.clone(),
121                )?;
122                codec::encode_json(&message)?;
123                Ok(ClientActionResult::Custom {
124                    name: "send_gelf_message".into(),
125                    data: serde_json::to_value(message)?,
126                })
127            }
128            Some("disconnect") => Ok(ClientActionResult::Disconnect),
129            _ => bail!("unknown GELF emitter action"),
130        }
131    }
132}
netget/client/gelf/actions.rs

netget/client/gelf/
actions.rs
