netget/client/statsd/
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::metadata::{DevelopmentState, ProtocolMetadataV2};
8use crate::protocol::{ConnectContext, EventType};
9use crate::server::statsd::{
10    actions::{dialect_parameter, parameter},
11    codec::{self, Dialect, Record},
12};
13use crate::state::app_state::AppState;
14use anyhow::{bail, Context, Result};
15use serde_json::{json, Value};
16use std::sync::LazyLock;
17
18pub struct StatsdClientProtocol;
19impl Default for StatsdClientProtocol {
20    fn default() -> Self {
21        Self::new()
22    }
23}
24impl StatsdClientProtocol {
25    pub fn new() -> Self {
26        Self
27    }
28}
29pub fn send_action() -> ActionDefinition {
30    ActionDefinition { name:"send_statsd_batch".into(), description:"Encode and emit 1..256 structured records in one UDP datagram, capped at 8192 bytes. No acknowledgment or delivery guarantee. A signed gauge value such as +2 is a delta; -2 also means a delta, not an absolute negative gauge.".into(), parameters:vec![parameter("records", "array", "Objects tagged kind: metric (name, value string, metric_type c/g/ms/s/h/d, optional sample_rate number (DogStatsD 0..1, classic StatsD >0..1) and tags string array); event (title/text, optional timestamp/hostname/aggregation_key/priority normal|low/source_type/alert_type error|warning|info|success/tags); service_check (name/status 0..3, optional timestamp/hostname/tags/message). DogStatsD extensions require dogstatsd dialect. No packed metric values or origin metadata.", true)], example:json!({"type":"send_statsd_batch","records":[{"kind":"metric","name":"requests","value":"1","metric_type":"c","tags":["env:test"]}]}), log_template: Some(LogTemplate::new().with_info("-> StatsD records={records_len}")) }
31}
32fn disconnect_action() -> ActionDefinition {
33    ActionDefinition {
34        name: "disconnect".into(),
35        description: "Close the emitter socket and remove its command handle".into(),
36        parameters: vec![],
37        example: json!({"type":"disconnect"}),
38        log_template: Some(LogTemplate::new().with_info("-> StatsD disconnect")),
39    }
40}
41pub static STATSD_CONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
42    EventType::new(
43        "statsd_connected",
44        "UDP emitter ready. No handshake or confirmation that a collector exists.",
45        send_action().example,
46    )
47    .with_parameters(vec![
48        parameter("remote_addr", "string", "Collector address", true),
49        parameter("local_addr", "string", "Emitter address", true),
50        parameter("dialect", "string", "statsd or dogstatsd", true),
51    ])
52    .with_actions(vec![send_action(), disconnect_action()])
53});
54
55impl Protocol for StatsdClientProtocol {
56    fn protocol_name(&self) -> &'static str {
57        "StatsD"
58    }
59    fn stack_name(&self) -> &'static str {
60        "ETH>IP>UDP>StatsD"
61    }
62    fn keywords(&self) -> Vec<&'static str> {
63        vec!["statsd", "dogstatsd"]
64    }
65    fn description(&self) -> &'static str {
66        "Emit structured StatsD or DogStatsD metric/event/service-check batches over UDP"
67    }
68    fn example_prompt(&self) -> &'static str {
69        "Send a requests counter with value 1 to StatsD at localhost:8125"
70    }
71    fn group_name(&self) -> &'static str {
72        "Network Services"
73    }
74    fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
75        vec![send_action(), disconnect_action()]
76    }
77    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
78        vec![]
79    }
80    fn get_event_types(&self) -> Vec<EventType> {
81        vec![STATSD_CONNECTED_EVENT.clone()]
82    }
83    fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
84        vec![dialect_parameter()]
85    }
86    fn metadata(&self) -> ProtocolMetadataV2 {
87        ProtocolMetadataV2::builder().state(DevelopmentState::Experimental).well_known_udp_port(8125)
88            .implementation("Native UTF-8 UDP codec shared with collector; no additional dependencies")
89            .llm_control("One connected event; structured batch actions also available via command injection")
90            .e2e_testing("Exact UDP wire fixtures, handler and lifecycle tests; independent receiver evidence in tests/client/statsd/AGENTS.md")
91            .notes("8 KiB / 256 records per datagram. No delivery acknowledgment, retries, TCP, Unix sockets, aggregation or sampling decisions; sample_rate is metadata. Default dialect dogstatsd.").build()
92    }
93    fn get_startup_examples(&self) -> StartupExamples {
94        let action = json!({"type":"send_statsd_batch","records":[{"kind":"metric","name":"requests","value":"1","metric_type":"c"}]});
95        StartupExamples::new(
96            json!({"type":"open_client","base_stack":"statsd","remote_addr":"localhost:8125","instruction":"Send one requests counter value 1"}),
97            json!({"type":"open_client","base_stack":"statsd","remote_addr":"localhost:8125","event_handlers":[{"event_pattern":"statsd_connected","handler":{"type":"script","language":"python","code":"import json, sys\njson.load(sys.stdin)\nprint(json.dumps({'actions': [{'type': 'send_statsd_batch', 'records': [{'kind': 'metric', 'name': 'requests', 'value': '1', 'metric_type': 'c'}]}]}))"}}]}),
98            json!({"type":"open_client","base_stack":"statsd","remote_addr":"localhost:8125","event_handlers":[{"event_pattern":"statsd_connected","handler":{"type":"static","actions":[action]}}]}),
99        )
100    }
101}
102impl Client for StatsdClientProtocol {
103    fn connect(
104        &self,
105        ctx: ConnectContext,
106    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
107    {
108        Box::pin(super::StatsdClient::connect(ctx))
109    }
110    fn execute_action(&self, action: Value) -> Result<ClientActionResult> {
111        match action["type"].as_str() {
112            Some("send_statsd_batch") => {
113                let records: Vec<Record> = serde_json::from_value(
114                    action.get("records").context("missing records")?.clone(),
115                )
116                .context("invalid structured records")?;
117                codec::encode_datagram(&records, Dialect::Dogstatsd)?;
118                Ok(ClientActionResult::Custom {
119                    name: "send_statsd_batch".into(),
120                    data: serde_json::to_value(records)?,
121                })
122            }
123            Some("disconnect") => Ok(ClientActionResult::Disconnect),
124            _ => bail!("unknown StatsD client action"),
125        }
126    }
127}
netget/client/statsd/actions.rs

netget/client/statsd/
actions.rs
