Batched PUT requests

[robagar]

This adds the ability to send multiple data points in a single PUT to InfluxDB, giving a huge performance gain for high frequency data.

Volume configuration:

• put_batch_size - Maximum number of data points per PUT request
• put_batch_timeout_ms - Milliseconds before sending a batch if not full

If put_batch_size is not set it reverts to the original behaviour of sending each data point in its own PUT request.

[gabrik]

Hi @robagar, thank you for your contribution.

In order to accept it we need you to sing the Eclipse Contribution Agreement.

[robagar]

ECA signed

[Mallets]

@Charles-Schleich could you review the PR?
@robagar I see you are only targeting v1, before merging this kind of changes it would be good to ensure that it is available also for v2.

[robagar]

We don't use InfluxDB v2, and have no intention to. It looks like a dead end, with Flux being abandoned for v3.

[Charles-Schleich]

Hi @robagar, thank you for the contribution !
A couple of things to note:
We strive to keep the different versions of the influxdb plugin aligned in terms of features and behavior, so to consider merging PR's we would need the same batching feature in version 2 as well.
The configuration options would also need to be added to the README for both versions so the users know they are available.
PR are generally acceptable if they cover config, documentation, tests, and ideally some performance tests/setup that we can reproduce our side.

[robagar]

Skimming the code it looks fairly simple to add the batching to v2 - the client write method takes a stream as input - but frankly that would be better done by someone who is going to use & test it properly.

[Charles-Schleich]

The plugins can be linked statically or dynamically to an instance of zenohd.
In the case that we link statically, we want to reuse the Tokio runtime of zenohd and not use the TOKIO_RUNTIME stored in the lazy static:
This can be achieved by a check on the tokio handle, and running it on the correct executor.

let batch_future = async move{...};
match tokio::runtime::Handle::try_current() {
    Ok(handle) => handle.spawn(batch_future),
    Err(_) => TOKIO_RUNTIME.spawn(batch_future),
};

[Charles-Schleich]

These config options must be documented in the README (there is a pending PR for an example Config file)

[Charles-Schleich]

I'm not 100% sure about returning aStorageInsertionResult::Inserted as the insertion could fail inside the batch, but it has been reported to zenoh as inserted. This will likely have implications inside replication and storage alignment.

The suggestion to add a Pending status could work, however then we will also have to add a Completed status that will have to be signaled to zenoh that all of those Puts have been successful as a group.
This may require an additional ID per Put when batching, and upon successful write, the ID's of the batch would have to be signaled back to zenoh.

@JEnoch, What do we think about adding a Pending and Complete system ?
it adds a little bit of complication, and i'd like to make sure we don't break alignment due to batching.

[Charles-Schleich]

Suggestion / nit: let mut put_batch_tx = None; above 322 and just over write with a Some() value on 410.

[robagar]

HIya.. any chance of this making it into the 1.0.0 release?

[Mallets]

After some discussion, the proper support of batching would require some work on the backend API.
There are internal mechanisms in Zenoh, like storage alignment, that requires having the knowledge on wether the put of a message succeeded or not. With the current backend API, it's not possible to bubble up this information when batching is in place. This would lead to an incorrect behaviour when replication is activated.

So I'm afraid some more work is required in Zenoh to properly support this use case.

[robagar]

OK, but please be aware that as it stands writing to the InfluxDB backend is just too slow - it falls behind here even with ~100Hz updates, which is hardly excessive.