driftluo / influxdbclient-rs Goto Github PK

Could you please add Python binding for this lib? Also add wheel package for ARM.

I would like to use it in ARM board, to replace the Python-pure client.

use influx_db_client::{
    Client, Point, Points, Value, Precision, point, points
};




fn main() {
    let mut client=Client::default().set_authentication("wizdata","whizhome_sg");
    let point = point!("log")
        .add_tag("level",Value::String(String::from("debug")))
        .add_field("value",Value::Integer(1));
    let points=points!(point);
    tokio::runtime::Runtime::new().unwrap().block_on(async move{
        client.write_points(points,Some(Precision::Seconds),None).await.unwrap();
    });

}

log

thread 'main' panicked at 'called `Result::unwrap()` on an `Err` value: Communication("error sending request for url (http://localhost:8086/write?u=wizdata&p=whizhome_sg&db=test&precision=s): connection closed before message completed")', src\main.rs:21:73
stack backtrace:
   0: std::panicking::begin_panic_handler
             at /rustc/18bf6b4f01a6feaf7259ba7cdae58031af1b7b39\/library\std\src\panicking.rs:475
   1: core::panicking::panic_fmt
             at /rustc/18bf6b4f01a6feaf7259ba7cdae58031af1b7b39\/library\core\src\panicking.rs:85
   2: core::option::expect_none_failed
             at /rustc/18bf6b4f01a6feaf7259ba7cdae58031af1b7b39\/library\core\src\option.rs:1221
   3: core::result::Result<tuple<>, influx_db_client::error::Error>::unwrap<tuple<>,influx_db_client::error::Error>
             at D:\software\rustup\toolchains\stable-x86_64-pc-windows-msvc\lib\rustlib\src\rust\library\core\src\result.rs:973
   4: w_gui::main::{{closure}}
             at .\src\main.rs:21
   5: core::future::from_generator::{{impl}}::poll<generator-0>
             at D:\software\rustup\toolchains\stable-x86_64-pc-windows-msvc\lib\rustlib\src\rust\library\core\src\future\mod.rs:79
   6: tokio::runtime::enter::{{impl}}::block_on::{{closure}}<core::future::from_generator::GenFuture<generator-0>>
             at D:\software\cargo\registry\src\github.com-1ecc6299db9ec823\tokio-0.2.22\src\runtime\enter.rs:160
   7: tokio::coop::with_budget::{{closure}}<core::task::poll::Poll<tuple<>>,closure-0>
             at D:\software\cargo\registry\src\github.com-1ecc6299db9ec823\tokio-0.2.22\src\coop.rs:127
   8: std::thread::local::LocalKey<core::cell::Cell<tokio::coop::Budget>>::try_with<core::cell::Cell<tokio::coop::Budget>,closure-0,core::task::poll::Poll<tuple<>>>
             at D:\software\rustup\toolchains\stable-x86_64-pc-windows-msvc\lib\rustlib\src\rust\library\std\src\thread\local.rs:265
   9: std::thread::local::LocalKey<core::cell::Cell<tokio::coop::Budget>>::with<core::cell::Cell<tokio::coop::Budget>,closure-0,core::task::poll::Poll<tuple<>>>
             at D:\software\rustup\toolchains\stable-x86_64-pc-windows-msvc\lib\rustlib\src\rust\library\std\src\thread\local.rs:241
  10: tokio::runtime::enter::Enter::block_on<core::future::from_generator::GenFuture<generator-0>>
             at D:\software\cargo\registry\src\github.com-1ecc6299db9ec823\tokio-0.2.22\src\runtime\enter.rs:160
  11: tokio::runtime::thread_pool::ThreadPool::block_on<core::future::from_generator::GenFuture<generator-0>>
             at D:\software\cargo\registry\src\github.com-1ecc6299db9ec823\tokio-0.2.22\src\runtime\thread_pool\mod.rs:82
  12: tokio::runtime::{{impl}}::block_on::{{closure}}<core::future::from_generator::GenFuture<generator-0>>
             at D:\software\cargo\registry\src\github.com-1ecc6299db9ec823\tokio-0.2.22\src\runtime\mod.rs:446
  13: tokio::runtime::context::enter<closure-0,tuple<>>
             at D:\software\cargo\registry\src\github.com-1ecc6299db9ec823\tokio-0.2.22\src\runtime\context.rs:72
  14: tokio::runtime::handle::Handle::enter<closure-0,tuple<>>
             at D:\software\cargo\registry\src\github.com-1ecc6299db9ec823\tokio-0.2.22\src\runtime\handle.rs:76
  15: tokio::runtime::Runtime::block_on<core::future::from_generator::GenFuture<generator-0>>
             at D:\software\cargo\registry\src\github.com-1ecc6299db9ec823\tokio-0.2.22\src\runtime\mod.rs:448
  16: w_gui::main
             at .\src\main.rs:20
  17: core::ops::function::FnOnce::call_once<fn(),tuple<>>
             at D:\software\rustup\toolchains\stable-x86_64-pc-windows-msvc\lib\rustlib\src\rust\library\core\src\ops\function.rs:227
