Merge pull request #2810 from spotify/freben/batch

feat(catalog-backend): introduce batching, speed up reading and writing of large datasets

plugins/catalog-backend/migrations/20201007201501_index_entity_search.js

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2020 Spotify AB
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// @ts-check
+
+/**
+ * @param {import('knex')} knex
+ */
+exports.up = async function up(knex) {
+  await knex.schema.alterTable('entities_search', table => {
+    table.index(['key'], 'entities_search_key');
+    table.index(['value'], 'entities_search_value');
+  });
+};
+
+/**
+ * @param {import('knex')} knex
+ */
+exports.down = async function down(knex) {
+  await knex.schema.alterTable('entities_search', table => {
+    table.dropIndex('', 'entities_search_key');
+    table.dropIndex('', 'entities_search_value');
+  });
+};

plugins/catalog-backend/src/catalog/CoalescedEntitiesCatalog.test.ts

@@ -37,6 +37,7 @@ describe('CoalescedEntitiesCatalog', () => {
     entityByUid: jest.fn(),
     entityByName: jest.fn(),
     addOrUpdateEntity: jest.fn(),
+    addEntities: jest.fn(),
     removeEntityByUid: jest.fn(),
   };
 
@@ -45,6 +46,7 @@ describe('CoalescedEntitiesCatalog', () => {
     entityByUid: jest.fn(),
     entityByName: jest.fn(),
     addOrUpdateEntity: jest.fn(),
+    addEntities: jest.fn(),
     removeEntityByUid: jest.fn(),
   };
 

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

@@ -90,6 +90,10 @@ export class CoalescedEntitiesCatalog implements EntitiesCatalog {
     throw new Error('Method not implemented.');
   }
 
+  addEntities(): Promise<void> {
+    throw new Error('Method not implemented.');
+  }
+
   removeEntityByUid(): Promise<void> {
     throw new Error('Method not implemented.');
   }

plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.test.ts

@@ -25,6 +25,7 @@ describe('DatabaseEntitiesCatalog', () => {
     db = {
       transaction: jest.fn(),
       addEntity: jest.fn(),
+      addEntities: jest.fn(),
       updateEntity: jest.fn(),
       entities: jest.fn(),
       entityByName: jest.fn(),

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

@@ -79,6 +79,15 @@ export class DatabaseEntitiesCatalog implements EntitiesCatalog {
     });
   }
 
+  async addEntities(entities: Entity[], locationId?: string): Promise<void> {
+    await this.database.transaction(async tx => {
+      await this.database.addEntities(
+        tx,
+        entities.map(entity => ({ locationId, entity })),
+      );
+    });
+  }
+
   async removeEntityByUid(uid: string): Promise<void> {
     return await this.database.transaction(async tx => {
       const entityResponse = await this.database.entityByUid(tx, uid);

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

@@ -50,6 +50,10 @@ export class StaticEntitiesCatalog implements EntitiesCatalog {
     throw new Error('Not supported');
   }
 
+  async addEntities(): Promise<void> {
+    throw new Error('Not supported');
+  }
+
   async removeEntityByUid(): Promise<void> {
     throw new Error('Not supported');
   }

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

@@ -26,6 +26,7 @@ export type EntitiesCatalog = {
   entityByUid(uid: string): Promise<Entity | undefined>;
   entityByName(name: EntityName): Promise<Entity | undefined>;
   addOrUpdateEntity(entity: Entity, locationId?: string): Promise<Entity>;
+  addEntities(entities: Entity[], locationId?: string): Promise<void>;
   removeEntityByUid(uid: string): Promise<void>;
 };
 

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

@@ -168,6 +168,79 @@ describe('CommonDatabase', () => {
     });
   });
 
+  describe('addEntities', () => {
+    it('happy path: adds entities to empty database', async () => {
+      await db.transaction(tx => db.addEntities(tx, [entityRequest]));
+      expect(true).toBeTruthy();
+    });
+
+    it('rejects adding the same-named entity twice', async () => {
+      const req: DbEntityRequest[] = [
+        {
+          entity: {
+            apiVersion: 'av1',
+            kind: 'k1',
+            metadata: { name: 'n1', namespace: 'ns1' },
+          },
+        },
+        {
+          entity: {
+            apiVersion: 'av1',
+            kind: 'k1',
+            metadata: { name: 'n1', namespace: 'ns1' },
+          },
+        },
+      ];
+      await expect(
+        db.transaction(tx => db.addEntities(tx, req)),
+      ).rejects.toThrow(ConflictError);
+    });
+
+    it('rejects adding the almost-same-namespace entity twice', async () => {
+      const req: DbEntityRequest[] = [
+        {
+          entity: {
+            apiVersion: 'av1',
+            kind: 'k1',
+            metadata: { name: 'n1', namespace: 'ns1' },
+          },
+        },
+        {
+          entity: {
+            apiVersion: 'av1',
+            kind: 'k1',
+            metadata: { name: 'n1', namespace: 'nS1' },
+          },
+        },
+      ];
+      await expect(
+        db.transaction(tx => db.addEntities(tx, req)),
+      ).rejects.toThrow(ConflictError);
+    });
+
+    it('accepts adding the same-named entity twice if on different namespaces', async () => {
+      const req: DbEntityRequest[] = [
+        {
+          entity: {
+            apiVersion: 'av1',
+            kind: 'k1',
+            metadata: { name: 'n1', namespace: 'ns1' },
+          },
+        },
+        {
+          entity: {
+            apiVersion: 'av1',
+            kind: 'k1',
+            metadata: { name: 'n1', namespace: 'ns2' },
+          },
+        },
+      ];
+      await expect(
+        db.transaction(tx => db.addEntities(tx, req)),
+      ).resolves.toBeUndefined();
+    });
+  });
+
   describe('locationHistory', () => {
     it('outputs the history correctly', async () => {
       const location: Location = {

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

@@ -45,6 +45,12 @@ import type {
   EntityFilters,
 } from './types';
 
+// The number of items that are sent per batch to the database layer, when
+// doing .batchInsert calls to knex. This needs to be low enough to not cause
+// errors in the underlying engine due to exceeding query limits, but large
+// enough to get the speed benefits.
+const BATCH_SIZE = 50;
+
 /**
  * The core database implementation.
  */
@@ -100,6 +106,48 @@ export class CommonDatabase implements Database {
     return { locationId: request.locationId, entity: newEntity };
   }
 
+  async addEntities(
+    txOpaque: unknown,
+    request: DbEntityRequest[],
+  ): Promise<void> {
+    const tx = txOpaque as Knex.Transaction<any, any>;
+
+    const entityRows: DbEntitiesRow[] = [];
+    const searchRows: DbEntitiesSearchRow[] = [];
+
+    for (const { entity, locationId } of request) {
+      if (entity.metadata.uid !== undefined) {
+        throw new InputError('May not specify uid for new entities');
+      } else if (entity.metadata.etag !== undefined) {
+        throw new InputError('May not specify etag for new entities');
+      } else if (entity.metadata.generation !== undefined) {
+        throw new InputError('May not specify generation for new entities');
+      }
+
+      const newEntity = {
+        ...entity,
+        metadata: {
+          ...entity.metadata,
+          uid: generateEntityUid(),
+          etag: generateEntityEtag(),
+          generation: 1,
+        },
+      };
+
+      entityRows.push(this.toEntityRow(locationId, newEntity));
+      searchRows.push(...buildEntitySearch(newEntity.metadata.uid, newEntity));
+    }
+
+    await tx.batchInsert('entities', entityRows, BATCH_SIZE);
+    await tx<DbEntitiesSearchRow>('entities_search')
+      .whereIn(
+        'entity_id',
+        entityRows.map(r => r.id),
+      )
+      .del();
+    await tx.batchInsert('entities_search', searchRows, BATCH_SIZE);
+  }
+
   async updateEntity(
     txOpaque: unknown,
     request: DbEntityRequest,
@@ -165,10 +213,10 @@ export class CommonDatabase implements Database {
   ): Promise<DbEntityResponse[]> {
     const tx = txOpaque as Knex.Transaction<any, any>;
 
-    let builder = tx<DbEntitiesRow>('entities');
-    for (const [indexU, filter] of (filters ?? []).entries()) {
-      const index = Number(indexU);
-      const key = filter.key.toLowerCase().replace(/\*/g, '%');
+    let entitiesQuery = tx<DbEntitiesRow>('entities');
+
+    for (const filter of filters || []) {
+      const key = filter.key.toLowerCase().replace(/[*]/g, '%');
       const keyOp = filter.key.includes('*') ? 'like' : '=';
 
       let matchNulls = false;
@@ -179,36 +227,54 @@ export class CommonDatabase implements Database {
         if (!value) {
           matchNulls = true;
         } else if (value.includes('*')) {
-          matchLike.push(value.toLowerCase().replace(/\*/g, '%'));
+          matchLike.push(value.toLowerCase().replace(/[*]/g, '%'));
         } else {
           matchIn.push(value.toLowerCase());
         }
       }
 
-      builder = builder
-        .leftOuterJoin(`entities_search as t${index}`, function joins() {
-          this.on('entities.id', '=', `t${index}.entity_id`);
-          this.andOn(`t${index}.key`, keyOp, tx.raw('?', [key]));
-        })
-        .where(function rules() {
-          if (matchIn.length) {
-            this.orWhereIn(`t${index}.value`, matchIn);
-          }
-          if (matchLike.length) {
-            for (const x of matchLike) {
-              this.orWhere(`t${index}.value`, 'like', tx.raw('?', [x]));
+      // NOTE(freben): This used to be a set of OUTER JOIN, which may seem to
+      // make a lot of sense. However, it had abysmal performance on sqlite
+      // when datasets grew large, so we're using IN instead.
+      const matchQuery = tx<DbEntitiesSearchRow>('entities_search')
+        .select('entity_id')
+        .where(function keyFilter() {
+          this.andWhere('key', keyOp, key);
+          this.andWhere(function valueFilter() {
+            if (matchIn.length === 1) {
+              this.orWhere({ value: matchIn[0] });
+            } else if (matchIn.length > 1) {
+              this.orWhereIn('value', matchIn);
             }
-          }
-          if (matchNulls) {
-            this.orWhereNull(`t${index}.value`);
-          }
+            if (matchLike.length) {
+              for (const x of matchLike) {
+                this.orWhere('value', 'like', tx.raw('?', [x]));
+              }
+            }
+            if (matchNulls) {
+              // Match explicit nulls, and then handle absence separately below
+              this.orWhereNull('value');
+            }
+          });
         });
+
+      // Handle absence as nulls as well
+      entitiesQuery = entitiesQuery.andWhere(function match() {
+        this.whereIn('id', matchQuery);
+        if (matchNulls) {
+          this.orWhereNotIn(
+            'id',
+            tx<DbEntitiesSearchRow>('entities_search')
+              .select('entity_id')
+              .where('key', keyOp, key),
+          );
+        }
+      });
     }
 
-    const rows = await builder
+    const rows = await entitiesQuery
       .select('entities.*')
-      .orderBy('full_name', 'asc')
-      .groupBy('id');
+      .orderBy('full_name', 'asc');
 
     return rows.map(row => this.toEntityResponse(row));
   }

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

@@ -97,6 +97,14 @@ export type Database = {
    */
   addEntity(tx: unknown, request: DbEntityRequest): Promise<DbEntityResponse>;
 
+  /**
+   * Adds a set of new entities to the catalog.
+   *
+   * @param tx An ongoing transaction
+   * @param request The entities being added
+   */
+  addEntities(tx: unknown, request: DbEntityRequest[]): Promise<void>;
+
   /**
    * Updates an existing entity in the catalog.
    *

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

@@ -39,6 +39,7 @@ describe('HigherOrderOperations', () => {
       entityByUid: jest.fn(),
       entityByName: jest.fn(),
       addOrUpdateEntity: jest.fn(),
+      addEntities: jest.fn(),
       removeEntityByUid: jest.fn(),
     };
     locationsCatalog = {
@@ -191,8 +192,8 @@ describe('HigherOrderOperations', () => {
         entities: [{ entity: desc, location }],
         errors: [],
       });
-      entitiesCatalog.entityByName.mockResolvedValue(undefined);
-      entitiesCatalog.addOrUpdateEntity.mockResolvedValue(desc);
+      entitiesCatalog.entities.mockResolvedValue([]);
+      entitiesCatalog.addEntities.mockResolvedValue(undefined);
 
       await expect(
         higherOrderOperation.refreshAllLocations(),
@@ -204,18 +205,19 @@ describe('HigherOrderOperations', () => {
         type: 'some',
         target: 'thing',
       });
-      expect(entitiesCatalog.entityByName).toHaveBeenCalledTimes(1);
-      expect(entitiesCatalog.entityByName).toHaveBeenNthCalledWith(1, {
-        kind: 'Component',
-        namespace: ENTITY_DEFAULT_NAMESPACE,
-        name: 'c1',
-      });
-      expect(entitiesCatalog.addOrUpdateEntity).toHaveBeenCalledTimes(1);
-      expect(entitiesCatalog.addOrUpdateEntity).toHaveBeenNthCalledWith(
+      expect(entitiesCatalog.entities).toHaveBeenCalledTimes(1);
+      expect(entitiesCatalog.entities).toHaveBeenNthCalledWith(
         1,
-        expect.objectContaining({
-          metadata: expect.objectContaining({ name: 'c1' }),
-        }),
+        expect.arrayContaining([
+          { key: 'kind', values: ['Component'] },
+          { key: 'metadata.namespace', values: [ENTITY_DEFAULT_NAMESPACE] },
+          { key: 'metadata.name', values: ['c1'] },
+        ]),
+      );
+      expect(entitiesCatalog.addEntities).toHaveBeenCalledTimes(1);
+      expect(entitiesCatalog.addEntities).toHaveBeenNthCalledWith(
+        1,
+        [expect.objectContaining({ metadata: { name: 'c1' } })],
         '123',
       );
     });
@@ -245,8 +247,8 @@ describe('HigherOrderOperations', () => {
         entities: [{ entity: desc, location }],
         errors: [],
       });
-      entitiesCatalog.entityByName.mockResolvedValue(undefined);
-      entitiesCatalog.addOrUpdateEntity.mockResolvedValue(desc);
+      entitiesCatalog.entities.mockResolvedValue([]);
+      entitiesCatalog.addEntities.mockResolvedValue(undefined);
 
       await expect(
         higherOrderOperation.refreshAllLocations(),

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

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-import { InputError } from '@backstage/backend-common';
+import { ConflictError, InputError } from '@backstage/backend-common';
 import {
   Entity,
   entityHasChanges,
@@ -23,15 +23,33 @@ import {
   LocationSpec,
   serializeEntityRef,
 } from '@backstage/catalog-model';
+import { chunk, groupBy } from 'lodash';
 import { v4 as uuidv4 } from 'uuid';
 import { Logger } from 'winston';
 import { EntitiesCatalog, LocationsCatalog } from '../catalog';
+import { durationText } from '../util/timing';
 import {
   AddLocationResult,
   HigherOrderOperation,
   LocationReader,
 } from './types';
 
+type BatchContext = {
+  kind: string;
+  namespace: string;
+  location: Location;
+};
+
+// Some locations return tens or hundreds of thousands of entities. To make
+// those payloads more manageable, we break work apart in batches of this
+// many entities and write them to storage per batch.
+const BATCH_SIZE = 100;
+
+// When writing large batches, there's an increasing chance of contention in
+// the form of conflicts where we compete with other writes. Each batch gets
+// this many attempts at being written before giving up.
+const BATCH_ATTEMPTS = 3;
+
 /**
  * Placeholder for operations that span several catalogs and/or stretches out
  * in time.
@@ -88,7 +106,7 @@ export class HigherOrderOperations implements HigherOrderOperation {
     if (readerOutput.errors.length) {
       const item = readerOutput.errors[0];
       throw new InputError(
-        `Failed to read location ${item.location.type} ${item.location.target}, ${item.error}`,
+        `Failed to read location ${item.location.type}:${item.location.target}, ${item.error}`,
       );
     }
 
@@ -121,85 +139,197 @@ export class HigherOrderOperations implements HigherOrderOperation {
    * without changes.
    */
   async refreshAllLocations(): Promise<void> {
-    const startTimestamp = new Date().valueOf();
+    const startTimestamp = process.hrtime();
     this.logger.info('Beginning locations refresh');
 
     const locations = await this.locationsCatalog.locations();
     this.logger.info(`Visiting ${locations.length} locations`);
 
     for (const { data: location } of locations) {
-      this.logger.debug(
-        `Refreshing location id="${location.id}" type="${location.type}" target="${location.target}"`,
+      this.logger.info(
+        `Refreshing location ${location.type}:${location.target}`,
       );
       try {
         await this.refreshSingleLocation(location);
         await this.locationsCatalog.logUpdateSuccess(location.id, undefined);
       } catch (e) {
-        this.logger.debug(
-          `Failed to refresh location id="${location.id}" type="${location.type}" target="${location.target}", ${e}`,
+        this.logger.warn(
+          `Failed to refresh location ${location.type}:${location.target}, ${e}`,
         );
         await this.locationsCatalog.logUpdateFailure(location.id, e);
       }
     }
 
-    const endTimestamp = new Date().valueOf();
-    const duration = ((endTimestamp - startTimestamp) / 1000).toFixed(1);
-    this.logger.debug(`Completed locations refresh in ${duration} seconds`);
+    this.logger.info(
+      `Completed locations refresh in ${durationText(startTimestamp)}`,
+    );
   }
 
   // Performs a full refresh of a single location
   private async refreshSingleLocation(location: Location) {
+    let startTimestamp = process.hrtime();
+
     const readerOutput = await this.locationReader.read({
       type: location.type,
       target: location.target,
     });
 
     for (const item of readerOutput.errors) {
-      this.logger.debug(
-        `Failed item in location type="${item.location.type}" target="${item.location.target}", ${item.error}`,
+      this.logger.warn(
+        `Failed item in location ${item.location.type}:${item.location.target}, ${item.error}`,
       );
     }
 
     this.logger.info(
-      `Read ${readerOutput.entities.length} entities from location ${location.type} ${location.target}`,
+      `Read ${readerOutput.entities.length} entities from location ${
+        location.type
+      }:${location.target} in ${durationText(startTimestamp)}`,
     );
 
-    const startTimestamp = process.hrtime();
-    for (const item of readerOutput.entities) {
-      const { entity } = item;
+    startTimestamp = process.hrtime();
 
-      try {
-        const previous = await this.entitiesCatalog.entityByName(
-          getEntityName(entity),
+    await this.batchAddOrUpdateEntities(
+      readerOutput.entities.map(e => e.entity),
+      location,
+    );
+
+    this.logger.info(
+      `Wrote ${readerOutput.entities.length} entities from location ${
+        location.type
+      }:${location.target} in ${durationText(startTimestamp)}`,
+    );
+  }
+
+  /**
+   * Writes a number of entities efficiently to storage.
+   *
+   * @param entities Some entities
+   * @param location The location that they all belong to
+   */
+  async batchAddOrUpdateEntities(entities: Entity[], location: Location) {
+    // Group the entities by unique kind+namespace combinations
+    const entitiesByKindAndNamespace = groupBy(entities, entity => {
+      const name = getEntityName(entity);
+      return `${name.kind}:${name.namespace}`.toLowerCase();
+    });
+
+    for (const groupEntities of Object.values(entitiesByKindAndNamespace)) {
+      const { kind, namespace } = getEntityName(groupEntities[0]);
+
+      // Go through the new entities in reasonable chunk sizes (sometimes,
+      // sources produce tens of thousands of entities, and those are too large
+      // batch sizes to reasonably send to the database)
+      for (const batch of chunk(groupEntities, BATCH_SIZE)) {
+        const first = serializeEntityRef(batch[0]);
+        const last = serializeEntityRef(batch[batch.length - 1]);
+        this.logger.debug(
+          `Considering batch ${first}-${last} (${batch.length} entries)`,
         );
 
-        if (!previous) {
-          await this.entitiesCatalog.addOrUpdateEntity(entity, location.id);
-        } else if (entityHasChanges(previous, entity)) {
-          await this.entitiesCatalog.addOrUpdateEntity(entity, location.id);
+        // Retry the batch write a few times to deal with contention
+        const context = { kind, namespace, location };
+        for (let attempt = 1; attempt <= BATCH_ATTEMPTS; ++attempt) {
+          try {
+            const { toAdd, toUpdate } = await this.analyzeBatch(batch, context);
+            if (toAdd.length) await this.batchAdd(toAdd, context);
+            if (toUpdate.length) await this.batchUpdate(toUpdate, context);
+            break;
+          } catch (e) {
+            if (e instanceof ConflictError && attempt < BATCH_ATTEMPTS) {
+              this.logger.warn(
+                `Failed to write batch at attempt ${attempt}/${BATCH_ATTEMPTS}, ${e}`,
+              );
+            } else {
+              throw e;
+            }
+          }
         }
+      }
+    }
+  }
 
-        await this.locationsCatalog.logUpdateSuccess(
-          location.id,
-          entity.metadata.name,
-        );
-      } catch (error) {
-        this.logger.info(
-          `Failed refresh of entity ${serializeEntityRef(entity)}, ${error}`,
-        );
+  // Given a batch of entities that were just read from a location, take them
+  // into consideration by comparing against the existing catalog entities and
+  // produce the list of entities to be added, and the list of entities to be
+  // updated
+  private async analyzeBatch(
+    newEntities: Entity[],
+    { kind, namespace }: BatchContext,
+  ): Promise<{
+    toAdd: Entity[];
+    toUpdate: Entity[];
+  }> {
+    const markTimestamp = process.hrtime();
 
-        await this.locationsCatalog.logUpdateFailure(
-          location.id,
-          error,
-          entity.metadata.name,
-        );
+    const names = newEntities.map(e => e.metadata.name);
+    const oldEntities = await this.entitiesCatalog.entities([
+      { key: 'kind', values: [kind] },
+      { key: 'metadata.namespace', values: [namespace] },
+      { key: 'metadata.name', values: names },
+    ]);
+
+    const oldEntitiesByName = new Map(
+      oldEntities.map(e => [e.metadata.name, e]),
+    );
+
+    const toAdd: Entity[] = [];
+    const toUpdate: Entity[] = [];
+
+    for (const newEntity of newEntities) {
+      const oldEntity = oldEntitiesByName.get(newEntity.metadata.name);
+      if (!oldEntity) {
+        toAdd.push(newEntity);
+      } else if (entityHasChanges(oldEntity, newEntity)) {
+        toUpdate.push(newEntity);
       }
     }
 
-    const delta = process.hrtime(startTimestamp);
-    const durationMs = ((delta[0] * 1e9 + delta[1]) / 1e9).toFixed(1);
-    this.logger.info(
-      `Wrote ${readerOutput.entities.length} entities from location ${location.type} ${location.target} in ${durationMs} seconds`,
+    this.logger.debug(
+      `Found ${toAdd.length} entities to add, ${
+        toUpdate.length
+      } entities to update in ${durationText(markTimestamp)}`,
+    );
+
+    return { toAdd, toUpdate };
+  }
+
+  // Efficiently adds the given entities to storage, under the assumption that
+  // they do not conflict with any existing entities
+  private async batchAdd(entities: Entity[], { location }: BatchContext) {
+    const markTimestamp = process.hrtime();
+
+    await this.entitiesCatalog.addEntities(entities, location.id);
+
+    // TODO(freben): Still not batched
+    for (const entity of entities) {
+      await this.locationsCatalog.logUpdateSuccess(
+        location.id,
+        entity.metadata.name,
+      );
+    }
+
+    this.logger.debug(
+      `Added ${entities.length} entities in ${durationText(markTimestamp)}`,
+    );
+  }
+
+  // Efficiently updates the given entities into storage, under the assumption
+  // that there already exist entities with the same names
+  private async batchUpdate(entities: Entity[], { location }: BatchContext) {
+    const markTimestamp = process.hrtime();
+
+    // TODO(freben): Still not batched
+    for (const entity of entities) {
+      await this.entitiesCatalog.addOrUpdateEntity(entity);
+
+      await this.locationsCatalog.logUpdateSuccess(
+        location.id,
+        entity.metadata.name,
+      );
+    }
+
+    this.logger.debug(
+      `Updated ${entities.length} entities in ${durationText(markTimestamp)}`,
     );
   }
 }

plugins/catalog-backend/src/service/router.test.ts

@@ -35,6 +35,7 @@ describe('createRouter', () => {
       entityByUid: jest.fn(),
       entityByName: jest.fn(),
       addOrUpdateEntity: jest.fn(),
+      addEntities: jest.fn(),
       removeEntityByUid: jest.fn(),
     };
     locationsCatalog = {

plugins/catalog-backend/src/util/timing.ts

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2020 Spotify AB
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Returns a string with the elapsed time since the start of an operation,
+ * with some human friendly precision, e.g. "133ms" or "14.5s".
+ *
+ * @param startTimestamp The timestamp (from process.hrtime()) at the start ot
+ *                       the operation
+ */
+export function durationText(startTimestamp: [number, number]): string {
+  const delta = process.hrtime(startTimestamp);
+  const seconds = delta[0] + delta[1] / 1e9;
+  if (seconds > 1) {
+    return `${seconds.toFixed(1)}s`;
+  }
+  return `${(seconds * 1000).toFixed(0)}ms`;
+}