netget/server/influxdb/
mod.rs

1//! One-request-per-connection HTTP/1.1 InfluxDB endpoint with owned tasks: v2 writes, Flux
2//! queries at `/api/v2/query`, InfluxQL at the v1-compatible `/query`, and `/ping`.
3pub mod actions;
4pub mod codec;
5pub mod query;
6use crate::protocol::{Event, SpawnContext};
7use crate::server::{
8    accept_bounded::{accept_bounded, ConnectionLimiter, DEFAULT_MAX_CONNECTIONS},
9    connection::ConnectionId,
10};
11use crate::state::{
12    server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo},
13    AccessLogOwner,
14};
15use crate::{console_error, console_info};
16use anyhow::{ensure, Context, Result};
17use bytes::Bytes;
18use http_body_util::{BodyExt, Full, Limited};
19use hyper::{
20    body::Incoming,
21    header::{HeaderValue, ALLOW, AUTHORIZATION, CONTENT_ENCODING, CONTENT_TYPE, RETRY_AFTER},
22    server::conn::http1,
23    service::service_fn,
24    Method, Request, Response, StatusCode,
25};
26use hyper_util::rt::{TokioIo, TokioTimer};
27use serde_json::json;
28use std::{collections::BTreeMap, convert::Infallible, net::SocketAddr, sync::Arc, time::Duration};
29use tokio::sync::Notify;
30
31pub const HEADER_TIMEOUT: Duration = Duration::from_secs(30);
32pub const BODY_TIMEOUT: Duration = Duration::from_secs(30);
33pub const WRITE_TIMEOUT: Duration = Duration::from_secs(10);
34pub const MAX_HEADER_BYTES: usize = 32 * 1024;
35pub const MAX_HEADERS: usize = 64;
36const CAP_REFUSAL:&[u8]=b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\nRetry-After: 5\r\n\r\n";
37#[derive(Clone)]
38struct Config {
39    token: Option<String>,
40    llm_fallback: bool,
41}
42pub struct InfluxDbServer;
43impl InfluxDbServer {
44    pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
45        let token = ctx
46            .startup_params
47            .as_ref()
48            .map(|p| p.get_optional_string("auth_token"))
49            .transpose()?
50            .flatten();
51        if let Some(token) = &token {
52            codec::validate_token(token)?;
53        }
54        let llm_fallback = ctx
55            .startup_params
56            .as_ref()
57            .map(|p| p.get_optional_bool("llm_fallback"))
58            .transpose()?
59            .flatten()
60            .unwrap_or(codec::DEFAULT_LLM_FALLBACK);
61        let config = Arc::new(Config {
62            token,
63            llm_fallback,
64        });
65        let listener = tokio::net::TcpListener::bind(ctx.legacy_listen_addr()).await?;
66        let local = listener.local_addr()?;
67        console_info!(
68            ctx.status_tx,
69            "InfluxDB v2 write listening on {} (auth_required={},llm_fallback={})",
70            local,
71            config.token.is_some(),
72            llm_fallback
73        );
74        let owner = ctx.state.clone();
75        let server_id = ctx.server_id;
76        owner.spawn_server_task(server_id,async move {
77            let limiter=ConnectionLimiter::new(DEFAULT_MAX_CONNECTIONS);
78            loop {
79                let (stream,peer,permit)=match accept_bounded(&listener,&limiter,CAP_REFUSAL,"InfluxDB",Some(&ctx.status_tx)).await {
80                    Ok(value)=>value,
81                    Err(error)=>{console_error!(ctx.status_tx,"InfluxDB accept failed: {}",error);break;}
82                };
83                let id=ConnectionId::new(ctx.state.get_next_unified_id().await);
84                let now=crate::utils::clock::Instant::now();
85                ctx.state.add_connection_to_server(server_id,ConnectionState {id,remote_addr:peer,local_addr:local,bytes_sent:0,bytes_received:0,packets_sent:0,packets_received:0,last_activity:now,status:ConnectionStatus::Active,status_changed_at:now,protocol_info:ProtocolConnectionInfo::empty()}).await;
86                let child_ctx=ctx.clone(); let config=config.clone();
87                ctx.state.spawn_server_task(server_id,async move {
88                    let _permit=permit;
89                    let write_ready=Arc::new(Notify::new());
90                    let service_ready=write_ready.clone();
91                    let request_ctx=child_ctx.clone();
92                    let service=service_fn(move |request| {
93                        let ctx=request_ctx.clone(); let config=config.clone(); let ready=service_ready.clone();
94                        async move {
95                            let response=handle_request(request,peer,id,&ctx,&config).await;
96                            ready.notify_one();
97                            Ok::<_,Infallible>(response)
98                        }
99                    });
100                    let mut builder=http1::Builder::new();
101                    builder.keep_alive(false).timer(TokioTimer::new()).header_read_timeout(HEADER_TIMEOUT).max_headers(MAX_HEADERS).max_buf_size(MAX_HEADER_BYTES);
102                    let connection=builder.serve_connection(TokioIo::new(stream),service);
103                    tokio::pin!(connection);
104                    tokio::select! {
105                        result=&mut connection=>{if let Err(error)=result {console_error!(child_ctx.status_tx,"InfluxDB HTTP framing/header error: {}",error);}}
106                        _=async {write_ready.notified().await;tokio::time::sleep(WRITE_TIMEOUT).await;}=>{console_error!(child_ctx.status_tx,"InfluxDB decision=fail_closed_response_write_timeout");}
107                    }
108                    child_ctx.state.remove_connection_from_server(server_id,id).await;
109                    let _=child_ctx.status_tx.send("__UPDATE_UI__".into());
110                }).await;
111            }
112        }).await;
113        Ok(local)
114    }
115}
116fn response(status: u16, body: serde_json::Value, retry: Option<u16>) -> Response<Full<Bytes>> {
117    let bytes = if status == 204 {
118        Vec::new()
119    } else {
120        serde_json::to_vec(&body).unwrap_or_default()
121    };
122    let mut r = Response::new(Full::new(Bytes::from(bytes)));
123    *r.status_mut() = StatusCode::from_u16(status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
124    r.headers_mut()
125        .insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
126    if let Some(n) = retry {
127        if let Ok(v) = HeaderValue::from_str(&n.to_string()) {
128            r.headers_mut().insert(RETRY_AFTER, v);
129        }
130    }
131    r
132}
133fn code(status: u16) -> &'static str {
134    match status {
135        400 => "invalid",
136        401 => "unauthorized",
137        403 => "forbidden",
138        404 => "not found",
139        405 => "method not allowed",
140        408 => "request timeout",
141        413 => "request too large",
142        415 => "unsupported media type",
143        422 => "unprocessable entity",
144        429 => "too many requests",
145        503 => "unavailable",
146        _ => "internal error",
147    }
148}
149fn error(status: u16, message: &str) -> Response<Full<Bytes>> {
150    response(status, json!({"code":code(status),"message":message}), None)
151}
152pub(crate) fn query_piece(s: &str) -> Result<String> {
153    let b = s.as_bytes();
154    let mut at = 0;
155    let mut out = Vec::new();
156    while at < b.len() {
157        match b[at] {
158            b'+' => {
159                out.push(b' ');
160                at += 1;
161            }
162            b'%' => {
163                let pair = b.get(at + 1..at + 3).context("truncated query escape")?;
164                let hex = std::str::from_utf8(pair)?;
165                out.push(u8::from_str_radix(hex, 16).context("invalid query escape")?);
166                at += 3;
167            }
168            c => {
169                out.push(c);
170                at += 1;
171            }
172        }
173    }
174    Ok(String::from_utf8(out)?)
175}
176fn targets(query: Option<&str>) -> Result<(String, String, codec::Precision)> {
177    let mut params = BTreeMap::new();
178    for field in query.context("org and bucket required")?.split('&') {
179        let (key, value) = field
180            .split_once('=')
181            .context("query parameter requires value")?;
182        let key = query_piece(key)?;
183        let value = query_piece(value)?;
184        ensure!(
185            ["org", "bucket", "precision"].contains(&key.as_str()),
186            "unsupported query parameter"
187        );
188        ensure!(
189            params.insert(key, value).is_none(),
190            "duplicate query parameter"
191        );
192    }
193    let org = params
194        .remove("org")
195        .context("org required (orgID query not implemented)")?;
196    let bucket = params.remove("bucket").context("bucket required")?;
197    codec::validate_target(&org)?;
198    codec::validate_target(&bucket)?;
199    let precision =
200        codec::Precision::parse(params.get("precision").map(String::as_str).unwrap_or("ns"))?;
201    Ok((org, bucket, precision))
202}
203fn authorized(headers: &hyper::HeaderMap, expected: Option<&str>) -> bool {
204    let Some(expected) = expected else {
205        return true;
206    };
207    let values = headers.get_all(AUTHORIZATION).iter().collect::<Vec<_>>();
208    if values.len() != 1 {
209        return false;
210    }
211    let Some((scheme, token)) = values[0].to_str().ok().and_then(|v| v.split_once(' ')) else {
212        return false;
213    };
214    if !["Token", "Bearer"].contains(&scheme) || codec::validate_token(token).is_err() {
215        return false;
216    }
217    same_secret(expected, token)
218}
219/// Compare in time independent of where the first difference is.
220fn same_secret(expected: &str, offered: &str) -> bool {
221    let mut difference = expected.len() ^ offered.len();
222    for at in 0..expected.len().max(offered.len()) {
223        difference |= usize::from(
224            expected.as_bytes().get(at).copied().unwrap_or(0)
225                ^ offered.as_bytes().get(at).copied().unwrap_or(0),
226        );
227    }
228    difference == 0
229}
230/// A query's credential: the `Token`/`Bearer` header, or for InfluxQL the v1 forms InfluxDB 2
231/// accepts in its compatibility API — the token as the `p` parameter, or as the password of
232/// HTTP Basic authentication. The user name is not checked, as InfluxDB 2 does not check it.
233fn query_authorized(
234    headers: &hyper::HeaderMap,
235    request: &query::QueryRequest,
236    expected: Option<&str>,
237) -> bool {
238    let Some(expected) = expected else {
239        return true;
240    };
241    if authorized(headers, Some(expected)) {
242        return true;
243    }
244    if request.language != query::Language::Influxql {
245        return false;
246    }
247    if let Some(p) = &request.v1_password {
248        return same_secret(expected, p);
249    }
250    let values = headers.get_all(AUTHORIZATION).iter().collect::<Vec<_>>();
251    let [value] = values.as_slice() else {
252        return false;
253    };
254    let Some(encoded) = value.to_str().ok().and_then(|v| v.strip_prefix("Basic ")) else {
255        return false;
256    };
257    use base64::Engine as _;
258    let Some(decoded) = base64::engine::general_purpose::STANDARD
259        .decode(encoded.trim())
260        .ok()
261        .and_then(|d| String::from_utf8(d).ok())
262    else {
263        return false;
264    };
265    decoded
266        .split_once(':')
267        .is_some_and(|(_, password)| same_secret(expected, password))
268}
269fn decide(
270    decision: actions::Decision,
271    parsed: &codec::ParsedBatch,
272) -> Result<Response<Full<Bytes>>> {
273    let total = parsed.points.len() + parsed.errors.len();
274    let (accepted, message) = match decision {
275        actions::Decision::Reject {
276            status,
277            message,
278            retry_after_seconds,
279        } => {
280            return Ok(response(
281                status,
282                json!({"code":code(status),"message":message,"accepted_points":0,"rejected_points":total}),
283                retry_after_seconds,
284            ))
285        }
286        actions::Decision::Accept => (
287            parsed.points.iter().map(|p| p.line).collect::<Vec<_>>(),
288            "Invalid line protocol".to_string(),
289        ),
290        actions::Decision::Partial {
291            accepted_lines,
292            message,
293        } => {
294            ensure!(
295                accepted_lines
296                    .iter()
297                    .all(|n| parsed.points.iter().any(|p| p.line == *n)),
298                "cannot accept a missing/invalid source line"
299            );
300            (accepted_lines, message)
301        }
302    };
303    if accepted.len() == total && parsed.errors.is_empty() {
304        return Ok(response(204, serde_json::Value::Null, None));
305    }
306    let line = parsed.errors.first().map(|e| e.line).or_else(|| {
307        parsed
308            .points
309            .iter()
310            .find(|p| !accepted.contains(&p.line))
311            .map(|p| p.line)
312    });
313    Ok(response(
314        400,
315        json!({"code":"invalid","message":format!("partial write: {message}; accepted={} rejected={}",accepted.len(),total-accepted.len()),"line":line,"accepted_points":accepted.len(),"rejected_points":total-accepted.len()}),
316        None,
317    ))
318}
319async fn handle_request(
320    request: Request<Incoming>,
321    peer: SocketAddr,
322    id: ConnectionId,
323    ctx: &SpawnContext,
324    config: &Config,
325) -> Response<Full<Bytes>> {
326    let (parts, body) = request.into_parts();
327    let bytes = match tokio::time::timeout(
328        BODY_TIMEOUT,
329        Limited::new(body, codec::MAX_BODY_BYTES).collect(),
330    )
331    .await
332    {
333        Err(_) => return error(408, "Request body deadline exceeded"),
334        Ok(Err(error)) => {
335            return if error
336                .downcast_ref::<http_body_util::LengthLimitError>()
337                .is_some()
338            {
339                self::error(413, "Request body limit exceeded")
340            } else {
341                self::error(400, "Incomplete HTTP body")
342            };
343        }
344        Ok(Ok(body)) => body.to_bytes(),
345    };
346    ctx.state
347        .update_connection_stats(
348            ctx.server_id,
349            id,
350            Some(bytes.len() as u64),
351            None,
352            Some(1),
353            None,
354        )
355        .await;
356    match parts.uri.path() {
357        "/api/v2/write" => {}
358        "/api/v2/query" => {
359            return handle_query(&parts, &bytes, peer, id, ctx, config, query::Language::Flux).await
360        }
361        "/query" => {
362            return handle_query(
363                &parts,
364                &bytes,
365                peer,
366                id,
367                ctx,
368                config,
369                query::Language::Influxql,
370            )
371            .await
372        }
373        "/ping" => return ping(&parts.method),
374        _ => {
375            return error(
376                404,
377                "Only /api/v2/write, /api/v2/query, /query and /ping are served",
378            )
379        }
380    }
381    if parts.method != Method::POST {
382        let mut r = error(405, "Only POST is supported");
383        r.headers_mut()
384            .insert(ALLOW, HeaderValue::from_static("POST"));
385        return r;
386    }
387    let (org,bucket,precision)=match targets(parts.uri.query()) {Ok(v)=>v,Err(_)=>return error(400,"org, bucket and optional ns/us/ms/s precision required; invalid/duplicate/unsupported query")};
388    if !authorized(&parts.headers, config.token.as_deref()) {
389        return error(401, "unauthorized access");
390    }
391    if parts.headers.get_all(CONTENT_ENCODING).iter().count() > 1
392        || parts.headers.get_all(CONTENT_TYPE).iter().count() > 1
393    {
394        return error(400, "Duplicate encoding/content type headers");
395    }
396    if let Some(ct) = parts.headers.get(CONTENT_TYPE) {
397        let valid = ct.to_str().ok().is_some_and(|s| {
398            let mut p = s.split(';');
399            p.next()
400                .is_some_and(|mime| mime.trim().eq_ignore_ascii_case("text/plain"))
401                && p.all(|x| x.trim().eq_ignore_ascii_case("charset=utf-8"))
402        });
403        if !valid {
404            return error(415, "Expected text/plain UTF-8 line protocol");
405        }
406    }
407    let encoding = match parts
408        .headers
409        .get(CONTENT_ENCODING)
410        .map(|v| v.to_str())
411        .transpose()
412    {
413        Ok(Some(v)) => v,
414        Ok(None) => "identity",
415        Err(_) => return error(415, "Invalid content encoding"),
416    };
417    if !["identity", "gzip"].contains(&encoding) {
418        return error(415, "Only identity/gzip encoding supported");
419    }
420    let body = match codec::decode_body(&bytes, encoding) {
421        Ok(b) => b,
422        Err(e) => {
423            return error(
424                if e.to_string().contains("byte limit") {
425                    413
426                } else {
427                    400
428                },
429                "Malformed or oversized compressed body",
430            )
431        }
432    };
433    let received_ns = crate::utils::clock::SystemTime::now()
434        .duration_since(crate::utils::clock::UNIX_EPOCH)
435        .ok()
436        .and_then(|t| i64::try_from(t.as_nanos()).ok());
437    let Some(received_ns) = received_ns else {
438        return error(500, "Receiver clock out of range");
439    };
440    let parsed = match codec::parse_batch(&body, precision, received_ns) {
441        Ok(batch) => batch,
442        Err(e) => {
443            return error(
444                if e.to_string().contains("limit") {
445                    413
446                } else {
447                    400
448                },
449                "Invalid/oversized line protocol batch",
450            )
451        }
452    };
453    let event = Event::new(
454        &actions::INFLUX_WRITE_EVENT,
455        json!({"org":org,"bucket":bucket,"precision":precision,"points":parsed.points,"errors":parsed.errors,"authenticated":true,"auth_required":config.token.is_some(),"source_addr":peer.to_string()}),
456    );
457    let configured = ctx
458        .state
459        .get_event_handler_config(ctx.server_id)
460        .await
461        .is_some_and(|c| c.find_handler("influx_write").is_some());
462    let decision = if configured || config.llm_fallback {
463        match crate::llm::action_helper::call_llm(
464            &ctx.llm_client,
465            &ctx.state,
466            ctx.server_id,
467            Some(id),
468            &event,
469            &actions::InfluxDbProtocol::new(),
470        )
471        .await
472        {
473            Ok(result) if result.failures.is_empty() => {
474                let decisions = result
475                    .protocol_results
476                    .iter()
477                    .filter_map(|r| match r {
478                        crate::llm::ActionResult::Custom { name, data }
479                            if name == "influx_write_decision" =>
480                        {
481                            Some(data.clone())
482                        }
483                        _ => None,
484                    })
485                    .collect::<Vec<_>>();
486                if decisions.len() > 1 {
487                    return failed(ctx, id, event.data, "fail_closed_multiple_decisions").await;
488                }
489                match decisions
490                    .first()
491                    .map(|d| serde_json::from_value::<actions::Decision>(d.clone()))
492                    .transpose()
493                {
494                    Ok(Some(d)) => d,
495                    Ok(None) => actions::Decision::Accept,
496                    Err(_) => {
497                        return failed(ctx, id, event.data, "fail_closed_invalid_decision").await
498                    }
499                }
500            }
501            Ok(_) => return failed(ctx, id, event.data, "fail_closed_handler_action_error").await,
502            Err(_) => return failed(ctx, id, event.data, "fail_closed_handler_error").await,
503        }
504    } else {
505        ctx.state
506            .record_access_log(
507                AccessLogOwner::Server(ctx.server_id.as_u32()),
508                "InfluxDB",
509                Some(id.as_u32()),
510                "influx_write",
511                event.data.clone(),
512                vec![json!({"type":"accept_influx_points"})],
513            )
514            .await;
515        actions::Decision::Accept
516    };
517    match decide(decision, &parsed) {
518        Ok(r) => r,
519        Err(_) => failed(ctx, id, event.data, "fail_closed_invalid_partial_decision").await,
520    }
521}
522async fn failed(
523    ctx: &SpawnContext,
524    id: ConnectionId,
525    data: serde_json::Value,
526    decision: &str,
527) -> Response<Full<Bytes>> {
528    console_error!(ctx.status_tx, "InfluxDB decision={}", decision);
529    ctx.state
530        .record_access_log(
531            AccessLogOwner::Server(ctx.server_id.as_u32()),
532            "InfluxDB",
533            Some(id.as_u32()),
534            "influx_handler_failed",
535            data,
536            vec![json!({"decision":decision})],
537        )
538        .await;
539    error(503, "Write handler failed; no acceptance confirmed")
540}
541
542/// `GET|HEAD /ping`: 204, as InfluxDB answers it, naming what answered.
543fn ping(method: &Method) -> Response<Full<Bytes>> {
544    if method != Method::GET && method != Method::HEAD {
545        let mut r = error(405, "Only GET and HEAD are supported");
546        r.headers_mut()
547            .insert(ALLOW, HeaderValue::from_static("GET, HEAD"));
548        return r;
549    }
550    let mut r = response(204, serde_json::Value::Null, None);
551    r.headers_mut().remove(CONTENT_TYPE);
552    r.headers_mut()
553        .insert("X-Influxdb-Version", HeaderValue::from_static("netget"));
554    r.headers_mut()
555        .insert("X-Influxdb-Build", HeaderValue::from_static("OSS"));
556    r
557}
558/// An error in the dialect's own shape: v2's `{"code","message"}` for Flux, v1's `{"error"}`
559/// for InfluxQL.
560fn query_error(language: query::Language, status: u16, message: &str) -> Response<Full<Bytes>> {
561    match language {
562        query::Language::Flux => error(status, message),
563        query::Language::Influxql => response(status, json!({ "error": message }), None),
564    }
565}
566fn body_response(status: u16, content_type: &'static str, body: Vec<u8>) -> Response<Full<Bytes>> {
567    let mut r = Response::new(Full::new(Bytes::from(body)));
568    *r.status_mut() = StatusCode::from_u16(status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
569    r.headers_mut()
570        .insert(CONTENT_TYPE, HeaderValue::from_static(content_type));
571    r
572}
573async fn handle_query(
574    parts: &hyper::http::request::Parts,
575    body: &[u8],
576    peer: SocketAddr,
577    id: ConnectionId,
578    ctx: &SpawnContext,
579    config: &Config,
580    language: query::Language,
581) -> Response<Full<Bytes>> {
582    let allowed = match language {
583        query::Language::Flux => parts.method == Method::POST,
584        query::Language::Influxql => parts.method == Method::POST || parts.method == Method::GET,
585    };
586    if !allowed {
587        let mut r = query_error(
588            language,
589            405,
590            "Method not supported for this query endpoint",
591        );
592        let allow = match language {
593            query::Language::Flux => "POST",
594            query::Language::Influxql => "GET, POST",
595        };
596        r.headers_mut()
597            .insert(ALLOW, HeaderValue::from_static(allow));
598        return r;
599    }
600    if parts
601        .headers
602        .get(CONTENT_ENCODING)
603        .is_some_and(|e| !e.as_bytes().eq_ignore_ascii_case(b"identity"))
604    {
605        return query_error(language, 415, "Compressed query bodies are not accepted");
606    }
607    let content_type = parts
608        .headers
609        .get(CONTENT_TYPE)
610        .and_then(|v| v.to_str().ok());
611    let parsed = match language {
612        query::Language::Flux => query::parse_flux(parts.uri.query(), content_type, body),
613        query::Language::Influxql => query::parse_influxql(parts.uri.query(), content_type, body),
614    };
615    let request = match parsed {
616        Ok(request) => request,
617        Err(e) => {
618            return query_error(
619                language,
620                400,
621                &crate::utils::truncate_for_log(&e.to_string(), 512),
622            )
623        }
624    };
625    if !query_authorized(&parts.headers, &request, config.token.as_deref()) {
626        return query_error(language, 401, "unauthorized access");
627    }
628    let mut data = serde_json::to_value(&request).unwrap_or_else(|_| json!({}));
629    data["authenticated"] = json!(true);
630    data["auth_required"] = json!(config.token.is_some());
631    data["source_addr"] = json!(peer.to_string());
632    let event = Event::new(&actions::INFLUX_QUERY_EVENT, data);
633    let configured = ctx
634        .state
635        .get_event_handler_config(ctx.server_id)
636        .await
637        .is_some_and(|c| c.find_handler("influx_query").is_some());
638    if !configured && !config.llm_fallback {
639        // Nothing here holds data, so without a handler or the model there is no answer — and
640        // an empty result would be an assertion that there is no data, which nobody made.
641        return query_failed(
642            ctx,
643            id,
644            event.data,
645            language,
646            "fail_closed_no_query_handler",
647        )
648        .await;
649    }
650    let answers = match crate::llm::action_helper::call_llm(
651        &ctx.llm_client,
652        &ctx.state,
653        ctx.server_id,
654        Some(id),
655        &event,
656        &actions::InfluxDbProtocol::new(),
657    )
658    .await
659    {
660        Ok(result) if result.failures.is_empty() => result
661            .protocol_results
662            .iter()
663            .filter_map(|r| match r {
664                crate::llm::ActionResult::Custom { name, data }
665                    if name == actions::QUERY_ANSWER =>
666                {
667                    Some(data.clone())
668                }
669                _ => None,
670            })
671            .collect::<Vec<_>>(),
672        Ok(_) => {
673            return query_failed(
674                ctx,
675                id,
676                event.data,
677                language,
678                "fail_closed_handler_action_error",
679            )
680            .await
681        }
682        Err(_) => {
683            return query_failed(ctx, id, event.data, language, "fail_closed_handler_error").await
684        }
685    };
686    let answer = match answers.as_slice() {
687        [answer] => answer,
688        [] => return query_failed(ctx, id, event.data, language, "fail_closed_no_answer").await,
689        _ => {
690            return query_failed(
691                ctx,
692                id,
693                event.data,
694                language,
695                "fail_closed_multiple_answers",
696            )
697            .await
698        }
699    };
700    match render_answer(answer, &request) {
701        Ok(response) => response,
702        Err(_) => query_failed(ctx, id, event.data, language, "fail_closed_invalid_answer").await,
703    }
704}
705/// Render the model's answer in the request's dialect.
706fn render_answer(
707    answer: &serde_json::Value,
708    request: &query::QueryRequest,
709) -> Result<Response<Full<Bytes>>> {
710    if answer["mode"] == "error" {
711        let status = u16::try_from(answer["status"].as_u64().context("status")?)?;
712        let message = answer["message"].as_str().context("message")?;
713        return Ok(query_error(request.language, status, message));
714    }
715    let action = &answer["action"];
716    match request.language {
717        query::Language::Flux => {
718            ensure!(
719                action.get("statements").is_none(),
720                "a Flux query is answered with 'tables', not 'statements'"
721            );
722            let tables =
723                query::tables_from_json(action.get("tables").context("tables")?, &mut 0, &mut 0)?;
724            Ok(body_response(
725                200,
726                "text/csv; charset=utf-8",
727                query::render_flux(&tables)?.into_bytes(),
728            ))
729        }
730        query::Language::Influxql => {
731            let results = query::statements_from_json(action)?;
732            ensure!(
733                results.len() == request.statements.len(),
734                "{} answers for {} statements",
735                results.len(),
736                request.statements.len()
737            );
738            Ok(body_response(
739                200,
740                "application/json",
741                query::render_influxql(&results, request.epoch.as_deref())?,
742            ))
743        }
744    }
745}
746async fn query_failed(
747    ctx: &SpawnContext,
748    id: ConnectionId,
749    data: serde_json::Value,
750    language: query::Language,
751    decision: &str,
752) -> Response<Full<Bytes>> {
753    console_error!(ctx.status_tx, "InfluxDB query decision={}", decision);
754    ctx.state
755        .record_access_log(
756            AccessLogOwner::Server(ctx.server_id.as_u32()),
757            "InfluxDB",
758            Some(id.as_u32()),
759            "influx_query_failed",
760            data,
761            vec![json!({ "decision": decision })],
762        )
763        .await;
764    query_error(
765        language,
766        503,
767        "Query handler failed; no answer is available",
768    )
769}
netget/server/influxdb/mod.rs

netget/server/influxdb/
mod.rs
