From 4e26ae5820ac89a94b2380d4e120bd15fca5e18c Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Wed, 31 Mar 2021 16:29:18 +0200 Subject: [PATCH] Initial catalog proccessing rework MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Patrik Oldsberg Co-authored-by: Fredrik Adelöw Co-authored-by: Tim Signed-off-by: Johan Haals --- .../20210302150147_refresh_state.js | 75 +++++++++ plugins/catalog-backend/package.json | 4 +- .../src/next/CatalogProcessingEngineImpl.ts | 125 ++++++++++++++ .../next/CatalogProcessingOrchestratorImpl.ts | 153 +++++++++++++++++ .../src/next/DatabaseLocationProvider.ts | 65 ++++++++ .../src/next/LocationServiceImpl.ts | 70 ++++++++ .../src/next/LocationStoreImpl.ts | 114 +++++++++++++ .../src/next/LocationToEntity.ts | 39 +++++ .../src/next/ProcessingStateManagerImpl.ts | 65 ++++++++ .../next/database/ProcessingDatabaseImpl.ts | 156 ++++++++++++++++++ .../src/next/database/types.ts | 68 ++++++++ plugins/catalog-backend/src/next/types.ts | 103 ++++++++++++ 12 files changed, 1036 insertions(+), 1 deletion(-) create mode 100644 plugins/catalog-backend/migrations/20210302150147_refresh_state.js create mode 100644 plugins/catalog-backend/src/next/CatalogProcessingEngineImpl.ts create mode 100644 plugins/catalog-backend/src/next/CatalogProcessingOrchestratorImpl.ts create mode 100644 plugins/catalog-backend/src/next/DatabaseLocationProvider.ts create mode 100644 plugins/catalog-backend/src/next/LocationServiceImpl.ts create mode 100644 plugins/catalog-backend/src/next/LocationStoreImpl.ts create mode 100644 plugins/catalog-backend/src/next/LocationToEntity.ts create mode 100644 plugins/catalog-backend/src/next/ProcessingStateManagerImpl.ts create mode 100644 plugins/catalog-backend/src/next/database/ProcessingDatabaseImpl.ts create mode 100644 plugins/catalog-backend/src/next/database/types.ts create mode 100644 plugins/catalog-backend/src/next/types.ts diff --git a/plugins/catalog-backend/migrations/20210302150147_refresh_state.js b/plugins/catalog-backend/migrations/20210302150147_refresh_state.js new file mode 100644 index 0000000000..ae8a02c785 --- /dev/null +++ b/plugins/catalog-backend/migrations/20210302150147_refresh_state.js @@ -0,0 +1,75 @@ +/* + * 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} knex + */ +exports.up = async function up(knex) { + await knex.schema.createTable('refresh_state', table => { + table.comment('Location Refresh states'); + table.text('id').primary().notNullable().comment('Primary id'); + table + .text('entity_ref') + .primary() + .notNullable() + .comment('A reference to the entity that the refresh state is tied to'); + table + .text('unprocessed_entity') + .notNullable() + .comment( + 'The unprocessed entity (in its source form, before being run through all of the processors) as JSON', + ); + table + .text('processed_entity') + .nullable() + .comment( + 'The processed entity (after running through all processors, but before being stitched together with state and relations) as JSON', + ); + table + .text('cache') + .nullable() + .comment( + 'Cache information tied to the refreshing of this entity, such as etag information or actual response caching', + ); + table + .text('errors') + .notNullable() + .comment('JSON array containing all errors related to entity'); + table + .dateTime('next_update_at') + .notNullable() + .comment('Timestamp of when entity should be updated'); + table + .dateTime('last_discovery_at') + .notNullable() + .comment('The last timestamp of which this entity was discovered'); + table.index('entity_ref', 'entity_ref_idx'); + table.index('next_update_at', 'next_update_at_idx'); + }); +}; + +/** + * @param {import('knex').Knex} knex + */ +exports.down = async function down(knex) { + await knex.schema.alterTable('refresh_state', table => { + table.dropIndex([], 'entity_ref_idx'); + table.dropIndex([], 'next_update_at_idx'); + }); + await knex.schema.dropTable('refresh_state'); +}; diff --git a/plugins/catalog-backend/package.json b/plugins/catalog-backend/package.json index 531caa9ea5..2661025000 100644 --- a/plugins/catalog-backend/package.json +++ b/plugins/catalog-backend/package.json @@ -33,6 +33,7 @@ "@backstage/backend-common": "^0.6.1", "@backstage/catalog-model": "^0.7.5", "@backstage/config": "^0.1.4", + "@backstage/core": "^0.7.3", "@backstage/errors": "^0.1.1", "@backstage/integration": "^0.5.1", "@backstage/plugin-search-backend-node": "^0.1.2", @@ -59,7 +60,8 @@ "winston": "^3.2.1", "yaml": "^1.9.2", "yn": "^4.0.0", - "yup": "^0.29.3" + "yup": "^0.29.3", + "zen-observable": "^0.8.15" }, "devDependencies": { "@backstage/cli": "^0.6.6", diff --git a/plugins/catalog-backend/src/next/CatalogProcessingEngineImpl.ts b/plugins/catalog-backend/src/next/CatalogProcessingEngineImpl.ts new file mode 100644 index 0000000000..e94cdf2f31 --- /dev/null +++ b/plugins/catalog-backend/src/next/CatalogProcessingEngineImpl.ts @@ -0,0 +1,125 @@ +/* + * Copyright 2021 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. + */ + +import { Subscription } from '@backstage/core'; +import { + CatalogProcessingEngine, + EntityProvider, + EntityMessage, + ProcessingStateManager, + CatalogProcessingOrchestrator, +} from './types'; + +import { JsonObject } from '@backstage/config'; +import { EntitiesCatalog } from '../catalog/types'; +import { Logger } from 'winston'; + +export class CatalogProcessingEngineImpl implements CatalogProcessingEngine { + private subscriptions: Subscription[] = []; + private running: boolean = false; + + constructor( + private readonly logger: Logger, + private readonly entityProviders: EntityProvider[], + private readonly stateManager: ProcessingStateManager, + private readonly orchestrator: CatalogProcessingOrchestrator, + private readonly entitiesCatalog: EntitiesCatalog, + ) {} + + async start() { + for (const provider of this.entityProviders) { + const subscription = provider + .entityChange$() + .subscribe({ next: m => this.onNext(m) }); + this.subscriptions.push(subscription); + } + + this.running = true; + + while (this.running) { + const { entity, state, eager } = await this.stateManager.pop(); + + const { + completeEntities, + deferredEntites, + errors, + } = await this.orchestrator.process({ + entity, + state, + eager, + }); + + for (const error of errors) { + this.logger.warn(error.message); + } + + for (const deferred of deferredEntites) { + this.stateManager.setResult({ + request: { entity: deferred, state: new Map() }, + nextRefresh: 'now()', + }); + } + + if (completeEntities.length) { + await this.entitiesCatalog.batchAddOrUpdateEntities( + completeEntities.map(e => ({ + entity: e, + relations: [], + })), + { + locationId: 'xyz', + dryRun: false, + outputEntities: true, + }, + ); + } + } + } + + async stop() { + this.running = false; + + for (const subscription of this.subscriptions) { + subscription.unsubscribe(); + } + } + private async onNext(message: EntityMessage) { + if ('all' in message) { + // TODO unhandled rejection + await Promise.all( + message.all.map(entity => + this.stateManager.setResult({ + request: { entity, state: new Map() }, + nextRefresh: 'now()', + }), + ), + ); + } + + if ('added' in message) { + await Promise.all( + message.added.map(entity => + this.stateManager.setResult({ + request: { entity, state: new Map() }, + nextRefresh: 'now()', + }), + ), + ); + + // TODO deletions of message.removed + } + } +} diff --git a/plugins/catalog-backend/src/next/CatalogProcessingOrchestratorImpl.ts b/plugins/catalog-backend/src/next/CatalogProcessingOrchestratorImpl.ts new file mode 100644 index 0000000000..f78ebfbef2 --- /dev/null +++ b/plugins/catalog-backend/src/next/CatalogProcessingOrchestratorImpl.ts @@ -0,0 +1,153 @@ +/* + * Copyright 2021 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. + */ + +import { + Entity, + LocationSpec, + stringifyLocationReference, +} from '@backstage/catalog-model'; +import { JsonObject } from '@backstage/config'; +import { + CatalogProcessor, + CatalogProcessorEmit, + CatalogProcessorParser, + CatalogProcessorResult, +} from '../ingestion/processors'; +import { CatalogProcessingOrchestrator, EntityProcessingError } from './types'; +import { Logger } from 'winston'; +import * as result from '../ingestion/processors/results'; +import { locationToEntity } from './LocationToEntity'; + +export class CatalogProcessingOrchestratorImpl + implements CatalogProcessingOrchestrator { + constructor( + private readonly options: { + processors: CatalogProcessor[]; + logger: Logger; + parser: CatalogProcessorParser; + }, + ) {} + + async process(request: { + entity: Entity; + eager?: boolean | undefined; + state: Map; + }): Promise<{ + state: Map; + completeEntities: Entity[]; + deferredEntites: Entity[]; + errors: EntityProcessingError[]; + }> { + const { entity, eager, state } = request; + const completedEntities: Entity[] = []; + const deferredEntites: Entity[] = []; + const errors: EntityProcessingError[] = []; + + if (eager) { + const stack = [entity]; + const emit = (i: CatalogProcessorResult) => stack.push(i); + while (stack.length) { + const item = stack.pop(); + stack.push(); + } + } else { + const emit = (i: CatalogProcessorResult) => { + if (i.type === 'entity') { + deferredEntites.push(i.entity); + } + if (i.type === 'location') { + deferredEntites.push( + locationToEntity(i.location.type, i.location.type), + ); + } + }; + if (entity.spec) { + if (entity.spec.kind === 'Location') { + this.handleLocation( + { + type: (entity.spec.type as unknown) as string, + target: (entity.spec.target as unknown) as string, + }, + emit, + ); + } + } + } + + return { + deferredEntites, + completeEntities: completedEntities, + errors, + state, + }; + } + + private async handleLocation( + location: LocationSpec, + emit: CatalogProcessorEmit, + ) { + const { processors, logger } = this.options; + + const validatedEmit: CatalogProcessorEmit = emitResult => { + if (emitResult.type === 'relation') { + throw new Error('readLocation may not emit entity relations'); + } + if ( + emitResult.type === 'location' && + emitResult.location.type === location.type && + emitResult.location.target === location.target + ) { + // Ignore self-referential locations silently (this can happen for + // example if you use a glob target like "**/*.yaml" in a Location + // entity) + return; + } + emit(emitResult); + }; + + for (const processor of processors) { + if (processor.readLocation) { + try { + if ( + await processor.readLocation( + location, + // TODO: figure out how this will work in the new processing system - locations from PRs might be optional + location.presence !== 'required', + validatedEmit, + this.options.parser, + ) + ) { + return; + } + } catch (e) { + const message = `Processor ${ + processor.constructor.name + } threw an error while reading location ${stringifyLocationReference( + location, + )}, ${e}`; + emit(result.generalError(location, message)); + logger.warn(message); + } + } + } + + const message = `No processor was able to read location ${stringifyLocationReference( + location, + )}`; + emit(result.inputError(location, message)); + logger.warn(message); + } +} diff --git a/plugins/catalog-backend/src/next/DatabaseLocationProvider.ts b/plugins/catalog-backend/src/next/DatabaseLocationProvider.ts new file mode 100644 index 0000000000..b56dfa1158 --- /dev/null +++ b/plugins/catalog-backend/src/next/DatabaseLocationProvider.ts @@ -0,0 +1,65 @@ +/* + * Copyright 2021 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. + */ + +import { Observable } from '@backstage/core'; +import { EntityProvider, LocationStore, EntityMessage } from './types'; +import ObservableImpl from 'zen-observable'; +import { locationToEntity, locationToEntityName } from './LocationToEntity'; + +export class DatabaseLocationProvider implements EntityProvider { + private subscribers = new Set< + ZenObservable.SubscriptionObserver + >(); + + constructor(private readonly store: LocationStore) { + store.location$().subscribe({ + next: locations => { + if ('all' in locations) { + this.notify({ + all: locations.all.map(l => locationToEntity(l.type, l.target)), + }); + } else { + this.notify({ + added: locations.added.map(l => locationToEntity(l.type, l.target)), + removed: locations.removed.map(l => + locationToEntityName(l.type, l.target), + ), + }); + } + }, + }); + } + + private notify(message: EntityMessage) { + for (const subscriber of this.subscribers) { + subscriber.next(message); + } + } + + entityChange$(): Observable { + return new ObservableImpl(subscriber => { + this.store.listLocations().then(locations => { + subscriber.next({ + all: locations.map(l => locationToEntity(l.type, l.target)), + }); + this.subscribers.add(subscriber); + }); + return () => { + this.subscribers.delete(subscriber); + }; + }); + } +} diff --git a/plugins/catalog-backend/src/next/LocationServiceImpl.ts b/plugins/catalog-backend/src/next/LocationServiceImpl.ts new file mode 100644 index 0000000000..29b41e0a09 --- /dev/null +++ b/plugins/catalog-backend/src/next/LocationServiceImpl.ts @@ -0,0 +1,70 @@ +/* + * Copyright 2021 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. + */ +import { LocationSpec, Location, Entity } from '@backstage/catalog-model'; +import { + LocationService, + LocationStore, + CatalogProcessingOrchestrator, +} from './types'; + +export class LocationServiceImpl implements LocationService { + constructor( + private readonly store: LocationStore, + private readonly orchestrator: CatalogProcessingOrchestrator, + ) {} + + async createLocation( + spec: LocationSpec, + dryRun: boolean, + ): Promise<{ location: Location; entities: Entity[] }> { + if (dryRun) { + const entity = { + apiVersion: 'backstage.io/v1alpha1', + kind: 'Location', + metadata: { + name: `${spec.type}:${spec.target}`, + namespace: 'default', + }, + spec: { + location: { type: spec.type, target: spec.target }, + }, + }; + const processed = await this.orchestrator.process({ + entity, + eager: true, + state: new Map(), + }); + + return { + location: { ...spec, id: `${spec.type}:${spec.target}` }, + entities: processed.completeEntities, + }; + } + + const location = await this.store.createLocation(spec); + return { location, entities: [] }; + } + + listLocations(): Promise { + return this.store.listLocations(); + } + getLocation(id: string): Promise { + return this.store.getLocation(id); + } + deleteLocation(id: string): Promise { + return this.store.deleteLocation(id); + } +} diff --git a/plugins/catalog-backend/src/next/LocationStoreImpl.ts b/plugins/catalog-backend/src/next/LocationStoreImpl.ts new file mode 100644 index 0000000000..89247056d6 --- /dev/null +++ b/plugins/catalog-backend/src/next/LocationStoreImpl.ts @@ -0,0 +1,114 @@ +/* + * Copyright 2021 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. + */ + +import { LocationSpec, Location } from '@backstage/catalog-model'; +import { CommonDatabase } from '../database'; +import { LocationStore } from './types'; +import { v4 as uuidv4 } from 'uuid'; +import { ConflictError } from '@backstage/errors'; +import { Observable } from '@backstage/core'; +import ObservableImpl from 'zen-observable'; + +export type LocationMessage = + | { all: Location[] } + | { added: Location[]; removed: Location[] }; + +export class LocationStoreImpl implements LocationStore { + private subscribers = new Set< + ZenObservable.SubscriptionObserver + >(); + + constructor(private readonly db: CommonDatabase) {} + + createLocation(spec: LocationSpec): Promise { + return this.db.transaction(async tx => { + // TODO: id should really be type and target combined and not a uuid. + + // Attempt to find a previous location matching the spec + const previousLocations = await this.listLocations(); + const previousLocation = previousLocations.some( + l => spec.type === l.type && spec.target === l.target, + ); + + if (previousLocation) { + throw new ConflictError( + `Location ${spec.type}:${spec.target} already exists`, + ); + } + + const location = await this.db.addLocation(tx, { + id: uuidv4(), + type: spec.type, + target: spec.target, + }); + + this.notifyAddition(location); + + return location; + }); + } + + async listLocations(): Promise { + const dbLocations = await this.db.locations(); + return dbLocations.map(item => ({ + id: item.id, + target: item.target, + type: item.type, + })); + } + + getLocation(id: string): Promise { + return this.db.location(id); + } + + deleteLocation(id: string): Promise { + return this.db.transaction(async tx => { + const location = await this.db.location(id); + if (!location) { + throw new ConflictError(`No location found with with id: ${id}`); + } + await this.db.removeLocation(tx, id); + this.notifyDeletion(location); + }); + } + + private notifyAddition(location: Location) { + for (const subscriber of this.subscribers) { + subscriber.next({ + added: [location], + removed: [], + }); + } + } + + private notifyDeletion(location: Location) { + for (const subscriber of this.subscribers) { + subscriber.next({ + added: [], + removed: [location], + }); + } + } + + location$(): Observable { + return new ObservableImpl(subscriber => { + this.subscribers.add(subscriber); + return () => { + this.subscribers.delete(subscriber); + }; + }); + } +} diff --git a/plugins/catalog-backend/src/next/LocationToEntity.ts b/plugins/catalog-backend/src/next/LocationToEntity.ts new file mode 100644 index 0000000000..28695e14af --- /dev/null +++ b/plugins/catalog-backend/src/next/LocationToEntity.ts @@ -0,0 +1,39 @@ +/* + * Copyright 2021 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. + */ + +import { Entity, EntityName } from '@backstage/catalog-model'; + +export function locationToEntity(type: string, target: string): Entity { + return { + apiVersion: 'backstage.io/v1alpha1', + kind: 'Location', + metadata: { + name: `${type}:${target}`, + namespace: 'default', + }, + spec: { + location: { type, target }, + }, + }; +} + +export function locationToEntityName(type: string, target: string): EntityName { + return { + kind: 'Location', + namespace: 'default', + name: `${type}:${target}`, + }; +} diff --git a/plugins/catalog-backend/src/next/ProcessingStateManagerImpl.ts b/plugins/catalog-backend/src/next/ProcessingStateManagerImpl.ts new file mode 100644 index 0000000000..d8736cfa77 --- /dev/null +++ b/plugins/catalog-backend/src/next/ProcessingStateManagerImpl.ts @@ -0,0 +1,65 @@ +/* + * Copyright 2021 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. + */ + +import { Entity } from '@backstage/catalog-model'; +import { JsonObject } from '@backstage/config'; +import { DbRefreshStateRow, ProcessingDatabase } from './database/types'; +import { ProcessingResult, ProcessingStateManager } from './types'; + +export class ProcessingStateManagerImpl implements ProcessingStateManager { + constructor(private readonly db: ProcessingDatabase) {} + + async setResult(result: ProcessingResult) { + await this.db.transaction(async tx => { + this.db.addEntityRefreshState(tx, [ + { + entity: result.request.entity, + nextRefresh: result.nextRefresh, + }, + ]); + }); + } + + async pop(): Promise<{ + entity: Entity; + eager?: boolean | undefined; + state: Map; + }> { + const entities = await new Promise(resolve => + this.popFromQueue(resolve), + ); + const result = entities[0]; + return { + entity: JSON.parse(result.unproccessed_entity) as Entity, + state: new Map(JSON.parse(result.cache)), + }; + } + + async popFromQueue(resolve: (rows: DbRefreshStateRow[]) => void) { + const entities = await this.db.transaction(async tx => { + return this.db.getProcessableEntities(tx, { + processBatchSize: 1, + }); + }); + + if (!entities.length) { + setTimeout(() => this.popFromQueue(resolve), 1000); + return; + } + + resolve(entities); + } +} diff --git a/plugins/catalog-backend/src/next/database/ProcessingDatabaseImpl.ts b/plugins/catalog-backend/src/next/database/ProcessingDatabaseImpl.ts new file mode 100644 index 0000000000..af0d331199 --- /dev/null +++ b/plugins/catalog-backend/src/next/database/ProcessingDatabaseImpl.ts @@ -0,0 +1,156 @@ +/* + * Copyright 2021 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. + */ + +import { ConflictError, NotFoundError } from '@backstage/errors'; +import { Knex } from 'knex'; +import { Transaction } from '../../database'; +import { + ProcessingDatabase, + AddUnprocessedEntitiesOptions, + UpdateProcessedEntityOptions, + GetProcessedEntitiesResult, +} from './types'; +import type { Logger } from 'winston'; +import { Entity } from '@backstage/catalog-model'; +import { v4 as uuid } from 'uuid'; + +export type DbRefreshStateRequest = { + entity: Entity; + nextRefresh: string; // TODO dateTime/ Date? +}; + +export type DbRefreshStateRow = { + id: string; + entity_ref: string; + unprocessed_entity: string; + processed_entity: string; + cache: string; + next_update_at: string; + last_discovery_at: string; // remove? + errors: string; +}; + +class ProcessingDatabaseImpl implements ProcessingDatabase { + constructor( + private readonly database: Knex, + private readonly logger: Logger, + ) {} + + async updateProcessedEntity( + txOpaque: Transaction, + options: UpdateProcessedEntityOptions, + ): Promise { + const tx = txOpaque as Knex.Transaction; + const { id, processedEntity, cache, errors } = options; + const result = await tx('refresh_state') + .update({ + processed_entity: processedEntity, + cache, + errors, + }) + .where('id', id); + if (result === 0) { + throw new NotFoundError(`Processing state not found for ${id}`); + } + + // Update refs table + // Update fragments + } + + async addUnprocessedEntities( + txOpaque: Transaction, + options: AddUnprocessedEntitiesOptions, + ): Promise { + const tx = txOpaque as Knex.Transaction; + + for (const entity of options.unprocessedEntities) { + await tx('refresh_state') + .insert({ + id: uuid(), + entity_ref: stringifyEntityRef(entity), + unprocessed_entity: JSON.stringify(entity), + next_update_at: tx.fn.now(), + last_discovery_at: tx.fn.now(), + }) + .onConflict('entity_ref') + .merge(['unprocessed_entity', 'last_discovery_at']); + } + } + + async getProcessableEntities( + txOpaque: Transaction, + request: { processBatchSize: number }, + ): Promise { + const tx = txOpaque as Knex.Transaction; + + const items = await tx('refresh_state') + .select() + .where('next_update_at', '<=', tx.fn.now()) + .limit(request.processBatchSize) + .orderBy('next_update_at', 'desc'); + + await tx('refresh_state') + .whereIn( + 'entity_ref', + items.map(i => i.entity_ref), + ) + .update({ + next_update_at: + tx.client.config.client === 'sqlite3' + ? tx.raw(`datetime('now', ?)`, [`10 seconds`]) // TODO: test this in sqlite3 + : tx.raw(`now() + interval '30 seconds'`), + }); + + return items; + } + + async transaction(fn: (tx: Transaction) => Promise): Promise { + try { + let result: T | undefined = undefined; + + await this.database.transaction( + async tx => { + // We can't return here, as knex swallows the return type in case the transaction is rolled back: + // https://github.com/knex/knex/blob/e37aeaa31c8ef9c1b07d2e4d3ec6607e557d800d/lib/transaction.js#L136 + result = await fn(tx); + }, + { + // If we explicitly trigger a rollback, don't fail. + doNotRejectOnRollback: true, + }, + ); + + return result!; + } catch (e) { + this.logger.debug(`Error during transaction, ${e}`); + + if ( + /SQLITE_CONSTRAINT: UNIQUE/.test(e.message) || + /unique constraint/.test(e.message) + ) { + throw new ConflictError(`Rejected due to a conflicting entity`, e); + } + + throw e; + } + } +} + +function stringifyEntityRef( + entity: Entity, +): Knex.MaybeRawColumn | undefined { + throw new Error('Function not implemented.'); +} diff --git a/plugins/catalog-backend/src/next/database/types.ts b/plugins/catalog-backend/src/next/database/types.ts new file mode 100644 index 0000000000..310b6b4b7e --- /dev/null +++ b/plugins/catalog-backend/src/next/database/types.ts @@ -0,0 +1,68 @@ +/* + * Copyright 2021 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. + */ + +import { Entity } from '@backstage/catalog-model'; +import { JsonObject } from '@backstage/config'; +import { Transaction } from '../../database/types'; + +export type AddUnprocessedEntitiesOptions = { + unprocessedEntities: Entity[]; +}; + +export type AddUnprocessedEntitiesResult = {}; + +export type UpdateProcessedEntityOptions = { + id: string; + processedEntity?: string; + cache?: string; + errors?: string; +}; + +export type RefreshStateItem = { + id: string; + entityRef: string; + unprocessedEntity: string; + processedEntity: string; + nextUpdateAt: string; + lastDiscoveryAt: string; // remove? + cache: JsonObject; + errors: string; +}; + +export type GetProcessedEntitiesResult = { + items: RefreshStateItem; +}; + +export interface ProcessingDatabase { + transaction(fn: (tx: Transaction) => Promise): Promise; + addUnprocessedEntities( + tx: Transaction, + options: AddUnprocessedEntitiesOptions, + ): Promise; + + getProcessableEntities( + txOpaque: Transaction, + request: { processBatchSize: number }, + ): Promise; + + /** + * Updates the + */ + updateProcessedEntity( + txOpaque: Transaction, + options: UpdateProcessedEntityOptions, + ): Promise; +} diff --git a/plugins/catalog-backend/src/next/types.ts b/plugins/catalog-backend/src/next/types.ts new file mode 100644 index 0000000000..c7f1e370e0 --- /dev/null +++ b/plugins/catalog-backend/src/next/types.ts @@ -0,0 +1,103 @@ +/* + * Copyright 2021 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. + */ +import { + Entity, + EntityName, + LocationSpec, + Location, +} from '@backstage/catalog-model'; +import { JsonObject } from '@backstage/config'; +import { Observable } from '@backstage/core'; // << nooo + +export interface LocationEntity { + apiVersion: 'backstage.io/v1alpha1'; + kind: 'Location'; + metadata: { + name: string; // type:target + namespace: 'default'; + }; + spec: { + location: { type: string; target: string }; + }; +} + +export interface LocationService { + createLocation( + spec: LocationSpec, + dryRun: boolean, + ): Promise<{ location: Location; entities: Entity[] }>; + listLocations(): Promise; + getLocation(id: string): Promise; + deleteLocation(id: string): Promise; +} + +export type EntityMessage = + | { all: Entity[] } + | { added: Entity[]; removed: EntityName[] }; + +export interface LocationStore { + // extends EntityProvider + createLocation(spec: LocationSpec): Promise; + listLocations(): Promise; + getLocation(id: string): Promise; + deleteLocation(id: string): Promise; + + location$(): Observable< + { all: Location[] } | { added: Location[]; removed: Location[] } + >; +} + +export interface CatalogProcessingEngine { + start(): Promise; + stop(): Promise; +} + +export interface EntityProvider { + entityChange$(): Observable; +} + +// interface CatalogProcessor {} +type EntityProcessingRequest = { + entity: Entity; + eager?: boolean; + state: Map; // Versions for multiple deployments etc +}; + +export type EntityProcessingError = { + // some error stuff here + message: String; +}; + +type EntityProcessingResult = { + state: Map; + completeEntities: Entity[]; + deferredEntites: Entity[]; + errors: EntityProcessingError[]; +}; + +export interface CatalogProcessingOrchestrator { + process(request: EntityProcessingRequest): Promise; +} + +export type ProcessingResult = { + nextRefresh: string; + request: EntityProcessingRequest; +}; + +export interface ProcessingStateManager { + setResult(result: ProcessingResult): Promise; + pop(): Promise; +}