netget/server/graphite/
actions.rs

1use super::codec::{DEFAULT_LLM_FALLBACK, MAX_LINE_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 GraphiteProtocol;
18impl GraphiteProtocol {
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: "collect_graphite_batch".into(), description: "Observe validated Carbon metrics in the bounded access log. No time-series database, aggregation or acknowledgment.".into(), parameters: vec![], example: json!({"type":"collect_graphite_batch"}), log_template: Some(LogTemplate::new().with_info("Graphite Carbon batch observed")) }
33}
34pub static GRAPHITE_BATCH_EVENT: LazyLock<EventType> = LazyLock::new(|| {
35    EventType::new("graphite_batch", "A bounded batch of complete Carbon plaintext lines. TCP read boundaries determine batching; sender action boundaries are not preserved. Unmatched events collect without model calls by default.", collect_action().example)
36        .with_parameters(vec![parameter("metrics", "array", "Objects with path (UTF-8 string), value (finite number), timestamp (UNIX seconds). Incoming -1 timestamps resolve to receiver time.", true), parameter("record_count", "number", "Number of metrics", true), parameter("source_addr", "string", "TCP peer address", true)])
37        .with_actions(vec![collect_action()])
38});
39impl Protocol for GraphiteProtocol {
40    fn protocol_name(&self) -> &'static str {
41        "Graphite"
42    }
43    fn stack_name(&self) -> &'static str {
44        "ETH>IP>TCP>Graphite"
45    }
46    fn keywords(&self) -> Vec<&'static str> {
47        vec!["graphite", "carbon"]
48    }
49    fn description(&self) -> &'static str {
50        "Bounded Carbon plaintext TCP metric collector; no model calls by default"
51    }
52    fn example_prompt(&self) -> &'static str {
53        "Collect Graphite Carbon plaintext metrics on TCP port 2003"
54    }
55    fn group_name(&self) -> &'static str {
56        "Network Services"
57    }
58    fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
59        vec![]
60    }
61    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
62        vec![collect_action()]
63    }
64    fn get_event_types(&self) -> Vec<EventType> {
65        vec![GRAPHITE_BATCH_EVENT.clone()]
66    }
67    fn metadata(&self) -> ProtocolMetadataV2 {
68        ProtocolMetadataV2::builder().request_only("Carbon plaintext collectors never send application replies or unsolicited peer messages").deliberately_silent().state(DevelopmentState::Experimental).well_known_port(2003)
69            .max_inbound_bytes(MAX_LINE_BYTES)
70            .request_only("Carbon plaintext is a one-way metric stream; the collector has no replies or unsolicited server messages")
71            .implementation("Native bounded Carbon plaintext TCP codec, no added dependencies")
72            .llm_control("Explicit handlers process bounded batches; llm_fallback=false collects unmatched batches without model calls")
73            .e2e_testing("Structured codec, fragmented/coalesced stream and lifecycle tests; independent Graphyte and official Carbon peers")
74            .notes("TCP only; 4096 bytes per line, 256 metrics per dispatch, 256 connections, 30s frame read deadline excluding handler time. No Pickle, UDP, Whisper storage, query API, TLS or authentication. Tags in metric paths pass through without interpretation.").build()
75    }
76    fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
77        vec![ParameterDefinition { name: "llm_fallback".into(), type_hint: "boolean".into(), description: "Opt unmatched metric 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)) }]
78    }
79    fn get_startup_examples(&self) -> StartupExamples {
80        StartupExamples::new(
81            json!({"type":"open_server","base_stack":"graphite","port":2003,"startup_params":{"llm_fallback":true},"instruction":"Summarize suspicious metric batches"}),
82            json!({"type":"open_server","base_stack":"graphite","port":2003,"event_handlers":[{"event_pattern":"graphite_batch","handler":{"type":"script","language":"python","code":"import json,sys\njson.load(sys.stdin)\nprint(json.dumps({'actions':[{'type':'collect_graphite_batch'}]}))"}}]}),
83            json!({"type":"open_server","base_stack":"graphite","port":2003,"event_handlers":[{"event_pattern":"graphite_batch","handler":{"type":"static","actions":[{"type":"collect_graphite_batch"}]}}]}),
84        )
85    }
86}
87impl Server for GraphiteProtocol {
88    fn spawn(
89        &self,
90        ctx: SpawnContext,
91    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
92    {
93        Box::pin(super::GraphiteServer::spawn(ctx))
94    }
95    fn execute_action(&self, action: Value) -> Result<ActionResult> {
96        match action["type"].as_str() {
97            Some("collect_graphite_batch") => Ok(ActionResult::NoAction),
98            _ => bail!("unknown Graphite collector action"),
99        }
100    }
101}
netget/server/graphite/actions.rs

netget/server/graphite/
actions.rs
