Merge pull request 'hotfix: `configuration property "influx.org_id" not found`' (#111) from influx-docker into main

Reviewed-on: #111

Dockerfile

@@ -0,0 +1,27 @@
+FROM rust:1.57.0-bullseye as builder
+
+RUN curl -sL https://deb.nodesource.com/setup_16.x | bash -
+
+RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install --no-install-recommends -y \
+    nodejs \
+    gcc \
+    g++ \
+    make \
+    musl-tools \
+    && rm -rf /var/lib/apt/lists/*
+
+RUN npm install --global yarn
+
+RUN rustup target add x86_64-unknown-linux-musl
+
+ADD . ./
+
+RUN make
+
+FROM debian:bullseye-slim
+
+COPY --from=builder /release/hastic /hastic
+COPY --from=builder /release/config.toml /config.toml
+COPY --from=builder /release/public /public
+
+CMD ["./hastic"]

README.md

@@ -17,6 +17,42 @@ instance for getting metrics.
 make
 ```
 
+### Configure
+Hastic can be configured using config-file or environment variables.
+
+At first, choose which datasource you'll be using: `prometheus` or `influx`. Only one can be used at a time.
+
+#### Config-file
+- copy the config example to the release directory:
+```bash
+cp config.example.toml release/config.toml
+```
+- edit the config file, e.g. using `nano`:
+```bash
+nano release/config.toml
+```
+
+#### Environment variables
+All config fields are also available as environment variables with `HASTIC_` prefix
+
+Variable name structure:
+- for high-level fields: `HASTIC_<field_name>`, e.g. `HASTIC_PORT`
+- for nested fields: `HASTIC_<category_name>__<field_name>`, e.g. `HASTIC_PROMETHEUS__URL`
+
+Environment variables can be set either by exporting them (they'll be actual until a bash-session is closed):
+```bash
+export HASTIC_PORT=8000
+export HASTIC_PROMETHEUS__URL=http://localhost:9090
+export HASTIC_PROMETHEUS__QUERY=rate(go_memstats_alloc_bytes_total[5m])
+```
+
+or specifing them in a run command (they'll be actual only for one run):
+```bash
+HASTIC_PORT=8000 HASTIC_PROMETHEUS__URL=http://localhost:9090 HASTIC_PROMETHEUS__QUERY=rate(go_memstats_alloc_bytes_total[5m]) ./release/hastic
+```
+
+### Run
+
 ```
 cd release
 ./hastic

client/src/components/AnlyticsStatus.vue

@@ -1,10 +1,12 @@
 <template>
   <div class="analytic-status">
-    analytic status: <strong> {{ status }} </strong>
+    analytic status: <strong> {{ status.message }} </strong>
   </div>
 </template>
 
 <script lang="ts">
+import { AnalyticStatus } from "@/store";
+
 import { defineComponent } from 'vue';
 
 
@@ -13,7 +15,7 @@ export default defineComponent({
   components: {
   },
   computed: {
-    status() {
+    status(): AnalyticStatus {
       return this.$store.state.analyticStatus;
     }
   }

client/src/components/pods/pattern_pod.ts

@@ -79,7 +79,6 @@ export class PatternPod extends HasticPod<UpdateDataCallback> {
     } else {
       console.log('took from range from default');
     }
-    console.log(from + " ---- " + to);
 
     this.udc({ from, to })
       .then(resp => {

client/src/services/analytics.service.ts

@@ -5,20 +5,33 @@ import axios from 'axios';
 
 import { getGenerator } from '@/utils';
 
-import _ from 'lodash';
-import { 
+import {
   AnalyticUnitType, AnlyticUnitConfig,
   PatternConfig, ThresholdConfig, AnomalyConfig
 } from "@/types/analytic_units";
 import { AnomalyHSR } from "@/types";
+import { AnalyticStatus } from "@/store";
+
+import _ from 'lodash';
+
 
 const ANALYTICS_API_URL = API_URL + "analytics/";
 
-export async function getStatus(): Promise<string> {
+export async function getStatus(): Promise<AnalyticStatus> {
   const uri = ANALYTICS_API_URL + `status`;
-  const res = await axios.get(uri);
-  const data = res['data'] as any;
-  return data.status;
+  try {
+    const res = await axios.get<{ status: string }>(uri);
+    const data = res.data;
+    return {
+      available: true,
+      message: data.status
+    };
+  } catch (e) {
+    return {
+      available: false,
+      message: e.message
+    };
+  }
 }
 
 export async function getConfig(): Promise<[AnalyticUnitType, AnlyticUnitConfig]> {
@@ -54,8 +67,8 @@ export async function patchConfig(patchObj: any) {
   await axios.patch(uri, patchObj);
 }
 
-export function getStatusGenerator(): AsyncIterableIterator<string> {
-  return getGenerator<string>(100, getStatus);
+export function getStatusGenerator(): AsyncIterableIterator<AnalyticStatus> {
+  return getGenerator<AnalyticStatus>(100, getStatus);
 }
 
 
@@ -68,6 +81,6 @@ export async function getHSRAnomaly(from: number, to: number): Promise<AnomalyHS
   const res = await axios.get(uri);
 
   const values = res["data"]["AnomalyHSR"];
-  
+
   return values as AnomalyHSR;
 }

client/src/store/index.ts

@@ -12,23 +12,30 @@ const _SET_STATUS_GENERATOR = '_SET_STATUS_GENERATOR';
 
 // TODO: consts for actions
 
+export type AnalyticStatus = {
+  available: boolean,
+  message: string,
+}
 
 type State = {
-  analyticStatus: string,
+  analyticStatus: AnalyticStatus,
   analyticUnitType?: AnalyticUnitType,
   analyticUnitConfig?: AnlyticUnitConfig,
-  _statusGenerator: AsyncIterableIterator<string>
+  _statusGenerator: AsyncIterableIterator<AnalyticStatus>
 }
 
 const store = createStore<State>({
   state: {
-    analyticStatus: 'loading...',
+    analyticStatus: {
+      available: false,
+      message: 'loading...',
+    },
     analyticUnitType: null,
     analyticUnitConfig: null,
     _statusGenerator: null
   },
   mutations: {
-    [SET_ANALYTICS_STATUS](state, status: string) {
+    [SET_ANALYTICS_STATUS](state, status: AnalyticStatus) {
       state.analyticStatus = status;
     },
     [SET_DETECTOR_CONFIG](state, { analyticUnitType, analyticUnitConfig }) {      
@@ -38,7 +45,7 @@ const store = createStore<State>({
     // [PATCH_CONFIG](state, patchObj) {
     //   patchConfig(patchConfig)
     // }
-    [_SET_STATUS_GENERATOR](state, generator: AsyncIterableIterator<string>) {
+    [_SET_STATUS_GENERATOR](state, generator: AsyncIterableIterator<AnalyticStatus>) {
       state._statusGenerator = generator;
     }
   },

client/src/views/Home.vue

@@ -2,50 +2,53 @@
   <div class="home">
     <img alt="Vue logo" src="../assets/logo.png">
     <graph ref="graph" />
-    
+
     <analytic-status />
-    <div>
-      Analytic unit type:
-      <select :value="analyticUnitType" @change="changeAnalyticUnitType">
-        <option disabled value="">Please Select</option>
-        <option v-bind:key="option" v-for="option in analyticUnitTypes" :value="option">{{option}}</option>
-      </select> <br/><br/>
-    </div>
-    <div id="controls">
-      <div v-if="analyticUnitType == analyticUnitTypes[0]">
-        Threshold: 
-        <input :value="analyticUnitConfig.threshold" @change="thresholdChange" /> <br/><br/>
-      </div>
-      <div v-if="analyticUnitType == analyticUnitTypes[1]">
-        Hold <pre>S</pre> to label patterns; 
-        Hold <pre>A</pre> to label anti patterns <br/>
-        Hold <pre>D</pre> to delete patterns
-        <br/>
-        <hr/>
-        Correlation score:
-        <input :value="analyticUnitConfig.correlation_score" @change="correlationScoreChange" /> <br/>
-        Anti correlation score: 
-        <input :value="analyticUnitConfig.anti_correlation_score" @change="antiCorrelationScoreChange" /> <br/>
-        Model score: 
-        <input :value="analyticUnitConfig.model_score" @change="modelScoreChange" /> <br/>
-        Threshold score: 
-        <input :value="analyticUnitConfig.threshold_score" @change="thresholdScoreChange" /> <br/><br/>
-        <button @click="clearAllLabeling"> clear all labeling </button>
+
+    <template v-if="analyticStatus.available">
+      <div>
+        Analytic unit type:
+        <select :value="analyticUnitType" @change="changeAnalyticUnitType">
+          <option disabled value="">Please Select</option>
+          <option v-bind:key="option" v-for="option in analyticUnitTypes" :value="option">{{option}}</option>
+        </select> <br/><br/>
       </div>
-      <div v-if="analyticUnitType == analyticUnitTypes[2]">
-        Hold <pre>Z</pre> to set seasonality timespan
-        <hr/>
-        <!-- Alpha:
-        <input :value="analyticUnitConfig.alpha" @change="alphaChange" /> <br/> -->
-        Confidence:
-        <input :value="analyticUnitConfig.confidence" @change="confidenceChange" /> <br/>
-        Seasonality:
-        <input :value="analyticUnitConfig.seasonality" @change="seasonalityChange" /> <br/>
-        Seasonality iterations:
-        <input :value="analyticUnitConfig.seasonality_iterations" @change="seasonalityIterationsChange" /> <br/>
-        <br/>
+      <div id="controls">
+        <div v-if="analyticUnitType == analyticUnitTypes[0]">
+          Threshold:
+          <input v-model="analyticUnitConfig.threshold" @change="thresholdChange" /> <br/><br/>
+        </div>
+        <div v-if="analyticUnitType == analyticUnitTypes[1]">
+          Hold <pre>S</pre> to label patterns;
+          Hold <pre>A</pre> to label anti patterns <br/>
+          Hold <pre>D</pre> to delete patterns
+          <br/>
+          <hr/>
+          Correlation score:
+          <input v-model="analyticUnitConfig.correlation_score" @change="correlationScoreChange" /> <br/>
+          Anti correlation score:
+          <input v-model="analyticUnitConfig.anti_correlation_score" @change="antiCorrelationScoreChange" /> <br/>
+          Model score:
+          <input v-model="analyticUnitConfig.model_score" @change="modelScoreChange" /> <br/>
+          Threshold score:
+          <input v-model="analyticUnitConfig.threshold_score" @change="thresholdScoreChange" /> <br/><br/>
+          <button @click="clearAllLabeling"> clear all labeling </button>
+        </div>
+        <div v-if="analyticUnitType == analyticUnitTypes[2]">
+          Hold <pre>Z</pre> to set seasonality timespan
+          <hr/>
+          <!-- Alpha:
+          <input :value="analyticUnitConfig.alpha" @change="alphaChange" /> <br/> -->
+          Confidence:
+          <input v-model="analyticUnitConfig.confidence" @change="confidenceChange" /> <br/>
+          Seasonality:
+          <input v-model="analyticUnitConfig.seasonality" @change="seasonalityChange" /> <br/>
+          Seasonality iterations:
+          <input v-model="analyticUnitConfig.seasonality_iterations" @change="seasonalityIterationsChange" /> <br/>
+          <br/>
+        </div>
       </div>
-    </div>
+    </template>
   </div>
 </template>
 
@@ -140,6 +143,9 @@ export default defineComponent({
     },
     analyticUnitConfig() {
       return this.$store.state.analyticUnitConfig;
+    },
+    analyticStatus() {
+      return this.$store.state.analyticStatus;
     }
   }
 });

docker-compose.yml

@@ -0,0 +1,12 @@
+
+version: '3'
+services:
+  app:
+    image: hastic/hastic:latest
+    restart: always
+    environment:
+      HASTIC_PORT: "4347"
+      HASTIC_PROMETHEUS__URL: "http://demo.robustperception.io:9090"
+      HASTIC_PROMETHEUS__QUERY: "rate(go_memstats_alloc_bytes_total[1m])"
+    ports:
+      - "4347:4347"

server/config.example.toml

@@ -1,8 +1,10 @@
 port = 4347
 
-[prometheus]
-url = "http://localhost:9090"
-query = "rate(go_memstats_alloc_bytes_total[5m])"
+# one of datasource sections (prometheus / influx) should be uncommented and edited corresponding to your environment
+
+# [prometheus]
+# url = "http://localhost:9090"
+# query = "rate(go_memstats_alloc_bytes_total[5m])"
 
 
 # [influx]
@@ -16,8 +18,7 @@ query = "rate(go_memstats_alloc_bytes_total[5m])"
 #   |> yield(name: "mean")
 # """
 
-[alerting]
-type = "webhook"
-interval = 10 # in seconds
-endpoint = "http://localhost:9092"
-
+# [alerting]
+# type = "webhook"
+# interval = 10 # in seconds
+# endpoint = "http://localhost:9092"

server/src/config.rs

@@ -27,6 +27,7 @@ pub struct Config {
 
 fn resolve_datasource(config: &config::Config) -> anyhow::Result<DatasourceConfig> {
     if config.get::<String>("prometheus.url").is_ok() {
+        println!("using Prometheus");
         return Ok(DatasourceConfig::Prometheus(PrometheusConfig {
             url: config.get("prometheus.url")?,
             query: config.get("prometheus.query")?,
@@ -34,6 +35,7 @@ fn resolve_datasource(config: &config::Config) -> anyhow::Result<DatasourceConfi
     }
 
     if config.get::<String>("influx.url").is_ok() {
+        println!("using Influx");
         return Ok(DatasourceConfig::Influx(InfluxConfig {
             url: config.get("influx.url")?,
             org_id: config.get("influx.org_id")?,
@@ -42,7 +44,7 @@ fn resolve_datasource(config: &config::Config) -> anyhow::Result<DatasourceConfi
         }));
     }
 
-    return Err(anyhow::format_err!("no datasource found"));
+    return Err(anyhow::format_err!("please configure a datasource"));
 }
 
 fn resolve_alerting(config: &config::Config) -> anyhow::Result<Option<AlertingConfig>> {
@@ -80,24 +82,38 @@ fn resolve_alerting(config: &config::Config) -> anyhow::Result<Option<AlertingCo
 // config::Environment doesn't support nested configs, e.g. `alerting.type`,
 // so I've copied this from:
 // https://github.com/rust-lang/mdBook/blob/f3e5fce6bf5e290c713f4015947dc0f0ad172d20/src/config.rs#L132
-// so that `__` can be used in env variables instead of `.`, 
+// so that `__` can be used in env variables instead of `.`,
 // e.g. `HASTIC_ALERTING__TYPE` -> alerting.type
 pub fn update_from_env(config: &mut config::Config) {
     let overrides =
         env::vars().filter_map(|(key, value)| parse_env(&key).map(|index| (index, value)));
 
     for (key, value) in overrides {
-        println!("{} => {}", key, value);
         config.set(&key, value).unwrap();
     }
 }
 
+pub fn print_config(config: config::Config) {
+    // TODO: support any nesting level
+    let sections = config.to_owned().cache.into_table().unwrap();
+    for (section_name, values) in sections {
+        match values.clone().into_table() {
+            Err(_) => println!("{} => {}", section_name, values),
+            Ok(section) => {
+                for (key, value) in section {
+                    println!("{}.{} => {}", section_name, key, value);
+                }
+            }
+        }
+    }
+}
+
 fn parse_env(key: &str) -> Option<String> {
     const PREFIX: &str = "HASTIC_";
 
     if key.starts_with(PREFIX) {
         let key = &key[PREFIX.len()..];
-        Some(key.to_lowercase().replace("__", ".").replace("_", "-"))
+        Some(key.to_lowercase().replace("__", "."))
     } else {
         None
     }
@@ -114,10 +130,11 @@ impl Config {
         update_from_env(&mut config);
 
         if config.get::<u16>("port").is_err() {
-            config.set("port", "8000").unwrap();
+            config.set("port", 4347).unwrap();
         }
 
-        // TODO: print resulted config (perfectly, it needs adding `derive(Debug)` in `subbeat`'s `DatasourceConfig`)
+        print_config(config.clone());
+
         Ok(Config {
             port: config.get::<u16>("port").unwrap(),
             datasource_config: resolve_datasource(&config)?,

server/src/main.rs

@@ -1,6 +1,6 @@
 mod api;
 
-use hastic::services::{analytic_service, analytic_unit_service, metric_service, segments_service};
+use hastic::services::{analytic_service, analytic_unit_service, metric_service, segments_service, data_service};
 
 use anyhow;
 
@@ -9,9 +9,10 @@ async fn main() -> anyhow::Result<()> {
     let config = hastic::config::Config::new()?;
     let cfg_clone = config.clone();
 
-    let analytic_unit_service = analytic_unit_service::AnalyticUnitService::new()?;
+    let data_service = data_service::DataService::new()?;
+    let analytic_unit_service = analytic_unit_service::AnalyticUnitService::new(&data_service)?;
     let metric_service = metric_service::MetricService::new(&config.datasource_config);
-    let segments_service = segments_service::SegmentsService::new()?;
+    let segments_service = segments_service::SegmentsService::new(&data_service)?;
 
     let mut analytic_service = analytic_service::AnalyticService::new(
         analytic_unit_service.clone(),

server/src/services/analytic_service/analytic_service.rs

@@ -137,7 +137,7 @@ impl AnalyticService {
         };
         let tx = self.tx.clone();
         let au = self.analytic_unit.as_ref().unwrap().clone();
-        let dr = DetectionRunner::new(self.metric_service.clone(), tx, drcfg, au);
+        let dr = DetectionRunner::new(tx,self.metric_service.clone(), drcfg, au);
         self.detection_runner = Some(dr);
         self.detection_runner.as_mut().unwrap().run(from);
 
@@ -245,6 +245,9 @@ impl AnalyticService {
                             .set_last_detection(id, timestamp)
                             .unwrap();
                     }
+                    ResponseType::DetectionRunnerDetection(from, to) => {
+                        println!("detection: {} {}", from, to);
+                    }
                     ResponseType::LearningStarted => {
                         self.analytic_unit_learning_status = LearningStatus::Learning
                     }
@@ -299,8 +302,9 @@ impl AnalyticService {
                     self.consume_request(RequestType::RunLearning);
                     match tx.send(()) {
                         Ok(_) => {}
-                        Err(_e) => {
+                        Err(e) => {
                             println!("Can`t send patch config notification");
+                            println!("{:?}", e);
                         }
                     }
                     return;
@@ -312,8 +316,9 @@ impl AnalyticService {
                             au.unwrap().write().await.set_config(cfg);
                             match tx.send(()) {
                                 Ok(_) => {}
-                                Err(_e) => {
+                                Err(e) => {
                                     println!("Can`t send patch config notification");
+                                    println!("{:?}", e);
                                 }
                             }
                         }
@@ -323,8 +328,9 @@ impl AnalyticService {
                 // TODO: check if we need this else
                 match tx.send(()) {
                     Ok(_) => {}
-                    Err(_e) => {
+                    Err(e) => {
                         println!("Can`t send patch config notification");
+                        println!("{:?}", e);
                     }
                 }
             }
@@ -337,8 +343,9 @@ impl AnalyticService {
             self.consume_request(RequestType::RunLearning);
             match tx.send(()) {
                 Ok(_) => {}
-                Err(_e) => {
+                Err(e) => {
                     println!("Can`t send patch config notification");
+                    println!("{:?}", e);
                 }
             }
         }

server/src/services/analytic_service/analytic_unit/pattern_analytic_unit.rs

@@ -220,8 +220,8 @@ impl AnalyticUnit for PatternAnalyticUnit {
         return self.id.to_owned();
     }
     fn get_detection_window(&self) -> u64 {
-        // TODO: return window based on real petterns info
-        return DETECTION_STEP;
+        let lr = self.learning_results.as_ref().unwrap();
+        return lr.avg_pattern_length as u64;
     }
     fn set_config(&mut self, config: AnalyticUnitConfig) {
         if let AnalyticUnitConfig::Pattern(cfg) = config {

server/src/services/analytic_service/detection_runner.rs

@@ -7,11 +7,10 @@ use crate::services::metric_service::MetricService;
 use super::types::{AnalyticServiceMessage, AnalyticUnitRF, DetectionRunnerConfig, ResponseType};
 use tokio::time::{sleep, Duration};
 
-const DETECTION_STEP: u64 = 10;
 
 pub struct DetectionRunner {
-    metric_service: MetricService,
     tx: mpsc::Sender<AnalyticServiceMessage>,
+    metric_service: MetricService,
     config: DetectionRunnerConfig,
     analytic_unit: AnalyticUnitRF,
     running_handler: Option<tokio::task::JoinHandle<()>>,
@@ -19,8 +18,8 @@ pub struct DetectionRunner {
 
 impl DetectionRunner {
     pub fn new(
-        metric_service: MetricService,
         tx: mpsc::Sender<AnalyticServiceMessage>,
+        metric_service: MetricService,
         config: DetectionRunnerConfig,
         analytic_unit: AnalyticUnitRF,
     ) -> DetectionRunner {
@@ -34,7 +33,6 @@ impl DetectionRunner {
     }
 
     pub fn run(&mut self, from: u64) {
-        // TODO: get last detection timestamp from persistance
         // TODO: set last detection from "now"
         if self.running_handler.is_some() {
             self.running_handler.as_mut().unwrap().abort();
@@ -47,9 +45,12 @@ impl DetectionRunner {
             async move {
                 // TODO: run detection "from" for big timespan
                 // TODO: parse detections to webhooks
-                // TODO: define window for detection
                 // TODO: handle case when detection is in the end and continues after "now"
+                //       it's better to make an issue on github
+                // TODO: find place to update analytic unit model
+
                 let window_size = au.as_ref().read().await.get_detection_window();
+                let detection_step = ms.get_detection_step();
                 let mut t_from = from - window_size;
                 let mut t_to = from;
 
@@ -68,11 +69,18 @@ impl DetectionRunner {
                     let detections = a.detect(ms.clone(), t_from, t_to).await.unwrap();
 
                     for d in detections {
-                        println!("detection: {} {}", d.0, d.1);
+                        match tx
+                        .send(AnalyticServiceMessage::Response(Ok(
+                            ResponseType::DetectionRunnerDetection(d.0, d.1),
+                        )))
+                        .await
+                        {
+                            Ok(_) => {}
+                            Err(_e) => println!("Fail to send detection runner detection notification"),
+                        }
                     }
 
-                    // TODO: run detection periodically
-                    // TODO: set info about detections to tx
+                    // TODO: send info about detections to tx
 
                     match tx
                         .send(AnalyticServiceMessage::Response(Ok(
@@ -88,6 +96,8 @@ impl DetectionRunner {
                     }
 
                     sleep(Duration::from_secs(cfg.interval)).await;
+                    t_from += detection_step;
+                    t_to += detection_step;
                 }
             }
         }));

server/src/services/analytic_service/types.rs

@@ -45,6 +45,7 @@ impl Default for LearningTrain {
 pub enum ResponseType {
     DetectionRunnerStarted(u64),
     DetectionRunnerUpdate(String, u64), // analytic_unit id and timestamp
+    DetectionRunnerDetection(u64, u64), // TODO: add more into about analytic unit and more
     LearningStarted,
     LearningFinished(Box<dyn AnalyticUnit + Send + Sync>),
     LearningFinishedEmpty,

server/src/services/analytic_unit_service.rs

@@ -1,5 +1,3 @@
-use serde::{Deserialize, Serialize};
-use serde_json::{Result, Value};
 use std::sync::{Arc, Mutex};
 
 use rusqlite::{params, Connection};
@@ -11,19 +9,20 @@ use super::analytic_service::analytic_unit::{
     types::{self, AnalyticUnitConfig},
 };
 
+use super::data_service::DataService;
+
 #[derive(Clone)]
 pub struct AnalyticUnitService {
     connection: Arc<Mutex<Connection>>,
 }
 
+// TODO: get DataService
 impl AnalyticUnitService {
-    pub fn new() -> anyhow::Result<AnalyticUnitService> {
-        // TODO: remove repetitoin with segment_service
-        std::fs::create_dir_all("./data").unwrap();
-        let conn = Connection::open("./data/analytic_units.db")?;
+    pub fn new(ds: &DataService) -> anyhow::Result<AnalyticUnitService> {
+        let conn = ds.analytic_units_connection.clone();
 
         // TODO: add learning results field
-        conn.execute(
+        conn.lock().unwrap().execute(
             "CREATE TABLE IF NOT EXISTS analytic_unit (
                       id              TEXT PRIMARY KEY,
                       last_detection  INTEGER,
@@ -35,7 +34,7 @@ impl AnalyticUnitService {
         )?;
 
         Ok(AnalyticUnitService {
-            connection: Arc::new(Mutex::new(conn)),
+            connection: conn
         })
     }
 

server/src/services/data_service.rs

@@ -0,0 +1,23 @@
+use std::sync::{Arc, Mutex};
+
+use rusqlite::{Connection};
+
+
+pub struct DataService {
+    pub analytic_units_connection: Arc<Mutex<Connection>>,
+    pub segments_connection: Arc<Mutex<Connection>>
+}
+
+impl DataService {
+    pub fn new() -> anyhow::Result<DataService> {
+        std::fs::create_dir_all("./data").unwrap();
+
+        let analytic_units_connection = Connection::open("./data/analytic_units.db")?;
+        let segments_connection = Connection::open("./data/segments.db")?;
+
+        Ok(DataService {
+            analytic_units_connection: Arc::new(Mutex::new(analytic_units_connection)),
+            segments_connection: Arc::new(Mutex::new(segments_connection))
+        })
+    }
+}

server/src/services/metric_service.rs

@@ -32,4 +32,11 @@ impl MetricService {
         }
         return Ok(mr);
     }
+
+    // TODO: it a hack for DetectionRunner: it should vary for different analytic units
+    //       and it's config         
+    pub fn get_detection_step(&self) -> u64 {
+        return 10;
+    }
+
 }

server/src/services/mod.rs

@@ -1,4 +1,5 @@
 pub mod analytic_service;
+pub mod data_service;
 pub mod analytic_unit_service;
 pub mod metric_service;
 pub mod segments_service;

server/src/services/segments_service.rs

@@ -6,9 +6,13 @@ use serde::{Deserialize, Serialize};
 
 use std::sync::{Arc, Mutex};
 
+use super::data_service::DataService;
+
 pub const ID_LENGTH: usize = 20;
 pub type SegmentId = String;
 
+
+
 // TODO: make logic with this enum shorter
 #[derive(Debug, Serialize, Deserialize, Clone, Copy, PartialEq)]
 pub enum SegmentType {
@@ -58,19 +62,19 @@ impl Segment {
     }
 }
 
+
+// TODO: get DataService
 #[derive(Clone)]
 pub struct SegmentsService {
     connection: Arc<Mutex<Connection>>,
 }
 
 impl SegmentsService {
-    pub fn new() -> anyhow::Result<SegmentsService> {
-        // TODO: move it to data service
-        std::fs::create_dir_all("./data").unwrap();
+    pub fn new(ds: &DataService) -> anyhow::Result<SegmentsService> {
 
         // TODO: add unilytic_unit id as a new field
-        let conn = Connection::open("./data/segments.db")?;
-        conn.execute(
+        let conn = ds.segments_connection.clone();
+        conn.lock().unwrap().execute(
             "CREATE TABLE IF NOT EXISTS segment (
                       id            TEXT PRIMARY KEY,
                       start         INTEGER NOT NULL,
@@ -81,7 +85,7 @@ impl SegmentsService {
         )?;
 
         Ok(SegmentsService {
-            connection: Arc::new(Mutex::new(conn)),
+            connection: conn
         })
     }