netget/client/prometheus_remote_write/
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::{
8    metadata::{DevelopmentState, ProtocolMetadataV2},
9    ConnectContext, EventType,
10};
11use crate::server::prometheus_remote_write::{
12    actions::parameter,
13    codec::{self, WriteBatch, DEFAULT_PATH, DEFAULT_RETRY_429},
14};
15use crate::state::app_state::AppState;
16use anyhow::{bail, Context, Result};
17use serde_json::{json, Value};
18use std::sync::LazyLock;
19#[derive(Default)]
20pub struct PrometheusRemoteWriteClientProtocol;
21impl PrometheusRemoteWriteClientProtocol {
22    pub fn new() -> Self {
23        Self
24    }
25}
26pub fn write_action() -> ActionDefinition {
27    ActionDefinition { name:"write_remote_samples".into(), description:"Submit one typed published1.0 float-sample batch. Retry5xx/transport failures with backoff while connected;429 opt-in. Caller must preserve per-series order across batches and supply stale markers when appropriate.".into(), parameters:vec![parameter("batch","object","series[{labels:{name:nonempty UTF-8 value},samples:[{timestamp_ms:i64,value:finite number|nan|+inf|-inf|stale}]}]; empty series array is a v1 probe",true)], example:json!({"type":"write_remote_samples","batch":{"series":[{"labels":{"__name__":"example_temperature","site":"one"},"samples":[{"timestamp_ms":1700000000000i64,"value":21.5}]}]}}), log_template:Some(LogTemplate::new().with_info("Remote write action {type} queued")) }
28}
29fn disconnect() -> ActionDefinition {
30    ActionDefinition { name:"disconnect".into(),description:"Cancel logical session, pending exchange/backoff and parked handler; unsent volatile samples are lost".into(),parameters:vec![],example:json!({"type":"disconnect"}),log_template:Some(LogTemplate::new().with_info("Remote write action {type} queued")) }
31}
32pub static REMOTE_WRITE_CONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
33    EventType::new(
34        "remote_write_connected",
35        "Logical HTTP origin ready; no TCP until a validated write",
36        write_action().example,
37    )
38    .with_parameters(vec![parameter(
39        "remote_addr",
40        "string",
41        "Collector HTTP origin address",
42        true,
43    )])
44    .with_actions(vec![write_action(), disconnect()])
45});
46pub static REMOTE_WRITE_RESPONSE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
47    EventType::new("remote_write_response","Final HTTP outcome: any2xx accepted;4xx other than opted-in429 terminal.5xx retry until acceptance/cancellation. Response body ignored. No durability/exactly-once promise.",write_action().example).with_parameters(vec![parameter("series_count","number","Submitted series",true),parameter("sample_count","number","Submitted float samples",true),parameter("status","number","Terminal HTTP status",true),parameter("accepted","boolean","True for any terminal 2xx response; durability is unconfirmed",true),parameter("attempts","number","HTTP attempts including retries",true),parameter("durable_storage_confirmed","boolean","Always false; HTTP acceptance alone is insufficient",true)]).with_actions(vec![write_action(),disconnect()])
48});
49impl Protocol for PrometheusRemoteWriteClientProtocol {
50    fn protocol_name(&self) -> &'static str {
51        "PrometheusRemoteWrite"
52    }
53    fn stack_name(&self) -> &'static str {
54        "ETH>IP>TCP>HTTP>PrometheusRemoteWrite"
55    }
56    fn keywords(&self) -> Vec<&'static str> {
57        vec![
58            "prometheus-remote-write",
59            "remote-write",
60            "prometheus-write",
61        ]
62    }
63    fn group_name(&self) -> &'static str {
64        "Network Services"
65    }
66    fn description(&self) -> &'static str {
67        "Typed published remote write1.0 float-sample emitter with cancellable retry backoff"
68    }
69    fn example_prompt(&self) -> &'static str {
70        "Send remote write float samples to localhost9090"
71    }
72    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
73        vec![]
74    }
75    fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
76        vec![write_action(), disconnect()]
77    }
78    fn get_event_types(&self) -> Vec<EventType> {
79        vec![
80            REMOTE_WRITE_CONNECTED_EVENT.clone(),
81            REMOTE_WRITE_RESPONSE_EVENT.clone(),
82        ]
83    }
84    fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
85        vec![
86            ParameterDefinition {
87                name: "auth_token".into(),
88                type_hint: "string".into(),
89                description: "Optional Bearer token<=1024 printable ASCII; excluded from handlers"
90                    .into(),
91                required: false,
92                example: json!("collector-secret"),
93                default: None,
94            },
95            ParameterDefinition {
96                name: "path".into(),
97                type_hint: "string".into(),
98                description: "Absolute HTTP write path<=1024 bytes, no query/fragment".into(),
99                required: false,
100                example: json!("/receive"),
101                default: Some(json!(DEFAULT_PATH)),
102            },
103            ParameterDefinition {
104                name: "retry_429".into(),
105                type_hint: "boolean".into(),
106                description: "Retry429 like5xx while connected; otherwise429 is terminal".into(),
107                required: false,
108                example: json!(true),
109                default: Some(json!(DEFAULT_RETRY_429)),
110            },
111        ]
112    }
113    fn metadata(&self) -> ProtocolMetadataV2 {
114        ProtocolMetadataV2::builder().state(DevelopmentState::Experimental).well_known_port(9090).implementation("Native bounded public1.0 protobuf/Snappy block encoding; owned Hyper HTTP/1.1 exchange and exponential backoff")
115            .llm_control("Connected/typed terminal-response events and common shared memory; injection/disconnect independent of parked handlers and retry IO")
116            .e2e_testing("Required pinned official Prometheus3.15.0 TSDB readback; actual independent Prometheus sender; literal codec, retry/errors and owned cancellation checks")
117            .notes("Published1.0 float-sample selected scope, not full Prometheus agent/conformance. Cleartext HTTP origin and configurable path(default/api/v1/write), optional Bearer; required headers fixed, no custom override.256KiB wire/decoded body,128series/2048samples/32labels,128byte legacy ASCII names/2048byte nonempty UTF-8 values. One in-flight batch,32queued events/actions,followup depth8,64KiB ignored response body,64/32KiB headers,10s each exchange.5xx and IO/framing failures retry indefinitely while connected with100ms..5s exponential backoff;429 optional, other HTTP statuses terminal,any2xx accepted,redirects not followed,Retry-After ignored. Identical retry body can duplicate remotely. Per-batch sample order validated; caller owns cross-batch order/stale lifecycle. Cancellation drops volatile samples; no WAL,dedup,storage,scraping/discovery,automatic stale detection,metadata/exemplars/native histograms,2.0,TLS/basic/cloud auth,queries,fuzz orpcap. Logical local address0.0.0.0:0;freshTCPperattempt.").build()
118    }
119    fn get_startup_examples(&self) -> StartupExamples {
120        StartupExamples::new(
121            json!({"type":"open_client","base_stack":"prometheus-remote-write","remote_addr":"http://localhost:9090","instruction":"Send one float batch"}),
122            json!({"type":"open_client","base_stack":"prometheus-remote-write","remote_addr":"http://localhost:9090","event_handlers":[{"event_pattern":"remote_write_connected","handler":{"type":"script","language":"python","code":"import json,sys\njson.load(sys.stdin)\nprint(json.dumps({'actions':[{'type':'disconnect'}]}))"}}]}),
123            json!({"type":"open_client","base_stack":"prometheus-remote-write","remote_addr":"http://localhost:9090","event_handlers":[{"event_pattern":"remote_write_connected","handler":{"type":"static","actions":[write_action().example]}}]}),
124        )
125    }
126}
127impl Client for PrometheusRemoteWriteClientProtocol {
128    fn connect(
129        &self,
130        ctx: ConnectContext,
131    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
132    {
133        Box::pin(super::PrometheusRemoteWriteClient::connect(ctx))
134    }
135    fn execute_action(&self, action: Value) -> Result<ClientActionResult> {
136        match action["type"].as_str() {
137            Some("write_remote_samples") => {
138                let batch: WriteBatch =
139                    serde_json::from_value(action.get("batch").context("batch required")?.clone())?;
140                codec::encode_batch(&batch)?;
141                Ok(ClientActionResult::Custom {
142                    name: "write_remote_samples".into(),
143                    data: serde_json::to_value(batch)?,
144                })
145            }
146            Some("disconnect") => Ok(ClientActionResult::Disconnect),
147            _ => bail!("unknown remote write emitter action"),
148        }
149    }
150}
netget/client/prometheus_remote_write/actions.rs

netget/client/prometheus_remote_write/
actions.rs
