Anomaly detector's segments concatenation #716 (#741)

server/spec/segments.jest.ts

@@ -1,18 +1,31 @@
 import { deleteNonDetectedSegments } from '../src/controllers/analytics_controller';
 import * as AnalyticUnit from '../src/models/analytic_units';
 import * as Segment from '../src/models/segment_model';
+import * as AnalyticUnitCache from '../src/models/analytic_unit_cache_model';
 
 import * as _ from 'lodash';
 
-let id: AnalyticUnit.AnalyticUnitId = 'testid';
-let baseSegments = segmentBuilder([[0,1], [2,3], [4,5]]);
+const TEST_ID: AnalyticUnit.AnalyticUnitId = 'testid';
+const INITIAL_SEGMENTS = segmentBuilder([[0, 1], [2, 3], [4, 5]]);
 
 beforeAll(async () => {
   clearDB();
+  await AnalyticUnit.create(
+    AnalyticUnit.createAnalyticUnitFromObject({
+      _id: TEST_ID,
+      name: 'name',
+      grafanaUrl: 'grafanaUrl',
+      panelId: 'panelId',
+      type: 'type',
+      detectorType: AnalyticUnit.DetectorType.ANOMALY
+    })
+  );
+  await AnalyticUnitCache.create(TEST_ID);
+  await AnalyticUnitCache.setData(TEST_ID, { timeStep: 1 });
 });
 
 beforeEach(async () => {
-  await Segment.insertSegments(baseSegments);
+  await Segment.insertSegments(INITIAL_SEGMENTS);
 });
 
 afterEach(async () => {
@@ -28,26 +41,25 @@ describe('Check deleted segments', function() {
 
   it('previous segments not found', async function() {
     payload.segments = segmentBuilder([[0, 1], [4, 5]]);
-    expect(await getDeletedSegments(id, payload)).toEqual(segmentBuilder([[2,3]]));
+    expect(await getDeletedSegments(TEST_ID, payload)).toEqual(segmentBuilder([[2, 3]]));
   });
 
   it('all previous segments found', async function() {
     payload.segments = segmentBuilder([[0, 1], [2, 3], [4, 5]]);
-    expect(await getDeletedSegments(id, payload)).toEqual([]);
+    expect(await getDeletedSegments(TEST_ID, payload)).toEqual([]);
   });
 
 });
 
-async function getDeletedSegments(id, payload): Promise<Segment.Segment[]> {
-  let preSegments = await Segment.findMany(id, {labeled: false, deleted:false});
-  await deleteNonDetectedSegments(id, payload);
-  let postSegments = await Segment.findMany(id, {labeled: false, deleted:false});
-  let deleted = setDifference(preSegments, postSegments);
-  deleted = deleted.map(s => {
+async function getDeletedSegments(TEST_ID, payload): Promise<Segment.Segment[]> {
+  const preSegments = await Segment.findMany(TEST_ID, { labeled: false, deleted: false });
+  await deleteNonDetectedSegments(TEST_ID, payload);
+  const postSegments = await Segment.findMany(TEST_ID, { labeled: false, deleted: false });
+  const deleted = setDifference(preSegments, postSegments);
+  return deleted.map(s => {
     s.id = undefined;
     return s;
   });
-  return deleted;
 }
 
 function setDifference(a, b: Segment.Segment[]): Segment.Segment[] {
@@ -56,11 +68,11 @@ function setDifference(a, b: Segment.Segment[]): Segment.Segment[] {
 
 function segmentBuilder(times) {
   return times.map(t => {
-    return new Segment.Segment(id, t[0], t[1], false, false, undefined);
+    return new Segment.Segment(TEST_ID, t[0], t[1], false, false, undefined);
   });
 }
 
 async function clearDB() {
-  let segments = await Segment.findMany(id, {labeled: false, deleted: false});
+  const segments = await Segment.findMany(TEST_ID, { labeled: false, deleted: false });
   await Segment.removeSegments(segments.map(s => s.id));
 }

server/src/models/analytic_unit_cache_model.ts

@@ -49,6 +49,10 @@ export class AnalyticUnitCache {
     return 3 * MILLISECONDS_IN_INDEX;
   }
 
+  public getTimeStep(): number {
+    return this.data.timeStep;
+  }
+
   public isCacheOutdated(analyticUnit: AnalyticUnit) {
     return !_.every(
       _.keys(analyticUnit.analyticProps).map(k => _.isEqual(analyticUnit.analyticProps[k], this.data[k]))

server/src/models/detection_model.ts

@@ -66,6 +66,9 @@ export class DetectionSpan {
 
 export type FindManyQuery = {
   status?: DetectionStatus,
+  // TODO: 
+  // from?: { $gte?: number, $lte?: number }
+  // to?: { $gte?: number, $lte?: number }
   timeFromLTE?: number,
   timeToGTE?: number,
   timeFromGTE?: number,

server/src/models/segment_model.ts

@@ -1,5 +1,6 @@
 import { AnalyticUnitId } from './analytic_units';
-
+import * as AnalyticUnit from '../models/analytic_units';
+import * as AnalyticUnitCache from '../models/analytic_unit_cache_model';
 import { Collection, makeDBQ } from '../services/data_service';
 
 import * as _ from 'lodash';
@@ -71,21 +72,14 @@ export class Segment {
 
 export type FindManyQuery = {
   $or?: any,
-  timeFromGTE?: number,
-  timeToLTE?: number,
-  intexGT?: number,
+  from?: { $gte?: number, $lte?: number },
+  to?: { $gte?: number, $lte?: number },
   labeled?: boolean,
   deleted?: boolean
 }
 
 export async function findMany(id: AnalyticUnitId, query: FindManyQuery): Promise<Segment[]> {
   var dbQuery: any = { analyticUnitId: id };
-  if(query.timeFromGTE !== undefined) {
-    dbQuery.from = { $gte: query.timeFromGTE };
-  }
-  if(query.timeToLTE !== undefined) {
-    dbQuery.to = { $lte: query.timeToLTE };
-  }
   if(query.labeled !== undefined) {
     dbQuery.labeled = query.labeled;
   }
@@ -104,8 +98,7 @@ export async function insertSegments(segments: Segment[]) {
     return [];
   }
   const analyticUnitId: AnalyticUnitId = segments[0].analyticUnitId;
-  const learningSegments: Segment[] = await db.findMany({
-    analyticUnitId,
+  const learningSegments = await findMany(analyticUnitId, {
     labeled: true,
     deleted: false
   });
@@ -122,8 +115,7 @@ export async function insertSegments(segments: Segment[]) {
     }
 
     if(!segment.deleted && !segment.labeled) {
-      const intersectedWithDeletedSegments = await db.findMany({
-        analyticUnitId,
+      const intersectedWithDeletedSegments = await findMany(analyticUnitId, {
         to: { $gte: segment.from },
         from: { $lte: segment.to },
         labeled: false,
@@ -135,8 +127,38 @@ export async function insertSegments(segments: Segment[]) {
       }
     }
 
-    const intersectedSegments = await db.findMany({
-      analyticUnitId,
+    let cache = await AnalyticUnitCache.findById(analyticUnitId);
+    const timeStep = cache.getTimeStep();
+    let unit = await AnalyticUnit.findById(analyticUnitId);
+    const detector = unit.detectorType;
+
+    if(detector !== AnalyticUnit.DetectorType.PATTERN) {
+      const intersectedWithLeftBound = await findMany(analyticUnitId, {
+        to: { $gte: segment.from - timeStep, $lte: segment.from },
+        labeled: false,
+        deleted: false
+      });
+
+      if(intersectedWithLeftBound.length > 0) {
+        const leftSegment = _.minBy(intersectedWithLeftBound, s => s.from);
+        segment.from = leftSegment.from;
+        segmentIdsToRemove.push(leftSegment.id);
+      }
+
+      const intersectedWithRightBound = await findMany(analyticUnitId, {
+        from: { $gte: segment.to, $lte: segment.to + timeStep },
+        labeled: false,
+        deleted: false
+      });
+
+      if(intersectedWithRightBound.length > 0) {
+        const rightSegment = _.maxBy(intersectedWithRightBound, s => s.to);
+        segment.to = rightSegment.to;
+        segmentIdsToRemove.push(rightSegment.id);
+      }
+    }
+
+    const intersectedSegments = await findMany(analyticUnitId, {
       to: { $gte: segment.from },
       from: { $lte: segment.to },
       labeled: segment.labeled,
@@ -149,7 +171,7 @@ export async function insertSegments(segments: Segment[]) {
       let newSegment = Segment.fromObject(segment.toObject());
       newSegment.from = from;
       newSegment.to = to;
-      segmentIdsToRemove = segmentIdsToRemove.concat(intersectedSegments.map(s => s._id));
+      segmentIdsToRemove = segmentIdsToRemove.concat(intersectedSegments.map(s => s.id));
       segmentsToInsert.push(newSegment);
     } else {
       segmentsToInsert.push(segment);

server/src/routes/segments_router.ts

@@ -13,14 +13,11 @@ async function getSegments(ctx: Router.IRouterContext) {
   }
   let query: Segment.FindManyQuery = {};
 
-  if(!isNaN(+ctx.request.query.lastSegmentId)) {
-    query.intexGT = +ctx.request.query.lastSegmentId;
-  }
   if(!isNaN(+ctx.request.query.from)) {
-    query.timeFromGTE = +ctx.request.query.from;
+    query.from = { $gte: +ctx.request.query.from };
   }
   if(!isNaN(+ctx.request.query.to)) {
-    query.timeToLTE = +ctx.request.query.to;
+    query.to = { $lte: +ctx.request.query.to };
   }
   let segments = await Segment.findMany(id, query);
   ctx.response.body = { segments };