netget/client/fluent_forward/
actions.rs

1use crate::llm::actions::{
2    client_trait::{Client, ClientActionResult},
3    protocol_trait::Protocol,
4    ActionDefinition, StartupExamples,
5};
6use crate::protocol::log_template::LogTemplate;
7use crate::protocol::{
8    metadata::{DevelopmentState, ProtocolMetadataV2},
9    ConnectContext, EventType,
10};
11use crate::server::fluent_forward::{
12    actions::parameter,
13    codec::{self, Batch},
14};
15use crate::state::app_state::AppState;
16use anyhow::{bail, Context, Result};
17use serde_json::{json, Value};
18use std::sync::LazyLock;
19
20#[derive(Default)]
21pub struct FluentForwardClientProtocol;
22impl FluentForwardClientProtocol {
23    pub fn new() -> Self {
24        Self
25    }
26}
27pub fn send_action() -> ActionDefinition {
28    ActionDefinition {name:"send_forward_batch".into(),description:"Send a typed Fluent Forward batch. require_ack generates an internal chunk ID and raises forward_ack after correlated receipt. Send outcome confirms local write only. No automatic retry.".into(),parameters:vec![parameter("batch","object","tag; 1..256 entries each timestamp={seconds, optional nanoseconds<1e9}, record=JSON object; mode message(1 entry), forward(default), packed or compressed_packed; require_ack bool. 256KiB encoded/decompressed, depth32/value16384 limits.",true)],example:json!({"type":"send_forward_batch","batch":{"tag":"demo.logs","entries":[{"timestamp":{"seconds":1700000000,"nanoseconds":250000000},"record":{"message":"Started"}}],"mode":"forward","require_ack":true}}),log_template: Some(LogTemplate::new().with_info("-> Fluent Forward tag={batch.tag} entries={batch.entries_len} mode={batch.mode} require_ack={batch.require_ack}"))}
29}
30fn disconnect_action() -> ActionDefinition {
31    ActionDefinition {
32        name: "disconnect".into(),
33        description: "Close the Forward TCP connection and remove its command handle".into(),
34        parameters: vec![],
35        example: json!({"type":"disconnect"}),
36        log_template: Some(LogTemplate::new().with_info("-> Fluent Forward disconnect")),
37    }
38}
39pub static FORWARD_CONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
40    EventType::new(
41        "forward_connected",
42        "TCP Forward transport ready; unauthenticated operation starts without greeting.",
43        send_action().example,
44    )
45    .with_parameters(vec![
46        parameter("remote_addr", "string", "Collector address", true),
47        parameter("local_addr", "string", "Emitter address", true),
48    ])
49    .with_actions(vec![send_action(), disconnect_action()])
50});
51pub static FORWARD_ACK_EVENT: LazyLock<EventType> = LazyLock::new(|| {
52    EventType::new("forward_ack","Collector sent a matching ACK for a pending batch. Correlation token stays inside the transport; receipt is not persistence.",send_action().example).with_parameters(vec![parameter("tag","string","Forward tag of the acknowledged batch",true),parameter("record_count","number","Accepted entries",true)]).with_actions(vec![send_action(),disconnect_action()])
53});
54impl Protocol for FluentForwardClientProtocol {
55    fn protocol_name(&self) -> &'static str {
56        "FluentForward"
57    }
58    fn stack_name(&self) -> &'static str {
59        "ETH>IP>TCP>FluentForward"
60    }
61    fn keywords(&self) -> Vec<&'static str> {
62        vec![
63            "fluent-forward",
64            "fluent_forward",
65            "fluentforward",
66            "fluentd",
67        ]
68    }
69    fn description(&self) -> &'static str {
70        "Emit typed Fluent Forward logs with optional correlated acknowledgments"
71    }
72    fn example_prompt(&self) -> &'static str {
73        "Send a structured Fluent Forward log to localhost:24224"
74    }
75    fn group_name(&self) -> &'static str {
76        "Network Services"
77    }
78    fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
79        vec![send_action(), disconnect_action()]
80    }
81    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
82        vec![]
83    }
84    fn get_event_types(&self) -> Vec<EventType> {
85        vec![FORWARD_CONNECTED_EVENT.clone(), FORWARD_ACK_EVENT.clone()]
86    }
87    fn metadata(&self) -> ProtocolMetadataV2 {
88        ProtocolMetadataV2::builder().state(DevelopmentState::Experimental).well_known_port(24224).implementation("Native bounded MessagePack/EventTime and gzip PackedForward encoder/ACK correlation").llm_control("Connected and correlated ACK events with standard memory, bounded follow-ups and live command injection").e2e_testing("Four event carriers against official Fluentd in_forward; independent fluent-logger emits to collector").notes("256KiB frame; 256 records; depth32/16384 values; 32 pending ACKs/events, 8 follow-up depth, 32 handler actions; 10s connect/write/ACK deadlines. No secure-forward handshake/TLS, UDP heartbeat, JSON framing, persistence or retry. ACK confirms handler acceptance, not durable delivery.").build()
89    }
90    fn get_startup_examples(&self) -> StartupExamples {
91        StartupExamples::new(
92            json!({"type":"open_client","base_stack":"fluent-forward","remote_addr":"localhost:24224","instruction":"Send an informational startup log"}),
93            json!({"type":"open_client","base_stack":"fluent-forward","remote_addr":"localhost:24224","event_handlers":[{"event_pattern":"forward_connected","handler":{"type":"script","language":"python","code":"import json,sys\njson.load(sys.stdin)\nprint(json.dumps({'actions':[{'type':'send_forward_batch','batch':{'tag':'demo.logs','entries':[{'timestamp':{'seconds':1700000000},'record':{'message':'Started'}}]}}]}))"}}]}),
94            json!({"type":"open_client","base_stack":"fluent-forward","remote_addr":"localhost:24224","event_handlers":[{"event_pattern":"forward_connected","handler":{"type":"static","actions":[send_action().example]}}]}),
95        )
96    }
97}
98impl Client for FluentForwardClientProtocol {
99    fn connect(
100        &self,
101        ctx: ConnectContext,
102    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
103    {
104        Box::pin(super::FluentForwardClient::connect(ctx))
105    }
106    fn execute_action(&self, action: Value) -> Result<ClientActionResult> {
107        match action["type"].as_str() {
108            Some("send_forward_batch") => {
109                let batch: Batch =
110                    serde_json::from_value(action.get("batch").context("missing batch")?.clone())?;
111                codec::encode_batch(&batch, None)?;
112                Ok(ClientActionResult::Custom {
113                    name: "send_forward_batch".into(),
114                    data: serde_json::to_value(batch)?,
115                })
116            }
117            Some("disconnect") => Ok(ClientActionResult::Disconnect),
118            _ => bail!("unknown FluentForward emitter action"),
119        }
120    }
121}
netget/client/fluent_forward/actions.rs

netget/client/fluent_forward/
actions.rs
