continuous downsampling using redis for checkpointing and notifications of new data

Author: michaelr524

Cargo.toml

@@ -19,5 +19,6 @@ rayon = "1.0.2"
 config = "0.9.1"
 serde_derive = "1.0.80"
 serde = "1.0.80"
-string_template = "0.1.0"
+string_template = "0.2.1"
 lazy_static = "1.2.0"
+redis = "0.9.1"

README.md

@@ -1,10 +1,32 @@
 # downsampler
 Utilities for transforming InfluxDB time series data
 
+## Get up and running
+The basics are:
+
+1. Install Rust (to build the project): https://rustup.rs/
+1. git clone [email protected]:michaelr524/downsampler.git
+1. cd downsampler
+1. cargo build --release
+1. Edit config.toml to your needs
+1. Run like this: target/release/downsampler downsample -s ‘2018-08-13 15:22:06’ -e ‘2018-08-17 19:03:17’
+
 The following operations are currently supported:
 
 #### downsample
 Create downsampled series from other series.
 
 #### split
 Creates many measurement with a single serie from measurements with multiple series.
+
+#### listen - continuous downsampling
+Continuously downsampling the configured series as new data arrives.
+You need to have Redis running. 
+Redis is used for notifying downsampler about new data. 
+Additionally, Redis is used to store a checkpoint so that downsampler knows how to restart from the point where it stopped last time it ran.
+The way to send updates to Redis is:
+```
+HSET downsampler_updates "binance_BTCUSDT" "1537710900000000000"
+```
+The key will be used in the downsampled series name (check `listen` > `measurement_template` in `config.toml`) and the value is the timestamp in nanoseconds format.
+Downsampler will downsample all the data from the previous checkpoint up to the given timestamp.

config.toml

@@ -1,3 +1,14 @@
+[listen]
+redis_url = "redis://127.0.0.1:6379"
+poll_sleep_ms = 250
+measurement_template = "trades_{{id}}_{{time_interval}}"
+query_template = """
+select price, amount
+from glukoz."glukoz-rp".trades_{{id}}_{{time_interval}}
+WHERE time >= {{start}} AND time < {{end}}
+limit {{limit}}
+"""
+
 [influxdb]
 url = "http://localhost:8086"
 db = "glukoz"

src/cmdargs.rs

@@ -1,4 +1,5 @@
 use chrono::{format::ParseError, offset::TimeZone, NaiveDateTime, Utc};
+use clap::ArgMatches;
 use clap::{crate_version, App, Arg, ArgGroup, SubCommand};
 use crate::utils::time::truncate_seconds;
 use failure_derive::Fail;
@@ -47,17 +48,17 @@ pub enum Error {
     DurationTooLong(OutOfRangeError),
 }
 
-pub enum Command {
-    Downsample,
-    Split,
-}
-
-pub struct CmdArgs {
+pub struct TimePeriod {
     pub raw_start: NaiveDateTime,
     pub raw_end: NaiveDateTime,
     pub start: NaiveDateTime,
     pub end: NaiveDateTime,
-    pub command: Command,
+}
+
+pub enum CmdArgs {
+    Downsample(TimePeriod),
+    Split(TimePeriod),
+    Listen,
 }
 
 fn args_definitions<'a, 'b>() -> App<'a, 'b> {
@@ -121,23 +122,32 @@ years, year, y -- defined as 365.25 days
                 .arg(duration_arg.clone())
                 .group(period_end_group.clone()),
         )
+        .subcommand(SubCommand::with_name("listen").about("Continuous downsampling"))
 }
 
 pub fn parse_args() -> Result<CmdArgs, Error> {
     let args = args_definitions().get_matches();
 
-    let (args, command) = match args.subcommand() {
-        ("downsample", Some(subcommand)) => (subcommand, Command::Downsample),
-        ("split", Some(subcommand)) => (subcommand, Command::Split),
+    match args.subcommand() {
+        ("downsample", Some(subcommand)) => {
+            let time_period = parse_time_period(subcommand)?;
+            Ok(CmdArgs::Downsample(time_period))
+        }
+        ("split", Some(subcommand)) => {
+            let time_period = parse_time_period(subcommand)?;
+            Ok(CmdArgs::Split(time_period))
+        }
+        ("listen", Some(_)) => Ok(CmdArgs::Listen),
         _ => {
             args_definitions().print_help().unwrap();
             return Err(Error::CommandMissing);
         }
-    };
+    }
+}
 
