chore: scheduled created-by migrations metrics (#6089)

## About the changes

the created_by_user_id data migration from resolving events.created_by
(for both events and features) now emits events on how many rows were
updated.

Adds listeners for these events that records these metrics with
prometheus


![image](https://github.com/Unleash/unleash/assets/707867/3bb02645-0919-4a9a-83fe-a07383ac0be1)

src/lib/features/events/event-created-by-migration.test.ts

@@ -3,6 +3,9 @@ import getLogger from '../../../test/fixtures/no-logger';
 import dbInit, { ITestDb } from '../../../test/e2e/helpers/database-init';
 import { defaultExperimentalOptions } from '../../types/experimental';
 import FlagResolver from '../../util/flag-resolver';
+import { EventEmitter } from 'stream';
+import EventService from './event-service';
+import { EVENTS_CREATED_BY_PROCESSED } from '../../metric-events';
 
 let db: ITestDb;
 let resolver: FlagResolver;
@@ -28,7 +31,7 @@ test('sets created_by_user_id on events with user username/email set as created_
     await db.rawDatabase('events').insert({
         type: 'feature-created',
         created_by: 'test1',
-        feature_name: `feature1`,
+        feature_name: 'feature1',
         data: `{"test": "data-migrate"}`,
     });
 
@@ -67,7 +70,7 @@ test('sets created_by_user_id on a mix of events and created_bys', async () => {
     await db.rawDatabase('events').insert({
         type: 'feature-created',
         created_by: 'test2',
-        feature_name: `feature1`,
+        feature_name: 'feature1',
         data: `{"test": "data-migrate"}`,
     });
 
@@ -92,14 +95,14 @@ test('sets created_by_user_id on a mix of events and created_bys', async () => {
     await db.rawDatabase('events').insert({
         type: 'feature-created',
         created_by: 'unknown',
-        feature_name: `feature2`,
+        feature_name: 'feature2',
         data: `{"test": "data-migrate"}`,
     });
 
     await db.rawDatabase('events').insert({
         type: 'feature-created',
         created_by: 'adm-token',
-        feature_name: `feature3`,
+        feature_name: 'feature3',
         data: `{"test": "data-migrate"}`,
     });
 
@@ -121,3 +124,40 @@ test('sets created_by_user_id on a mix of events and created_bys', async () => {
     expect(notSet).toHaveLength(1);
     expect(test).toHaveLength(1);
 });
+
+test('emits events with details on amount of updated rows', async () => {
+    const store = new EventStore(db.rawDatabase, getLogger, resolver);
+
+    const eventBus = new EventEmitter();
+    const service = new EventService(
+        { eventStore: store, featureTagStore: db.stores.featureTagStore },
+        { getLogger, eventBus },
+    );
+    let triggered = false;
+
+    eventBus.on(EVENTS_CREATED_BY_PROCESSED, ({ updated }) => {
+        expect(updated).toBe(2);
+        triggered = true;
+    });
+
+    await db.rawDatabase('users').insert({ username: 'events-test-1' });
+    await db.rawDatabase('events').insert({
+        type: 'feature-created',
+        created_by: 'events-test-1',
+        feature_name: 'feature1',
+    });
+    await db.rawDatabase('events').insert({
+        type: 'feature-created',
+        created_by: 'events-test-1',
+        feature_name: 'feature2',
+    });
+    await db.rawDatabase('events').insert({
+        type: 'feature-created',
+        created_by: 'doesnt-exist',
+        feature_name: 'not-counted',
+    });
+
+    await service.setEventCreatedByUserId();
+
+    expect(triggered).toBeTruthy();
+});

src/lib/features/events/event-service.ts

@@ -11,6 +11,7 @@ import {
     extractUserIdFromUser,
     extractUsernameFromUser,
 } from '../../util/extract-user';
+import { EVENTS_CREATED_BY_PROCESSED } from '../../metric-events';
 
 export default class EventService {
     private logger: Logger;
@@ -19,16 +20,19 @@ export default class EventService {
 
     private featureTagStore: IFeatureTagStore;
 
+    private eventBus: EventEmitter;
+
     constructor(
         {
             eventStore,
             featureTagStore,
         }: Pick<IUnleashStores, 'eventStore' | 'featureTagStore'>,
-        { getLogger }: Pick<IUnleashConfig, 'getLogger'>,
+        { getLogger, eventBus }: Pick<IUnleashConfig, 'getLogger' | 'eventBus'>,
     ) {
         this.logger = getLogger('services/event-service.ts');
         this.eventStore = eventStore;
         this.featureTagStore = featureTagStore;
+        this.eventBus = eventBus;
     }
 
     async getEvents(): Promise<IEventList> {
@@ -136,6 +140,11 @@ export default class EventService {
     }
 
     async setEventCreatedByUserId(): Promise<void> {
-        return this.eventStore.setCreatedByUserId(100);
+        const updated = await this.eventStore.setCreatedByUserId(100);
+        if (updated !== undefined) {
+            this.eventBus.emit(EVENTS_CREATED_BY_PROCESSED, {
+                updated,
+            });
+        }
     }
 }

src/lib/features/events/event-store.ts

@@ -433,16 +433,16 @@ class EventStore implements IEventStore {
         events.forEach((e) => this.eventEmitter.emit(e.type, e));
     }
 
-    async setCreatedByUserId(batchSize: number): Promise<void> {
+    async setCreatedByUserId(batchSize: number): Promise<number | undefined> {
         const API_TOKEN_TABLE = 'api_tokens';
 
         if (!this.flagResolver.isEnabled('createdByUserIdDataMigration')) {
-            return;
+            return undefined;
         }
 
         const toUpdate = await this.db(`${TABLE} as e`)
             .joinRaw(
-                `LEFT OUTER JOIN users AS u ON e.created_by = u.username OR e.created_by = u.email`,
+                'LEFT OUTER JOIN users AS u ON e.created_by = u.username OR e.created_by = u.email',
             )
             .joinRaw(
                 `LEFT OUTER JOIN ${API_TOKEN_TABLE} AS t on e.created_by = t.username`,
@@ -469,21 +469,23 @@ class EventStore implements IEventStore {
                 return this.db(TABLE)
                     .update({ created_by_user_id: SYSTEM_USER_ID })
                     .where({ id: row.id });
-            } else if (row.userid) {
+            }
+            if (row.userid) {
                 return this.db(TABLE)
                     .update({ created_by_user_id: row.userid })
                     .where({ id: row.id });
-            } else if (row.username) {
+            }
+            if (row.username) {
                 return this.db(TABLE)
                     .update({ created_by_user_id: ADMIN_TOKEN_USER.id })
                     .where({ id: row.id });
-            } else {
-                this.logger.warn(`Could not find user for event ${row.id}`);
-                return Promise.resolve();
             }
+            this.logger.warn(`Could not find user for event ${row.id}`);
+            return Promise.resolve();
         });
 
         await Promise.all(updatePromises);
+        return toUpdate.length;
     }
 }
 

src/lib/features/feature-toggle/createFeatureToggleService.ts

@@ -52,6 +52,7 @@ import {
     createFakeDependentFeaturesService,
 } from '../dependent-features/createDependentFeaturesService';
 import { createEventsService } from '../events/createEventsService';
+import { EventEmitter } from 'stream';
 
 export const createFeatureToggleService = (
     db: Db,
@@ -134,7 +135,7 @@ export const createFeatureToggleService = (
             contextFieldStore,
             strategyStore,
         },
-        { getLogger, flagResolver },
+        { getLogger, flagResolver, eventBus },
         segmentService,
         accessService,
         eventService,
@@ -166,7 +167,7 @@ export const createFakeFeatureToggleService = (
     const environmentStore = new FakeEnvironmentStore();
     const eventService = new EventService(
         { eventStore, featureTagStore },
-        { getLogger },
+        { getLogger, eventBus: new EventEmitter() },
     );
     const groupService = new GroupService(
         { groupStore, accountStore },
@@ -195,7 +196,7 @@ export const createFakeFeatureToggleService = (
             contextFieldStore,
             strategyStore,
         },
-        { getLogger, flagResolver },
+        { getLogger, flagResolver, eventBus: new EventEmitter() },
         segmentService,
         accessService,
         eventService,

src/lib/features/feature-toggle/fakes/fake-feature-toggle-store.ts

@@ -166,7 +166,7 @@ export default class FakeFeatureToggleStore implements IFeatureToggleStore {
     async getFeatureToggleList(
         query?: IFeatureToggleQuery,
         userId?: number,
-        archived: boolean = false,
+        archived = false,
     ): Promise<FeatureToggle[]> {
         return this.features.filter((feature) => feature.archived !== archived);
     }
@@ -329,7 +329,7 @@ export default class FakeFeatureToggleStore implements IFeatureToggleStore {
         throw new Error('Method not implemented.');
     }
 
-    setCreatedByUserId(batchSize: number): Promise<void> {
+    setCreatedByUserId(batchSize: number): Promise<number | undefined> {
         throw new Error('Method not implemented.');
     }
 }

src/lib/features/feature-toggle/feature-toggle-service.ts

@@ -105,6 +105,8 @@ import EventService from '../events/event-service';
 import { DependentFeaturesService } from '../dependent-features/dependent-features-service';
 import { FeatureToggleInsert } from './feature-toggle-store';
 import ArchivedFeatureError from '../../error/archivedfeature-error';
+import { FEATURES_CREATED_BY_PROCESSED } from '../../metric-events';
+import { EventEmitter } from 'stream';
 
 interface IFeatureContext {
     featureName: string;
@@ -172,6 +174,8 @@ class FeatureToggleService {
 
     private dependentFeaturesService: DependentFeaturesService;
 
+    private eventBus: EventEmitter;
+
     constructor(
         {
             featureStrategiesStore,
@@ -196,7 +200,8 @@ class FeatureToggleService {
         {
             getLogger,
             flagResolver,
-        }: Pick<IUnleashConfig, 'getLogger' | 'flagResolver'>,
+            eventBus,
+        }: Pick<IUnleashConfig, 'getLogger' | 'flagResolver' | 'eventBus'>,
         segmentService: ISegmentService,
         accessService: AccessService,
         eventService: EventService,
@@ -222,6 +227,7 @@ class FeatureToggleService {
         this.privateProjectChecker = privateProjectChecker;
         this.dependentFeaturesReadModel = dependentFeaturesReadModel;
         this.dependentFeaturesService = dependentFeaturesService;
+        this.eventBus = eventBus;
     }
 
     async validateFeaturesContext(
@@ -2444,7 +2450,12 @@ class FeatureToggleService {
     }
 
     async setFeatureCreatedByUserIdFromEvents(): Promise<void> {
-        await this.featureToggleStore.setCreatedByUserId(100);
+        const updated = await this.featureToggleStore.setCreatedByUserId(100);
+        if (updated !== undefined) {
+            this.eventBus.emit(FEATURES_CREATED_BY_PROCESSED, {
+                updated,
+            });
+        }
     }
 }
 

src/lib/features/feature-toggle/feature-toggle-store.ts

@@ -723,13 +723,13 @@ export default class FeatureToggleStore implements IFeatureToggleStore {
         return result?.potentially_stale ?? false;
     }
 
-    async setCreatedByUserId(batchSize: number): Promise<void> {
+    async setCreatedByUserId(batchSize: number): Promise<number | undefined> {
         const EVENTS_TABLE = 'events';
         const USERS_TABLE = 'users';
         const API_TOKEN_TABLE = 'api_tokens';
 
         if (!this.flagResolver.isEnabled('createdByUserIdDataMigration')) {
-            return;
+            return undefined;
         }
         const toUpdate = await this.db(`${TABLE} as f`)
             .joinRaw(`JOIN ${EVENTS_TABLE} AS ev ON ev.feature_name = f.name`)
@@ -757,6 +757,7 @@ export default class FeatureToggleStore implements IFeatureToggleStore {
         });
 
         await Promise.all(updatePromises);
+        return toUpdate.length;
     }
 }
 

src/lib/features/feature-toggle/types/feature-toggle-store-type.ts

@@ -104,5 +104,5 @@ export interface IFeatureToggleStore extends Store<FeatureToggle, string> {
         params: IFeatureProjectUserParams,
     ): Promise<IFeatureTypeCount[]>;
 
-    setCreatedByUserId(batchSize: number): Promise<void>;
+    setCreatedByUserId(batchSize: number): Promise<number | undefined>;
 }

src/lib/metric-events.ts

@@ -1,5 +1,13 @@
 const REQUEST_TIME = 'request_time';
 const DB_TIME = 'db_time';
 const SCHEDULER_JOB_TIME = 'scheduler_job_time';
+const FEATURES_CREATED_BY_PROCESSED = 'features_created_by_processed';
+const EVENTS_CREATED_BY_PROCESSED = 'events_created_by_processed';
 
-export { REQUEST_TIME, DB_TIME, SCHEDULER_JOB_TIME };
+export {
+    REQUEST_TIME,
+    DB_TIME,
+    SCHEDULER_JOB_TIME,
+    FEATURES_CREATED_BY_PROCESSED,
+    EVENTS_CREATED_BY_PROCESSED,
+};

src/lib/metrics.ts

@@ -224,6 +224,14 @@ export default class MetricsMonitor {
             help: 'Rate limits (per minute) for METHOD/ENDPOINT pairs',
             labelNames: ['endpoint', 'method'],
         });
+        const featureCreatedByMigration = createCounter({
+            name: 'feature_created_by_migration_count',
+            help: 'Feature createdBy migration count',
+        });
+        const eventCreatedByMigration = createCounter({
+            name: 'event_created_by_migration_count',
+            help: 'Event createdBy migration count',
+        });
 
         async function collectStaticCounters() {
             try {
@@ -374,6 +382,14 @@ export default class MetricsMonitor {
             schedulerDuration.labels(jobId).observe(time);
         });
 
+        eventBus.on(events.EVENTS_CREATED_BY_PROCESSED, ({ updated }) => {
+            eventCreatedByMigration.inc(updated);
+        });
+
+        eventBus.on(events.FEATURES_CREATED_BY_PROCESSED, ({ updated }) => {
+            featureCreatedByMigration.inc(updated);
+        });
+
         eventBus.on(events.DB_TIME, ({ store, action, time }) => {
             dbDuration.labels({ store, action }).observe(time);
         });

src/lib/services/addon-service.test.ts

@@ -16,6 +16,7 @@ import SimpleAddon from './addon-service-test-simple-addon';
 import { IAddonProviders } from '../addons';
 import EventService from '../features/events/event-service';
 import { SYSTEM_USER } from '../types';
+import { EventEmitter } from 'stream';
 
 const MASKED_VALUE = '*****';
 
@@ -25,7 +26,10 @@ let addonProvider: IAddonProviders;
 
 function getSetup() {
     const stores = createStores();
-    const eventService = new EventService(stores, { getLogger });
+    const eventService = new EventService(stores, {
+        getLogger,
+        eventBus: new EventEmitter(),
+    });
     const tagTypeService = new TagTypeService(
         stores,
         { getLogger },

src/lib/services/state-service.test.ts

@@ -15,12 +15,16 @@ import { GLOBAL_ENV } from '../types/environment';
 import variantsExportV3 from '../../test/examples/variantsexport_v3.json';
 import EventService from '../features/events/event-service';
 import { SYSTEM_USER_ID } from '../types';
+import { EventEmitter } from 'stream';
 const oldExportExample = require('./state-service-export-v1.json');
 const TESTUSERID = 3333;
 
 function getSetup() {
     const stores = createStores();
-    const eventService = new EventService(stores, { getLogger });
+    const eventService = new EventService(stores, {
+        getLogger,
+        eventBus: new EventEmitter(),
+    });
     return {
         stateService: new StateService(
             stores,
@@ -70,7 +74,10 @@ async function setupV3VariantsCompatibilityScenario(
             ],
         );
     });
-    const eventService = new EventService(stores, { getLogger });
+    const eventService = new EventService(stores, {
+        getLogger,
+        eventBus: new EventEmitter(),
+    });
     return {
         stateService: new StateService(
             stores,
@@ -645,7 +652,10 @@ test('Should export projects', async () => {
 
 test('exporting to new format works', async () => {
     const stores = createStores();
-    const eventService = new EventService(stores, { getLogger });
+    const eventService = new EventService(stores, {
+        getLogger,
+        eventBus: new EventEmitter(),
+    });
     const stateService = new StateService(
         stores,
         {

src/lib/types/stores/event-store.ts

@@ -17,5 +17,5 @@ export interface IEventStore
     getMaxRevisionId(currentMax?: number): Promise<number>;
     query(operations: IQueryOperations[]): Promise<IEvent[]>;
     queryCount(operations: IQueryOperations[]): Promise<number>;
-    setCreatedByUserId(batchSize: number): Promise<void>;
+    setCreatedByUserId(batchSize: number): Promise<number | undefined>;
 }

src/test/e2e/api/admin/event.e2e.test.ts

@@ -7,6 +7,7 @@ import getLogger from '../../../fixtures/no-logger';
 import { FEATURE_CREATED, IBaseEvent } from '../../../../lib/types/events';
 import { randomId } from '../../../../lib/util/random-id';
 import { EventService } from '../../../../lib/services';
+import { EventEmitter } from 'stream';
 
 let app: IUnleashTest;
 let db: ITestDb;
@@ -21,7 +22,10 @@ beforeAll(async () => {
             },
         },
     });
-    eventService = new EventService(db.stores, { getLogger });
+    eventService = new EventService(db.stores, {
+        getLogger,
+        eventBus: new EventEmitter(),
+    });
 });
 
 beforeEach(async () => {

src/test/e2e/features-created-by-user-id-data-migration.e2e.test.ts

@@ -1,4 +1,6 @@
+import { EventEmitter } from 'stream';
 import { createFeatureToggleService } from '../../lib/features';
+import { FEATURES_CREATED_BY_PROCESSED } from '../../lib/metric-events';
 import { EventService, FeatureToggleService } from '../../lib/services';
 import {
     ADMIN_TOKEN_USER,
@@ -11,11 +13,13 @@ import dbInit, { ITestDb } from './helpers/database-init';
 let stores: IUnleashStores;
 let db: ITestDb;
 let service: FeatureToggleService;
+let eventBus: EventEmitter;
 let eventService: EventService;
 let unleashConfig: IUnleashConfig;
 
 beforeAll(async () => {
     const config = createTestConfig();
+    eventBus = config.eventBus;
     db = await dbInit(
         'features_created_by_user_id_migration',
         config.getLogger,
@@ -193,3 +197,72 @@ test('admin tokens get populated to admin token user', async () => {
     expect(test1).toHaveLength(2);
     expect(test2).toHaveLength(2);
 });
+
+test('emits event with updated rows count', async () => {
+    for (let i = 0; i < 5; i++) {
+        await db.rawDatabase('features').insert({
+            name: `feature${i}`,
+            type: 'release',
+            project: 'default',
+            description: '--created_by_test--',
+        });
+    }
+
+    await db.rawDatabase('users').insert({
+        username: 'input2',
+    });
+
+    await db.rawDatabase('api_tokens').insert({
+        secret: 'token2',
+        username: 'adm-token2',
+        type: 'admin',
+        environment: 'default',
+        token_name: 'admin-token2',
+    });
+
+    await db.rawDatabase('events').insert({
+        type: 'feature-created',
+        created_by: 'input2',
+        feature_name: 'feature0',
+        data: `{"name":"feature0","description":null,"type":"release","project":"default","stale":false,"createdAt":"2024-01-08T10:36:32.866Z","lastSeenAt":null,"impressionData":false,"archivedAt":null,"archived":false}`,
+    });
+
+    await db.rawDatabase('events').insert({
+        type: 'feature-created',
+        created_by: 'input2',
+        feature_name: 'feature1',
+        data: `{"name":"feature1","description":null,"type":"release","project":"default","stale":false,"createdAt":"2024-01-08T10:36:32.866Z","lastSeenAt":null,"impressionData":false,"archivedAt":null,"archived":false}`,
+    });
+
+    await db.rawDatabase('events').insert({
+        type: 'feature-created',
+        created_by: 'adm-token2',
+        feature_name: 'feature2',
+        data: `{"name":"feature2","description":null,"type":"release","project":"default","stale":false,"createdAt":"2024-01-08T10:36:32.866Z","lastSeenAt":null,"impressionData":false,"archivedAt":null,"archived":false}`,
+    });
+
+    await db.rawDatabase('events').insert({
+        type: 'feature-created',
+        created_by: 'deleted-user',
+        feature_name: 'feature3',
+        data: `{"name":"feature3","description":null,"type":"release","project":"default","stale":false,"createdAt":"2024-01-08T10:36:32.866Z","lastSeenAt":null,"impressionData":false,"archivedAt":null,"archived":false}`,
+    });
+
+    await db.rawDatabase('events').insert({
+        type: 'feature-created',
+        created_by: 'adm-token2',
+        feature_name: 'feature4',
+        data: `{"name":"feature4","description":null,"type":"release","project":"default","stale":false,"createdAt":"2024-01-08T10:36:32.866Z","lastSeenAt":null,"impressionData":false,"archivedAt":null,"archived":false}`,
+    });
+
+    let triggered = false;
+
+    eventBus.on(FEATURES_CREATED_BY_PROCESSED, ({ updated }) => {
+        expect(updated).toBe(4);
+        triggered = true;
+    });
+
+    await service.setFeatureCreatedByUserIdFromEvents();
+
+    expect(triggered).toBeTruthy();
+});

src/test/fixtures/fake-event-store.ts

@@ -126,7 +126,7 @@ class FakeEventStore implements IEventStore {
         throw new Error('Method not implemented.');
     }
 
-    setCreatedByUserId(batchSize: number): Promise<void> {
+    setCreatedByUserId(batchSize: number): Promise<number | undefined> {
         throw new Error('Method not implemented.');
     }
 }