netget/server/statsd/
actions.rs

1use super::codec::{DEFAULT_DIALECT, DEFAULT_LLM_FALLBACK, MAX_DATAGRAM_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::metadata::{DevelopmentState, ProtocolMetadataV2};
8use crate::protocol::{EventType, SpawnContext};
9use crate::state::app_state::AppState;
10use anyhow::{bail, Result};
11use serde_json::{json, Value};
12use std::sync::LazyLock;
13
14pub struct StatsdProtocol;
15impl Default for StatsdProtocol {
16    fn default() -> Self {
17        Self::new()
18    }
19}
20impl StatsdProtocol {
21    pub fn new() -> Self {
22        Self
23    }
24}
25
26pub fn parameter(name: &str, type_hint: &str, description: &str, required: bool) -> Parameter {
27    Parameter {
28        name: name.into(),
29        type_hint: type_hint.into(),
30        description: description.into(),
31        required,
32    }
33}
34pub fn dialect_parameter() -> ParameterDefinition {
35    ParameterDefinition { name: "dialect".into(), type_hint: "string".into(), description: "dogstatsd accepts StatsD metrics plus histogram/distribution, tags, events and service checks. statsd rejects these extensions.".into(), required: false, example: json!("statsd"), default: Some(json!(DEFAULT_DIALECT)) }
36}
37pub fn collect_action() -> ActionDefinition {
38    ActionDefinition { name: "collect_statsd_batch".into(), description: "Observe this batch in the bounded access log. Does not aggregate, persist, or acknowledge metrics; UDP has no response.".into(), parameters: vec![], example: json!({"type":"collect_statsd_batch"}), log_template: Some(LogTemplate::new().with_info("StatsD batch observed")) }
39}
40pub static STATSD_BATCH_EVENT: LazyLock<EventType> = LazyLock::new(|| {
41    EventType::new("statsd_batch", "One validated UDP datagram containing typed metric/event/service_check records. Defaults to access-log collection without model calls; explicit handlers always run.", json!({"type":"collect_statsd_batch"}))
42        .with_parameters(vec![parameter("records", "array", "Typed records. Metric: kind, name, value (string preserving signed gauge delta), metric_type, optional sample_rate/tags. Event: kind, title, text and optional timestamp/hostname/aggregation_key/priority/source_type/alert_type/tags. Service check: kind, name, status (0..3), optional timestamp/hostname/tags/message.", true), parameter("source_addr", "string", "UDP sender address", true), parameter("dialect", "string", "statsd or dogstatsd", true), parameter("record_count", "number", "Number of records", true)])
43        .with_actions(vec![collect_action()])
44});
45impl Protocol for StatsdProtocol {
46    fn protocol_name(&self) -> &'static str {
47        "StatsD"
48    }
49    fn stack_name(&self) -> &'static str {
50        "ETH>IP>UDP>StatsD"
51    }
52    fn keywords(&self) -> Vec<&'static str> {
53        vec!["statsd", "dogstatsd"]
54    }
55    fn description(&self) -> &'static str {
56        "Bounded UDP StatsD and DogStatsD batch collector; no model calls by default"
57    }
58    fn example_prompt(&self) -> &'static str {
59        "Collect StatsD and DogStatsD metrics on UDP port 8125"
60    }
61    fn group_name(&self) -> &'static str {
62        "Network Services"
63    }
64    fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
65        vec![]
66    }
67    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
68        vec![collect_action()]
69    }
70    fn get_event_types(&self) -> Vec<EventType> {
71        vec![STATSD_BATCH_EVENT.clone()]
72    }
73    fn metadata(&self) -> ProtocolMetadataV2 {
74        ProtocolMetadataV2::builder().connectionless()
75            // StatsD/DogStatsD is one-way UDP: no acknowledgment or negative response exists.
76            .deliberately_silent().state(DevelopmentState::Experimental).well_known_udp_port(8125)
77            .max_inbound_bytes(MAX_DATAGRAM_BYTES)
78            .implementation("Native bounded UTF-8 UDP codec, no additional dependencies")
79            .llm_control("Explicit handlers process entire batches; llm_fallback=false collects unmatched batches without model calls")
80            .e2e_testing("Typed codec vectors and real UDP wire/lifecycle tests; see tests/server/statsd/AGENTS.md for independent implementation evidence")
81            .notes("8 KiB / 256 records per datagram. No aggregation, persistence, replies, origin metadata or packed values. DogStatsD is the default dialect.")
82            .build()
83    }
84    fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
85        vec![dialect_parameter(), ParameterDefinition { name: "llm_fallback".into(), type_hint: "boolean".into(), description: "Opt unmatched batches into LLM reasoning. False records them without model calls. Configured script/static/manual/LLM handlers always run.".into(), required: false, example: json!(true), default: Some(json!(DEFAULT_LLM_FALLBACK)) }]
86    }
87    fn get_startup_examples(&self) -> StartupExamples {
88        StartupExamples::new(
89            json!({"type":"open_server","base_stack":"statsd","port":8125,"startup_params":{"llm_fallback":true},"instruction":"Summarize suspicious metric batches"}),
90            json!({"type":"open_server","base_stack":"statsd","port":8125,"event_handlers":[{"event_pattern":"statsd_batch","handler":{"type":"script","language":"python","code":"import json, sys\nbatch = json.load(sys.stdin)\nprint(json.dumps({'actions': [{'type': 'collect_statsd_batch'}]}))"}}]}),
91            json!({"type":"open_server","base_stack":"statsd","port":8125,"event_handlers":[{"event_pattern":"statsd_batch","handler":{"type":"static","actions":[{"type":"collect_statsd_batch"}]}}]}),
92        )
93    }
94}
95impl Server for StatsdProtocol {
96    fn spawn(
97        &self,
98        ctx: SpawnContext,
99    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
100    {
101        Box::pin(super::StatsdServer::spawn(ctx))
102    }
103    fn execute_action(&self, action: Value) -> Result<ActionResult> {
104        match action["type"].as_str() {
105            Some("collect_statsd_batch") => Ok(ActionResult::NoAction),
106            _ => bail!("unknown StatsD server action"),
107        }
108    }
109}
netget/server/statsd/actions.rs

netget/server/statsd/
actions.rs