+fn parse_time_period(args: &ArgMatches) -> Result<TimePeriod, Error> {
     let raw_start = parse_datetime(args.value_of("start"))
         .map_err(|e| Error::InvalidStartArgument(Box::new(e)))?;
-
     let raw_end = if args.is_present("duration") {
         let duration = parse_duration(args.value_of("duration"))
             .map_err(|e| Error::InvalidDurationArgument(Box::new(e)))?;
@@ -147,27 +157,37 @@ pub fn parse_args() -> Result<CmdArgs, Error> {
             .map_err(|e| Error::InvalidEndArgument(Box::new(e)))?;
         datetime
     };
-
     let start = truncate_seconds(raw_start);
     let end = truncate_seconds(raw_end);
 
-    Ok(CmdArgs {
+    Ok(TimePeriod {
         start,
         end,
         raw_start,
         raw_end,
-        command,
     })
 }
 
 pub fn print_args_info(settings: &CmdArgs) {
-    println!("Period {:?} - {:?}", settings.raw_start, settings.raw_end);
-    println!("Period truncated {:?} - {:?}", settings.start, settings.end);
-    println!(
-        "Period in nanos {:?} - {:?}",
-        settings.start.timestamp_nanos(),
-        settings.end.timestamp_nanos()
-    );
+    if let Some(time_period) = match settings {
+        CmdArgs::Downsample(time_period) => Some(time_period),
+        CmdArgs::Split(time_period) => Some(time_period),
+        CmdArgs::Listen => None,
+    } {
+        println!(
+            "Period {:?} - {:?}",
+            time_period.raw_start, time_period.raw_end
+        );
+        println!(
+            "Period truncated {:?} - {:?}",
+            time_period.start, time_period.end
+        );
+        println!(
+            "Period in nanos {:?} - {:?}",
+            time_period.start.timestamp_nanos(),
+            time_period.end.timestamp_nanos()
+        );
+    }
 }
 
 fn parse_duration(date_string: Option<&str>) -> Result<Duration, Error> {

src/downsampler.rs

@@ -1,28 +1,14 @@
-use chrono::{NaiveDateTime, TimeZone, Utc};
+use crate::cmdargs::TimePeriod;
+use crate::downsampling::is_downsampling_interval;
 use crate::{
-    cmdargs::CmdArgs,
-    influx::{
-        extract_float_value, from_json_values, get_range, influx_client, save_points, to_point,
-        Error, FieldValue,
-    },
-    lttb::{lttb_downsample, DataPoint},
-    settings::{Config, Field},
-    utils::{error::print_err_and_exit, time::intervals},
+    downsampling::downsample_period, influx::influx_client, settings::Config,
+    utils::time::intervals,
 };
-use influx_db_client::Client;
-use influx_db_client::Point;
-use lazy_static::lazy_static;
 use rayon::prelude::*;
 use std::collections::HashMap;
-use std::ops::Sub;
-use std::time::Duration as StdDuration;
 use string_template::Template;
 use time::Duration;
 
-lazy_static! {
-    static ref UNIX_EPOCH: NaiveDateTime = Utc.timestamp(0, 0).naive_utc();
-}
-
 pub fn pre_render_names(config: &Config, template: Template) -> HashMap<(u64, &str), String> {
     let mut map: HashMap<(u64, &str), String> =
         HashMap::with_capacity(config.vars.ids.len() * config.downsampler.intervals.len());
@@ -40,7 +26,7 @@ pub fn pre_render_names(config: &Config, template: Template) -> HashMap<(u64, &s
     map
 }
 
-pub fn downsample(args: &CmdArgs, config: &Config) -> () {
+pub fn downsample(args: &TimePeriod, config: &Config) -> () {
     let client = influx_client(
         &config.influxdb.url,
         &config.influxdb.db,
@@ -58,116 +44,24 @@ pub fn downsample(args: &CmdArgs, config: &Config) -> () {
 
         for (start, _end) in intervals(args.start, args.end, Duration::seconds(1)) {
             for interval_period in config.downsampler.intervals.iter() {
-                if start.signed_duration_since(*UNIX_EPOCH).num_seconds()
-                    % (interval_period.duration_secs as i64)
-                    == 0
-                    {
-                        let measurement_name = measurements
-                            .get(&(interval_period.duration_secs, id))
-                            .unwrap();
-
-                        downsample_period(
-                            config,
-                            &client,
-                            &query_template,
-                            id,
-                            start,
-                            interval_period.duration_secs,
-                            measurement_name,
-                        );
-                    }
+                if is_downsampling_interval(&start, interval_period) {
+                    let measurement_name = measurements
+                        .get(&(interval_period.duration_secs, id))
+                        .unwrap();
+
+                    downsample_period(
+                        config,
+                        &client,
+                        &query_template,
+                        id,
+                        start,
+                        interval_period.duration_secs,
+                        measurement_name,
+                    );
+                }
             }
         }
 
         println!("end {}", id);
     });
 }
-
-pub fn downsample_period(
-    config: &Config,
-    client: &Client,
-    query_template: &Template,
-    id: &str,
-    end: NaiveDateTime,
-    interval_duration_secs: u64,
-    measurement_name: &str,
-) {
-    let duration = Duration::from_std(StdDuration::from_secs(interval_duration_secs)).unwrap();
-    let begin = end.sub(duration);
-
-    let query_str = build_query(&query_template, id, begin, end, 0, "raw");
-    let series = match get_range(&client, &query_str) {
-        Ok(series) => series,
-        Err(err) => match err {
-            Error::NoResult => return,
-            e => print_err_and_exit(e),
-        },
-    };
-    let vals = from_json_values(series.values, &config.downsampler.fields)
-        .unwrap_or_else(|e| print_err_and_exit(e));
-    //                let _count = vals.iter().count();
-    //                println!("{} - [{} - {}] ({})", i, start, end, _count);
-    let subset = lttb_downsample(
-        &vals,
-        60,
-        config.downsampler.x_field_index,
-        config.downsampler.y_field_index,
-    );
-    let points = to_influx_points(measurement_name, &vals, &subset, &config.downsampler.fields);
-    //                println!("{:#?}", &points);
-    // TODO: handle errors
-    save_points(&client, &config.influxdb.retention_policy, points).unwrap();
-}
-
-pub fn to_influx_points(
-    measurement_name: &str,
-    raw: &[Vec<FieldValue>],
-    downsampled: &Option<Vec<&Vec<FieldValue>>>,
-    fields: &[Field],
-) -> Vec<Point> {
-    match downsampled {
-        Some(downsampled) => downsampled
-            .iter()
-            .map(|v| to_point(v, measurement_name, fields))
-            .collect(),
-        _ => raw
-            .iter()
-            .map(|v| to_point(v, measurement_name, fields))
-            .collect(),
-    }
-}
-
-// pass `limit: 0` to disable limit
-pub fn build_query(
-    query_template: &Template,
-    id: &str,
-    start: NaiveDateTime,
-    end: NaiveDateTime,
-    limit: i64,
-    time_interval: &str,
-) -> String {
-    let start_str = start.timestamp_nanos().to_string();
-    let end_str = end.timestamp_nanos().to_string();
-    let limit_str = limit.to_string();
-
-    let mut map = HashMap::new();
-    map.insert("id", id);
-    map.insert("start", &start_str);
-    map.insert("end", &end_str);
-    map.insert("limit", &limit_str);
-    map.insert("time_interval", time_interval);
-
-    query_template.render(&map)
-}
-
-impl DataPoint for Vec<FieldValue> {
-    fn get_x(&self, index: usize) -> f64 {
-        let field_value = self.get(index).unwrap();
-        extract_float_value(field_value)
-    }
-
-    fn get_y(&self, index: usize) -> f64 {
-        let field_value = self.get(index).unwrap();
-        extract_float_value(field_value)
-    }
-}