From a1540e72866b3b9b83aff57e32ef2a72ab8b71a6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Wed, 7 Oct 2020 14:21:00 +0200 Subject: [PATCH] feat(catalog-backend): introduce batching, speed up reading and writing of large datasets --- .../20201007201501_index_entity_search.js | 37 +++ .../catalog/CoalescedEntitiesCatalog.test.ts | 2 + .../src/catalog/CoalescedEntitiesCatalog.ts | 4 + .../catalog/DatabaseEntitiesCatalog.test.ts | 1 + .../src/catalog/DatabaseEntitiesCatalog.ts | 9 + .../src/catalog/StaticEntitiesCatalog.ts | 4 + plugins/catalog-backend/src/catalog/types.ts | 1 + .../src/database/CommonDatabase.test.ts | 73 ++++++ .../src/database/CommonDatabase.ts | 114 ++++++++-- plugins/catalog-backend/src/database/types.ts | 8 + .../ingestion/HigherOrderOperations.test.ts | 32 +-- .../src/ingestion/HigherOrderOperations.ts | 210 ++++++++++++++---- .../src/service/router.test.ts | 1 + plugins/catalog-backend/src/util/timing.ts | 31 +++ 14 files changed, 448 insertions(+), 79 deletions(-) create mode 100644 plugins/catalog-backend/migrations/20201007201501_index_entity_search.js create mode 100644 plugins/catalog-backend/src/util/timing.ts diff --git a/plugins/catalog-backend/migrations/20201007201501_index_entity_search.js b/plugins/catalog-backend/migrations/20201007201501_index_entity_search.js new file mode 100644 index 0000000000..b902073e64 --- /dev/null +++ b/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'); + }); +}; diff --git a/plugins/catalog-backend/src/catalog/CoalescedEntitiesCatalog.test.ts b/plugins/catalog-backend/src/catalog/CoalescedEntitiesCatalog.test.ts index 1cfea23e43..f127c2e0bd 100644 --- a/plugins/catalog-backend/src/catalog/CoalescedEntitiesCatalog.test.ts +++ b/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(), }; diff --git a/plugins/catalog-backend/src/catalog/CoalescedEntitiesCatalog.ts b/plugins/catalog-backend/src/catalog/CoalescedEntitiesCatalog.ts index 7eaf1868cb..705a210f60 100644 --- a/plugins/catalog-backend/src/catalog/CoalescedEntitiesCatalog.ts +++ b/plugins/catalog-backend/src/catalog/CoalescedEntitiesCatalog.ts @@ -90,6 +90,10 @@ export class CoalescedEntitiesCatalog implements EntitiesCatalog { throw new Error('Method not implemented.'); } + addEntities(): Promise { + throw new Error('Method not implemented.'); + } + removeEntityByUid(): Promise { throw new Error('Method not implemented.'); } diff --git a/plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.test.ts b/plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.test.ts index b94365a9ba..a9444af0f9 100644 --- a/plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.test.ts +++ b/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(), diff --git a/plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.ts b/plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.ts index 750d728c67..82ef680b09 100644 --- a/plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.ts +++ b/plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.ts @@ -79,6 +79,15 @@ export class DatabaseEntitiesCatalog implements EntitiesCatalog { }); } + async addEntities(entities: Entity[], locationId?: string): Promise { + await this.database.transaction(async tx => { + await this.database.addEntities( + tx, + entities.map(entity => ({ locationId, entity })), + ); + }); + } + async removeEntityByUid(uid: string): Promise { return await this.database.transaction(async tx => { const entityResponse = await this.database.entityByUid(tx, uid); diff --git a/plugins/catalog-backend/src/catalog/StaticEntitiesCatalog.ts b/plugins/catalog-backend/src/catalog/StaticEntitiesCatalog.ts index 65fbfb9071..b9d50cfc62 100644 --- a/plugins/catalog-backend/src/catalog/StaticEntitiesCatalog.ts +++ b/plugins/catalog-backend/src/catalog/StaticEntitiesCatalog.ts @@ -50,6 +50,10 @@ export class StaticEntitiesCatalog implements EntitiesCatalog { throw new Error('Not supported'); } + async addEntities(): Promise { + throw new Error('Not supported'); + } + async removeEntityByUid(): Promise { throw new Error('Not supported'); } diff --git a/plugins/catalog-backend/src/catalog/types.ts b/plugins/catalog-backend/src/catalog/types.ts index 37618a2440..5d84cbcebc 100644 --- a/plugins/catalog-backend/src/catalog/types.ts +++ b/plugins/catalog-backend/src/catalog/types.ts @@ -26,6 +26,7 @@ export type EntitiesCatalog = { entityByUid(uid: string): Promise; entityByName(name: EntityName): Promise; addOrUpdateEntity(entity: Entity, locationId?: string): Promise; + addEntities(entities: Entity[], locationId?: string): Promise; removeEntityByUid(uid: string): Promise; }; diff --git a/plugins/catalog-backend/src/database/CommonDatabase.test.ts b/plugins/catalog-backend/src/database/CommonDatabase.test.ts index 2bacec4026..2883d81b3d 100644 --- a/plugins/catalog-backend/src/database/CommonDatabase.test.ts +++ b/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 = { diff --git a/plugins/catalog-backend/src/database/CommonDatabase.ts b/plugins/catalog-backend/src/database/CommonDatabase.ts index 890a64781a..c583217711 100644 --- a/plugins/catalog-backend/src/database/CommonDatabase.ts +++ b/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 { + const tx = txOpaque as Knex.Transaction; + + 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('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 { const tx = txOpaque as Knex.Transaction; - let builder = tx('entities'); - for (const [indexU, filter] of (filters ?? []).entries()) { - const index = Number(indexU); - const key = filter.key.toLowerCase().replace(/\*/g, '%'); + let entitiesQuery = tx('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('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('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)); } diff --git a/plugins/catalog-backend/src/database/types.ts b/plugins/catalog-backend/src/database/types.ts index 5a577f2223..36de5c2152 100644 --- a/plugins/catalog-backend/src/database/types.ts +++ b/plugins/catalog-backend/src/database/types.ts @@ -97,6 +97,14 @@ export type Database = { */ addEntity(tx: unknown, request: DbEntityRequest): Promise; + /** + * 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; + /** * Updates an existing entity in the catalog. * diff --git a/plugins/catalog-backend/src/ingestion/HigherOrderOperations.test.ts b/plugins/catalog-backend/src/ingestion/HigherOrderOperations.test.ts index 32fcc27909..15d9168732 100644 --- a/plugins/catalog-backend/src/ingestion/HigherOrderOperations.test.ts +++ b/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(), diff --git a/plugins/catalog-backend/src/ingestion/HigherOrderOperations.ts b/plugins/catalog-backend/src/ingestion/HigherOrderOperations.ts index 7232b876c7..a9ad03302f 100644 --- a/plugins/catalog-backend/src/ingestion/HigherOrderOperations.ts +++ b/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 { - 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)}`, ); } } diff --git a/plugins/catalog-backend/src/service/router.test.ts b/plugins/catalog-backend/src/service/router.test.ts index 32d7b6aa63..ecf4b0e614 100644 --- a/plugins/catalog-backend/src/service/router.test.ts +++ b/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 = { diff --git a/plugins/catalog-backend/src/util/timing.ts b/plugins/catalog-backend/src/util/timing.ts new file mode 100644 index 0000000000..1abc446ab4 --- /dev/null +++ b/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`; +}