note: Some details are omitted, run with `RUST_BACKTRACE=full` for a verbose backtrace.
error: process didn't exit successfully: `target\debug\w-gui.exe` (exit code: 101)

Hi, I came across this lib, thanks for your work! I have some questions:

1. Does it support influxdb v2?
2. Will influxdb v2 be supported?

if field value contain \" it will fail with error SyntaxError("{error:unable to parse...

Solution can be:

point.add_field("message", InflValue::String(message.replace("\\\"", "\\\\\"")));

The code as is in the Readme panicks with below message:

thread 'main' panicked at 'not currently running on the Tokio runtime.', /home/stoxalpha/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.2.22/src/runtime/handle.rs:118:28

I'm querying and then putting back into the DB a subset of the results set. So for my use case it makes sense to get the results from the DB in the same type as used to insert records into the DB.
I wonder if it would make sense to have this:
https://github.com/driftluo/InfluxDBClient-rs/blob/master/src/keys.rs#L141
Deserialize to this directly:
https://github.com/driftluo/InfluxDBClient-rs/blob/master/src/keys.rs#L9
Instead of using the JSON serde Value type?

The 0.5.2 release of this crate adds lifetime parameters to several structs (Value and Point) in #62. This is an API breaking change, as code which uses these types may no longer compile without code changes. E.g. I had a function like

fn point_for_property_value(
    device: &Device,
    node: &Node,
) -> Point {
  // ...
}

which worked fine with influx-db-client 0.5.1, but with 0.5.2 it fails to build due to the missing lifetime parameter on Point which can't be inferred.

I suggest you yank 0.5.2 and re-release it as 0.6.0.

pub fn new<T: Into<String>>(address: T) -> Self {
        UdpClient {
            hosts: vec![address.into().to_socket_addrs().unwrap().next().unwrap()],
        }
    }

If the hostname isn't resolvable then the udp client panics on unwrap().
It would be great if new returned a Result instead (or alternatively had a try_new fn).

Workaround:

match against address.into().to_socket_addrs() and check that doesn't error first before calling new.

It seems the library currently does not support chunked responses. How would we go about implementing this? Should chunked responses be exposed to the user, or should chunked responses simply be merged?

Specifically, right now query returns a Vec<Node>. What we get from a server with chunked response we would be getting a data structure similar to Vec<Vec<Node>>, or possibly Iter<Vec<Node>>.

influxDB supports bind parameters, this lib should support them as well.

When I try to query with the GROUP BY statement, I always return the following error, but I have no problem with the same statement query in the CLI.

    let sql = format!(
        "select SPREAD(imp) from zydl_01_1 where time > '{}' AND time < '{}'",
        start.to_rfc3339(),
        end.to_rfc3339()
    );
    let q = influxdb.query(&sql, Some(Precision::Seconds)).await;

This will return "data did not match any variant of untagged enum Value at line 1 column 115"

I see that you can set_authentication using a username and password. But, in many cases an authentication token is preferred.

I have a lot of data from 2019 to 2020.
When I wrote data time field contains current time.
How can I write this "old" points?

I've checked the UDP example tests/client_test.rs and I found it a bit confusing.

Do UDP messages go to the "udp" and "telegraf" databases by default?
Or is there a way to specify where do the UDP messages go?

    udp.add_host("127.0.0.1:8090");
    let mut client = Client::default().set_authentication("root", "root");

    let mut point = point!("test");
    point.add_field("foo", Value::String(String::from("bar")));

    let _ = udp.write_point(point).unwrap(); // DOES THIS GO TO UDP/TELEGRAF BY DEFAULT?

    let _ = sleep(Duration::from_secs(1));
    client.switch_database("udp");
    let _ = client.drop_measurement("test").unwrap();
    client.switch_database("telegraf");
    let _ = client.drop_measurement("test").unwrap();

It seems that you assume that Influxdb authentication mechanism will always be enabled, but it also can be disabled, and there is no way no to passe credentials, or even the ability to not pass the "u" and "p" arguments in the url.

Do you have any plans in supporting disabled authentication?

We are using this library to query an influx db version 1.8. container.

When running client.query("SHOW RETENTION POLICIES ON database", None) the HTTP response looks as follows:

{
    "results": [
        {
            "statement_id": 0,
            "series": [
                {
                    "columns": [
                        "name",
                        "duration",
                        "shardGroupDuration",
                        "replicaN",
                        "default"
                    ],
                    "values": [
                        [
                            "autogen",
                            "0s",
                            "168h0m0s",
                            1,
                            false
                        ],
                        [
                            "online_policy",
                            "8736h0m0s",
                            "168h0m0s",
                            1,
                            true
                        ]
                    ]
                }
            ]
        }
    ]
}

The field name is missing which seems to be correct behaviour. However, this library expects it to be present and panics with:

Err(Communication("error decoding response body: missing field 'name' at line 1 column 208"))

If we are not mistaken, this is due to this line and serde trying to match this:

Don't know much about serde but probably one can define the field as optional like I did here? Unfortunately, this would be a breaking change as pub name: String changes. Let me know if you would be interested in a pull request.

I just came to click on the link in the "About" section on GitHub and saw that it points to a specific (and old) version. I think https://docs.rs/influx_db_client/ would be better as docs.rs auto-redirects to the latest version.

To reduce the amount of points we're entering into influx_db one could only send a point (identified by set of tag values) if any of the values are different to last time a point was sent.

I'm thinking others will need this also - this seems pretty generic functionality? Is there already functionality to do this? If not I can take a stab at a PR and we can take it from there.

Default precision of InfluxDB is nanosecond (ref).

Default precision of this crate is second (ref).

hyperium/hyper#1443

Currently, hyper 0.12 is waiting for the release of futures 0.2. Tokio has explicitly supported the work-steal thread pool.

I also hope that this crate can use features to select whether the client uses asynchronous mode, that is say that support synchronous and asynchronous at the same time, the default synchronization mode.

However, currently, Rust/Cargo doesn't support two versions of a library at the same time, which makes it very difficult for me to do it. Maybe 2018-epoch RFC can help me?

Maybe use reqwest? But it currently only supports uses system-native TLS....

Or wait for hyper 0.12 to give up sync mode directly?

I tried to speed up writing points using threadpool. But not at all points is written.

let n_workers = 4;
thread_local!(static CLIENT: Client = Client::new(Url::parse("http://localhost:8086").expect("Cannot parse url"), "data") );
let pool = ThreadPool::new(n_workers);

for chunk in data.chunks(60000) {
    let points = Points::create_new(chunk.to_vec());
    pool.execute(move || {
        CLIENT.with(|c| {
            tokio::runtime::Runtime::new().unwrap().block_on(async move {
                c.write_points(points, Some(Precision::Seconds), None).await.expect("Cannot write points");
            });
        });
            // CLIENT.write_points(points, Some(Precision::Milliseconds), None).await.expect("Cannot write points");
    });
    
}

Why did this happen?

I assume that this is an artefact from early development.

This causes wrong integer inputs.