InfluxDB

use influx.qweruse influx_line, influx_lines from influx      // Quillo 0.6: only these

ureq = "2"flate2 = "1"

// Greenhouse readings in InfluxDB: a line of line protocol built and read back, readings written to// InfluxDB 2 and the latest read with Flux, request timings buffered and written in one gzipped// batch, and the same readings in InfluxDB 3, read with SQL and a parameter.////   INFLUX_URL=http://127.0.0.1:8086 INFLUX_TOKEN=... INFLUX_ORG=shop INFLUX_BUCKET=metrics quillo run influx.qwe      then open http://localhost:8080/record?temp=21.5//   INFLUX3_URL=http://127.0.0.1:8181 INFLUX3_TOKEN=apiv3_... INFLUX3_DATABASE=shop      for /v3//// Needs InfluxDB 2 (influxd, then influx setup) and, for /v3, InfluxDB 3 (influxdb3 serve, then// influxdb3 create token --admin).use "influx.qwer"let servers = [    influx_connect("v2", env("INFLUX_URL", "http://127.0.0.1:8086"), {        token: env("INFLUX_TOKEN", ""), org: env("INFLUX_ORG", "shop"), bucket: env("INFLUX_BUCKET", "metrics"), precision: "ms",    }),    influx_connect("v3", env("INFLUX3_URL", "http://127.0.0.1:8181"), { token: env("INFLUX3_TOKEN", ""), database: env("INFLUX3_DATABASE", "shop") }),]// What influx_add keeps goes out once a second (or as soon as 5000 lines wait).every(1000, () => influx_flush("v2"))route GET /line {    let line = influx_line("cpu load", { host: "web 1", region: "eu,west", rack: "" }, { usage: 12.5, cores: 8, note: "said \"hi\"", ok: true }, 1761800000000)?    { line, back: influx_parse(line)?[0] }}// Two readings written, then the latest of each field read back with Flux.route GET /record ? { sensor: str = "north", temp: float = 21.5 } {    let t = now()    let written = influx_write("v2", [        { measurement: "climate", tags: { sensor }, fields: { temp, humidity: 61 }, time: t - 1000 },        { measurement: "climate", tags: { sensor }, fields: { temp: temp + 0.25, humidity: 60 }, time: t },    ])?    let rows = influx_query("v2", `from(bucket: "${env("INFLUX_BUCKET", "metrics")}")        |> range(start: -1h)        |> filter(fn: (r) => r._measurement == "climate" and r.sensor == "${sensor}")        |> last()`, {})?    { written: written.lines, latest: rows | map((r) => { field: r._field, value: r._value, sensor: r.sensor }) | sort_by((r) => r.field) }}// 2000 timings buffered, then written at once: the body is gzipped.route GET /batch {    let t = now()    for i in range(0, 2000) {        influx_add("v2", { measurement: "http", tags: { route: i % 2 == 0 ? "/home" : "/cart" }, fields: { ms: 3 + i % 17 }, time: t - i })?    }    let sent = influx_flush("v2")?    let counted = influx_query("v2", `from(bucket: "${env("INFLUX_BUCKET", "metrics")}") |> range(start: -1h) |> filter(fn: (r) => r._measurement == "http") |> count()`, {})?    { sent, counted: counted | map((r) => { route: r.route, n: r._value }) | sort_by((r) => r.route) }}route GET /v3 ? { sensor: str = "south" } {    let t = now()    influx_write("v3", `climate,sensor=${sensor} temp=19.5,humidity=55i ${t - 1000}\nclimate,sensor=${sensor} temp=19.75,humidity=54i ${t}`)?    let rows = influx_query("v3", "SELECT sensor, temp, humidity FROM climate WHERE sensor = $sensor ORDER BY time LIMIT 10", { params: { sensor } })?    { rows, refused: influx_write("v3", "climate,sensor=x temp=oops").error }}

