Update Data Puller state #294 (#295)

* Get analytic unit from db each puller tick

* Use async iterator to pull and push data

* Fix webpack config for proper babel-polyfill usage

* Add MetricDataChunk type

* Add pattern field to payload

server/build/webpack.614.prod.conf.js

@@ -7,7 +7,7 @@ module.exports = {
     __dirname: false,
     __filename: false,
   },
-  entry: [ 'babel-polyfill', './dist/server-dev.js' ],
+  entry: [ './dist/server-dev.js' ],
   output: {
     filename: "server.js",
     path: path.join(__dirname, '../dist')

server/build/webpack.base.conf.js

@@ -14,7 +14,7 @@ module.exports = {
     __dirname: false,
     __filename: false,
   },
-  entry: [ './src/index.ts' ],
+  entry: [ 'babel-polyfill', './src/index.ts' ],
   output: {
     filename: "server-dev.js",
     path: resolve('dist')

server/src/controllers/analytics_controller.ts

@@ -286,13 +286,20 @@ export async function createAnalyticUnitFromObject(obj: any): Promise<AnalyticUn
   }
   let unit: AnalyticUnit.AnalyticUnit = AnalyticUnit.AnalyticUnit.fromObject(obj);
   let id = await AnalyticUnit.create(unit);
-  unit.id = id;
 
-  if(dataPuller !== undefined) {
+  return id;
-    dataPuller.addUnit(unit);
 }
 
-  return id;
+export async function setAlert(analyticUnitId: AnalyticUnit.AnalyticUnitId, alert: boolean) {
+  AnalyticUnit.setAlert(analyticUnitId, alert);
+  if(dataPuller !== undefined) {
+    if(alert) {
+      const analyticUnit = await AnalyticUnit.findById(analyticUnitId);
+      dataPuller.addUnit(analyticUnit);
+    } else {
+      dataPuller.deleteUnit(analyticUnitId);
+    }
+  }
 }
 
 export async function updateSegments(

server/src/routes/analytic_units_router.ts

@@ -113,7 +113,7 @@ async function setAlert(ctx: Router.IRouterContext) {
       throw new Error('Cannot set undefined alert status');
     }
     
-    await AnalyticUnit.setAlert(analyticUnitId, alert);
+    await AnalyticsController.setAlert(analyticUnitId, alert);
 
     ctx.response.body = {
       code: 200,

server/src/services/data_puller.ts

@@ -8,32 +8,28 @@ import { queryByMetric } from 'grafana-datasource-kit';
 import * as _ from 'lodash';
 
 
-declare type UnitTime = {
+type MetricDataChunk = { values: [number, number][], columns: string[] };
-  unit: AnalyticUnit.AnalyticUnit,
+
-  time: number
+const PULL_PERIOD_MS = 5000;
-};
 
 export class DataPuller {
 
-  private PULL_PERIOD_MS: number = 5000;
+  private _unitTimes: { [analyticUnitId: string]: number } = {};
-  private _interval: number = 1000;
-  private _timer: any = null;
-  private _unitTimes: { [id: string]: UnitTime } = {};
 
   constructor(private analyticsService: AnalyticsService) {};
 
-  public addUnit(unit: AnalyticUnit.AnalyticUnit) {
+  public addUnit(analyticUnit: AnalyticUnit.AnalyticUnit) {
-    let time = unit.lastDetectionTime || Date.now();
+    this._runAnalyticUnitPuller(analyticUnit);
-    let unitTime: UnitTime = {unit, time };
-    this._unitTimes[unit.id] = unitTime;
   }
 
-  public deleteUnit(id: AnalyticUnit.AnalyticUnitId) {
+  public deleteUnit(analyticUnitId: AnalyticUnit.AnalyticUnitId) {
-    delete this._unitTimes[id];
+    if(_.has(this._unitTimes, analyticUnitId)) {
+      delete this._unitTimes[analyticUnitId];
+    }
   }
 
-  private pullData(unit: AnalyticUnit.AnalyticUnit, from: number, to: number) {
+  private async pullData(unit: AnalyticUnit.AnalyticUnit, from: number, to: number): Promise<MetricDataChunk> {
-    if(!unit) {
+    if(unit === undefined) {
       throw Error(`puller: can't pull undefined unit`);
     }
     return queryByMetric(unit.metric, unit.panelUrl, from, to, HASTIC_API_KEY);
@@ -48,46 +44,65 @@ export class DataPuller {
   }
 
   //TODO: group analyticUnits by panelID and send same dataset for group
-  public runPuller() {
+  public async runPuller() {
-    this._timer = setTimeout(this.puller.bind(this), this._interval);
+    const analyticUnits = await AnalyticUnit.findMany({ alert: true });
+
+    _.each(analyticUnits, analyticUnit => {
+      this._runAnalyticUnitPuller(analyticUnit);
+    });
+
     console.log('Data puller runned');
   }
 
   public stopPuller() {
-    if(this._timer) {
+    this._unitTimes = {};
-      clearTimeout(this._timer);
-      this._timer = null;
-      this._interval = 0;
-      console.log('Data puller stopped');
   }
-    console.log('Data puller already stopped');
+
+  private async _runAnalyticUnitPuller(analyticUnit: AnalyticUnit.AnalyticUnit) {
+    const time = analyticUnit.lastDetectionTime || Date.now();
+    this._unitTimes[analyticUnit.id] = time;
+
+    const dataGenerator = this.getDataGenerator(
+      analyticUnit, PULL_PERIOD_MS
+    );
+
+    for await (const data of dataGenerator) {
+      if(!_.has(this._unitTimes, analyticUnit.id)) {
+        break;
       }
 
-  private async puller() {
+      if(data.values.length === 0) {
+        continue;
+      }
 
-    if(_.isEmpty(this._unitTimes)) {
+      const now = Date.now();
-      this._interval = this.PULL_PERIOD_MS;
+      let payload = { data, from: time, to: now, pattern: analyticUnit.type };
-      this._timer = setTimeout(this.puller.bind(this), this._interval);
+      this._unitTimes[analyticUnit.id] = now;
-      return;
+      this.pushData(analyticUnit, payload);
+    }
   }
 
-    let now = Date.now();
+  async * getDataGenerator(analyticUnit: AnalyticUnit.AnalyticUnit, duration: number):
+    AsyncIterableIterator<MetricDataChunk> {
 
-    _.forOwn(this._unitTimes, async value => {
+    const getData = async () => {
-      if(!value.unit.alert) {
+      try {
-        return;
+        const time = this._unitTimes[analyticUnit.id]
+        const now = Date.now();
+        return await this.pullData(analyticUnit, time, now);
+      } catch(err) {
+        throw new Error(`Error while pulling data: ${err.message}`);
       }
-      let data = await this.pullData(value.unit, value.time, now);
-      if(data.values.length === 0) {
-        return;
     }
 
-      let payload = { data, from: value.time, to: now};
+    const timeout = async () => new Promise(
-      value.time = now;
+      resolve => setTimeout(resolve, duration)
-      this.pushData(value.unit, payload); 
+    );
-    });
 
-    this._timer = setTimeout(this.puller.bind(this), this._interval);
+    while(true) {
+      yield await getData();
+      await timeout();
+    }
   }
 
 }