netget/server/fluent_forward/
actions.rs

1use super::codec::{DEFAULT_LLM_FALLBACK, MAX_FRAME_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
16#[derive(Default)]
17pub struct FluentForwardProtocol;
18impl FluentForwardProtocol {
19    pub fn new() -> Self {
20        Self
21    }
22}
23pub fn parameter(name: &str, type_hint: &str, description: &str, required: bool) -> Parameter {
24    Parameter {
25        name: name.into(),
26        type_hint: type_hint.into(),
27        description: description.into(),
28        required,
29    }
30}
31fn collect_action() -> ActionDefinition {
32    ActionDefinition { name: "accept_forward_batch".into(), description: "Accept a validated Forward batch; send a native correlated ACK when requested. No persistent event store.".into(), parameters: vec![], example: json!({"type":"accept_forward_batch"}), log_template: Some(LogTemplate::new().with_info("Fluent Forward batch accepted")) }
33}
34fn reject_action() -> ActionDefinition {
35    ActionDefinition {
36        name: "reject_forward_batch".into(),
37        description: "Reject this batch by closing the peer without ACK".into(),
38        parameters: vec![],
39        example: json!({"type":"reject_forward_batch"}),
40        log_template: Some(
41            LogTemplate::new().with_info("Fluent Forward batch rejected; close without ACK"),
42        ),
43    }
44}
45pub static FORWARD_BATCH_EVENT: LazyLock<EventType> = LazyLock::new(|| {
46    EventType::new("forward_batch","A complete Message/Forward/PackedForward/gzip PackedForward batch. Each entry has typed timestamp seconds/optional nanoseconds and a JSON record. Opaque chunk IDs stay inside the transport.",collect_action().example)
47 .with_parameters(vec![parameter("tag","string","Fluent event tag",true),parameter("entries","array","Timestamp/record entries",true),parameter("record_count","number","Number of records in this Forward batch",true),parameter("mode","string","message, forward, packed, compressed_packed",true),parameter("ack_requested","boolean","Peer requested correlated acceptance",true),parameter("source_addr","string","Remote IP and TCP port of the Forward emitter",true)])
48 .with_actions(vec![collect_action(),reject_action()])
49});
50impl Protocol for FluentForwardProtocol {
51    fn protocol_name(&self) -> &'static str {
52        "FluentForward"
53    }
54    fn stack_name(&self) -> &'static str {
55        "ETH>IP>TCP>FluentForward"
56    }
57    fn keywords(&self) -> Vec<&'static str> {
58        vec![
59            "fluent-forward",
60            "fluent_forward",
61            "fluentforward",
62            "fluentd",
63        ]
64    }
65    fn description(&self) -> &'static str {
66        "Bounded Fluent Forward MessagePack collector with correlated ACKs; no model calls by default"
67    }
68    fn example_prompt(&self) -> &'static str {
69        "Collect Fluent Forward events on TCP port 24224"
70    }
71    fn group_name(&self) -> &'static str {
72        "Network Services"
73    }
74    fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
75        vec![]
76    }
77    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
78        vec![collect_action(), reject_action()]
79    }
80    fn get_event_types(&self) -> Vec<EventType> {
81        vec![FORWARD_BATCH_EVENT.clone()]
82    }
83    fn metadata(&self) -> ProtocolMetadataV2 {
84        // Forward has no negative-ACK grammar; a failed handler closes without ACK.
85        ProtocolMetadataV2::builder().request_only("Forward acceptance or rejection requires the current inbound batch; ACKs must echo its transport-owned chunk token, so unsolicited replies are not supported.").deliberately_silent().state(DevelopmentState::Experimental).well_known_port(24224)
86            .max_inbound_bytes(MAX_FRAME_BYTES)
87            .implementation("Native bounded MessagePack/EventTime and gzip PackedForward via flate2")
88            .llm_control("Explicit handlers process bounded batches; llm_fallback=false collects unmatched batches without model calls")
89            .e2e_testing("Four carrier modes, correlated ACK and bounds/lifecycle tests; independent fluent-logger and official Fluentd peers")
90            .notes("256KiB frame/decompressed entries; 256 records, nesting depth 32, 16384 MessagePack values, tag 1024 bytes, 256 TCP peers, 30s absolute frame deadline. Typed JSON record values only. No secure-forward handshake, TLS, JSON convenience framing, UDP heartbeat, persistence, automatic retry or deduplication.").build()
91    }
92    fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
93        vec![ParameterDefinition { name: "llm_fallback".into(), type_hint: "boolean".into(), description: "Opt unmatched event batches into model reasoning. False collects without model calls. Configured static/script/manual/LLM handlers always run.".into(), required: false, example: json!(true), default: Some(json!(DEFAULT_LLM_FALLBACK)) }]
94    }
95    fn get_startup_examples(&self) -> StartupExamples {
96        StartupExamples::new(
97            json!({"type":"open_server","base_stack":"fluent-forward","port":24224,"startup_params":{"llm_fallback":true},"instruction":"Summarize suspicious event batches"}),
98            json!({"type":"open_server","base_stack":"fluent-forward","port":24224,"event_handlers":[{"event_pattern":"forward_batch","handler":{"type":"script","language":"python","code":"import json,sys\njson.load(sys.stdin)\nprint(json.dumps({'actions':[{'type':'accept_forward_batch'}]}))"}}]}),
99            json!({"type":"open_server","base_stack":"fluent-forward","port":24224,"event_handlers":[{"event_pattern":"forward_batch","handler":{"type":"static","actions":[{"type":"accept_forward_batch"}]}}]}),
100        )
101    }
102}
103impl Server for FluentForwardProtocol {
104    fn spawn(
105        &self,
106        ctx: SpawnContext,
107    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
108    {
109        Box::pin(super::FluentForwardServer::spawn(ctx))
110    }
111    fn execute_action(&self, action: Value) -> Result<ActionResult> {
112        match action["type"].as_str() {
113            Some("accept_forward_batch") => Ok(ActionResult::NoAction),
114            Some("reject_forward_batch") => Ok(ActionResult::CloseConnection),
115            _ => bail!("unknown FluentForward collector action"),
116        }
117    }
118}
netget/server/fluent_forward/actions.rs

netget/server/fluent_forward/
actions.rs
