Ensure project event collections have indexes before grouper writes

workers/grouper/src/index.ts

@@ -20,6 +20,7 @@ import DataFilter from './data-filter';
 import RedisHelper from './redisHelper';
 import { computeDelta } from './utils/repetitionDiff';
 import { bucketTimestampMs } from './utils/bucketTimestamp';
+import { ensureEventIndexes } from './utils/eventIndexes';
 import GrouperMetrics from './metrics/grouperMetrics';
 import GrouperMemoryMonitor from './metrics/memoryMonitor';
 import SlowHandleDiagnostics, { SlowHandleSession } from './metrics/slowHandleDiagnostics';
@@ -43,6 +44,11 @@ const MEMOIZATION_TTL = 60_000;
  */
 const CACHE_CLEANUP_INTERVAL_SECONDS = 30;
 
+/**
+ * How long a project's event collections are considered checked
+ */
+const EVENT_INDEXES_CHECK_TTL = TimeMs.HOUR;
+
 /**
  * Error code of MongoDB key duplication error
  */
@@ -107,6 +113,11 @@ export default class GrouperWorker extends Worker {
    */
   private handledTasksCount = 0;
 
+  /**
+   * Running or finished event collections checks by project id
+   */
+  private eventIndexesChecks: Map<string, { expiresAt: number; done: Promise<void> }> = new Map();
+
   /**
    * Start consuming messages
    */
@@ -202,6 +213,8 @@ export default class GrouperWorker extends Worker {
      */
     this.dataFilter.processEvent(task.payload);
 
+    await session.measureStep('ensureEventIndexes', () => this.ensureEventIndexes(task.projectId));
+
     let uniqueEventHash = await session.measureStep('hash', () => this.getUniqueEventHash(task));
     let existedEvent: GroupedEventDBScheme;
     let repetitionId = null;
@@ -498,6 +511,46 @@ export default class GrouperWorker extends Worker {
     }
   }
 
+  /**
+   * Makes sure project's event collections have indexes before the first write, once per TTL.
+   * Failures are logged and do not stop event processing.
+   *
+   * @param projectId - project to check
+   */
+  private ensureEventIndexes(projectId: string): Promise<void> {
+    if (!projectId || !mongodb.ObjectID.isValid(projectId)) {
+      return Promise.resolve();
+    }
+
+    const cached = this.eventIndexesChecks.get(projectId);
+
+    if (cached && cached.expiresAt > Date.now()) {
+      return cached.done;
+    }
+
+    const done = ensureEventIndexes(this.eventsDb.getConnection(), projectId)
+      .then(({ created, missing }) => {
+        if (created.length > 0) {
+          this.logger.info(`[ensureEventIndexes] project=${projectId} created collections: ${created.join(', ')}`);
+        }
+
+        if (missing.length > 0) {
+          this.logger.warn(`[ensureEventIndexes] project=${projectId} missing indexes: ${missing.join(', ')}`);
+        }
+      })
+      .catch((error) => {
+        this.eventIndexesChecks.delete(projectId);
+        this.logger.error(`[ensureEventIndexes] project=${projectId} failed: ${error.message}`);
+      });
+
+    this.eventIndexesChecks.set(projectId, {
+      expiresAt: Date.now() + EVENT_INDEXES_CHECK_TTL,
+      done,
+    });
+
+    return done;
+  }
+
   /**
    * Get unique hash based on event type and title
    *

workers/grouper/src/metrics/grouperMetrics.ts

@@ -19,6 +19,7 @@ export type GrouperStep =
   'computeDelta' |
   'decodeEvent' |
   'enqueueNotifier' |
+  'ensureEventIndexes' |
   'findSimilarEvent' |
   'getEvent' |
   'hash' |

workers/grouper/src/utils/eventIndexes.ts

@@ -0,0 +1,130 @@
+import type { Db, IndexSpecification } from 'mongodb';
+
+const NAMESPACE_NOT_FOUND = 26;
+
+/**
+ * Must match the indexes hawk.api.nodejs creates for a new project
+ */
+export const EVENT_COLLECTIONS_INDEXES: Record<string, IndexSpecification[]> = {
+  events: [
+    {
+      key: { 'payload.release': 1 },
+      name: 'payloadRelease',
+      sparse: true,
+    },
+    {
+      key: { groupHash: 1 },
+      name: 'groupHashUnique',
+      unique: true,
+    },
+    {
+      key: { timestamp: 1 },
+      name: 'timestamp',
+      sparse: true,
+    },
+  ],
+  repetitions: [
+    {
+      key: { groupHash: 'hashed' },
+      name: 'groupHash_hashed',
+    },
+    {
+      key: {
+        groupHash: 1,
+        release: 1,
+      },
+      name: 'groupHashAndRelease',
+    },
+    {
+      key: { 'payload.user.id': 1 },
+      name: 'userId',
+      sparse: true,
+    },
+    {
+      key: { timestamp: 1 },
+      name: 'timestamp',
+      sparse: true,
+    },
+  ],
+  dailyEvents: [
+    {
+      key: { groupingTimestamp: 1 },
+      name: 'groupingTimestamp',
+    },
+    {
+      key: {
+        groupingTimestamp: 1,
+        groupHash: 1,
+      },
+      name: 'groupingTimestampAndGroupHash',
+    },
+    {
+      key: {
+        groupingTimestamp: -1,
+        lastRepetitionTime: -1,
+        _id: -1,
+      },
+      name: 'groupingTimestampAndLastRepetitionTimeAndId',
+    },
+    {
+      key: { groupHash: 1 },
+      name: 'groupHash',
+    },
+  ],
+};
+
+export interface EventIndexesReport {
+  /**
+   * Collections that did not exist and were created with indexes
+   */
+  created: string[];
+
+  /**
+   * Expected indexes absent in already existing collections
+   */
+  missing: string[];
+}
+
+/**
+ * Creates missing event collections of a project with their indexes.
+ * Existing collections are only compared with the expected set: building an index on live data is not a hot path job.
+ *
+ * @param db - events database
+ * @param projectId - project to check
+ */
+export async function ensureEventIndexes(db: Db, projectId: string): Promise<EventIndexesReport> {
+  const report: EventIndexesReport = {
+    created: [],
+    missing: [],
+  };
+
+  for (const [prefix, specs] of Object.entries(EVENT_COLLECTIONS_INDEXES)) {
+    const collection = db.collection(`${prefix}:${projectId}`);
+    const existing: IndexSpecification[] = await collection.indexes().catch((error) => {
+      if (error.code !== NAMESPACE_NOT_FOUND) {
+        throw error;
+      }
+
+      return [];
+    });
+
+    if (existing.length === 0) {
+      await collection.createIndexes(specs);
+      report.created.push(prefix);
+
+      continue;
+    }
+
+    for (const spec of specs) {
+      const found = existing.some((index) => {
+        return JSON.stringify(index.key) === JSON.stringify(spec.key) && !!index.unique === !!spec.unique;
+      });
+
+      if (!found) {
+        report.missing.push(`${prefix}.${spec.name}`);
+      }
+    }
+  }
+
+  return report;
+}

workers/grouper/tests/eventIndexes.test.ts

@@ -0,0 +1,61 @@
+import '../../../env-test';
+import { MongoClient, ObjectID } from 'mongodb';
+import type { Db } from 'mongodb';
+import { ensureEventIndexes, EVENT_COLLECTIONS_INDEXES } from '../src/utils/eventIndexes';
+
+describe('ensureEventIndexes', () => {
+  let connection: MongoClient;
+  let db: Db;
+
+  beforeAll(async () => {
+    connection = await MongoClient.connect(process.env.MONGO_EVENTS_DATABASE_URI, {
+      useNewUrlParser: true,
+      useUnifiedTopology: true,
+    });
+    db = connection.db();
+  });
+
+  afterAll(async () => {
+    await connection.close();
+  });
+
+  it('creates missing collections with all indexes and reports nothing on the next run', async () => {
+    const projectId = new ObjectID().toString();
+
+    expect(await ensureEventIndexes(db, projectId)).toEqual({
+      created: ['events', 'repetitions', 'dailyEvents'],
+      missing: [],
+    });
+
+    for (const [prefix, specs] of Object.entries(EVENT_COLLECTIONS_INDEXES)) {
+      const names = (await db.collection(`${prefix}:${projectId}`).indexes()).map(index => index.name);
+
+      expect(names.sort()).toEqual([...specs.map(spec => spec.name), '_id_'].sort());
+    }
+
+    const events = db.collection(`events:${projectId}`);
+
+    await events.insertOne({ groupHash: 'hash' });
+    await expect(events.insertOne({ groupHash: 'hash' })).rejects.toMatchObject({ code: 11000 });
+
+    expect(await ensureEventIndexes(db, projectId)).toEqual({
+      created: [],
+      missing: [],
+    });
+  });
+
+  it('reports indexes missing in an existing collection without building them', async () => {
+    const projectId = new ObjectID().toString();
+    const events = db.collection(`events:${projectId}`);
+
+    await events.createIndex({ groupHash: 1 });
+
+    const report = await ensureEventIndexes(db, projectId);
+
+    expect(report).toEqual({
+      created: ['repetitions', 'dailyEvents'],
+      missing: ['events.payloadRelease', 'events.groupHashUnique', 'events.timestamp'],
+    });
+    expect(await events.indexes()).toHaveLength(2);
+  });
+});