Stop limiter from opening every events collection each hour

.env.sample

@@ -57,3 +57,6 @@ IS_NOTIFIER_WORKER_ENABLED=false
 
 ## Url for telegram notifications about workspace blocks and unblocks 
 TELEGRAM_LIMITER_CHAT_URL=
+
+## Workspace ids to compare old and new limiter counts in telegram
+LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS=

workers/limiter/src/dbHelper.ts

@@ -184,13 +184,91 @@ export class DbHelper {
    * increments `count` for originals and repetitions alike); only the
    * partial day containing `since` is counted from the raw collections,
    * since dailyEvents buckets have day granularity and lastChargeDate does not.
+   * Raw collections are skipped if that day's bucket is empty.
    *
    * @param project - project to check
    * @param since - timestamp of the time from which we count the events
    */
   public async getEventsCountByProjectUsingDailyEvents(
     project: ProjectDBScheme,
     since: number
+  ): Promise<number> {
+    try {
+      const projectId = project._id.toString();
+      const dailyEventsCollection = this.eventsDbConnection.collection('dailyEvents:' + projectId);
+      const boundaryDayTimestamp = since - (since % SEC_IN_DAY);
+      const firstFullDayTimestamp = this.getFirstFullDailyEventsTimestamp(since);
+
+      const [ counters ] = await dailyEventsCollection
+        .aggregate<{ boundaryDay: number; fullDays: number }>([
+          /** buckets from the day containing `since` onwards */
+          { $match: { groupingTimestamp: { $gte: boundaryDayTimestamp } } },
+          {
+            $group: {
+              _id: null,
+              /** whole boundary day, only gates the raw count below */
+              boundaryDay: {
+                $sum: { $cond: [ { $lt: ['$groupingTimestamp', firstFullDayTimestamp] }, '$count', 0] },
+              },
+              /** days after the boundary day */
+              fullDays: {
+                $sum: { $cond: [ { $gte: ['$groupingTimestamp', firstFullDayTimestamp] }, '$count', 0] },
+              },
+            },
+          },
+        ], {
+          /** one table per project instead of racing all groupingTimestamp indexes */
+          hint: { $natural: 1 },
+        })
+        .toArray();
+
+      /** no buckets in the billing period */
+      if (!counters) {
+        return 0;
+      }
+
+      /** the bucket spans the whole day, so the part after `since` is counted from raw events */
+      const boundaryDayCount = counters.boundaryDay > 0
+        ? await this.getRawEventsCountByProject(project, {
+          timestamp: {
+            $gt: since,
+            $lt: firstFullDayTimestamp,
+          },
+        })
+        : 0;
+
+      return boundaryDayCount + counters.fullDays;
+    } catch (e) {
+      HawkCatcher.send(e);
+      throw new CriticalError(e);
+    }
+  }
+
+  /**
+   * Calculates total events count for all provided projects since the specific date
+   * using dailyEvents counters for full days.
+   *
+   * @param projects - projects to calculate for
+   * @param since - timestamp of the time from which we count the events
+   */
+  public async getEventsCountByProjectsUsingDailyEvents(projects: ProjectDBScheme[], since: number): Promise<number> {
+    const sum = (array: number[]): number => array.reduce((acc, val) => acc + val, 0);
+
+    return Promise.all(projects.map(
+      project => this.getEventsCountByProjectUsingDailyEvents(project, since)
+    ))
+      .then(sum);
+  }
+
+  /**
+   * Previous query, kept for rollout comparison
+   *
+   * @param project - project to check
+   * @param since - timestamp of the time from which we count the events
+   */
+  public async getEventsCountByProjectUsingDailyEventsOld(
+    project: ProjectDBScheme,
+    since: number
   ): Promise<number> {
     try {
       const projectId = project._id.toString();
@@ -231,17 +309,16 @@ export class DbHelper {
   }
 
   /**
-   * Calculates total events count for all provided projects since the specific date
-   * using dailyEvents counters for full days.
+   * Previous query, kept for rollout comparison
    *
    * @param projects - projects to calculate for
    * @param since - timestamp of the time from which we count the events
    */
-  public async getEventsCountByProjectsUsingDailyEvents(projects: ProjectDBScheme[], since: number): Promise<number> {
+  public async getEventsCountByProjectsUsingDailyEventsOld(projects: ProjectDBScheme[], since: number): Promise<number> {
     const sum = (array: number[]): number => array.reduce((acc, val) => acc + val, 0);
 
     return Promise.all(projects.map(
-      project => this.getEventsCountByProjectUsingDailyEvents(project, since)
+      project => this.getEventsCountByProjectUsingDailyEventsOld(project, since)
     ))
       .then(sum);
   }

workers/limiter/src/index.ts

@@ -266,7 +266,7 @@ export default class LimiterWorker extends Worker {
 
     const since = Math.floor(new Date(workspace.lastChargeDate).getTime() / MS_IN_SEC);
 
-    const workspaceEventsCount = await this.dbHelper.getEventsCountByProjectsUsingDailyEvents(projects, since);
+    const workspaceEventsCount = await this.getWorkspaceEventsCount(workspace, projects, since);
 
     this.logger.info(`workspace ${workspace._id} events count since last charge date: ${workspaceEventsCount}`);
 
@@ -328,6 +328,42 @@ export default class LimiterWorker extends Worker {
     };
   }
 
+  /**
+   * Counts workspace events, comparing with the previous query for LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS
+   *
+   * @param workspace - workspace to count events for
+   * @param projects - workspace projects
+   * @param since - timestamp of the time from which we count the events
+   */
+  private async getWorkspaceEventsCount(
+    workspace: WorkspaceWithTariffPlan,
+    projects: ProjectDBScheme[],
+    since: number
+  ): Promise<number> {
+    const compareWorkspaceIds = (process.env.LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS || '').split(',').map(id => id.trim());
+
+    if (!compareWorkspaceIds.includes(workspace._id.toString())) {
+      return this.dbHelper.getEventsCountByProjectsUsingDailyEvents(projects, since);
+    }
+
+    const oldAlgoStartedAt = Date.now();
+    const oldAlgoCount = await this.dbHelper.getEventsCountByProjectsUsingDailyEventsOld(projects, since);
+    const oldAlgoTook = (Date.now() - oldAlgoStartedAt) / MS_IN_SEC;
+
+    const newAlgoStartedAt = Date.now();
+    const newAlgoCount = await this.dbHelper.getEventsCountByProjectsUsingDailyEvents(projects, since);
+    const newAlgoTook = (Date.now() - newAlgoStartedAt) / MS_IN_SEC;
+
+    telegram.sendMessage(
+      `Workspace <b>${workspace.name}</b> event count:\n` +
+      `Old algo: ${oldAlgoCount}, took ${oldAlgoTook}sec\n` +
+      `New algo: ${newAlgoCount}, took ${newAlgoTook}sec`,
+      telegram.TelegramBotURLs.Limiter
+    );
+
+    return newAlgoCount;
+  }
+
   // Old raw counter with the opt-in switch, kept in case we need to roll back
   //
   // /**

workers/limiter/tests/dbHelper.test.ts

@@ -23,6 +23,9 @@ const BOUNDARY_DAY_TIMESTAMP = 1585756800;
  */
 const NEXT_MIDNIGHT_AFTER_LAST_CHARGE = 1585785600;
 
+/** 2020-04-01T00:00:00Z */
+const BOUNDARY_DAY_MIDNIGHT = NEXT_MIDNIGHT_AFTER_LAST_CHARGE - 86400;
+
 describe('DbHelper', () => {
   let connection: MongoClient;
   let db: Db;
@@ -132,6 +135,17 @@ describe('DbHelper', () => {
       await repetitionsCollection.insertMany(mockedEvents);
     }
 
+    /** as grouper does */
+    const boundaryDayEventsCount = parameters.eventsToMock + (parameters.repetitionsToMock ?? 0);
+
+    if (boundaryDayEventsCount > 0) {
+      await dailyEventsCollection.insertOne({
+        groupHash: 'ade987831d0d0d167aeea685b49db164eb4e113fd027858eef7f69d049357f62',
+        groupingTimestamp: BOUNDARY_DAY_MIDNIGHT,
+        count: boundaryDayEventsCount,
+      });
+    }
+
     if (parameters.dailyEventsToMock?.length > 0) {
       await dailyEventsCollection.insertMany(parameters.dailyEventsToMock.map(bucket => ({
         groupHash: 'ade987831d0d0d167aeea685b49db164eb4e113fd027858eef7f69d049357f62',
@@ -711,7 +725,7 @@ describe('DbHelper', () => {
         dailyEventsToMock: [
           /** bucket of the boundary day itself must not be counted */
           {
-            groupingTimestamp: NEXT_MIDNIGHT_AFTER_LAST_CHARGE - 86400,
+            groupingTimestamp: BOUNDARY_DAY_MIDNIGHT,
             count: 100,
           },
         ],
@@ -779,6 +793,80 @@ describe('DbHelper', () => {
        */
       expect(count).toBe(7);
     });
+
+    test('Should not query raw collections when the boundary day bucket is empty', async () => {
+      /**
+       * Arrange
+       */
+      const workspace = createWorkspaceMock({
+        plan: mockedPlans.eventsLimit10,
+        billingPeriodEventsCount: 0,
+        lastChargeDate: new Date(),
+      });
+      const project = createProjectMock({ workspaceId: workspace._id });
+      const since = Math.floor(LAST_CHARGE_DATE.getTime() / MS_IN_SEC);
+
+      await fillDatabaseWithMockedData({
+        workspace,
+        project,
+        eventsToMock: 0,
+        dailyEventsToMock: [
+          {
+            groupingTimestamp: NEXT_MIDNIGHT_AFTER_LAST_CHARGE,
+            count: 4,
+          },
+        ],
+      });
+
+      const collectionSpy = jest.spyOn(db, 'collection');
+
+      /**
+       * Act
+       */
+      const count = await dbHelper.getEventsCountByProjectUsingDailyEvents(project, since);
+
+      /**
+       * Assert
+       */
+      expect(count).toBe(4);
+      expect(collectionSpy.mock.calls.map(([ name ]) => name)).toEqual([ `dailyEvents:${project._id.toString()}` ]);
+
+      collectionSpy.mockRestore();
+    });
+
+    test('Should return zero for a project without dailyEvents collection', async () => {
+      const project = createProjectMock({ workspaceId: new ObjectId() });
+      const since = Math.floor(LAST_CHARGE_DATE.getTime() / MS_IN_SEC);
+
+      const count = await dbHelper.getEventsCountByProjectUsingDailyEvents(project, since);
+
+      expect(count).toBe(0);
+    });
+  });
+
+  describe('getEventsCountByProjectUsingDailyEventsOld', () => {
+    test('Should count raw boundary-day events even without a dailyEvents bucket', async () => {
+      const project = createProjectMock({ workspaceId: new ObjectId() });
+      const since = Math.floor(LAST_CHARGE_DATE.getTime() / MS_IN_SEC);
+
+      await fillDatabaseWithMockedData({
+        project,
+        eventsToMock: 0,
+        dailyEventsToMock: [
+          {
+            groupingTimestamp: NEXT_MIDNIGHT_AFTER_LAST_CHARGE,
+            count: 3,
+          },
+        ],
+      });
+      await db.collection(`events:${project._id.toString()}`).insertMany([createEventMock(), createEventMock()]);
+
+      const oldCount = await dbHelper.getEventsCountByProjectUsingDailyEventsOld(project, since);
+      const newCount = await dbHelper.getEventsCountByProjectUsingDailyEvents(project, since);
+
+      expect(oldCount).toBe(5);
+      expect(newCount).toBe(3);
+    });
   });
 
   describe('getEventsCountByProjectsUsingDailyEvents', () => {

workers/limiter/tests/index.test.ts

@@ -138,6 +138,13 @@ describe('Limiter worker', () => {
       }
       await repetitionsCollection.insertMany(mockedEvents);
     }
+
+    /** as grouper does */
+    await db.collection(`dailyEvents:${parameters.project._id.toString()}`).insertOne({
+      groupHash: 'ade987831d0d0d167aeea685b49db164eb4e113fd027858eef7f69d049357f62',
+      groupingTimestamp: NEXT_MIDNIGHT_AFTER_LAST_CHARGE - 86400,
+      count: parameters.eventsToMock + (parameters.repetitionsToMock ?? 0),
+    });
   };
 
   beforeAll(async () => {
@@ -381,6 +388,47 @@ describe('Limiter worker', () => {
       expect(telegram.sendMessage).not.toHaveBeenCalled();
     });
 
+    test('Should report old and new algo counts for workspaces listed for comparison', async () => {
+      const workspace = createWorkspaceMock({
+        plan: mockedPlans.eventsLimit10000,
+        billingPeriodEventsCount: 0,
+        lastChargeDate: LAST_CHARGE_DATE,
+      });
+      const project = createProjectMock({ workspaceId: workspace._id });
+
+      await fillDatabaseWithMockedData({
+        workspace,
+        project,
+        eventsToMock: 5,
+      });
+
+      await db.collection(`dailyEvents:${project._id.toString()}`).deleteMany({});
+
+      process.env.LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS = workspace._id.toString();
+
+      const worker = new LimiterWorker();
+
+      try {
+        await worker.start();
+        await worker.handle(REGULAR_WORKSPACES_CHECK_EVENT);
+        await worker.finish();
+      } finally {
+        delete process.env.LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS;
+      }
+
+      const workspaceInDatabase = await workspaceCollection.findOne({
+        _id: workspace._id,
+      });
+
+      expect(workspaceInDatabase.billingPeriodEventsCount).toBe(0);
+      expect(telegram.sendMessage).toHaveBeenCalledTimes(1);
+
+      const reportMessage = (telegram.sendMessage as jest.Mock).mock.calls[0][0];
+
+      expect(reportMessage).toContain('Old algo: 5');
+      expect(reportMessage).toContain('New algo: 0');
+    });
+
     test('Should not send a report when no projects are blocked or unblocked', async () => {
       /**
        * Arrange