Initial catalog proccessing rework
Co-authored-by: Patrik Oldsberg <poldsberg@gmail.com> Co-authored-by: Fredrik Adelöw <freben@users.noreply.github.com> Co-authored-by: Tim <timbonicus@users.noreply.github.com> Signed-off-by: Johan Haals <johan.haals@gmail.com>
This commit is contained in:
Johan Haals
@@ -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');
|
||||
};
|
||||
@@ -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",
|
||||
|
||||
@@ -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<string, JsonObject>() },
|
||||
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
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<string, JsonObject>;
|
||||
}): Promise<{
|
||||
state: Map<string, JsonObject>;
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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<EntityMessage>
|
||||
>();
|
||||
|
||||
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<EntityMessage> {
|
||||
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);
|
||||
};
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -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<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,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<LocationMessage>
|
||||
>();
|
||||
|
||||
constructor(private readonly db: CommonDatabase) {}
|
||||
|
||||
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 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,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}`,
|
||||
};
|
||||
}
|
||||
@@ -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<string, JsonObject>;
|
||||
}> {
|
||||
const entities = await new Promise<DbRefreshStateRow[]>(resolve =>
|
||||
this.popFromQueue(resolve),
|
||||
);
|
||||
const result = entities[0];
|
||||
return {
|
||||
entity: JSON.parse(result.unproccessed_entity) as Entity,
|
||||
state: new Map<string, JsonObject>(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);
|
||||
}
|
||||
}
|
||||
@@ -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<void> {
|
||||
const tx = txOpaque as Knex.Transaction;
|
||||
const { id, processedEntity, cache, errors } = options;
|
||||
const result = await tx<DbRefreshStateRow>('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<void> {
|
||||
const tx = txOpaque as Knex.Transaction;
|
||||
|
||||
for (const entity of options.unprocessedEntities) {
|
||||
await tx<DbRefreshStateRow>('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<GetProcessedEntitiesResult> {
|
||||
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', 'desc');
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function stringifyEntityRef(
|
||||
entity: Entity,
|
||||
): Knex.MaybeRawColumn<string> | undefined {
|
||||
throw new Error('Function not implemented.');
|
||||
}
|
||||
@@ -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<T>(fn: (tx: Transaction) => Promise<T>): Promise<T>;
|
||||
addUnprocessedEntities(
|
||||
tx: Transaction,
|
||||
options: AddUnprocessedEntitiesOptions,
|
||||
): Promise<void>;
|
||||
|
||||
getProcessableEntities(
|
||||
txOpaque: Transaction,
|
||||
request: { processBatchSize: number },
|
||||
): Promise<GetProcessedEntitiesResult>;
|
||||
|
||||
/**
|
||||
* Updates the
|
||||
*/
|
||||
updateProcessedEntity(
|
||||
txOpaque: Transaction,
|
||||
options: UpdateProcessedEntityOptions,
|
||||
): Promise<void>;
|
||||
}
|
||||
@@ -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<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>;
|
||||
}
|
||||
|
||||
// interface CatalogProcessor {}
|
||||
type EntityProcessingRequest = {
|
||||
entity: Entity;
|
||||
eager?: boolean;
|
||||
state: Map<string, JsonObject>; // Versions for multiple deployments etc
|
||||
};
|
||||
|
||||
export type EntityProcessingError = {
|
||||
// some error stuff here
|
||||
message: String;
|
||||
};
|
||||
|
||||
type EntityProcessingResult = {
|
||||
state: Map<string, JsonObject>;
|
||||
completeEntities: Entity[];
|
||||
deferredEntites: Entity[];
|
||||
errors: EntityProcessingError[];
|
||||
};
|
||||
|
||||
export interface CatalogProcessingOrchestrator {
|
||||
process(request: EntityProcessingRequest): Promise<EntityProcessingResult>;
|
||||
}
|
||||
|
||||
export type ProcessingResult = {
|
||||
nextRefresh: string;
|
||||
request: EntityProcessingRequest;
|
||||
};
|
||||
|
||||
export interface ProcessingStateManager {
|
||||
setResult(result: ProcessingResult): Promise<void>;
|
||||
pop(): Promise<EntityProcessingRequest>;
|
||||
}
|
||||
Reference in New Issue
Block a user