Merge pull request #5372 from backstage/mob/catalog-refactor
Initial iteration of the next software catalog
This commit is contained in:
Patrik Oldsberg
@@ -18,6 +18,7 @@ import { useHotCleanup } from '@backstage/backend-common';
|
||||
import {
|
||||
CatalogBuilder,
|
||||
createRouter,
|
||||
NextCatalogBuilder,
|
||||
runPeriodically,
|
||||
} from '@backstage/plugin-catalog-backend';
|
||||
import { Router } from 'express';
|
||||
@@ -26,6 +27,32 @@ import { PluginEnvironment } from '../types';
|
||||
export default async function createPlugin(
|
||||
env: PluginEnvironment,
|
||||
): Promise<Router> {
|
||||
/*
|
||||
* ** WARNING **
|
||||
* DO NOT enable the experimental catalog, it will brick your database migrations.
|
||||
* This is solely for internal backstage development.
|
||||
*/
|
||||
if (process.env.EXPERIMENTAL_CATALOG === '1') {
|
||||
const builder = new NextCatalogBuilder(env);
|
||||
const {
|
||||
entitiesCatalog,
|
||||
locationsCatalog,
|
||||
locationAnalyzer,
|
||||
processingEngine,
|
||||
} = await builder.build();
|
||||
|
||||
// TODO(jhaals): run and manage in background.
|
||||
processingEngine.start();
|
||||
|
||||
return await createRouter({
|
||||
entitiesCatalog,
|
||||
locationsCatalog,
|
||||
locationAnalyzer,
|
||||
logger: env.logger,
|
||||
config: env.config,
|
||||
});
|
||||
}
|
||||
|
||||
const builder = new CatalogBuilder(env);
|
||||
const {
|
||||
entitiesCatalog,
|
||||
|
||||
@@ -0,0 +1,182 @@
|
||||
/*
|
||||
* 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. Every individual location (that was ever directly or indirectly discovered) and entity has an entry in this table. It therefore represents the entire live set of things that the refresh loop considers.',
|
||||
);
|
||||
table
|
||||
.text('entity_id')
|
||||
.primary()
|
||||
.notNullable()
|
||||
.comment(
|
||||
'Primary ID, which will also be used as the uid of the resulting entity',
|
||||
);
|
||||
table
|
||||
.text('entity_ref')
|
||||
.unique()
|
||||
.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', 'refresh_state_entity_ref_idx');
|
||||
table.index('next_update_at', 'refresh_state_next_update_at_idx');
|
||||
});
|
||||
|
||||
await knex.schema.createTable('final_entities', table => {
|
||||
table.comment(
|
||||
'This table contains the final entity result after processing and stitching',
|
||||
);
|
||||
table
|
||||
.text('entity_id')
|
||||
.primary()
|
||||
.notNullable()
|
||||
.references('entity_id')
|
||||
.inTable('refresh_state')
|
||||
.onDelete('CASCADE')
|
||||
.comment(
|
||||
'Entity ID which correspond to the ID in the refresh_state table',
|
||||
);
|
||||
table.text('etag').notNullable().comment('Etag to be used for caching');
|
||||
table.text('finalized_entity').notNullable().comment('The final entity');
|
||||
table.index('entity_id', 'final_entities_entity_id_idx');
|
||||
});
|
||||
|
||||
await knex.schema.createTable('refresh_state_references', table => {
|
||||
table.comment(
|
||||
'Holds edges between refresh state rows. Every time when an entity is processed and emits another entity, an edge will be stored to represent that fact. This is used to detect orphans and ultimately deletions.',
|
||||
);
|
||||
table
|
||||
.text('source_special_key')
|
||||
.nullable()
|
||||
.comment(
|
||||
'When the reference source is not an entity, this is an opaque identifier for that source.',
|
||||
);
|
||||
table
|
||||
.text('source_entity_id')
|
||||
.nullable()
|
||||
.references('entity_id')
|
||||
.inTable('refresh_state')
|
||||
.onDelete('CASCADE')
|
||||
.comment(
|
||||
'When the reference source is an entity, this is the ID of the source entity.',
|
||||
);
|
||||
table
|
||||
.text('target_entity_id')
|
||||
.notNullable()
|
||||
.references('entity_id')
|
||||
.inTable('refresh_state')
|
||||
.onDelete('CASCADE')
|
||||
.comment('The ID of the target entity.');
|
||||
table.index(
|
||||
'source_special_key',
|
||||
'refresh_state_references_source_special_key_idx',
|
||||
);
|
||||
table.index(
|
||||
'source_entity_id',
|
||||
'refresh_state_references_source_entity_id_idx',
|
||||
);
|
||||
table.index(
|
||||
'target_entity_id',
|
||||
'refresh_state_references_target_entity_id_idx',
|
||||
);
|
||||
});
|
||||
|
||||
await knex.schema.createTable('relations', table => {
|
||||
table.comment('All relations between entities in the catalog');
|
||||
table
|
||||
.text('originating_entity_id')
|
||||
.references('entity_id')
|
||||
.inTable('refresh_state')
|
||||
.onDelete('CASCADE')
|
||||
.notNullable()
|
||||
.comment('The entity that provided the relation');
|
||||
table
|
||||
.text('source_entity_ref')
|
||||
.notNullable()
|
||||
.comment('The entity reference of the source entity of the relation');
|
||||
table
|
||||
.text('type')
|
||||
.notNullable()
|
||||
.comment('The type of the relation between the entities');
|
||||
table
|
||||
.text('target_entity_ref')
|
||||
.notNullable()
|
||||
.comment('The entity reference of the target entity of the relation');
|
||||
table.index('source_entity_ref', 'relations_source_entity_ref_idx');
|
||||
table.index('originating_entity_id', 'relations_source_entity_id_idx');
|
||||
});
|
||||
};
|
||||
|
||||
/**
|
||||
* @param {import('knex').Knex} knex
|
||||
*/
|
||||
exports.down = async function down(knex) {
|
||||
await knex.schema.alterTable('refresh_state_references', table => {
|
||||
table.dropIndex([], 'refresh_state_references_source_special_key_idx');
|
||||
table.dropIndex([], 'refresh_state_references_source_entity_id_idx');
|
||||
table.dropIndex([], 'refresh_state_references_target_entity_id_idx');
|
||||
});
|
||||
await knex.schema.alterTable('refresh_state', table => {
|
||||
table.dropIndex([], 'refresh_state_entity_ref_idx');
|
||||
table.dropIndex([], 'refresh_state_next_update_at_idx');
|
||||
});
|
||||
await knex.schema.alterTable('final_entities', table => {
|
||||
table.dropIndex([], 'final_entities_entity_id_idx');
|
||||
});
|
||||
await knex.schema.alterTable('relations', table => {
|
||||
table.index('source_entity_ref', 'relations_source_entity_ref_idx');
|
||||
table.index('originating_entity_id', 'relations_source_entity_id_idx');
|
||||
});
|
||||
await knex.schema.dropTable('final_entities');
|
||||
await knex.schema.dropTable('relations');
|
||||
await knex.schema.dropTable('references');
|
||||
await knex.schema.dropTable('refresh_state');
|
||||
};
|
||||
@@ -33,6 +33,7 @@
|
||||
"@backstage/backend-common": "^0.6.1",
|
||||
"@backstage/catalog-model": "^0.7.7",
|
||||
"@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",
|
||||
@@ -46,6 +47,7 @@
|
||||
"cross-fetch": "^3.0.6",
|
||||
"express": "^4.17.1",
|
||||
"express-promise-router": "^4.1.0",
|
||||
"fast-json-stable-stringify": "^2.1.0",
|
||||
"fs-extra": "^9.0.0",
|
||||
"git-url-parse": "^11.4.4",
|
||||
"glob": "^7.1.6",
|
||||
@@ -59,7 +61,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.8",
|
||||
@@ -77,6 +80,7 @@
|
||||
"files": [
|
||||
"dist",
|
||||
"migrations/**/*.{js,d.ts}",
|
||||
"migrationsv2/**/*.{js,d.ts}",
|
||||
"config.d.ts"
|
||||
],
|
||||
"configSchema": "config.d.ts"
|
||||
|
||||
@@ -20,3 +20,4 @@ export * from './ingestion';
|
||||
export * from './search';
|
||||
export * from './service';
|
||||
export * from './util';
|
||||
export * from './next';
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
/*
|
||||
* 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 { ENTITY_DEFAULT_NAMESPACE } from '@backstage/catalog-model';
|
||||
import { EntityProvider, LocationStore, EntityMessage } from './types';
|
||||
import ObservableImpl from 'zen-observable';
|
||||
import {
|
||||
locationSpecToLocationEntity,
|
||||
locationSpecToMetadataName,
|
||||
} from './util';
|
||||
|
||||
export class DatabaseLocationProvider implements EntityProvider {
|
||||
private subscribers = new Set<
|
||||
ZenObservable.SubscriptionObserver<EntityMessage>
|
||||
>();
|
||||
|
||||
constructor(private readonly store: LocationStore) {
|
||||
store.location$().subscribe({
|
||||
next: locations => {
|
||||
if ('all' in locations) {
|
||||
this.notify({
|
||||
all: locations.all.map(l => locationSpecToLocationEntity(l)),
|
||||
});
|
||||
} else {
|
||||
this.notify({
|
||||
added: locations.added.map(l => locationSpecToLocationEntity(l)),
|
||||
removed: locations.removed.map(l => ({
|
||||
kind: 'Location',
|
||||
namespace: ENTITY_DEFAULT_NAMESPACE,
|
||||
name: locationSpecToMetadataName(l),
|
||||
})),
|
||||
});
|
||||
}
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
private notify(message: EntityMessage) {
|
||||
for (const subscriber of this.subscribers) {
|
||||
subscriber.next(message);
|
||||
}
|
||||
}
|
||||
|
||||
entityChange$(): Observable<EntityMessage> {
|
||||
return new ObservableImpl(subscriber => {
|
||||
this.store.listLocations().then(locations => {
|
||||
subscriber.next({
|
||||
all: locations.map(l => locationSpecToLocationEntity(l)),
|
||||
});
|
||||
this.subscribers.add(subscriber);
|
||||
});
|
||||
return () => {
|
||||
this.subscribers.delete(subscriber);
|
||||
};
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,121 @@
|
||||
/*
|
||||
* 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 { Logger } from 'winston';
|
||||
import { stringifyEntityRef } from '@backstage/catalog-model';
|
||||
import { Stitcher } from './Stitcher';
|
||||
|
||||
export class DefaultCatalogProcessingEngine 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 stitcher: Stitcher,
|
||||
) {}
|
||||
|
||||
async start() {
|
||||
for (const provider of this.entityProviders) {
|
||||
const id = 'databaseProvider';
|
||||
const subscription = provider
|
||||
.entityChange$()
|
||||
.subscribe({ next: m => this.onNext(id, m) });
|
||||
this.subscriptions.push(subscription);
|
||||
}
|
||||
|
||||
this.running = true;
|
||||
|
||||
while (this.running) {
|
||||
const {
|
||||
id,
|
||||
entity,
|
||||
state: intialState,
|
||||
} = await this.stateManager.getNextProcessingItem();
|
||||
|
||||
const result = await this.orchestrator.process({
|
||||
entity,
|
||||
state: intialState,
|
||||
});
|
||||
|
||||
for (const error of result.errors) {
|
||||
this.logger.warn(error.message);
|
||||
}
|
||||
|
||||
if (!result.ok) {
|
||||
return;
|
||||
}
|
||||
|
||||
result.completedEntity.metadata.uid = id;
|
||||
await this.stateManager.setProcessingItemResult({
|
||||
id,
|
||||
entity: result.completedEntity,
|
||||
state: result.state,
|
||||
errors: result.errors,
|
||||
relations: result.relations,
|
||||
deferredEntities: result.deferredEntites,
|
||||
});
|
||||
|
||||
const setOfThingsToStitch = new Set<string>([
|
||||
stringifyEntityRef(result.completedEntity),
|
||||
...result.relations.map(relation =>
|
||||
stringifyEntityRef(relation.source),
|
||||
),
|
||||
]);
|
||||
await this.stitcher.stitch(setOfThingsToStitch);
|
||||
}
|
||||
}
|
||||
|
||||
async stop() {
|
||||
this.running = false;
|
||||
|
||||
for (const subscription of this.subscriptions) {
|
||||
subscription.unsubscribe();
|
||||
}
|
||||
}
|
||||
|
||||
private async onNext(id: string, message: EntityMessage) {
|
||||
if ('all' in message) {
|
||||
// TODO unhandled rejection
|
||||
await this.stateManager.addProcessingItems({
|
||||
id,
|
||||
type: 'provider',
|
||||
entities: message.all,
|
||||
});
|
||||
}
|
||||
|
||||
if ('added' in message) {
|
||||
await this.stateManager.addProcessingItems({
|
||||
id,
|
||||
type: 'provider',
|
||||
entities: message.added,
|
||||
});
|
||||
|
||||
// TODO deletions of message.removed
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,318 @@
|
||||
/*
|
||||
* 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,
|
||||
EntityRelationSpec,
|
||||
stringifyEntityRef,
|
||||
LOCATION_ANNOTATION,
|
||||
LocationSpec,
|
||||
LocationEntity,
|
||||
EntityPolicy,
|
||||
ORIGIN_LOCATION_ANNOTATION,
|
||||
stringifyLocationReference,
|
||||
parseLocationReference,
|
||||
} from '@backstage/catalog-model';
|
||||
import {
|
||||
CatalogProcessor,
|
||||
CatalogProcessorParser,
|
||||
CatalogProcessorResult,
|
||||
} from '../ingestion/processors';
|
||||
import {
|
||||
CatalogProcessingOrchestrator,
|
||||
EntityProcessingRequest,
|
||||
EntityProcessingResult,
|
||||
} from './types';
|
||||
import { Logger } from 'winston';
|
||||
import { InputError } from '@backstage/errors';
|
||||
import { locationSpecToLocationEntity } from './util';
|
||||
import path from 'path';
|
||||
import * as results from '../ingestion/processors/results';
|
||||
import { ScmIntegrationRegistry } from '@backstage/integration';
|
||||
|
||||
function isLocationEntity(entity: Entity): entity is LocationEntity {
|
||||
return entity.kind === 'Location';
|
||||
}
|
||||
|
||||
function getEntityOriginLocationRef(entity: Entity): string {
|
||||
const ref = entity.metadata.annotations?.[ORIGIN_LOCATION_ANNOTATION];
|
||||
if (!ref) {
|
||||
const entityRef = stringifyEntityRef(entity);
|
||||
throw new InputError(
|
||||
`Entity '${entityRef}' does not have an origin location`,
|
||||
);
|
||||
}
|
||||
return ref;
|
||||
}
|
||||
|
||||
function toAbsoluteUrl(
|
||||
integrations: ScmIntegrationRegistry,
|
||||
base: LocationSpec,
|
||||
type: string,
|
||||
target: string,
|
||||
): string {
|
||||
if (base.type !== type) {
|
||||
return target;
|
||||
}
|
||||
try {
|
||||
if (type === 'file') {
|
||||
if (target.startsWith('.')) {
|
||||
return path.join(path.dirname(base.target), target);
|
||||
}
|
||||
return target;
|
||||
} else if (type === 'url') {
|
||||
return integrations.resolveUrl({ url: target, base: base.target });
|
||||
}
|
||||
return target;
|
||||
} catch (e) {
|
||||
return target;
|
||||
}
|
||||
}
|
||||
|
||||
export class DefaultCatalogProcessingOrchestrator
|
||||
implements CatalogProcessingOrchestrator {
|
||||
constructor(
|
||||
private readonly options: {
|
||||
processors: CatalogProcessor[];
|
||||
integrations: ScmIntegrationRegistry;
|
||||
logger: Logger;
|
||||
parser: CatalogProcessorParser;
|
||||
policy: EntityPolicy;
|
||||
},
|
||||
) {}
|
||||
|
||||
async process(
|
||||
request: EntityProcessingRequest,
|
||||
): Promise<EntityProcessingResult> {
|
||||
const { entity } = request;
|
||||
|
||||
const result = await this.processSingleEntity(entity);
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
private async processSingleEntity(
|
||||
unprocessedEntity: Entity,
|
||||
): Promise<EntityProcessingResult> {
|
||||
// TODO: validate that this doesn't change during processing
|
||||
const entityRef = stringifyEntityRef(unprocessedEntity);
|
||||
// TODO: which one do we actually use here? source-location? - maybe probably doesn't exist yet?
|
||||
const locationRef =
|
||||
unprocessedEntity.metadata?.annotations?.[LOCATION_ANNOTATION];
|
||||
if (!locationRef) {
|
||||
throw new InputError(`Entity '${entityRef}' does not have a location`);
|
||||
}
|
||||
const location = parseLocationReference(locationRef);
|
||||
const originLocation = parseLocationReference(
|
||||
getEntityOriginLocationRef(unprocessedEntity),
|
||||
);
|
||||
|
||||
const emitter = createEmitter(this.options.logger, unprocessedEntity);
|
||||
try {
|
||||
// Pre-process phase, used to populate entities with data that is required during main processing step
|
||||
let entity = unprocessedEntity;
|
||||
for (const processor of this.options.processors) {
|
||||
if (processor.preProcessEntity) {
|
||||
try {
|
||||
entity = await processor.preProcessEntity(
|
||||
entity,
|
||||
location,
|
||||
emitter.emit,
|
||||
originLocation,
|
||||
);
|
||||
} catch (e) {
|
||||
throw new Error(
|
||||
`Processor ${processor.constructor.name} threw an error while preprocessing entity ${entityRef} at ${locationRef}, ${e}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Enforce entity policies making sure that entities conform to a general schema
|
||||
let policyEnforcedEntity;
|
||||
try {
|
||||
policyEnforcedEntity = await this.options.policy.enforce(entity);
|
||||
} catch (e) {
|
||||
throw new InputError(
|
||||
`Policy check failed while analyzing entity ${entityRef} at ${locationRef}, ${e}`,
|
||||
);
|
||||
}
|
||||
if (!policyEnforcedEntity) {
|
||||
throw new Error(
|
||||
`Policy unexpectedly returned no data while analyzing entity ${entityRef} at ${locationRef}`,
|
||||
);
|
||||
}
|
||||
entity = policyEnforcedEntity;
|
||||
|
||||
// Validate the given entity kind against its schema
|
||||
let handled = false;
|
||||
for (const processor of this.options.processors) {
|
||||
if (processor.validateEntityKind) {
|
||||
try {
|
||||
handled = await processor.validateEntityKind(entity);
|
||||
if (handled) {
|
||||
break;
|
||||
}
|
||||
} catch (e) {
|
||||
throw new InputError(
|
||||
`Processor ${processor.constructor.name} threw an error while validating the entity ${entityRef} at ${locationRef}, ${e}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (!handled) {
|
||||
throw new InputError(
|
||||
`No processor recognized the entity ${entityRef} at ${locationRef}`,
|
||||
);
|
||||
}
|
||||
|
||||
// Backwards compatible processing of location entites
|
||||
if (isLocationEntity(entity)) {
|
||||
const { type = location.type } = entity.spec;
|
||||
const targets = new Array<string>();
|
||||
if (entity.spec.target) {
|
||||
targets.push(entity.spec.target);
|
||||
}
|
||||
if (entity.spec.targets) {
|
||||
targets.push(...entity.spec.targets);
|
||||
}
|
||||
|
||||
for (const maybeRelativeTarget of targets) {
|
||||
if (type === 'file' && maybeRelativeTarget.endsWith(path.sep)) {
|
||||
emitter.emit(
|
||||
results.inputError(
|
||||
location,
|
||||
`LocationEntityProcessor cannot handle ${type} type location with target ${location.target} that ends with a path separator`,
|
||||
),
|
||||
);
|
||||
continue;
|
||||
}
|
||||
const target = toAbsoluteUrl(
|
||||
this.options.integrations,
|
||||
location,
|
||||
type,
|
||||
maybeRelativeTarget,
|
||||
);
|
||||
|
||||
for (const processor of this.options.processors) {
|
||||
if (processor.readLocation) {
|
||||
try {
|
||||
const read = await processor.readLocation(
|
||||
{
|
||||
type,
|
||||
target,
|
||||
presence: 'required',
|
||||
},
|
||||
false,
|
||||
emitter.emit,
|
||||
this.options.parser,
|
||||
);
|
||||
if (read) {
|
||||
break;
|
||||
}
|
||||
} catch (e) {
|
||||
throw new Error(
|
||||
`Processor ${processor.constructor.name} threw an error while postprocessing entity ${entityRef} at ${locationRef}, ${e}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Main processing step of the entity
|
||||
for (const processor of this.options.processors) {
|
||||
if (processor.postProcessEntity) {
|
||||
try {
|
||||
entity = await processor.postProcessEntity(
|
||||
entity,
|
||||
location,
|
||||
emitter.emit,
|
||||
);
|
||||
} catch (e) {
|
||||
throw new Error(
|
||||
`Processor ${processor.constructor.name} threw an error while postprocessing entity ${entityRef} at ${locationRef}, ${e}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
...emitter.results(),
|
||||
completedEntity: entity,
|
||||
state: new Map(),
|
||||
ok: true,
|
||||
};
|
||||
} catch (error) {
|
||||
this.options.logger.warn(error.message);
|
||||
return { ok: false, errors: emitter.results().errors.concat(error) };
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function createEmitter(logger: Logger, parentEntity: Entity) {
|
||||
let done = false;
|
||||
|
||||
const errors = new Array<Error>();
|
||||
const relations = new Array<EntityRelationSpec>();
|
||||
const deferredEntites = new Array<Entity>();
|
||||
|
||||
const emit = (i: CatalogProcessorResult) => {
|
||||
if (done) {
|
||||
logger.warn(
|
||||
`Item if type ${i.type} was emitted after processing had completed at ${
|
||||
new Error().stack
|
||||
}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (i.type === 'entity') {
|
||||
const originLocation = getEntityOriginLocationRef(parentEntity);
|
||||
|
||||
deferredEntites.push({
|
||||
...i.entity,
|
||||
metadata: {
|
||||
...i.entity.metadata,
|
||||
annotations: {
|
||||
...i.entity.metadata.annotations,
|
||||
[ORIGIN_LOCATION_ANNOTATION]: originLocation,
|
||||
[LOCATION_ANNOTATION]: stringifyLocationReference(i.location),
|
||||
},
|
||||
},
|
||||
});
|
||||
} else if (i.type === 'location') {
|
||||
deferredEntites.push(
|
||||
locationSpecToLocationEntity(i.location, parentEntity),
|
||||
);
|
||||
} else if (i.type === 'relation') {
|
||||
relations.push(i.relation);
|
||||
} else if (i.type === 'error') {
|
||||
errors.push(i.error);
|
||||
}
|
||||
};
|
||||
|
||||
return {
|
||||
emit,
|
||||
results() {
|
||||
done = true;
|
||||
return {
|
||||
errors,
|
||||
relations,
|
||||
deferredEntites,
|
||||
};
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -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 { Database } 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 DefaultLocationStore implements LocationStore {
|
||||
private subscribers = new Set<
|
||||
ZenObservable.SubscriptionObserver<LocationMessage>
|
||||
>();
|
||||
|
||||
constructor(private readonly db: Database) {}
|
||||
|
||||
createLocation(spec: LocationSpec): Promise<Location> {
|
||||
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<Location[]> {
|
||||
const dbLocations = await this.db.locations();
|
||||
return dbLocations.map(item => ({
|
||||
id: item.id,
|
||||
target: item.target,
|
||||
type: item.type,
|
||||
}));
|
||||
}
|
||||
|
||||
getLocation(id: string): Promise<Location> {
|
||||
return this.db.location(id);
|
||||
}
|
||||
|
||||
deleteLocation(id: string): Promise<void> {
|
||||
return this.db.transaction(async tx => {
|
||||
const location = await this.db.location(id);
|
||||
if (!location) {
|
||||
throw new ConflictError(`No location found 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<LocationMessage> {
|
||||
return new ObservableImpl<LocationMessage>(subscriber => {
|
||||
this.subscribers.add(subscriber);
|
||||
return () => {
|
||||
this.subscribers.delete(subscriber);
|
||||
};
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
/*
|
||||
* 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 { ProcessingDatabase, RefreshStateItem } from './database/types';
|
||||
import {
|
||||
AddProcessingItemRequest,
|
||||
ProccessingItem,
|
||||
ProcessingItemResult,
|
||||
ProcessingStateManager,
|
||||
} from './types';
|
||||
|
||||
export class DefaultProcessingStateManager implements ProcessingStateManager {
|
||||
constructor(private readonly db: ProcessingDatabase) {}
|
||||
|
||||
async setProcessingItemResult(result: ProcessingItemResult) {
|
||||
return this.db.transaction(async tx => {
|
||||
await this.db.updateProcessedEntity(tx, {
|
||||
id: result.id,
|
||||
processedEntity: result.entity,
|
||||
errors: JSON.stringify(result.errors),
|
||||
state: result.state,
|
||||
relations: result.relations,
|
||||
deferredEntities: result.deferredEntities,
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
async addProcessingItems(request: AddProcessingItemRequest) {
|
||||
return this.db.transaction(async tx => {
|
||||
await this.db.addUnprocessedEntities(tx, request);
|
||||
});
|
||||
}
|
||||
|
||||
async getNextProcessingItem(): Promise<ProccessingItem> {
|
||||
const entities = await new Promise<RefreshStateItem[]>(resolve =>
|
||||
this.popFromQueue(resolve),
|
||||
);
|
||||
const { id, state, unprocessedEntity } = entities[0];
|
||||
return {
|
||||
id,
|
||||
entity: unprocessedEntity,
|
||||
state,
|
||||
};
|
||||
}
|
||||
|
||||
async popFromQueue(resolve: (rows: RefreshStateItem[]) => void) {
|
||||
const entities = await this.db.transaction(async tx => {
|
||||
return this.db.getProcessableEntities(tx, {
|
||||
processBatchSize: 1,
|
||||
});
|
||||
});
|
||||
|
||||
// No entities require refresh, wait and try again.
|
||||
if (!entities.items.length) {
|
||||
setTimeout(() => this.popFromQueue(resolve), 1000);
|
||||
return;
|
||||
}
|
||||
|
||||
resolve(entities.items);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,83 @@
|
||||
/*
|
||||
* 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,
|
||||
LOCATION_ANNOTATION,
|
||||
ORIGIN_LOCATION_ANNOTATION,
|
||||
} from '@backstage/catalog-model';
|
||||
import {
|
||||
LocationService,
|
||||
LocationStore,
|
||||
CatalogProcessingOrchestrator,
|
||||
} from './types';
|
||||
|
||||
export class DefaultLocationService 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',
|
||||
annotations: {
|
||||
[LOCATION_ANNOTATION]: `${spec.type}:${spec.target}`,
|
||||
[ORIGIN_LOCATION_ANNOTATION]: `${spec.type}:${spec.target}`,
|
||||
},
|
||||
},
|
||||
spec: {
|
||||
location: { type: spec.type, target: spec.target },
|
||||
},
|
||||
};
|
||||
const processed = await this.orchestrator.process({
|
||||
entity,
|
||||
eager: true,
|
||||
state: new Map(),
|
||||
});
|
||||
if (processed.ok) {
|
||||
return {
|
||||
location: { ...spec, id: `${spec.type}:${spec.target}` },
|
||||
entities: [processed.completedEntity],
|
||||
};
|
||||
}
|
||||
|
||||
throw Error('error handling not implemented.');
|
||||
}
|
||||
|
||||
const location = await this.store.createLocation(spec);
|
||||
return { location, entities: [] };
|
||||
}
|
||||
|
||||
listLocations(): Promise<Location[]> {
|
||||
return this.store.listLocations();
|
||||
}
|
||||
getLocation(id: string): Promise<Location> {
|
||||
return this.store.getLocation(id);
|
||||
}
|
||||
deleteLocation(id: string): Promise<void> {
|
||||
return this.store.deleteLocation(id);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,379 @@
|
||||
/*
|
||||
* 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.
|
||||
*/
|
||||
|
||||
import {
|
||||
PluginDatabaseManager,
|
||||
resolvePackagePath,
|
||||
UrlReader,
|
||||
} from '@backstage/backend-common';
|
||||
import fs from 'fs-extra';
|
||||
import {
|
||||
DefaultNamespaceEntityPolicy,
|
||||
EntityPolicies,
|
||||
EntityPolicy,
|
||||
FieldFormatEntityPolicy,
|
||||
makeValidator,
|
||||
NoForeignRootFieldsEntityPolicy,
|
||||
SchemaValidEntityPolicy,
|
||||
Validators,
|
||||
} from '@backstage/catalog-model';
|
||||
import { Config } from '@backstage/config';
|
||||
import { ScmIntegrations } from '@backstage/integration';
|
||||
import lodash from 'lodash';
|
||||
import { Logger } from 'winston';
|
||||
import {
|
||||
DatabaseLocationsCatalog,
|
||||
EntitiesCatalog,
|
||||
LocationsCatalog,
|
||||
} from '../catalog';
|
||||
import {
|
||||
AnnotateLocationEntityProcessor,
|
||||
BitbucketDiscoveryProcessor,
|
||||
BuiltinKindsEntityProcessor,
|
||||
CatalogProcessor,
|
||||
CatalogProcessorParser,
|
||||
CodeOwnersProcessor,
|
||||
FileReaderProcessor,
|
||||
GithubDiscoveryProcessor,
|
||||
GithubOrgReaderProcessor,
|
||||
LdapOrgReaderProcessor,
|
||||
MicrosoftGraphOrgReaderProcessor,
|
||||
PlaceholderProcessor,
|
||||
PlaceholderResolver,
|
||||
StaticLocationProcessor,
|
||||
UrlReaderProcessor,
|
||||
} from '../ingestion';
|
||||
import { RepoLocationAnalyzer } from '../ingestion/LocationAnalyzer';
|
||||
import {
|
||||
jsonPlaceholderResolver,
|
||||
textPlaceholderResolver,
|
||||
yamlPlaceholderResolver,
|
||||
} from '../ingestion/processors/PlaceholderProcessor';
|
||||
import { defaultEntityDataParser } from '../ingestion/processors/util/parse';
|
||||
import { LocationAnalyzer } from '../ingestion/types';
|
||||
import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine';
|
||||
import { DefaultCatalogProcessingOrchestrator } from './DefaultCatalogProcessingOrchestrator';
|
||||
import { DefaultProcessingDatabase } from './database/DefaultProcessingDatabase';
|
||||
import { DatabaseLocationProvider } from '../next/DatabaseLocationProvider';
|
||||
import { DefaultLocationStore } from './DefaultLocationStore';
|
||||
import { DefaultProcessingStateManager } from './DefaultProcessingStateManager';
|
||||
import { CatalogProcessingEngine } from '../next/types';
|
||||
import { NextEntitiesCatalog } from './NextEntitiesCatalog';
|
||||
import { Stitcher } from './Stitcher';
|
||||
import { CommonDatabase } from '../database/CommonDatabase';
|
||||
|
||||
export type CatalogEnvironment = {
|
||||
logger: Logger;
|
||||
database: PluginDatabaseManager;
|
||||
config: Config;
|
||||
reader: UrlReader;
|
||||
};
|
||||
|
||||
/**
|
||||
* A builder that helps wire up all of the component parts of the catalog.
|
||||
*
|
||||
* The touch points where you can replace or extend behavior are as follows:
|
||||
*
|
||||
* - Entity policies can be added or replaced. These are automatically run
|
||||
* after the processors' pre-processing steps. All policies are given the
|
||||
* chance to inspect the entity, and all of them have to pass in order for
|
||||
* the entity to be considered valid from an overall point of view.
|
||||
* - Placeholder resolvers can be replaced or added. These run on the raw
|
||||
* structured data between the parsing and pre-processing steps, to replace
|
||||
* dollar-prefixed entries with their actual values (like $file).
|
||||
* - Field format validators can be replaced. These check the format of
|
||||
* individual core fields such as metadata.name, to ensure that they adhere
|
||||
* to certain rules.
|
||||
* - Processors can be added or replaced. These implement the functionality of
|
||||
* reading, parsing, validating, and processing the entity data before it is
|
||||
* persisted in the catalog.
|
||||
*/
|
||||
export class NextCatalogBuilder {
|
||||
private readonly env: CatalogEnvironment;
|
||||
private entityPolicies: EntityPolicy[];
|
||||
private entityPoliciesReplace: boolean;
|
||||
private placeholderResolvers: Record<string, PlaceholderResolver>;
|
||||
private fieldFormatValidators: Partial<Validators>;
|
||||
private processors: CatalogProcessor[];
|
||||
private processorsReplace: boolean;
|
||||
private parser: CatalogProcessorParser | undefined;
|
||||
|
||||
constructor(env: CatalogEnvironment) {
|
||||
this.env = env;
|
||||
this.entityPolicies = [];
|
||||
this.entityPoliciesReplace = false;
|
||||
this.placeholderResolvers = {};
|
||||
this.fieldFormatValidators = {};
|
||||
this.processors = [];
|
||||
this.processorsReplace = false;
|
||||
this.parser = undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Adds policies that are used to validate entities between the pre-
|
||||
* processing and post-processing stages. All such policies must pass for the
|
||||
* entity to be considered valid.
|
||||
*
|
||||
* If what you want to do is to replace the rules for what format is allowed
|
||||
* in various core entity fields (such as metadata.name), you may want to use
|
||||
* {@link NextCatalogBuilder#setFieldFormatValidators} instead.
|
||||
*
|
||||
* @param policies One or more policies
|
||||
*/
|
||||
addEntityPolicy(...policies: EntityPolicy[]): NextCatalogBuilder {
|
||||
this.entityPolicies.push(...policies);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets what policies to use for validation of entities between the pre-
|
||||
* processing and post-processing stages. All such policies must pass for the
|
||||
* entity to be considered valid.
|
||||
*
|
||||
* If what you want to do is to replace the rules for what format is allowed
|
||||
* in various core entity fields (such as metadata.name), you may want to use
|
||||
* {@link NextCatalogBuilder#setFieldFormatValidators} instead.
|
||||
*
|
||||
* This function replaces the default set of policies; use with care.
|
||||
*
|
||||
* @param policies One or more policies
|
||||
*/
|
||||
replaceEntityPolicies(policies: EntityPolicy[]): NextCatalogBuilder {
|
||||
this.entityPolicies = [...policies];
|
||||
this.entityPoliciesReplace = true;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Adds, or overwrites, a handler for placeholders (e.g. $file) in entity
|
||||
* definition files.
|
||||
*
|
||||
* @param key The key that identifies the placeholder, e.g. "file"
|
||||
* @param resolver The resolver that gets values for this placeholder
|
||||
*/
|
||||
setPlaceholderResolver(
|
||||
key: string,
|
||||
resolver: PlaceholderResolver,
|
||||
): NextCatalogBuilder {
|
||||
this.placeholderResolvers[key] = resolver;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the validator function to use for one or more special fields of an
|
||||
* entity. This is useful if the default rules for formatting of fields are
|
||||
* not sufficient.
|
||||
*
|
||||
* This function has no effect if used together with
|
||||
* {@link NextCatalogBuilder#replaceEntityPolicies}.
|
||||
*
|
||||
* @param validators The (subset of) validators to set
|
||||
*/
|
||||
setFieldFormatValidators(
|
||||
validators: Partial<Validators>,
|
||||
): NextCatalogBuilder {
|
||||
lodash.merge(this.fieldFormatValidators, validators);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Adds entity processors. These are responsible for reading, parsing, and
|
||||
* processing entities before they are persisted in the catalog.
|
||||
*
|
||||
* @param processors One or more processors
|
||||
*/
|
||||
addProcessor(...processors: CatalogProcessor[]): NextCatalogBuilder {
|
||||
this.processors.push(...processors);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets what entity processors to use. These are responsible for reading,
|
||||
* parsing, and processing entities before they are persisted in the catalog.
|
||||
*
|
||||
* This function replaces the default set of processors; use with care.
|
||||
*
|
||||
* @param processors One or more processors
|
||||
*/
|
||||
replaceProcessors(processors: CatalogProcessor[]): NextCatalogBuilder {
|
||||
this.processors = [...processors];
|
||||
this.processorsReplace = true;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets up the catalog to use a custom parser for entity data.
|
||||
*
|
||||
* This is the function that gets called immediately after some raw entity
|
||||
* specification data has been read from a remote source, and needs to be
|
||||
* parsed and emitted as structured data.
|
||||
*
|
||||
* @param parser The custom parser
|
||||
*/
|
||||
setEntityDataParser(parser: CatalogProcessorParser): NextCatalogBuilder {
|
||||
this.parser = parser;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Wires up and returns all of the component parts of the catalog
|
||||
*/
|
||||
async build(): Promise<{
|
||||
entitiesCatalog: EntitiesCatalog;
|
||||
locationsCatalog: LocationsCatalog;
|
||||
locationAnalyzer: LocationAnalyzer;
|
||||
processingEngine: CatalogProcessingEngine;
|
||||
}> {
|
||||
const { config, database, logger } = this.env;
|
||||
|
||||
const policy = this.buildEntityPolicy();
|
||||
const processors = this.buildProcessors();
|
||||
const parser = this.parser || defaultEntityDataParser;
|
||||
|
||||
const dbClient = await database.getClient();
|
||||
const allMigrations = resolvePackagePath(
|
||||
'@backstage/plugin-catalog-backend',
|
||||
'migrations',
|
||||
);
|
||||
|
||||
const migrationsDir = resolvePackagePath(
|
||||
'@backstage/plugin-catalog-backend',
|
||||
'migrationsv2',
|
||||
);
|
||||
await fs.copy(allMigrations, migrationsDir);
|
||||
await dbClient.migrate.latest({
|
||||
directory: migrationsDir,
|
||||
});
|
||||
const db = new CommonDatabase(dbClient, logger);
|
||||
|
||||
const processingDatabase = new DefaultProcessingDatabase(dbClient, logger);
|
||||
const stateManager = new DefaultProcessingStateManager(processingDatabase);
|
||||
const integrations = ScmIntegrations.fromConfig(config);
|
||||
const orchestrator = new DefaultCatalogProcessingOrchestrator({
|
||||
processors,
|
||||
integrations,
|
||||
logger,
|
||||
parser,
|
||||
policy,
|
||||
});
|
||||
const entitiesCatalog = new NextEntitiesCatalog(dbClient);
|
||||
|
||||
const locationStore = new DefaultLocationStore(db);
|
||||
const dbLocationProvider = new DatabaseLocationProvider(locationStore);
|
||||
const stitcher = new Stitcher(dbClient, logger);
|
||||
const processingEngine = new DefaultCatalogProcessingEngine(
|
||||
logger,
|
||||
[dbLocationProvider], // entityproviders
|
||||
stateManager,
|
||||
orchestrator,
|
||||
stitcher,
|
||||
);
|
||||
|
||||
const locationsCatalog = new DatabaseLocationsCatalog(db);
|
||||
const locationAnalyzer = new RepoLocationAnalyzer(logger);
|
||||
|
||||
return {
|
||||
entitiesCatalog,
|
||||
locationsCatalog,
|
||||
locationAnalyzer,
|
||||
processingEngine,
|
||||
};
|
||||
}
|
||||
|
||||
private buildEntityPolicy(): EntityPolicy {
|
||||
const entityPolicies: EntityPolicy[] = this.entityPoliciesReplace
|
||||
? [new SchemaValidEntityPolicy(), ...this.entityPolicies]
|
||||
: [
|
||||
new SchemaValidEntityPolicy(),
|
||||
new DefaultNamespaceEntityPolicy(),
|
||||
new NoForeignRootFieldsEntityPolicy(),
|
||||
new FieldFormatEntityPolicy(
|
||||
makeValidator(this.fieldFormatValidators),
|
||||
),
|
||||
...this.entityPolicies,
|
||||
];
|
||||
|
||||
return EntityPolicies.allOf(entityPolicies);
|
||||
}
|
||||
|
||||
private buildProcessors(): CatalogProcessor[] {
|
||||
const { config, logger, reader } = this.env;
|
||||
const integrations = ScmIntegrations.fromConfig(config);
|
||||
|
||||
this.checkDeprecatedReaderProcessors();
|
||||
|
||||
const placeholderResolvers: Record<string, PlaceholderResolver> = {
|
||||
json: jsonPlaceholderResolver,
|
||||
yaml: yamlPlaceholderResolver,
|
||||
text: textPlaceholderResolver,
|
||||
...this.placeholderResolvers,
|
||||
};
|
||||
|
||||
// These are always there no matter what
|
||||
const processors: CatalogProcessor[] = [
|
||||
StaticLocationProcessor.fromConfig(config),
|
||||
new PlaceholderProcessor({ resolvers: placeholderResolvers, reader }),
|
||||
new BuiltinKindsEntityProcessor(),
|
||||
];
|
||||
|
||||
// These are only added unless the user replaced them all
|
||||
if (!this.processorsReplace) {
|
||||
processors.push(
|
||||
new FileReaderProcessor(),
|
||||
BitbucketDiscoveryProcessor.fromConfig(config, { logger }),
|
||||
GithubDiscoveryProcessor.fromConfig(config, { logger }),
|
||||
GithubOrgReaderProcessor.fromConfig(config, { logger }),
|
||||
LdapOrgReaderProcessor.fromConfig(config, { logger }),
|
||||
MicrosoftGraphOrgReaderProcessor.fromConfig(config, { logger }),
|
||||
new UrlReaderProcessor({ reader, logger }),
|
||||
CodeOwnersProcessor.fromConfig(config, { logger, reader }),
|
||||
// new LocationEntityProcessor({ integrations }),
|
||||
new AnnotateLocationEntityProcessor({ integrations }),
|
||||
);
|
||||
}
|
||||
|
||||
// Add the ones (if any) that the user added
|
||||
processors.push(...this.processors);
|
||||
|
||||
return processors;
|
||||
}
|
||||
|
||||
// TODO(Rugvip): These old processors are removed, for a while we'll be throwing
|
||||
// errors here to make sure people know where to move the config
|
||||
private checkDeprecatedReaderProcessors() {
|
||||
const pc = this.env.config.getOptionalConfig('catalog.processors');
|
||||
if (pc?.has('github')) {
|
||||
throw new Error(
|
||||
`Using deprecated configuration for catalog.processors.github, move to using integrations.github instead`,
|
||||
);
|
||||
}
|
||||
if (pc?.has('gitlabApi')) {
|
||||
throw new Error(
|
||||
`Using deprecated configuration for catalog.processors.gitlabApi, move to using integrations.gitlab instead`,
|
||||
);
|
||||
}
|
||||
if (pc?.has('bitbucketApi')) {
|
||||
throw new Error(
|
||||
`Using deprecated configuration for catalog.processors.bitbucketApi, move to using integrations.bitbucket instead`,
|
||||
);
|
||||
}
|
||||
if (pc?.has('azureApi')) {
|
||||
throw new Error(
|
||||
`Using deprecated configuration for catalog.processors.azureApi, move to using integrations.azure instead`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
/*
|
||||
* 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.
|
||||
*/
|
||||
|
||||
import { Knex } from 'knex';
|
||||
import { DbFinalEntitiesRow } from './Stitcher';
|
||||
import { EntitiesCatalog } from '../catalog';
|
||||
import { EntitiesRequest, EntitiesResponse } from '../catalog/types';
|
||||
|
||||
export class NextEntitiesCatalog implements EntitiesCatalog {
|
||||
constructor(private readonly database: Knex) {}
|
||||
|
||||
async entities(request?: EntitiesRequest): Promise<EntitiesResponse> {
|
||||
if (request?.fields) {
|
||||
throw new Error('Fields extraction is not implemented');
|
||||
}
|
||||
if (request?.pagination) {
|
||||
throw new Error('Pagination is not implemented');
|
||||
}
|
||||
if (request?.filter) {
|
||||
throw new Error('Filters are not implemented');
|
||||
}
|
||||
|
||||
const dbResponse = await this.database<DbFinalEntitiesRow>(
|
||||
'final_entities',
|
||||
).select();
|
||||
|
||||
const entities = dbResponse.map(e => JSON.parse(e.finalized_entity));
|
||||
|
||||
return {
|
||||
entities,
|
||||
pageInfo: {
|
||||
hasNextPage: false,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async removeEntityByUid(_uid: string): Promise<void> {
|
||||
throw new Error('Not implemented');
|
||||
}
|
||||
|
||||
async batchAddOrUpdateEntities(): Promise<never> {
|
||||
throw new Error('Not implemented');
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,150 @@
|
||||
/*
|
||||
* 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 { Knex } from 'knex';
|
||||
import { Logger } from 'winston';
|
||||
import { Transaction } from '../database';
|
||||
import { ConflictError } from '@backstage/errors';
|
||||
import {
|
||||
DbRefreshStateReferences,
|
||||
DbRefreshStateRow,
|
||||
DbRelationsRow,
|
||||
} from './database/DefaultProcessingDatabase';
|
||||
import { Entity, parseEntityRef } from '@backstage/catalog-model';
|
||||
import { createHash } from 'crypto';
|
||||
import stableStringify from 'fast-json-stable-stringify';
|
||||
|
||||
export type DbFinalEntitiesRow = {
|
||||
entity_id: string;
|
||||
etag: string;
|
||||
finalized_entity: string;
|
||||
};
|
||||
|
||||
function generateEntityEtag(entity: Entity) {
|
||||
return createHash('sha1')
|
||||
.update(stableStringify({ ...entity }))
|
||||
.digest('hex');
|
||||
}
|
||||
|
||||
export class Stitcher {
|
||||
constructor(
|
||||
private readonly database: Knex,
|
||||
private readonly logger: Logger,
|
||||
) {}
|
||||
|
||||
async stitch(entityRefs: Set<string>) {
|
||||
for (const entityRef of entityRefs) {
|
||||
await this.transaction(async txOpaque => {
|
||||
const tx = txOpaque as Knex.Transaction;
|
||||
const [result] = await tx<DbRefreshStateRow>('refresh_state')
|
||||
.select('entity_id', 'processed_entity')
|
||||
.where({ entity_ref: entityRef });
|
||||
|
||||
if (!result) {
|
||||
this.logger.debug(
|
||||
`Unable to stitch ${entityRef}, item does not exist in refresh state table`,
|
||||
);
|
||||
return;
|
||||
} else if (!result.processed_entity) {
|
||||
this.logger.debug(
|
||||
`Unable to stitch ${entityRef}, the entity has not yet been processed`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const entity: Entity = JSON.parse(result.processed_entity);
|
||||
|
||||
const entityId = entity?.metadata?.uid;
|
||||
if (!entityId) {
|
||||
this.logger.error(`missing ID in entity ${JSON.stringify(entity)}`);
|
||||
return;
|
||||
}
|
||||
|
||||
const [reference_count_result] = await tx<DbRefreshStateReferences>(
|
||||
'refresh_state_references',
|
||||
)
|
||||
.where({ target_entity_id: entity.metadata.uid })
|
||||
.count({ reference_count: 'target_entity_id' });
|
||||
|
||||
if (Number(reference_count_result.reference_count) === 0) {
|
||||
this.logger.debug(`${entityRef} is orphan`);
|
||||
entity.metadata.annotations = {
|
||||
...entity.metadata.annotations,
|
||||
['backstage.io/orphan']: 'true',
|
||||
};
|
||||
}
|
||||
|
||||
const relationResults = await tx<DbRelationsRow>('relations')
|
||||
.where({ source_entity_ref: entityRef })
|
||||
.select();
|
||||
|
||||
// TODO: entityRef is lower case and should be uppercase in the final result.
|
||||
entity.relations = relationResults.map(relation => ({
|
||||
type: relation.type,
|
||||
target: parseEntityRef(relation.target_entity_ref),
|
||||
}));
|
||||
entity.metadata.generation = 1;
|
||||
const etag = generateEntityEtag(entity);
|
||||
entity.metadata.etag = etag;
|
||||
await tx<DbFinalEntitiesRow>('final_entities')
|
||||
.insert({
|
||||
finalized_entity: JSON.stringify(entity),
|
||||
entity_id: entityId,
|
||||
etag,
|
||||
})
|
||||
.onConflict('entity_id')
|
||||
.merge(['finalized_entity', 'etag']);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private async transaction<T>(
|
||||
fn: (tx: Transaction) => Promise<T>,
|
||||
): Promise<T> {
|
||||
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,
|
||||
isolationLevel:
|
||||
this.database.client.config.client === 'sqlite3'
|
||||
? undefined // sqlite3 only supports serializable transactions, ignoring the isolation level param
|
||||
: 'serializable',
|
||||
},
|
||||
);
|
||||
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,244 @@
|
||||
/*
|
||||
* 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 lodash from 'lodash';
|
||||
|
||||
import {
|
||||
ProcessingDatabase,
|
||||
AddUnprocessedEntitiesOptions,
|
||||
UpdateProcessedEntityOptions,
|
||||
GetProcessableEntitiesResult,
|
||||
} from './types';
|
||||
import type { Logger } from 'winston';
|
||||
import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
|
||||
import { v4 as uuid } from 'uuid';
|
||||
|
||||
export type DbRefreshStateRow = {
|
||||
entity_id: string;
|
||||
entity_ref: string;
|
||||
unprocessed_entity: string;
|
||||
processed_entity: string;
|
||||
cache: string;
|
||||
next_update_at: string;
|
||||
last_discovery_at: string; // remove?
|
||||
errors: string;
|
||||
};
|
||||
|
||||
export type DbRelationsRow = {
|
||||
originating_entity_id: string;
|
||||
source_entity_ref: string;
|
||||
target_entity_ref: string;
|
||||
type: string;
|
||||
};
|
||||
|
||||
export type DbRefreshStateReferences = {
|
||||
source_special_key?: string;
|
||||
source_entity_id?: string;
|
||||
target_entity_id: string;
|
||||
};
|
||||
|
||||
// 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;
|
||||
|
||||
export class DefaultProcessingDatabase implements ProcessingDatabase {
|
||||
constructor(
|
||||
private readonly database: Knex,
|
||||
private readonly logger: Logger,
|
||||
) {}
|
||||
|
||||
async updateProcessedEntity(
|
||||
txOpaque: Transaction,
|
||||
options: UpdateProcessedEntityOptions,
|
||||
): Promise<void> {
|
||||
const tx = txOpaque as Knex.Transaction;
|
||||
const {
|
||||
id,
|
||||
processedEntity,
|
||||
state,
|
||||
errors,
|
||||
relations,
|
||||
deferredEntities,
|
||||
} = options;
|
||||
|
||||
const refreshResult = await tx<DbRefreshStateRow>('refresh_state')
|
||||
.update({
|
||||
processed_entity: JSON.stringify(processedEntity),
|
||||
cache: JSON.stringify(state),
|
||||
errors,
|
||||
})
|
||||
.where('entity_id', id);
|
||||
|
||||
if (refreshResult === 0) {
|
||||
throw new NotFoundError(`Processing state not found for ${id}`);
|
||||
}
|
||||
|
||||
// Schedule all deferred entities for future processing.
|
||||
await this.addUnprocessedEntities(tx, {
|
||||
entities: deferredEntities,
|
||||
id,
|
||||
type: 'entity',
|
||||
});
|
||||
|
||||
// Update fragments
|
||||
|
||||
// Delete old relations
|
||||
await tx<DbRelationsRow>('relations')
|
||||
.where({ originating_entity_id: id })
|
||||
.delete();
|
||||
|
||||
// Batch insert new relations
|
||||
const relationRows: DbRelationsRow[] = relations.map(
|
||||
({ source, target, type }) => ({
|
||||
originating_entity_id: id,
|
||||
source_entity_ref: stringifyEntityRef(source),
|
||||
target_entity_ref: stringifyEntityRef(target),
|
||||
type,
|
||||
}),
|
||||
);
|
||||
await tx.batchInsert(
|
||||
'relations',
|
||||
this.deduplicateRelations(relationRows),
|
||||
BATCH_SIZE,
|
||||
);
|
||||
}
|
||||
|
||||
private deduplicateRelations(rows: DbRelationsRow[]): DbRelationsRow[] {
|
||||
return lodash.uniqBy(
|
||||
rows,
|
||||
r => `${r.source_entity_ref}:${r.target_entity_ref}:${r.type}`,
|
||||
);
|
||||
}
|
||||
|
||||
async addUnprocessedEntities(
|
||||
txOpaque: Transaction,
|
||||
options: AddUnprocessedEntitiesOptions,
|
||||
): Promise<void> {
|
||||
const tx = txOpaque as Knex.Transaction;
|
||||
const entityIds = new Array<string>();
|
||||
|
||||
for (const entity of options.entities) {
|
||||
const entityRef = stringifyEntityRef(entity);
|
||||
await tx<DbRefreshStateRow>('refresh_state')
|
||||
.insert({
|
||||
entity_id: uuid(),
|
||||
entity_ref: entityRef,
|
||||
unprocessed_entity: JSON.stringify(entity),
|
||||
errors: '',
|
||||
next_update_at: tx.fn.now(),
|
||||
last_discovery_at: tx.fn.now(),
|
||||
})
|
||||
.onConflict('entity_ref')
|
||||
.merge(['unprocessed_entity', 'last_discovery_at']);
|
||||
|
||||
const [{ entity_id: entityId }] = await tx<DbRefreshStateRow>(
|
||||
'refresh_state',
|
||||
).where({ entity_ref: entityRef });
|
||||
entityIds.push(entityId);
|
||||
}
|
||||
|
||||
const key =
|
||||
options.type === 'provider'
|
||||
? { source_special_key: options.id }
|
||||
: { source_entity_id: options.id };
|
||||
// copied from update refs
|
||||
await tx<DbRefreshStateReferences>('refresh_state_references')
|
||||
.where(key)
|
||||
.delete();
|
||||
|
||||
const referenceRows: DbRefreshStateReferences[] = entityIds.map(
|
||||
entityId => ({
|
||||
...key,
|
||||
target_entity_id: entityId,
|
||||
}),
|
||||
);
|
||||
await tx.batchInsert('refresh_state_references', referenceRows, BATCH_SIZE);
|
||||
}
|
||||
|
||||
async getProcessableEntities(
|
||||
txOpaque: Transaction,
|
||||
request: { processBatchSize: number },
|
||||
): Promise<GetProcessableEntitiesResult> {
|
||||
const tx = txOpaque as Knex.Transaction;
|
||||
|
||||
const items = await tx<DbRefreshStateRow>('refresh_state')
|
||||
.select()
|
||||
.where('next_update_at', '<=', tx.fn.now())
|
||||
.limit(request.processBatchSize)
|
||||
.orderBy('next_update_at', 'asc');
|
||||
|
||||
await tx<DbRefreshStateRow>('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: items.map(i => ({
|
||||
id: i.entity_id,
|
||||
entityRef: i.entity_ref,
|
||||
unprocessedEntity: JSON.parse(i.unprocessed_entity) as Entity,
|
||||
processedEntity: JSON.parse(i.processed_entity) as Entity,
|
||||
nextUpdateAt: i.next_update_at,
|
||||
lastDiscoveryAt: i.last_discovery_at,
|
||||
state: JSON.parse(i.cache),
|
||||
errors: i.errors,
|
||||
})),
|
||||
};
|
||||
}
|
||||
|
||||
async transaction<T>(fn: (tx: Transaction) => Promise<T>): Promise<T> {
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,73 @@
|
||||
/*
|
||||
* 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, EntityRelationSpec } from '@backstage/catalog-model';
|
||||
import { JsonObject } from '@backstage/config';
|
||||
import { Transaction } from '../../database/types';
|
||||
|
||||
export type AddUnprocessedEntitiesOptions = {
|
||||
type: 'entity' | 'provider';
|
||||
id: string;
|
||||
entities: Entity[];
|
||||
};
|
||||
|
||||
export type AddUnprocessedEntitiesResult = {};
|
||||
|
||||
export type UpdateProcessedEntityOptions = {
|
||||
id: string;
|
||||
processedEntity: Entity;
|
||||
state?: Map<string, JsonObject>;
|
||||
errors?: string;
|
||||
relations: EntityRelationSpec[];
|
||||
deferredEntities: Entity[];
|
||||
};
|
||||
|
||||
export type RefreshStateItem = {
|
||||
id: string;
|
||||
entityRef: string;
|
||||
unprocessedEntity: Entity;
|
||||
processedEntity: Entity;
|
||||
nextUpdateAt: string;
|
||||
lastDiscoveryAt: string; // remove?
|
||||
state: Map<string, JsonObject>;
|
||||
errors: string;
|
||||
};
|
||||
|
||||
export type GetProcessableEntitiesResult = {
|
||||
items: RefreshStateItem[];
|
||||
};
|
||||
|
||||
export interface ProcessingDatabase {
|
||||
transaction<T>(fn: (tx: Transaction) => Promise<T>): Promise<T>;
|
||||
|
||||
addUnprocessedEntities(
|
||||
tx: Transaction,
|
||||
options: AddUnprocessedEntitiesOptions,
|
||||
): Promise<void>;
|
||||
|
||||
getProcessableEntities(
|
||||
txOpaque: Transaction,
|
||||
request: { processBatchSize: number },
|
||||
): Promise<GetProcessableEntitiesResult>;
|
||||
|
||||
/**
|
||||
* Updates the
|
||||
*/
|
||||
updateProcessedEntity(
|
||||
txOpaque: Transaction,
|
||||
options: UpdateProcessedEntityOptions,
|
||||
): Promise<void>;
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
/*
|
||||
* 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.
|
||||
*/
|
||||
|
||||
export { NextCatalogBuilder } from './NextCatalogBuilder';
|
||||
@@ -0,0 +1,122 @@
|
||||
/*
|
||||
* 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,
|
||||
EntityRelationSpec,
|
||||
} 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<Location[]>;
|
||||
getLocation(id: string): Promise<Location>;
|
||||
deleteLocation(id: string): Promise<void>;
|
||||
}
|
||||
|
||||
export type EntityMessage =
|
||||
| { all: Entity[] }
|
||||
| { added: Entity[]; removed: EntityName[] };
|
||||
|
||||
export interface LocationStore {
|
||||
// extends EntityProvider
|
||||
createLocation(spec: LocationSpec): Promise<Location>;
|
||||
listLocations(): Promise<Location[]>;
|
||||
getLocation(id: string): Promise<Location>;
|
||||
deleteLocation(id: string): Promise<void>;
|
||||
|
||||
location$(): Observable<
|
||||
{ all: Location[] } | { added: Location[]; removed: Location[] }
|
||||
>;
|
||||
}
|
||||
|
||||
export interface CatalogProcessingEngine {
|
||||
start(): Promise<void>;
|
||||
stop(): Promise<void>;
|
||||
}
|
||||
|
||||
export interface EntityProvider {
|
||||
entityChange$(): Observable<EntityMessage>;
|
||||
}
|
||||
|
||||
export type EntityProcessingRequest = {
|
||||
entity: Entity;
|
||||
eager?: boolean;
|
||||
state: Map<string, JsonObject>; // Versions for multiple deployments etc
|
||||
};
|
||||
|
||||
export type EntityProcessingResult =
|
||||
| {
|
||||
ok: true;
|
||||
state: Map<string, JsonObject>;
|
||||
completedEntity: Entity;
|
||||
deferredEntites: Entity[];
|
||||
relations: EntityRelationSpec[];
|
||||
errors: Error[];
|
||||
}
|
||||
| {
|
||||
ok: false;
|
||||
errors: Error[];
|
||||
};
|
||||
|
||||
export interface CatalogProcessingOrchestrator {
|
||||
process(request: EntityProcessingRequest): Promise<EntityProcessingResult>;
|
||||
}
|
||||
|
||||
export type ProcessingItemResult = {
|
||||
id: string;
|
||||
entity: Entity;
|
||||
state: Map<string, JsonObject>;
|
||||
errors: Error[];
|
||||
relations: EntityRelationSpec[];
|
||||
deferredEntities: Entity[];
|
||||
};
|
||||
|
||||
export type AddProcessingItemRequest = {
|
||||
type: 'entity' | 'provider';
|
||||
id: string;
|
||||
entities: Entity[];
|
||||
};
|
||||
|
||||
export type ProccessingItem = {
|
||||
id: string;
|
||||
entity: Entity;
|
||||
state: Map<string, JsonObject>;
|
||||
};
|
||||
|
||||
export interface ProcessingStateManager {
|
||||
setProcessingItemResult(result: ProcessingItemResult): Promise<void>;
|
||||
getNextProcessingItem(): Promise<ProccessingItem>;
|
||||
addProcessingItems(request: AddProcessingItemRequest): Promise<void>;
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
/*
|
||||
* 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,
|
||||
LocationEntityV1alpha1,
|
||||
LOCATION_ANNOTATION,
|
||||
ORIGIN_LOCATION_ANNOTATION,
|
||||
stringifyEntityRef,
|
||||
stringifyLocationReference,
|
||||
} from '@backstage/catalog-model';
|
||||
import { createHash } from 'crypto';
|
||||
|
||||
export function locationSpecToMetadataName(location: LocationSpec) {
|
||||
const hash = createHash('sha1')
|
||||
.update(`${location.type}:${location.target}`)
|
||||
.digest('hex');
|
||||
|
||||
return `generated-${hash}`;
|
||||
}
|
||||
|
||||
export function locationSpecToLocationEntity(
|
||||
location: LocationSpec,
|
||||
parentEntity?: Entity,
|
||||
): LocationEntityV1alpha1 {
|
||||
let ownLocation: string;
|
||||
let originLocation: string;
|
||||
if (parentEntity) {
|
||||
const maybeOwnLocation =
|
||||
parentEntity.metadata.annotations?.[LOCATION_ANNOTATION];
|
||||
if (!maybeOwnLocation) {
|
||||
throw new Error(
|
||||
`Parent entity '${stringifyEntityRef(
|
||||
parentEntity,
|
||||
)}' of location '${stringifyLocationReference(
|
||||
location,
|
||||
)}' does not have a location annotation`,
|
||||
);
|
||||
}
|
||||
ownLocation = maybeOwnLocation;
|
||||
const maybeOriginLocation =
|
||||
parentEntity.metadata.annotations?.[ORIGIN_LOCATION_ANNOTATION];
|
||||
if (!maybeOriginLocation) {
|
||||
throw new Error(
|
||||
`Parent entity '${stringifyEntityRef(
|
||||
parentEntity,
|
||||
)}' of location '${stringifyLocationReference(
|
||||
location,
|
||||
)}' does not have an origin location annotation`,
|
||||
);
|
||||
}
|
||||
originLocation = maybeOriginLocation;
|
||||
} else {
|
||||
ownLocation = stringifyLocationReference(location);
|
||||
originLocation = ownLocation;
|
||||
}
|
||||
|
||||
const result: LocationEntityV1alpha1 = {
|
||||
apiVersion: 'backstage.io/v1alpha1',
|
||||
kind: 'Location',
|
||||
metadata: {
|
||||
name: locationSpecToMetadataName(location),
|
||||
annotations: {
|
||||
[LOCATION_ANNOTATION]: ownLocation,
|
||||
[ORIGIN_LOCATION_ANNOTATION]: originLocation,
|
||||
},
|
||||
},
|
||||
spec: {
|
||||
type: location.type,
|
||||
target: location.target,
|
||||
},
|
||||
};
|
||||
|
||||
return result;
|
||||
}
|
||||
@@ -13039,7 +13039,7 @@ fast-json-patch@^3.0.0-1:
|
||||
resolved "https://registry.npmjs.org/fast-json-patch/-/fast-json-patch-3.0.0-1.tgz#4c68f2e7acfbab6d29d1719c44be51899c93dabb"
|
||||
integrity sha512-6pdFb07cknxvPzCeLsFHStEy+MysPJPgZQ9LbQ/2O67unQF93SNqfdSqnPPl71YMHX+AD8gbl7iuoGFzHEdDuw==
|
||||
|
||||
fast-json-stable-stringify@2.x, fast-json-stable-stringify@^2.0.0:
|
||||
fast-json-stable-stringify@2.x, fast-json-stable-stringify@^2.0.0, fast-json-stable-stringify@^2.1.0:
|
||||
version "2.1.0"
|
||||
resolved "https://registry.npmjs.org/fast-json-stable-stringify/-/fast-json-stable-stringify-2.1.0.tgz#874bf69c6f404c2b5d99c481341399fd55892633"
|
||||
integrity sha512-lhd/wF+Lk98HZoTCtlVraHtfh5XYijIjalXck7saUtuanSDyLMxnHhSXEDJqHxD7msR8D0uCmqlkwjCV8xvwHw==
|
||||
|
||||
Reference in New Issue
Block a user