//! InfluxDB line protocol built and read (escaping, tags, typed fields, timestamps), points written to InfluxDB 2 or 3 over HTTP (org, bucket or database, token, precision, gzip, a buffer flushed by size or call), and queries in Flux, SQL or InfluxQL given back as rows.//! version: 1.0.0//!//! [dependencies]//! ureq = "2"//! flate2 = "1"use serde_json::{json, Map, Value};use std::collections::HashMap;use std::io::Write;use std::sync::{Arc, Mutex};use std::time::Duration;// ---- Line protocol ----/// Escapes `chars` with a backslash; a backslash itself only where it would be read as an escape (before one of them, another backslash, or at the end).fn esc(s: &str, chars: &[char]) -> String {    let v: Vec<char> = s.chars().collect();    let mut o = String::with_capacity(s.len());    for (i, &c) in v.iter().enumerate() {        if chars.contains(&c) {            o.push('\\');        } else if c == '\\' {            let next = v.get(i + 1);            if next.is_none() || next.map(|n| chars.contains(n) || *n == '\\').unwrap_or(false) {                o.push('\\');            }        }        o.push(c);    }    o}fn no_newline(s: &str, what: &str) -> Result<(), String> {    if s.contains('\n') || s.contains('\r') {        return Err(format!("a {} cannot hold a line break: `{}`", what, s.escape_debug()));    }    Ok(())}fn field_value(v: &Value) -> Result<Option<String>, String> {    Ok(Some(match v {        Value::Null => return Ok(None),        Value::Bool(b) => b.to_string(),        Value::Number(n) if n.is_i64() => format!("{}i", n),        Value::Number(n) if n.is_u64() => format!("{}u", n),        Value::Number(n) => {            let f = n.as_f64().unwrap_or(0.0);            if !f.is_finite() {                return Err("a field cannot be NaN or infinite".into());            }            format!("{}", f)        }        Value::String(s) => format!("\"{}\"", s.replace('\\', "\\\\").replace('"', "\\\"")),        Value::Object(m) if m.len() == 1 && m.contains_key("uint") => format!("{}u", m["uint"].as_u64().ok_or("{ uint: n } takes a whole number of 0 or more")?),        Value::Object(m) if m.len() == 1 && m.contains_key("float") => format!("{}", m["float"].as_f64().ok_or("{ float: n } takes a number")?),        other => return Err(format!("a field is a number, text or true/false, not {}", other)),    }))}fn time_text(t: &Value) -> Result<Option<String>, String> {    match t {        Value::Null => Ok(None),        Value::Number(n) if n.is_i64() || n.is_u64() => Ok(Some(n.to_string())),        Value::Number(n) => Ok(Some(format!("{}", n.as_f64().unwrap_or(0.0).round() as i64))),        Value::String(s) if !s.is_empty() && s.trim_start_matches('-').chars().all(|c| c.is_ascii_digit()) => Ok(Some(s.clone())),        _ => Err(format!("the time is a whole number in the write's precision, not {}", t)),    }}fn build(measurement: &str, tags: &Value, fields: &Value, time: &Value) -> Result<String, String> {    if measurement.is_empty() {        return Err("a point needs a measurement".into());    }    no_newline(measurement, "measurement")?;    let mut line = esc(measurement, &[',', ' ']);    if let Value::Object(t) = tags {        let mut keys: Vec<&String> = t.keys().collect();        keys.sort();        for k in keys {            let v = match &t[k] {                Value::Null => continue,                Value::String(s) => s.clone(),                other => other.to_string(),            };            if v.is_empty() || k.is_empty() {                continue;            }            no_newline(k, "tag key")?;            no_newline(&v, "tag value")?;            line.push(',');            line.push_str(&esc(k, &[',', '=', ' ']));            line.push('=');            line.push_str(&esc(&v, &[',', '=', ' ']));        }    } else if !tags.is_null() {        return Err("tags are a map: { host: \"web-1\" }".into());    }    let f = fields.as_object().ok_or("fields are a map: { value: 1.5 }")?;    let mut parts = vec![];    for (k, v) in f {        if let Some(text) = field_value(v)? {            no_newline(k, "field key")?;            parts.push(format!("{}={}", esc(k, &[',', '=', ' ']), text));        }    }    if parts.is_empty() {        return Err(format!("the point `{}` has no fields: InfluxDB needs at least one", measurement));    }    line.push(' ');    line.push_str(&parts.join(","));    if let Some(t) = time_text(time)? {        line.push(' ');        line.push_str(&t);    }    Ok(line)}fn point_line(p: &Value) -> Result<String, String> {    match p {        Value::String(s) => Ok(s.trim_end_matches('\n').to_string()),        Value::Object(_) => build(p["measurement"].as_str().ok_or("a point is { measurement, tags, fields, time }")?, &p["tags"], &p["fields"], &p["time"]),        _ => Err("a point is { measurement, tags, fields, time }, or a line of line protocol".into()),    }}fn lines_of(data: &Value) -> Result<Vec<String>, String> {    match data {        Value::Array(a) => a.iter().map(point_line).collect(),        Value::String(s) => Ok(split_lines(s).into_iter().map(|(_, l)| l).collect()),        other => Ok(vec![point_line(other)?]),    }}/// One line of line protocol: `measurement,tag=v field=1.5,count=3i,ok=true,note="x" time`. Ints become `3i`, floats stay floats (`{ uint: 3 }` is `3u`, `{ float: 3 }` is `3`), text is quoted; names and tags are escaped; tags are sorted and empty ones left out. `time` is a whole number in the precision you write with, or null for the server's clock.pub fn influx_line(measurement: &str, tags: serde_json::Value, fields: serde_json::Value, time: serde_json::Value) -> Result<String, String> {    build(measurement, &tags, &fields, &time)}/// Lines for a list of points (`{ measurement, tags, fields, time }` each), one per line.pub fn influx_lines(points: Vec<serde_json::Value>) -> Result<String, String> {    Ok(points.iter().map(point_line).collect::<Result<Vec<_>, _>>()?.join("\n"))}/// Splits text into lines as InfluxDB reads it: a line break inside a quoted string field belongs to the field. Comments and blank lines are left out.fn split_lines(text: &str) -> Vec<(usize, String)> {    let mut out = vec![];    let mut cur = String::new();    let (mut section, mut in_q, mut line_no, mut start_no) = (0u8, false, 1usize, 1usize);    let mut chars = text.chars();    let mut end = |cur: &mut String, start: usize, out: &mut Vec<(usize, String)>| {        let l = cur.trim_end_matches('\r').trim_start().to_string();        if !l.trim().is_empty() && !l.starts_with('#') {            out.push((start, l));        }        cur.clear();    };    while let Some(c) = chars.next() {        if c == '\\' {            cur.push(c);            if let Some(n) = chars.next() {                if n == '\n' {                    line_no += 1;                }                cur.push(n);            }            continue;        }        if c == '\n' {            line_no += 1;            if !in_q {                end(&mut cur, start_no, &mut out);                section = 0;                start_no = line_no;                continue;            }        }        if section == 0 && c == ' ' && !cur.trim().is_empty() && !cur.trim_start().starts_with('#') {            section = 1;        } else if section == 1 && c == '"' {            in_q = !in_q;        } else if section == 1 && c == ' ' && !in_q {            section = 2;        }        cur.push(c);    }    end(&mut cur, start_no, &mut out);    out}/// Splits at an unescaped `sep`, outside quotes when `quotes`.fn split_esc(s: &str, sep: char, quotes: bool) -> Vec<String> {    let mut out = vec![];    let mut cur = String::new();    let mut in_q = false;    let mut chars = s.chars().peekable();    while let Some(c) = chars.next() {        if c == '\\' {            cur.push(c);            if let Some(n) = chars.next() {                cur.push(n);            }            continue;        }        if quotes && c == '"' {            in_q = !in_q;        }        if c == sep && !in_q {            out.push(std::mem::take(&mut cur));        } else {            cur.push(c);        }    }    out.push(cur);    out}fn unesc(s: &str) -> String {    let mut o = String::with_capacity(s.len());    let mut it = s.chars().peekable();    while let Some(c) = it.next() {        if c == '\\' {            if let Some(&n) = it.peek() {                if matches!(n, ',' | '=' | ' ' | '\\' | '"') {                    o.push(n);                    it.next();                    continue;                }            }        }        o.push(c);    }    o}fn parse_field(v: &str) -> Result<Value, String> {    if let Some(inner) = v.strip_prefix('"') {        let inner = inner.strip_suffix('"').ok_or(format!("the text `{}` is not closed with \"", v))?;        return Ok(json!(inner.replace("\\\"", "\"").replace("\\\\", "\\")));    }    match v {        "t" | "T" | "true" | "True" | "TRUE" => return Ok(json!(true)),        "f" | "F" | "false" | "False" | "FALSE" => return Ok(json!(false)),        _ => {}    }    if let Some(i) = v.strip_suffix('i') {        return i.parse::<i64>().map(|n| json!(n)).map_err(|_| format!("`{}` is not an integer", v));    }    if let Some(u) = v.strip_suffix('u') {        return u.parse::<u64>().map(|n| json!(n)).map_err(|_| format!("`{}` is not an unsigned integer", v));    }    v.parse::<f64>().ok().filter(|f| f.is_finite()).map(|f| json!(f)).ok_or(format!("`{}` is not a field value (number, 1i, 1u, true, \"text\")", v))}fn parse_line(line: &str) -> Result<Value, String> {    // measurement+tags, then fields, then the time, separated by unescaped spaces outside quotes.    let parts: Vec<String> = split_esc(line, ' ', true).into_iter().filter(|p| !p.is_empty()).collect();    if parts.len() < 2 || parts.len() > 3 {        return Err("a line is `measurement,tags fields time`".into());    }    let head = split_esc(&parts[0], ',', false);    let measurement = unesc(&head[0]);    if measurement.is_empty() {        return Err("the line has no measurement".into());    }    let mut tags = Map::new();    for t in &head[1..] {        let kv = split_esc(t, '=', false);        if kv.len() != 2 || kv[0].is_empty() || kv[1].is_empty() {            return Err(format!("`{}` is not a tag (key=value)", t));        }        tags.insert(unesc(&kv[0]), json!(unesc(&kv[1])));    }    let mut fields = Map::new();    for f in split_esc(&parts[1], ',', true) {        let kv = split_esc(&f, '=', true);        if kv.len() < 2 || kv[0].is_empty() {            return Err(format!("`{}` is not a field (key=value)", f));        }        let value = kv[1..].join("=");        fields.insert(unesc(&kv[0]), parse_field(&value)?);    }    let time = match parts.get(2) {        Some(t) => json!(t.parse::<i64>().map_err(|_| format!("`{}` is not a timestamp", t))?),        None => Value::Null,    };    Ok(json!({ "measurement": measurement, "tags": tags, "fields": fields, "time": time }))}/// Reads line protocol into points `{ measurement, tags, fields, time }` (fields typed: `3i` an int, `1.5` a float, `true`, text); comments and blank lines skipped. A line that cannot be read is an error naming its number.pub fn influx_parse(text: &str) -> Result<Vec<Value>, String> {    let mut out = vec![];    for (n, l) in split_lines(text) {        out.push(parse_line(&l).map_err(|e| format!("line {}: {}", n, e))?);    }    Ok(out)}// ---- Writing and querying ----struct Client {    url: String,    version: u8,    token: Option<String>,    org: String,    bucket: String,    precision: String,    gzip: bool,    batch: usize,    timeout: Duration,    buf: Vec<String>,    written: u64,}/// The servers, by the name given to `influx_connect`.static CLIENTS: Mutex<Option<HashMap<String, Arc<Mutex<Client>>>>> = Mutex::new(None);fn client(name: &str) -> Result<Arc<Mutex<Client>>, String> {    CLIENTS.lock().unwrap_or_else(|e| e.into_inner()).get_or_insert_with(HashMap::new).get(name).cloned().ok_or(format!(        "no InfluxDB named `{}`: call influx_connect(\"{}\", \"http://127.0.0.1:8086\", {{ token, org, bucket }}) first",        name, name    ))}fn enc(s: &str) -> String {    s.bytes().map(|b| if b.is_ascii_alphanumeric() || b"-_.~".contains(&b) { (b as char).to_string() } else { format!("%{:02X}", b) }).collect()}fn auth(c: &Client, r: ureq::Request) -> ureq::Request {    match &c.token {        Some(t) if c.version == 3 => r.set("Authorization", &format!("Bearer {}", t)),        Some(t) => r.set("Authorization", &format!("Token {}", t)),        None => r,    }}fn agent(c: &Client) -> ureq::Agent {    ureq::AgentBuilder::new().timeout(c.timeout).build()}fn answer(r: Result<ureq::Response, ureq::Error>, what: &str) -> Result<(u16, String), String> {    match r {        Ok(r) => {            let code = r.status();            Ok((code, r.into_string().map_err(|e| e.to_string())?))        }        Err(ureq::Error::Status(code, r)) => {            let body = r.into_string().unwrap_or_default();            let v: Value = serde_json::from_str(&body).unwrap_or(Value::Null);            let mut msg = v["message"].as_str().or(v["error"].as_str()).map(str::to_string).unwrap_or(body.trim().chars().take(500).collect());            // InfluxDB 3 says which lines it could not take.            if let Some(d) = v["data"].as_array().and_then(|a| a.first()) {                if let Some(e) = d["error_message"].as_str() {                    msg = format!("{}: line {}: {}", msg, d["line_number"], e);                }            }            let hint = match code {                401 | 403 => " (check the token, and that it may write to this bucket or database)",                404 => " (check the org, bucket or database name)",                413 => " (send fewer lines at once: a smaller batch)",                _ => "",            };            Err(format!("InfluxDB refused {}: {} {}{}", what, code, msg, hint))        }        Err(e) => Err(format!("cannot reach InfluxDB for {}: {}", what, e)),    }}fn post_lines(c: &Client, lines: &[String]) -> Result<Value, String> {    if lines.is_empty() {        return Ok(json!({ "lines": 0, "bytes": 0, "gzip": false }));    }    let mut body = lines.join("\n");    body.push('\n');    let url = if c.version == 3 {        let p = match c.precision.as_str() { "ns" => "nanosecond", "us" => "microsecond", "ms" => "millisecond", _ => "second" };        format!("{}/api/v3/write_lp?db={}&precision={}", c.url, enc(&c.bucket), p)    } else {        format!("{}/api/v2/write?org={}&bucket={}&precision={}", c.url, enc(&c.org), enc(&c.bucket), c.precision)    };    let req = auth(c, agent(c).post(&url)).set("Content-Type", "text/plain; charset=utf-8");    let zip = c.gzip && body.len() > 1024;    let r = if zip {        let mut e = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());        e.write_all(body.as_bytes()).map_err(|e| e.to_string())?;        let z = e.finish().map_err(|e| e.to_string())?;        req.set("Content-Encoding", "gzip").send_bytes(&z)    } else {        req.send_bytes(body.as_bytes())    };    answer(r, "the write")?;    Ok(json!({ "lines": lines.len(), "bytes": body.len(), "gzip": zip }))}/// Remembers the server `name`: `url` (http://127.0.0.1:8086 for InfluxDB 2, :8181 for 3). Options: `token`, `org` and `bucket` (InfluxDB 2), `database` (InfluxDB 3: its writes and SQL go there), `version` (2 or 3; 3 when `database` is given), `precision` of the times you write ("ms", like Quillo's `now()`; "s", "us", "ns"), `gzip` (true: bodies over 1 KB compressed), `batch` (lines `influx_add` keeps before it writes; 5000), `timeout_ms` (10000).pub fn influx_connect(name: &str, url: &str, options: serde_json::Value) -> Result<Value, String> {    let u = url.trim().trim_end_matches('/').to_string();    if !(u.starts_with("http://") || u.starts_with("https://")) {        return Err(format!("`{}` is not InfluxDB's address, like http://127.0.0.1:8086", url));    }    let version = match options["version"].as_i64() {        Some(v @ (2 | 3)) => v as u8,        Some(v) => return Err(format!("the version is 2 or 3, not {}", v)),        None => if options["database"].is_string() { 3 } else { 2 },    };    let bucket = options[if version == 3 { "database" } else { "bucket" }].as_str().or(options["bucket"].as_str()).or(options["database"].as_str()).unwrap_or("").to_string();    if bucket.is_empty() {        return Err(if version == 3 { "InfluxDB 3 needs { database }".into() } else { "InfluxDB 2 needs { org, bucket }".into() });    }    let org = options["org"].as_str().unwrap_or("").to_string();    if version == 2 && org.is_empty() {        return Err("InfluxDB 2 needs { org } (its name or ID) with the bucket".into());    }    let precision = options["precision"].as_str().unwrap_or("ms").to_string();    if !matches!(precision.as_str(), "ns" | "us" | "ms" | "s") {        return Err(format!("the precision is \"ns\", \"us\", \"ms\" or \"s\", not \"{}\"", precision));    }    let c = Client {        url: u.clone(),        version,        token: options["token"].as_str().filter(|t| !t.is_empty()).map(str::to_string),        org,        bucket: bucket.clone(),        precision: precision.clone(),        gzip: options["gzip"].as_bool().unwrap_or(true),        batch: options["batch"].as_u64().unwrap_or(5000).clamp(1, 1_000_000) as usize,        timeout: Duration::from_millis(options["timeout_ms"].as_u64().unwrap_or(10_000).clamp(100, 300_000)),        buf: vec![],        written: 0,    };    let old = CLIENTS.lock().unwrap_or_else(|e| e.into_inner()).get_or_insert_with(HashMap::new).insert(name.to_string(), Arc::new(Mutex::new(c)));    if let Some(o) = old {        let mut o = o.lock().unwrap_or_else(|e| e.into_inner());        let lines = std::mem::take(&mut o.buf);        let _ = post_lines(&o, &lines);    }    Ok(json!({ "name": name, "url": u, "version": version, "bucket": bucket, "precision": precision }))}/// Writes now: `data` is a point `{ measurement, tags, fields, time }`, a list of points, or line protocol text. Gives `{ lines, bytes, gzip }`.pub fn influx_write(name: &str, data: serde_json::Value) -> Result<Value, String> {    let lines = lines_of(&data)?;    let c = client(name)?;    let mut c = c.lock().unwrap_or_else(|e| e.into_inner());    let r = post_lines(&c, &lines)?;    c.written += lines.len() as u64;    Ok(r)}/// Adds points (as `influx_write` takes them) to the buffer, which is written once it holds `batch` lines; gives how many lines wait. Call `influx_flush` from an `every` tick for the rest.pub fn influx_add(name: &str, data: serde_json::Value) -> Result<i64, String> {    let lines = lines_of(&data)?;    let c = client(name)?;    let mut c = c.lock().unwrap_or_else(|e| e.into_inner());    c.buf.extend(lines);    if c.buf.len() >= c.batch {        let lines = std::mem::take(&mut c.buf);        if let Err(e) = post_lines(&c, &lines) {            // Kept for the next try, up to ten batches.            let keep = c.batch * 10;            let mut back = lines;            back.append(&mut c.buf);            let start = back.len().saturating_sub(keep);            c.buf = back.split_off(start);            return Err(e);        }        c.written += lines.len() as u64;    }    Ok(c.buf.len() as i64)}/// Writes what the buffer holds: `{ lines, bytes, gzip }`. On an error the lines stay for the next flush.pub fn influx_flush(name: &str) -> Result<Value, String> {    let c = client(name)?;    let mut c = c.lock().unwrap_or_else(|e| e.into_inner());    let lines = std::mem::take(&mut c.buf);    match post_lines(&c, &lines) {        Ok(v) => {            c.written += lines.len() as u64;            Ok(v)        }        Err(e) => {            let mut back = lines;            back.append(&mut c.buf);            c.buf = back;            Err(e)        }    }}/// Flushes and forgets the server `name`; whether there was one.pub fn influx_close(name: &str) -> Result<bool, String> {    let c = CLIENTS.lock().unwrap_or_else(|e| e.into_inner()).get_or_insert_with(HashMap::new).remove(name);    match c {        Some(c) => {            let mut c = c.lock().unwrap_or_else(|e| e.into_inner());            let lines = std::mem::take(&mut c.buf);            post_lines(&c, &lines)?;            Ok(true)        }        None => Ok(false),    }}/// Whether the server answers, and its version: `{ ok, version }` (GET /ping).pub fn influx_ping(name: &str) -> Result<Value, String> {    let c = client(name)?;    let c = c.lock().unwrap_or_else(|e| e.into_inner());    match auth(&c, agent(&c).get(&format!("{}/ping", c.url))).call() {        Ok(r) => {            let header = r.header("X-Influxdb-Version").map(str::to_string);            let body: Value = r.into_string().ok().and_then(|t| serde_json::from_str(&t).ok()).unwrap_or(Value::Null);            Ok(json!({ "ok": true, "version": header.or(body["version"].as_str().map(str::to_string)) }))        }        Err(ureq::Error::Status(code, _)) => Ok(json!({ "ok": false, "status": code })),        Err(e) => Err(format!("cannot reach InfluxDB: {}", e)),    }}/// Splits CSV text into records, a record that is one empty field marking a blank line.fn csv_records(text: &str) -> Vec<Vec<String>> {    let mut out = vec![];    let mut rec = vec![];    let mut cur = String::new();    let mut in_q = false;    let mut chars = text.chars().peekable();    while let Some(c) = chars.next() {        if in_q {            // Flux writes a line break inside a value as CRLF.            if c == '\r' && chars.peek() == Some(&'\n') {                continue;            }            if c == '"' {                if chars.peek() == Some(&'"') {                    cur.push('"');                    chars.next();                } else {                    in_q = false;                }            } else {                cur.push(c);            }            continue;        }        match c {            '"' => in_q = true,            ',' => rec.push(std::mem::take(&mut cur)),            '\r' => {}            '\n' => {                rec.push(std::mem::take(&mut cur));                out.push(std::mem::take(&mut rec));            }            c => cur.push(c),        }    }    if !cur.is_empty() || !rec.is_empty() {        rec.push(cur);        out.push(rec);    }    out}fn typed(v: &str, t: &str) -> Value {    if v.is_empty() && t != "string" {        return Value::Null;    }    match t {        "long" => v.parse::<i64>().map(Value::from).unwrap_or(json!(v)),        "unsignedLong" => v.parse::<u64>().map(Value::from).unwrap_or(json!(v)),        "double" => v.parse::<f64>().ok().filter(|f| f.is_finite()).map(Value::from).unwrap_or(json!(v)),        "boolean" => json!(v == "true"),        _ => json!(v),    }}/// Flux's annotated CSV as rows: types from #datatype, one map per record, `table` kept, the empty and `result` columns left out.fn flux_rows(text: &str) -> Result<Vec<Value>, String> {    let mut rows = vec![];    let mut types: Vec<String> = vec![];    let mut header: Option<Vec<String>> = None;    for rec in csv_records(text) {        if rec.len() == 1 && rec[0].is_empty() {            header = None;            types.clear();            continue;        }        if rec[0] == "#datatype" {            types = rec.clone();            header = None;            continue;        }        if rec[0].starts_with('#') {            continue;        }        match &header {            None => header = Some(rec),            Some(h) => {                if rec.len() > 1 && rec[1] == "error" && h.get(1).map(|x| x == "error").unwrap_or(false) {                    return Err(format!("the Flux query failed: {}", rec.get(1).cloned().unwrap_or_default()));                }                let mut m = Map::new();                for (i, k) in h.iter().enumerate() {                    if k.is_empty() || k == "result" {                        continue;                    }                    let v = rec.get(i).cloned().unwrap_or_default();                    let t = types.get(i).map(String::as_str).unwrap_or(if k == "table" { "long" } else { "string" });                    m.insert(k.clone(), typed(&v, t));                }                rows.push(Value::Object(m));            }        }    }    Ok(rows)}/// InfluxQL's JSON (InfluxDB 1 and 2's /query) as rows: each value with its column names, `measurement` and the series' tags.fn influxql_rows(v: &Value) -> Result<Vec<Value>, String> {    let mut rows = vec![];    for r in v["results"].as_array().cloned().unwrap_or_default() {        if let Some(e) = r["error"].as_str() {            return Err(format!("the InfluxQL query failed: {}", e));        }        for s in r["series"].as_array().cloned().unwrap_or_default() {            let cols: Vec<String> = s["columns"].as_array().map(|a| a.iter().map(|c| c.as_str().unwrap_or("").to_string()).collect()).unwrap_or_default();            for vals in s["values"].as_array().cloned().unwrap_or_default() {                let mut m = Map::new();                m.insert("measurement".into(), s["name"].clone());                if let Some(t) = s["tags"].as_object() {                    for (k, x) in t {                        m.insert(k.clone(), x.clone());                    }                }                for (i, c) in cols.iter().enumerate() {                    m.insert(c.clone(), vals.get(i).cloned().unwrap_or(Value::Null));                }                rows.push(Value::Object(m));            }        }    }    Ok(rows)}/// Runs a query and gives its rows (maps). InfluxDB 2: Flux (rows with `_time`, `_measurement`, `_field`, `_value`, the tags and `table`), or `{ lang: "influxql" }` through its 1.x API (the bucket as the database). InfluxDB 3: SQL, or `{ lang: "influxql" }`. Options: `lang`, `params` (a map, for SQL `$name` placeholders on InfluxDB 3).pub fn influx_query(name: &str, query: &str, options: serde_json::Value) -> Result<Vec<Value>, String> {    let c = client(name)?;    let c = c.lock().unwrap_or_else(|e| e.into_inner());    let lang = options["lang"].as_str().unwrap_or(if c.version == 3 { "sql" } else { "flux" }).to_ascii_lowercase();    match (c.version, lang.as_str()) {        (2, "flux") => {            let body = json!({ "query": query, "type": "flux", "dialect": { "header": true, "annotations": ["datatype", "group", "default"], "delimiter": ",", "dateTimeFormat": "RFC3339" } });            let r = auth(&c, agent(&c).post(&format!("{}/api/v2/query?org={}", c.url, enc(&c.org))))                .set("Content-Type", "application/json")                .set("Accept", "application/csv")                .send_string(&body.to_string());            flux_rows(&answer(r, "the Flux query")?.1)        }        (2, "influxql") => {            let url = format!("{}/query?db={}&q={}&epoch={}", c.url, enc(&c.bucket), enc(query), c.precision);            let text = answer(auth(&c, agent(&c).get(&url)).call(), "the InfluxQL query")?.1;            influxql_rows(&serde_json::from_str(&text).map_err(|e| format!("InfluxDB's answer is not JSON: {}", e))?)        }        (3, "sql") | (3, "influxql") => {            let mut body = json!({ "db": c.bucket, "q": query, "format": "json" });            if options["params"].is_object() {                body["params"] = options["params"].clone();            }            let path = if lang == "sql" { "query_sql" } else { "query_influxql" };            let r = auth(&c, agent(&c).post(&format!("{}/api/v3/{}", c.url, path))).set("Content-Type", "application/json").send_string(&body.to_string());            let (_, text) = answer(r, "the query")?;            if text.trim().is_empty() {                return Ok(vec![]);            }            match serde_json::from_str::<Value>(&text).map_err(|e| format!("InfluxDB's answer is not JSON: {}", e))? {                Value::Array(a) => Ok(a),                other => Ok(vec![other]),            }        }        (v, l) => Err(format!("InfluxDB {} does not take {} queries here: {}", v, l, if v == 3 { "use sql or influxql" } else { "use flux or influxql" })),    }}