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::influxdb::{
12 actions::parameter,
13 codec::{self, WriteBatch},
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 InfluxDbClientProtocol;
21impl InfluxDbClientProtocol {
22 pub fn new() -> Self {
23 Self
24 }
25}
26pub fn write_action() -> ActionDefinition {
27 ActionDefinition {name:"write_influx_points".into(),description:"Submit one typed InfluxDB v2 batch. Completion reports HTTP status; 204 is handler acceptance, not durable storage. No automatic retries.".into(),parameters:vec![parameter("batch","object","org,bucket,precision ns/us/ms/s(default ns),points(1..256) with measurement,tags,fields typed {type:float/integer/unsigned/boolean/string,value},optional timestamp;gzip(default false)",true)],example:json!({"type":"write_influx_points","batch":{"org":"example","bucket":"metrics","points":[{"measurement":"cpu","tags":{"host":"localhost"},"fields":{"load":{"type":"float","value":0.42}},"timestamp":1700000000000000000i64}]}}),log_template: Some(LogTemplate::new().with_info("Submit typed InfluxDB write batch"))}
28}
29pub const QUERY_RESULT: &str = "influx_query";
31pub fn flux_action() -> ActionDefinition {
32 ActionDefinition {
33 name: "query_influx_flux".into(),
34 description: "Run a Flux query (POST /api/v2/query). The answer arrives as influx_query_response with typed tables: each table's columns (name, Flux datatype, group-key membership) and rows of JSON values, converted from InfluxDB's annotated CSV.".into(),
35 parameters: vec![
36 parameter("org", "string", "Organization name the query runs in", true),
37 parameter("query", "string", "The Flux script, at most 16 KiB, e.g. from(bucket:\"metrics\") |> range(start: -1h)", true),
38 ],
39 example: json!({"type":"query_influx_flux","org":"example","query":"from(bucket: \"metrics\") |> range(start: -1h) |> filter(fn: (r) => r._measurement == \"cpu\")"}),
40 log_template: Some(LogTemplate::new().with_info("Run Flux query in org {org}")),
41 }
42}
43pub fn influxql_action() -> ActionDefinition {
44 ActionDefinition {
45 name: "query_influxql".into(),
46 description: "Run an InfluxQL query (GET /query, the v1-compatible API; on InfluxDB 2 the database is a bucket's DBRP mapping). Several statements may be separated by ';'. The answer arrives as influx_query_response with one entry per statement: its series (name, tags, columns, values) or its error.".into(),
47 parameters: vec![
48 parameter("query", "string", "The InfluxQL text, at most 16 KiB, e.g. SELECT mean(usage) FROM cpu WHERE time > now() - 1h GROUP BY host", true),
49 parameter("db", "string", "Database (on InfluxDB 2, the bucket's DBRP database name)", false),
50 parameter("rp", "string", "Retention policy; absent means the database default", false),
51 parameter("epoch", "string", "Return times as integers in ns, u, ms, s, m or h; absent means RFC 3339 strings", false),
52 ],
53 example: json!({"type":"query_influxql","db":"metrics","query":"SELECT mean(usage) FROM cpu WHERE time > now() - 1h GROUP BY host"}),
54 log_template: Some(LogTemplate::new().with_info("Run InfluxQL query on database {db}")),
55 }
56}
57fn disconnect_action() -> ActionDefinition {
58 ActionDefinition {
59 name: "disconnect".into(),
60 description: "Cancel this HTTP write session, parked handlers and an in-flight exchange"
61 .into(),
62 parameters: vec![],
63 example: json!({"type":"disconnect"}),
64 log_template: Some(
65 LogTemplate::new().with_info("Cancel InfluxDB write session and active exchange"),
66 ),
67 }
68}
69pub static INFLUX_CONNECTED_EVENT: LazyLock<EventType> =
70 LazyLock::new(|| {
71 EventType::new(
72 "influx_connected",
73 "Logical HTTP write session ready; no TCP connection is opened until a validated write.",
74 write_action().example,
75 )
76 .with_parameters(vec![parameter(
77 "remote_addr",
78 "string",
79 "Configured HTTP origin for InfluxDB writes",
80 true,
81 )])
82 .with_actions(vec![write_action(), flux_action(), influxql_action(), disconnect_action()])
83 });
84pub static INFLUX_RESPONSE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
85 EventType::new("influx_write_response","Typed HTTP v2 write outcome. 204 is acceptance; error contains code/message and optional partial counts/source line. Tokens/raw bodies are excluded; retry advice does not trigger retries.",write_action().example).with_parameters(vec![parameter("org","string","Submitted organization",true),parameter("bucket","string","Submitted bucket",true),parameter("point_count","number","Submitted points",true),parameter("status","number","HTTP status returned for the submitted write batch",true),parameter("error","object|null","Typed code,message,optional line/accepted_points/rejected_points",true),parameter("retry_after_seconds","number|null","Numeric retry advice",true)]).with_actions(vec![write_action(),flux_action(),influxql_action(),disconnect_action()])
86});
87pub static INFLUX_QUERY_RESPONSE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
88 EventType::new("influx_query_response","The answer to a Flux or InfluxQL query, read into structured tables (never raw CSV). At most 1000 rows and 64 tables are reported; row_count says how many the answer held.",flux_action().example).with_parameters(vec![parameter("language","string","flux or influxql",true),parameter("query","string","The query as sent",true),parameter("org","string","Flux: the org it ran in",false),parameter("db","string","InfluxQL: the database",false),parameter("status","number","HTTP status of the answer",true),parameter("tables","array","Flux: [{result, table, columns:[{name,type,group}], rows:[[typed values]]}]",false),parameter("statements","array","InfluxQL: [{statement_id, series:[{name,tags,columns,values}]} or {statement_id, error}]",false),parameter("row_count","number","Rows in the whole answer",true),parameter("reported_rows","number","Rows this event carries",true),parameter("truncated","boolean","Rows or tables were left out of this event",true),parameter("error","object|null","Non-2xx answers: {code, message} from the server",true)]).with_actions(vec![write_action(),flux_action(),influxql_action(),disconnect_action()])
89});
90impl Protocol for InfluxDbClientProtocol {
91 fn protocol_name(&self) -> &'static str {
92 "InfluxDB"
93 }
94 fn stack_name(&self) -> &'static str {
95 "ETH>IP>TCP>HTTP>InfluxDB"
96 }
97 fn keywords(&self) -> Vec<&'static str> {
98 vec!["influxdb", "influxdb2", "influx-write", "flux", "influxql"]
99 }
100 fn group_name(&self) -> &'static str {
101 "Network Services"
102 }
103 fn description(&self) -> &'static str {
104 "Typed InfluxDB v2 HTTP write emitter and Flux/InfluxQL query client with explicit partial-write/error outcomes"
105 }
106 fn example_prompt(&self) -> &'static str {
107 "Write a typed CPU metric to InfluxDB at http://localhost:8086"
108 }
109 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
110 vec![]
111 }
112 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
113 vec![
114 write_action(),
115 flux_action(),
116 influxql_action(),
117 disconnect_action(),
118 ]
119 }
120 fn get_event_types(&self) -> Vec<EventType> {
121 vec![
122 INFLUX_CONNECTED_EVENT.clone(),
123 INFLUX_RESPONSE_EVENT.clone(),
124 INFLUX_QUERY_RESPONSE_EVENT.clone(),
125 ]
126 }
127 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
128 vec![ParameterDefinition {
129 name: "auth_token".into(),
130 type_hint: "string".into(),
131 description:
132 "Optional printable ASCII Token credential,<=1024bytes,never exposed to handlers"
133 .into(),
134 required: false,
135 example: json!("collector-secret"),
136 default: None,
137 }]
138 }
139 fn metadata(&self) -> ProtocolMetadataV2 {
140 ProtocolMetadataV2::builder().state(DevelopmentState::Experimental).well_known_port(8086)
141 .implementation("Native typed line protocol/gzip with owned Hyper HTTP/1.1 exchanges; Flux annotated-CSV reader and InfluxQL v1-JSON reader (src/client/influxdb/query.rs)")
142 .llm_control("Connected, write-response and query-response events, ordinary memory and live injection during parked handlers or IO; a query's answer reaches handlers as typed tables and their follow-ups run")
143 .e2e_testing("Official Python1.50.0 emitter; official InfluxDB 2.9.1 daemon: native writes read back, then Flux and InfluxQL queries read into typed tables (doubles, longs, booleans, quoted strings, group keys, multi-statement series, epoch, the daemon's own 400/401 refusals) and a write→query handler chain; MIT line-protocol2.2.1 decoder-backed HTTP receiver; native pair/negative/lifecycle checks")
144 .notes("HTTP origin only; POST /api/v2/write: org, bucket, ns/us/ms/s precision; optional Token auth; identity/gzip. POST /api/v2/query (Flux, annotated CSV back) and GET /query (InfluxQL, v1 JSON back; on InfluxDB 2 through DBRP mappings). Bounds: 256 KiB body, 16 KiB line, 256 points, 64 tags/fields, 1024-byte names/token, 64 KiB write response, 16 KiB query, 4 MiB query answer, 1000 reported rows and 64 reported tables per event (row_count gives the whole), 256 columns, 32 KiB/64 response headers, 128-byte code/1024-byte message, 10s write and 30s query exchange, one in-flight request, 32 queued events/actions, 8 follow-up depth. An unreadable 2xx answer or an in-band Flux error is reported as an error event. Strict single separators; no CR/LF/NUL strings or trailing-backslash names. No TLS, orgID query, redirects, proxy, chunked InfluxQL, schema/persistence/retry, HTTP-date Retry-After, fuzz or pcap. Fresh TCP per request; local logical address0.0.0.0:0.").build()
145 }
146 fn get_startup_examples(&self) -> StartupExamples {
147 StartupExamples::new(
148 json!({"type":"open_client","base_stack":"influxdb","remote_addr":"http://localhost:8086","startup_params":{"auth_token":"collector-secret"},"instruction":"Write one CPU load point"}),
149 json!({"type":"open_client","base_stack":"influxdb","remote_addr":"http://localhost:8086","event_handlers":[{"event_pattern":"influx_connected","handler":{"type":"script","language":"python","code":"import json,sys\njson.load(sys.stdin)\nprint(json.dumps({'actions':[{'type':'disconnect'}]}))"}}]}),
150 json!({"type":"open_client","base_stack":"influxdb","remote_addr":"http://localhost:8086","event_handlers":[{"event_pattern":"influx_connected","handler":{"type":"static","actions":[write_action().example]}}]}),
151 )
152 }
153}
154impl Client for InfluxDbClientProtocol {
155 fn connect(
156 &self,
157 ctx: ConnectContext,
158 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
159 {
160 Box::pin(super::InfluxDbClient::connect(ctx))
161 }
162 fn execute_action(&self, action: Value) -> Result<ClientActionResult> {
163 match action["type"].as_str() {
164 Some("write_influx_points") => {
165 let batch: WriteBatch =
166 serde_json::from_value(action.get("batch").context("batch required")?.clone())?;
167 codec::encode_batch(&batch)?;
168 Ok(ClientActionResult::Custom {
169 name: "write_influx_points".into(),
170 data: serde_json::to_value(batch)?,
171 })
172 }
173 Some(kind @ ("query_influx_flux" | "query_influxql")) => {
174 let spec = super::query::QuerySpec {
175 language: if kind == "query_influx_flux" {
176 super::query::Language::Flux
177 } else {
178 super::query::Language::Influxql
179 },
180 query: action["query"].as_str().context("query required")?.into(),
181 org: action["org"].as_str().map(String::from),
182 db: action["db"].as_str().map(String::from),
183 rp: action["rp"].as_str().map(String::from),
184 epoch: action["epoch"].as_str().map(String::from),
185 };
186 spec.validate()?;
187 Ok(ClientActionResult::Custom {
188 name: QUERY_RESULT.into(),
189 data: serde_json::to_value(spec)?,
190 })
191 }
192 Some("disconnect") => Ok(ClientActionResult::Disconnect),
193 _ => bail!("unknown InfluxDB emitter action"),
194 }
195 }
196}