Rust InfluxDB Library

A library for querying and sending data to InfluxDB.

https://gitlab.com/claudiomattera/rinfluxdb

Features

Serialize Data to InfluxDB Line Protocol

The [line protocol] is used to send data to InfluxDB.

~~~~rust use rinfluxdb::line_protocol::LineBuilder; use chrono::{TimeZone, Utc};

let line = LineBuilder::new("measurement") .insertfield("latitude", 55.383333) .insertfield("longitude", 10.383333) .inserttag("city", "Odense") .settimestamp(Utc.ymd(2014, 7, 8).and_hms(9, 10, 11)) .build();

asserteq!(line.measurement(), &"measurement".into()); asserteq!(line.field("latitude"), Some(&55.383333.into())); asserteq!(line.field("longitude"), Some(&10.383333.into())); asserteq!(line.tag("city"), Some(&"Odense".into())); asserteq!(line.timestamp(), Some(&Utc.ymd(2014, 7, 8).andhms(9, 10, 11)));

asserteq!( line.tostring(), "location,city=Odense latitude=55.383333,longitude=10.383333 1404810611000000000" ); ~~~~

Build InfluxQL Queries in Rust

InfluxQL queries can be built using influxql::QueryBuilder.

~~~~rust use rinfluxdb::influxql::QueryBuilder; use chrono::{TimeZone, Utc};

let query = QueryBuilder::from("indoorenvironment") .field("temperature") .field("humidity") .start(Utc.ymd(2021, 3, 7).andhms(21, 0, 0)) .build();

asserteq!( query.asref(), "SELECT temperature, humidity \ FROM indoor_environment \ WHERE time > '2021-03-07T21:00:00Z'", ); ~~~~

Build FLUX Queries in Rust

FLUX queries can be built using flux::QueryBuilder.

~~~~rust use rinfluxdb::types::Duration; use rinfluxdb::flux::QueryBuilder;

let query = QueryBuilder::from("telegraf/autogen") .rangestart(Duration::Minutes(-15)) .filter( r#"r.measurement == "cpu" and r.field == "usagesystem" and r.cpu == "cpu-total""# ) .build();

asserteq!( query.asref(), r#"from(bucket: "telegraf/autogen") |> range(start: -15m) |> filter(fn: (r) => r.measurement == "cpu" and r.field == "usage_system" and r.cpu == "cpu-total" ) |> yield()"#, ); ~~~~

Parse Responses from InfluxDB

When sending a query to InfluxDB, it will reply with either JSON or annotated CSV content containing a list of dataframes. This library allows to parse such replies to user-defined dataframe types.

A dataframe must be constructable from its name (a string), its index (a vector of instants) and its columns (a mapping of column names to vector of values).

A dataframe implementation only needs to implement this trait to be used with this crate. I. e., as long as trait TryFrom<(String, Vec<DateTime<Utc>>, HashMap<String, Vec<Value>>), Error = E> is implemented for a given type DF (and type E implements Into<ParseError>), the parser can use it to construct the final objects.

A dummy implementation of a dataframe is available as dataframe::DataFrame, but the trait can be implemented for many other existing libraries.

Parse JSON Returned for InfluxQL Queries

JSON responses to InfluxQL queries can be parsed to dataframes.

~~~~rust use rinfluxdb::influxql::{ResponseError, from_str}; use rinfluxdb::dataframe::DataFrame;

let response: String = ...;

let response: Vec)>, ResponseError>> = from_str(input)?; ~~~~

Parse Annotated CSV Returned for FLUX Queries

Annotated CSV responses to FLUX queries can be parsed.

~~~~rust use rinfluxdb::flux::{ResponseError, from_str}; use rinfluxdb::dataframe::DataFrame;

let response: String = ...;

let response: Vec)>, ResponseError>> = from_str(input)?; ~~~~

(Optional) Use a Client Based on Reqwest to Perform Common Queries

The functions shown above can be used to serialize and deserialize queries and data to raw text, and they can be integrated into existing applications. In alternative, this library also implements an optional client API based on [Reqwest] to directly interact with an InfluxDB instance. Both blocking and asynchronous clients are available, and they support common queries.

Clients are enabled using the client Cargo feature.

Query InfluxDB with InfluxQL

