1use super::codec::{DEFAULT_LLM_FALLBACK, DEFAULT_PATH, MAX_BODY_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, ensure, Context, Result};
13use serde::{Deserialize, Serialize};
14use serde_json::{json, Value};
15use std::sync::LazyLock;
16
17#[derive(Clone, Debug, Serialize, Deserialize)]
18#[serde(tag = "mode", rename_all = "snake_case")]
19pub enum Decision {
20 Accept,
21 Reject {
22 status: u16,
23 message: String,
24 retry_after_seconds: Option<u16>,
25 },
26}
27pub fn parameter(name: &str, type_hint: &str, description: &str, required: bool) -> Parameter {
28 Parameter {
29 name: name.into(),
30 type_hint: type_hint.into(),
31 description: description.into(),
32 required,
33 }
34}
35fn accept() -> ActionDefinition {
36 ActionDefinition { name:"accept_remote_write_samples".into(), description:"Accept all validated v1 float samples with204; handler observation only, no durable TSDB or protocol storage.".into(), parameters:vec![], example:json!({"type":"accept_remote_write_samples"}), log_template:Some(LogTemplate::new().with_info("Remote write action {type} queued")) }
37}
38fn reject() -> ActionDefinition {
39 ActionDefinition { name:"reject_remote_write_samples".into(), description:"Reject the whole batch:400 invalid/non-retryable;429 optional backoff;500/503 retryable. Common-action failure overrides acceptance.".into(), parameters:vec![parameter("status","number","400,429,500 or503",true),parameter("message","string","Nonempty UTF-8 text<=4096 bytes",true),parameter("retry_after_seconds","number","Optional1..3600 on429/503",false)], example:json!({"type":"reject_remote_write_samples","status":503,"message":"Temporarily unavailable","retry_after_seconds":1}), log_template:Some(LogTemplate::new().with_info("Remote write action {type} queued")) }
40}
41pub static REMOTE_WRITE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
42 EventType::new("remote_write_request","Validated published remote write1.0 float series; labels, signed millisecond timestamps, finite values or nan/+inf/-inf/stale. Unknown optional fields counted and discarded; no credentials/raw body.", accept().example).with_parameters(vec![parameter("series","array","[{labels:{name:value},samples:[{timestamp_ms:i64,value:number|nan|+inf|-inf|stale}]}]",true),parameter("series_count","number","Series in this request",true),parameter("sample_count","number","Float samples in this request",true),parameter("ignored_fields","number","Unknown/reserved fields discarded, not ingested",true),parameter("version","string","Published remote-write wire version 1.0",true),parameter("source_addr","string","HTTP sender socket address",true),parameter("authenticated","boolean","Token verified or anonymous",true),parameter("auth_required","boolean","Bearer token configured",true),parameter("durable_storage","boolean","Always false; common handler observation only",true)]).with_actions(vec![accept(),reject()])
43});
44#[derive(Default)]
45pub struct PrometheusRemoteWriteProtocol;
46impl PrometheusRemoteWriteProtocol {
47 pub fn new() -> Self {
48 Self
49 }
50}
51impl Protocol for PrometheusRemoteWriteProtocol {
52 fn protocol_name(&self) -> &'static str {
53 "PrometheusRemoteWrite"
54 }
55 fn stack_name(&self) -> &'static str {
56 "ETH>IP>TCP>HTTP>PrometheusRemoteWrite"
57 }
58 fn keywords(&self) -> Vec<&'static str> {
59 vec![
60 "prometheus-remote-write",
61 "remote-write",
62 "prometheus-write",
63 ]
64 }
65 fn group_name(&self) -> &'static str {
66 "Network Services"
67 }
68 fn description(&self) -> &'static str {
69 "Published remote write1.0 typed protobuf/Snappy float-sample collector"
70 }
71 fn example_prompt(&self) -> &'static str {
72 "Collect remote write samples on localhost9090"
73 }
74 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
75 vec![accept(), reject()]
76 }
77 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
78 vec![]
79 }
80 fn get_event_types(&self) -> Vec<EventType> {
81 vec![REMOTE_WRITE_EVENT.clone()]
82 }
83 fn metadata(&self) -> ProtocolMetadataV2 {
84 ProtocolMetadataV2::builder().answers_on_failure().request_only("HTTP write request/response; no unsolicited messages").state(DevelopmentState::Experimental).well_known_port(9090).max_inbound_bytes(MAX_BODY_BYTES)
85 .implementation("Native bounded published1.0 prometheus.WriteRequest protobuf and Snappy block codec, owned Hyper HTTP/1.1 tasks")
86 .llm_control("Typed float samples and exact stale-marker meaning through common handlers/shared memory; default no-model observation, optional llm_fallback")
87 .e2e_testing("Required pinned official Prometheus3.15.0 sender/receiver services, literal dual wire oracle and native bounds/error/lifecycle checks")
88 .notes("Selected published1.0 float-sample scope, not a full monitoring agent/TSDB. POST configured path(default/api/v1/write), required0.1.0 version/User-Agent/protobuf/Snappy headers; optional Bearer token.256KiB wire/decoded body,128series/2048samples/32labels,128byte legacy ASCII names/2048byte nonempty UTF-8 values,32768protobuf fields,256connections,32KiB/64headers,30s header/body,10s response write. Sorted unique labels; per-request sample ordering; duplicate series rejected; empty whole negotiation request accepted. Reserved/unknown optional fields discarded with count; exemplars/native histograms rejected. Successful empty/common-only handler accepts; failed actions/multiple decisions503.204means handler observation only, not durable persistence or deduplication. No TLS/basic/cloud auth,2.0,metadata ingestion,queries,protocol store,cross-request timestamp enforcement,fuzz orpcap.").build()
89 }
90 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
91 vec![
92 ParameterDefinition {
93 name: "auth_token".into(),
94 type_hint: "string".into(),
95 description: "Optional Bearer token<=1024 printable ASCII; excluded from handlers"
96 .into(),
97 required: false,
98 example: json!("collector-secret"),
99 default: None,
100 },
101 ParameterDefinition {
102 name: "path".into(),
103 type_hint: "string".into(),
104 description: "Absolute HTTP write path<=1024 ASCII bytes, no query/fragment".into(),
105 required: false,
106 example: json!("/receive"),
107 default: Some(json!(DEFAULT_PATH)),
108 },
109 ParameterDefinition {
110 name: "llm_fallback".into(),
111 type_hint: "boolean".into(),
112 description: "Opt unmatched writes into model calls; explicit handlers always run"
113 .into(),
114 required: false,
115 example: json!(true),
116 default: Some(json!(DEFAULT_LLM_FALLBACK)),
117 },
118 ]
119 }
120 fn get_startup_examples(&self) -> StartupExamples {
121 StartupExamples::new(
122 json!({"type":"open_server","base_stack":"prometheus-remote-write","port":9090,"startup_params":{"llm_fallback":true},"instruction":"Decide which samples to accept"}),
123 json!({"type":"open_server","base_stack":"prometheus-remote-write","port":9090,"event_handlers":[{"event_pattern":"remote_write_request","handler":{"type":"script","language":"python","code":"import json,sys\njson.load(sys.stdin)\nprint(json.dumps({'actions':[{'type':'accept_remote_write_samples'}]}))"}}]}),
124 json!({"type":"open_server","base_stack":"prometheus-remote-write","port":9090,"event_handlers":[{"event_pattern":"remote_write_request","handler":{"type":"static","actions":[accept().example]}}]}),
125 )
126 }
127}
128impl Server for PrometheusRemoteWriteProtocol {
129 fn spawn(
130 &self,
131 ctx: SpawnContext,
132 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
133 {
134 Box::pin(super::PrometheusRemoteWriteServer::spawn(ctx))
135 }
136 fn execute_action(&self, action: Value) -> Result<ActionResult> {
137 let decision = match action["type"].as_str() {
138 Some("accept_remote_write_samples") => Decision::Accept,
139 Some("reject_remote_write_samples") => {
140 let status = u16::try_from(action["status"].as_u64().context("status required")?)?;
141 ensure!(
142 [400, 429, 500, 503].contains(&status),
143 "unsupported rejection status"
144 );
145 let message = action["message"].as_str().context("message required")?;
146 ensure!(
147 !message.is_empty() && message.len() <= 4096,
148 "rejection message byte limit"
149 );
150 let retry = action
151 .get("retry_after_seconds")
152 .map(|v| {
153 u16::try_from(v.as_u64().context("integer retry advice required")?)
154 .map_err(anyhow::Error::from)
155 })
156 .transpose()?;
157 if let Some(n) = retry {
158 ensure!(
159 (1..=3600).contains(&n) && [429, 503].contains(&status),
160 "invalid retry advice range/status"
161 );
162 }
163 Decision::Reject {
164 status,
165 message: message.into(),
166 retry_after_seconds: retry,
167 }
168 }
169 _ => bail!("unknown remote write collector action"),
170 };
171 Ok(ActionResult::Custom {
172 name: "remote_write_decision".into(),
173 data: serde_json::to_value(decision)?,
174 })
175 }
176}