From def0eef285491deb3b575f121f24c10eba78a85e Mon Sep 17 00:00:00 2001 From: Kiss Miklos Date: Mon, 20 Jun 2022 11:56:20 +0200 Subject: [PATCH] add a potential implementation for refreshKeys Signed-off-by: Kiss Miklos --- .../migrations/20220616202842_refresh_keys.js | 41 +++++++++++++++++++ plugins/catalog-backend/src/api/processor.ts | 10 ++++- .../src/database/DefaultProcessingDatabase.ts | 36 ++++++++++++++++ .../catalog-backend/src/database/tables.ts | 5 +++ plugins/catalog-backend/src/database/types.ts | 13 ++++++ .../src/modules/core/FileReaderProcessor.ts | 7 ++++ .../src/modules/core/PlaceholderProcessor.ts | 8 +++- .../src/modules/core/UrlReaderProcessor.ts | 7 ++++ .../DefaultCatalogProcessingEngine.ts | 6 +++ .../processing/ProcessorOutputCollector.ts | 7 ++++ .../catalog-backend/src/processing/types.ts | 2 +- 11 files changed, 139 insertions(+), 3 deletions(-) create mode 100644 plugins/catalog-backend/migrations/20220616202842_refresh_keys.js diff --git a/plugins/catalog-backend/migrations/20220616202842_refresh_keys.js b/plugins/catalog-backend/migrations/20220616202842_refresh_keys.js new file mode 100644 index 0000000000..dfa0bf5794 --- /dev/null +++ b/plugins/catalog-backend/migrations/20220616202842_refresh_keys.js @@ -0,0 +1,41 @@ +/* + * Copyright 2022 The Backstage Authors + * + * 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. + */ + +exports.up = async function up(knex) { + await knex.schema.createTable('refresh_keys', table => { + table.comment( + 'This table contains relations between entities and keys to trigger refreshes with', + ); + table + .text('entity_ref') + .notNullable() + .comment('A reference to the entity that the refresh key is tied to'); + table + .text('key') + .notNullable() + .comment( + 'A reference to a key which should be used to trigger a refresh on this entity', + ); + table.unique(['entity_ref', 'key']); + }); +}; + +/** + * @param { import("knex").Knex } knex + */ +exports.down = async function down(knex) { + await knex.schema.dropTable('refresh_keys'); +}; diff --git a/plugins/catalog-backend/src/api/processor.ts b/plugins/catalog-backend/src/api/processor.ts index 3f274d13ea..d5a08edd44 100644 --- a/plugins/catalog-backend/src/api/processor.ts +++ b/plugins/catalog-backend/src/api/processor.ts @@ -169,9 +169,17 @@ export type CatalogProcessorErrorResult = { location: LocationSpec; }; +/** @public */ +export type CatalogProcessorRefreshKeysResult = { + type: 'refresh'; + entity: Entity; + key: String; +}; + /** @public */ export type CatalogProcessorResult = | CatalogProcessorLocationResult | CatalogProcessorEntityResult | CatalogProcessorRelationResult - | CatalogProcessorErrorResult; + | CatalogProcessorErrorResult + | CatalogProcessorRefreshKeysResult; diff --git a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts index dd28cf1290..e4a5ab3d85 100644 --- a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts @@ -33,12 +33,15 @@ import { UpdateEntityCacheOptions, ListParentsOptions, ListParentsResult, + RefreshKeyOptions, + RefreshByKeyOptions, } from './types'; import { DeferredEntity } from '../processing/types'; import { ProcessingIntervalFunction } from '../processing/refresh'; import { rethrowError, timestampToDateTime } from './conversion'; import { initDatabaseMetrics } from './metrics'; import { + DbRefreshKeysRow, DbRefreshStateReferencesRow, DbRefreshStateRow, DbRelationsRow, @@ -515,7 +518,40 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { throw new NotFoundError(`Failed to schedule ${entityRef} for refresh`); } } + async refreshByRefreshKey( + txOpaque: Transaction, + options: RefreshByKeyOptions, + ) { + const tx = txOpaque as Knex.Transaction; + const { key } = options; + const rows = await tx('refresh_keys') + .where({ key }) + .select({ + entity_ref: 'refresh_keys.entity_ref', + }); + + await Promise.all(rows.map(r => this.refresh(tx, r.entity_ref))); + } + async addRefreshKeys( + txOpaque: Transaction, + options: RefreshKeyOptions, + ): Promise { + const tx = txOpaque as Knex.Transaction; + const { keys } = options; + + await Promise.all( + keys.map(k => { + return tx('refresh_keys') + .insert({ + entity_ref: stringifyEntityRef(k.entity), + key: k.key, + }) + .onConflict(['entity_ref', 'key']) + .ignore(); + }), + ); + } async transaction(fn: (tx: Transaction) => Promise): Promise { try { let result: T | undefined = undefined; diff --git a/plugins/catalog-backend/src/database/tables.ts b/plugins/catalog-backend/src/database/tables.ts index 40b4cb9a12..33c3ab0ef3 100644 --- a/plugins/catalog-backend/src/database/tables.ts +++ b/plugins/catalog-backend/src/database/tables.ts @@ -43,6 +43,11 @@ export type DbRefreshStateRow = { location_key?: string; }; +export type DbRefreshKeysRow = { + entity_ref: string; + key: string; +}; + export type DbRefreshStateReferencesRow = { source_key?: string; source_entity_ref?: string; diff --git a/plugins/catalog-backend/src/database/types.ts b/plugins/catalog-backend/src/database/types.ts index ce45a88938..bf3f765f4a 100644 --- a/plugins/catalog-backend/src/database/types.ts +++ b/plugins/catalog-backend/src/database/types.ts @@ -81,6 +81,14 @@ export type ReplaceUnprocessedEntitiesOptions = type: 'delta'; }; +export type RefreshKeyOptions = { + keys: { key: String; entity: Entity }[]; +}; + +export type RefreshByKeyOptions = { + key: string; +}; + export type RefreshOptions = { entityRef: string; }; @@ -149,6 +157,11 @@ export interface ProcessingDatabase { */ refresh(txOpaque: Transaction, options: RefreshOptions): Promise; + addRefreshKeys( + txOpaque: Transaction, + options: RefreshKeyOptions, + ): Promise; + /** * Lists all ancestors of a given entityRef. * diff --git a/plugins/catalog-backend/src/modules/core/FileReaderProcessor.ts b/plugins/catalog-backend/src/modules/core/FileReaderProcessor.ts index 66c6479fa7..768d763a24 100644 --- a/plugins/catalog-backend/src/modules/core/FileReaderProcessor.ts +++ b/plugins/catalog-backend/src/modules/core/FileReaderProcessor.ts @@ -61,6 +61,13 @@ export class FileReaderProcessor implements CatalogProcessor { }, })) { emit(parseResult); + if (parseResult.type === 'entity') { + emit({ + type: 'refresh', + key: path.normalize(fileMatch), + entity: parseResult.entity, + }); + } } } } else if (!optional) { diff --git a/plugins/catalog-backend/src/modules/core/PlaceholderProcessor.ts b/plugins/catalog-backend/src/modules/core/PlaceholderProcessor.ts index ddd5f951db..5089ee2637 100644 --- a/plugins/catalog-backend/src/modules/core/PlaceholderProcessor.ts +++ b/plugins/catalog-backend/src/modules/core/PlaceholderProcessor.ts @@ -19,7 +19,11 @@ import { Entity } from '@backstage/catalog-model'; import { JsonValue } from '@backstage/types'; import { ScmIntegrationRegistry } from '@backstage/integration'; import yaml from 'yaml'; -import { CatalogProcessor, LocationSpec } from '../../api'; +import { + CatalogProcessor, + CatalogProcessorEmit, + LocationSpec, +} from '../../api'; /** @public */ export type PlaceholderResolverRead = (url: string) => Promise; @@ -66,6 +70,7 @@ export class PlaceholderProcessor implements CatalogProcessor { async preProcessEntity( entity: Entity, location: LocationSpec, + emit: CatalogProcessorEmit, ): Promise { const process = async (data: any): Promise<[any, boolean]> => { if (!data || !(data instanceof Object)) { @@ -102,6 +107,7 @@ export class PlaceholderProcessor implements CatalogProcessor { const resolverKey = keys[0].substr(1); const resolverValue = data[keys[0]]; + emit({ type: 'refresh', key: resolverValue, entity }); const resolver = this.options.resolvers[resolverKey]; if (!resolver || typeof resolverValue !== 'string') { // If there was no such placeholder resolver or if the value was not a diff --git a/plugins/catalog-backend/src/modules/core/UrlReaderProcessor.ts b/plugins/catalog-backend/src/modules/core/UrlReaderProcessor.ts index 7f27bcd19d..a6c5bd74d8 100644 --- a/plugins/catalog-backend/src/modules/core/UrlReaderProcessor.ts +++ b/plugins/catalog-backend/src/modules/core/UrlReaderProcessor.ts @@ -83,6 +83,13 @@ export class UrlReaderProcessor implements CatalogProcessor { })) { parseResults.push(parseResult); emit(parseResult); + if (parseResult.type === 'entity') { + emit({ + type: 'refresh', + key: item.url, + entity: parseResult.entity, + }); + } } } diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts index b7ef5c9344..d3d679d2b7 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts @@ -123,6 +123,12 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { let hashBuilder = this.createHash().update(errorsString); if (result.ok) { + await this.processingDatabase.transaction(tx => + this.processingDatabase.addRefreshKeys(tx, { + keys: result.refreshKeys, + }), + ); + const { entityRefs: parents } = await this.processingDatabase.transaction(tx => this.processingDatabase.listParents(tx, { diff --git a/plugins/catalog-backend/src/processing/ProcessorOutputCollector.ts b/plugins/catalog-backend/src/processing/ProcessorOutputCollector.ts index db7a050dce..058d04ff2d 100644 --- a/plugins/catalog-backend/src/processing/ProcessorOutputCollector.ts +++ b/plugins/catalog-backend/src/processing/ProcessorOutputCollector.ts @@ -38,6 +38,10 @@ export class ProcessorOutputCollector { private readonly errors = new Array(); private readonly relations = new Array(); private readonly deferredEntities = new Array(); + private readonly refreshKeys = new Array<{ + key: String; + entity: Entity; + }>(); private done = false; constructor( @@ -54,6 +58,7 @@ export class ProcessorOutputCollector { return { errors: this.errors, relations: this.relations, + refreshKeys: this.refreshKeys, deferredEntities: this.deferredEntities, }; } @@ -116,6 +121,8 @@ export class ProcessorOutputCollector { this.relations.push(i.relation); } else if (i.type === 'error') { this.errors.push(i.error); + } else if (i.type === 'refresh') { + this.refreshKeys.push({ key: i.key, entity: i.entity }); } } } diff --git a/plugins/catalog-backend/src/processing/types.ts b/plugins/catalog-backend/src/processing/types.ts index b178522c14..b93d57863f 100644 --- a/plugins/catalog-backend/src/processing/types.ts +++ b/plugins/catalog-backend/src/processing/types.ts @@ -26,7 +26,6 @@ export type EntityProcessingRequest = { entity: Entity; state?: JsonObject; // Versions for multiple deployments etc }; - /** * The result of processing an entity. * @public @@ -38,6 +37,7 @@ export type EntityProcessingResult = completedEntity: Entity; deferredEntities: DeferredEntity[]; relations: EntityRelationSpec[]; + refreshKeys: { key: String; entity: Entity }[]; errors: Error[]; } | {