~~~~rust use std::collections::HashMap; use url::Url; use rinfluxdb::influxql::QueryBuilder; use rinfluxdb::influxql::blocking::Client; use rinfluxdb::dataframe::DataFrame;

let client = Client::new( Url::parse("https://example.com/")?, Some(("username", "password")), )?;

let query = QueryBuilder::from("indoorenvironment") .database("house") .field("temperature") .field("humidity") .build(); let dataframe: DataFrame = client.fetchdataframe(query)?; println!("{}", dataframe);

let query = QueryBuilder::from("indoorenvironment") .database("house") .field("temperature") .field("humidity") .groupby("room") .build(); let taggeddataframes: HashMap = client.fetchdataframesbytag(query, "room")?; for (tag, dataframe) in tagged_dataframes { println!("{}: {}", tag, dataframe); } ~~~~

Query InfluxDB with FLUX

~~~~rust unimplemented!() ~~~~

Send Data to InfluxDB using the Line Protocol

~~~~rust use url::Url; use rinfluxdb::lineprotocol::LineBuilder; use rinfluxdb::lineprotocol::blocking::Client;

let client = Client::new( Url::parse("https://example.com/")?, Some(("username", "password")), )?;

let lines = vec![ LineBuilder::new("measurement") .insertfield("field", 42.0) .build(), LineBuilder::new("measurement") .insertfield("field", 43.0) .insert_tag("tag", "value") .build(), ];

client.send("database", &lines)?; ~~~~

(Optional) Wrapper around Reqwest Objects to Construct Requests and Parse Responses

This crate communicates with InfluxDB instances over HTTP(s). The content of HTTP requests and responses must follow InfluxDB conventions and protocols, but they can otherwise be customized, e.g. by adding basic authentication.

In order to ensure maximal freedom, a tiny wrapper is constructed around Reqwest's Client, so that it can create a request builder already prepared to communicate with InfluxDB. Such builder can be then converted to a regular request builder and executed.

~~~~rust // Bring into scope the trait implementation use rinfluxdb::influxql::blocking::InfluxqlClientWrapper;

// Create Reqwest client let client = reqwest::blocking::Client::new();

// Create InfluxQL request let baseurl = Url::parse("https://example.com")?; let mut builder = client // (this is a function added by the trait above) .influxql(&baseurl)? // (this functions are defined on influxql::RequestBuilder) .database("house") .query(Query::new("SELECT temperature FROM indoortemperature")) // (this function returns a regular Reqwest builder) .toreqwest_builder();

// Now this is a regular Reqwest builder, and can be customized as usual if let Some((username, password)) = Some(("username", "password")) { builder = builder.basic_auth(username, Some(password)); }

// Create a request from the builder let request = builder.build()?;

// Execute the request through Reqwest and obtain a response let response = client.execute(request)?; ~~~~

Similarly, a tiny wrapper is constructed around Reqwest's Response, so that a new function is added to parse dataframes from it.

~~~~rust use rinfluxdb::influxql::blocking::InfluxqlClientWrapper; use rinfluxdb::dataframe::DataFrame;

// Bring into scope the trait implementation use rinfluxdb::influxql::blocking::InfluxqlResponseWrapper;

// Create Reqwest client let client = reqwest::blocking::Client::new();

// Create InfluxQL request let baseurl = Url::parse("https://example.com")?; let mut request = client .influxql(&baseurl)? .database("house") .query(Query::new("SELECT temperature FROM indoortemperature")) .toreqwest_builder() .build()?;

// Execute the request through Reqwest and obtain a response let response = client.execute(request)?;

// Return an error if response status is not 200 // (this is a function from Reqwest's response) let response = response.errorforstatus()?;

// Parse the response from JSON to a list of dataframes // (this is a function added by the trait above) let results: Vec>)>, eError>> = response.dataframes()?; ~~~~

Wrappers are defined for both Reqwest's blocking API (influxql::blocking::InfluxqlClientWrapper, influxql::blocking::InfluxqlResponseWrapper) and Reqwest's asynchronous API (influxql::r#async::InfluxqlClientWrapper, influxql::r#async::InfluxqlResponseWrapper), and are enabled using the client Cargo feature.

License

Copyright Claudio Mattera 2021

You are free to copy, modify, and distribute this application with attribution under the terms of the [MIT license]. See the License.txt file for details.

This project is entirely original work, and it is not affiliated with nor endorsed in any way by InfluxData.