Merge pull request #3682 from backstage/freben/batch-markers

catalog-backend: batch the writing of statuses after refreshes

plugins/catalog-backend/src/catalog/DatabaseLocationsCatalog.ts

@@ -70,7 +70,7 @@ export class DatabaseLocationsCatalog implements LocationsCatalog {
 
   async logUpdateSuccess(
     locationId: string,
-    entityName?: string,
+    entityName?: string | string[],
   ): Promise<void> {
     await this.database.addLocationUpdateLogEvent(
       locationId,

plugins/catalog-backend/src/catalog/types.ts

@@ -80,7 +80,10 @@ export type LocationsCatalog = {
   locations(): Promise<LocationResponse[]>;
   location(id: string): Promise<LocationResponse>;
   locationHistory(id: string): Promise<LocationUpdateLogEvent[]>;
-  logUpdateSuccess(locationId: string, entityName?: string): Promise<void>;
+  logUpdateSuccess(
+    locationId: string,
+    entityName?: string | string[],
+  ): Promise<void>;
   logUpdateFailure(
     locationId: string,
     error?: Error,

plugins/catalog-backend/src/database/CommonDatabase.ts

@@ -432,7 +432,7 @@ export class CommonDatabase implements Database {
   async addLocationUpdateLogEvent(
     locationId: string,
     status: DatabaseLocationUpdateLogStatus,
-    entityName?: string,
+    entityName?: string | string[],
     message?: string,
   ): Promise<void> {
     // Remove log entries older than a day
@@ -442,14 +442,20 @@ export class CommonDatabase implements Database {
       .where('created_at', '<', cutoff.toISOString())
       .del();
 
-    await this.database<DatabaseLocationUpdateLogEvent>(
-      'location_update_log',
-    ).insert({
-      status,
-      location_id: locationId,
-      entity_name: entityName,
-      message,
-    });
+    const items: Partial<DatabaseLocationUpdateLogEvent>[] = [entityName]
+      .flat()
+      .map(n => ({
+        status,
+        location_id: locationId,
+        entity_name: n,
+        message,
+      }));
+
+    for (const chunk of lodash.chunk(items, BATCH_SIZE)) {
+      await this.database<DatabaseLocationUpdateLogEvent>(
+        'location_update_log',
+      ).insert(chunk);
+    }
   }
 
   private async updateEntitiesSearch(

plugins/catalog-backend/src/database/types.ts

@@ -208,7 +208,7 @@ export type Database = {
   addLocationUpdateLogEvent(
     locationId: string,
     status: DatabaseLocationUpdateLogStatus,
-    entityName?: string,
+    entityName?: string | string[],
     message?: string,
   ): Promise<void>;
 };

plugins/catalog-backend/src/ingestion/HigherOrderOperations.test.ts

@@ -372,10 +372,9 @@ describe('HigherOrderOperations', () => {
         '123',
         undefined,
       );
-      expect(locationsCatalog.logUpdateSuccess).toHaveBeenCalledWith(
-        '123',
+      expect(locationsCatalog.logUpdateSuccess).toHaveBeenCalledWith('123', [
         'c1',
-      );
+      ]);
     });
 
     it('logs unsuccessful updates when reader fails', async () => {

plugins/catalog-backend/src/ingestion/HigherOrderOperations.ts

@@ -188,12 +188,10 @@ export class HigherOrderOperations implements HigherOrderOperation {
 
     this.logger.info(`Posting update success markers`);
 
-    for (const entity of readerOutput.entities) {
-      await this.locationsCatalog.logUpdateSuccess(
-        location.id,
-        entity.entity.metadata.name,
-      );
-    }
+    await this.locationsCatalog.logUpdateSuccess(
+      location.id,
+      readerOutput.entities.map(e => e.entity.metadata.name),
+    );
 
     this.logger.info(
       `Wrote ${readerOutput.entities.length} entities from location ${