paulononato/backstage
1
0
Fork 0
Code Issues Pull Requests Actions 44 Packages Projects Releases Wiki Activity

Refactor refreshes into RefreshService

Browse Source 
Co-authored-by: Patrik Oldsberg <poldsberg@gmail.com>
Co-authored-by: Fredrik Adelöw <freben@users.noreply.github.com>
Signed-off-by: Johan Haals <johan.haals@gmail.com>
This commit is contained in:
Johan Haals
2021-09-16 09:31:14 +02:00
parent 7b40b45126
commit 57d462d5ed
10 changed files with 416 additions and 283 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts
+2 -240
View File
@@ -15,27 +15,13 @@
*/
import { getVoidLogger } from '@backstage/backend-common';
import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils';
import { createHash, Hash } from 'crypto';
import { Knex } from 'knex';
import { Logger } from 'winston';
import { Hash } from 'crypto';
import { DateTime } from 'luxon';
import { DatabaseManager } from './database/DatabaseManager';
import waitForExpect from 'wait-for-expect';
import { DefaultProcessingDatabase } from './database/DefaultProcessingDatabase';
import {
DbRefreshStateReferencesRow,
DbRefreshStateRow,
} from './database/tables';
import { ProcessingDatabase } from './database/types';
import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine';
import {
CatalogProcessingOrchestrator,
EntityProcessingRequest,
} from './processing/types';
import { CatalogProcessingOrchestrator } from './processing/types';
import { Stitcher } from './stitching/Stitcher';
import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
import { v4 as uuid } from 'uuid';
describe('DefaultCatalogProcessingEngine', () => {
const db = {
@@ -250,227 +236,3 @@ describe('DefaultCatalogProcessingEngine', () => {
await engine.stop();
});
});
describe('DefaultCatalogProcessingEngine integration', () => {
const defaultLogger = getVoidLogger();
const databases = TestDatabases.create({
ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
});
async function createDatabase(
databaseId: TestDatabaseId,
logger: Logger = defaultLogger,
) {
const knex = await databases.init(databaseId);
await DatabaseManager.createDatabase(knex);
return {
knex,
db: new DefaultProcessingDatabase({
database: knex,
logger,
refreshInterval: () => 100,
}),
};
}
const createPopulatedEngine = async (options: {
db: ProcessingDatabase;
knex: Knex;
entities: Entity[];
references: { [source: string]: string[] };
}) => {
const { db, knex, entities, references } = options;
const entityMap = new Map(
entities.map(entity => [stringifyEntityRef(entity), entity]),
);
for (const entity of entities) {
await knex<DbRefreshStateRow>('refresh_state').insert({
entity_id: uuid(),
entity_ref: stringifyEntityRef(entity),
unprocessed_entity: JSON.stringify(entity),
errors: '[]',
next_update_at: '2031-01-01 23:00:00',
last_discovery_at: '2021-04-01 13:37:00',
});
}
for (const entityRef of entityMap.keys()) {
if (!(entityRef in references)) {
await knex<DbRefreshStateReferencesRow>(
'refresh_state_references',
).insert({
source_key: 'ConfigLocationProvider',
target_entity_ref: entityRef,
});
}
}
for (const [sourceRef, targetRefs] of Object.entries(references)) {
for (const targetRef of targetRefs) {
await knex<DbRefreshStateReferencesRow>(
'refresh_state_references',
).insert({
source_entity_ref: sourceRef,
target_entity_ref: targetRef,
});
}
}
const engine = new DefaultCatalogProcessingEngine(
defaultLogger,
[],
db,
{
async process(request: EntityProcessingRequest) {
const entityRef = stringifyEntityRef(request.entity);
const entity = entityMap.get(entityRef);
if (!entity) {
throw new Error(`Unexpected entity: ${entityRef}`);
}
const deferredEntities =
references[entityRef]?.map(ref => {
const e = entityMap.get(ref);
if (!e) {
throw new Error(`Target entity not found: ${ref}`);
}
return { entity: e, locationKey: ref };
}) || [];
return {
ok: true,
completedEntity: {
...entity,
metadata: {
...entity.metadata,
annotations: {
...entity.metadata.annotations,
'refresh-completed': 'true',
},
},
},
relations: [],
errors: [],
deferredEntities,
state: new Map(),
};
},
},
new Stitcher(knex, defaultLogger),
() => createHash('sha1'),
50,
);
return engine;
};
const waitForRefresh = async (knex: Knex, entityRef: string) => {
for (;;) {
const [result] = await knex<DbRefreshStateRow>('refresh_state')
.where('entity_ref', entityRef)
.select();
const entity = result.processed_entity
? (JSON.parse(result.processed_entity) as Entity)
: undefined;
if (entity?.metadata?.annotations?.['refresh-completed']) {
return true;
}
await new Promise(resolve => setTimeout(resolve, 500));
}
};
it.each(databases.eachSupportedId())(
'should refresh the parent location, %p',
async databaseId => {
const { knex, db } = await createDatabase(databaseId);
const engine = await createPopulatedEngine({
db,
knex,
entities: [
{
kind: 'Location',
apiVersion: '1.0.0',
metadata: {
name: 'myloc',
},
},
{
kind: 'Component',
apiVersion: '1.0.0',
metadata: {
name: 'mycomp',
},
},
],
references: {
'location:default/myloc': ['component:default/mycomp'],
},
});
await engine.start();
await engine.refresh({
entityRef: 'component:default/mycomp',
});
await expect(
waitForRefresh(knex, 'component:default/mycomp'),
).resolves.toBe(true);
await engine.stop();
},
);
it.each(databases.eachSupportedId())(
'should refresh the location further up the tree, %p',
async databaseId => {
const { knex, db } = await createDatabase(databaseId);
const engine = await createPopulatedEngine({
db,
knex,
entities: [
{
kind: 'Location',
apiVersion: '1.0.0',
metadata: {
name: 'myloc',
},
},
{
kind: 'Component',
apiVersion: '1.0.0',
metadata: {
name: 'mycomp',
},
},
{
kind: 'Api',
apiVersion: '1.0.0',
metadata: {
name: 'myapi',
},
},
],
references: {
'location:default/myloc': ['component:default/mycomp'],
'component:default/mycomp': ['api:default/myapi'],
},
});
await engine.start();
await engine.refresh({
entityRef: 'api:default/myapi',
});
await expect(waitForRefresh(knex, 'api:default/myapi')).resolves.toBe(
true,
);
await engine.stop();
},
);
});
plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts
-16
View File
@@ -36,7 +36,6 @@ import {
EntityProvider,
EntityProviderConnection,
EntityProviderMutation,
CatalogProcessingEngineRefreshOptions,
} from './types';
class Connection implements EntityProviderConnection {
@@ -239,21 +238,6 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
this.stopFunc = undefined;
}
}
async refresh(options: CatalogProcessingEngineRefreshOptions) {
await this.processingDatabase.transaction(async tx => {
const { entityRefs } = await this.processingDatabase.listAncestors(tx, {
entityRef: options.entityRef,
});
const locationAncestor = entityRefs.find(ref =>
ref.startsWith('location:'),
);
await this.processingDatabase.refresh(tx, {
entityRef: locationAncestor ?? options.entityRef,
});
});
}
}
// Helps wrap the timing and logging behaviors
plugins/catalog-backend/src/next/DefaultRefreshService.test.ts
+328
View File
@@ -0,0 +1,328 @@
/*
* Copyright 2021 The Backstage Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import { getVoidLogger } from '@backstage/backend-common';
import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils';
import { createHash } from 'crypto';
import { Knex } from 'knex';
import { Logger } from 'winston';
import { DatabaseManager } from './database/DatabaseManager';
import { DefaultProcessingDatabase } from './database/DefaultProcessingDatabase';
import {
DbRefreshStateReferencesRow,
DbRefreshStateRow,
} from './database/tables';
import { ProcessingDatabase } from './database/types';
import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine';
import { EntityProcessingRequest } from './processing/types';
import { Stitcher } from './stitching/Stitcher';
import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
import { v4 as uuid } from 'uuid';
import { DefaultRefreshService } from './DefaultRefreshService';
describe('Refresh integration', () => {
const defaultLogger = getVoidLogger();
const databases = TestDatabases.create({
ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
});
async function createDatabase(
databaseId: TestDatabaseId,
logger: Logger = defaultLogger,
) {
const knex = await databases.init(databaseId);
await DatabaseManager.createDatabase(knex);
return {
knex,
db: new DefaultProcessingDatabase({
database: knex,
logger,
refreshInterval: () => 100,
}),
};
}
const createPopulatedEngine = async (options: {
db: ProcessingDatabase;
knex: Knex;
entities: Entity[];
references: { [source: string]: string[] };
entityProcessor?: (entity: Entity) => void;
}) => {
const { db, knex, entities, references, entityProcessor } = options;
const entityMap = new Map(
entities.map(entity => [stringifyEntityRef(entity), entity]),
);
for (const entity of entities) {
await knex<DbRefreshStateRow>('refresh_state').insert({
entity_id: uuid(),
entity_ref: stringifyEntityRef(entity),
unprocessed_entity: JSON.stringify(entity),
errors: '[]',
next_update_at: '2031-01-01 23:00:00',
last_discovery_at: '2021-04-01 13:37:00',
});
}
const entitiesWithParent = new Set(Object.values(references).flat());
for (const entityRef of entityMap.keys()) {
if (!entitiesWithParent.has(entityRef)) {
await knex<DbRefreshStateReferencesRow>(
'refresh_state_references',
).insert({
source_key: 'ConfigLocationProvider',
target_entity_ref: entityRef,
});
}
}
for (const [sourceRef, targetRefs] of Object.entries(references)) {
for (const targetRef of targetRefs) {
await knex<DbRefreshStateReferencesRow>(
'refresh_state_references',
).insert({
source_entity_ref: sourceRef,
target_entity_ref: targetRef,
});
}
}
const engine = new DefaultCatalogProcessingEngine(
defaultLogger,
[],
db,
{
async process(request: EntityProcessingRequest) {
const entityRef = stringifyEntityRef(request.entity);
const entity = entityMap.get(entityRef);
if (!entity) {
throw new Error(`Unexpected entity: ${entityRef}`);
}
const deferredEntities =
references[entityRef]?.map(ref => {
const e = entityMap.get(ref);
if (!e) {
throw new Error(`Target entity not found: ${ref}`);
}
return { entity: e, locationKey: ref };
}) || [];
entityProcessor?.(entity);
return {
ok: true,
completedEntity: {
...entity,
metadata: {
...entity.metadata,
annotations: {
...entity.metadata.annotations,
'refresh-completed': 'true',
},
},
},
relations: [],
errors: [],
deferredEntities,
state: new Map(),
};
},
},
new Stitcher(knex, defaultLogger),
() => createHash('sha1'),
50,
);
return engine;
};
const waitForRefresh = async (knex: Knex, entityRef: string) => {
for (;;) {
const [result] = await knex<DbRefreshStateRow>('refresh_state')
.where('entity_ref', entityRef)
.select();
const entity = result.processed_entity
? (JSON.parse(result.processed_entity) as Entity)
: undefined;
if (entity?.metadata?.annotations?.['refresh-completed']) {
// Reset the annotation so that we can run another verification
delete entity.metadata.annotations['refresh-completed'];
await knex<DbRefreshStateRow>('refresh_state')
.update({
processed_entity: JSON.stringify(entity),
})
.where('entity_ref', entityRef);
return true;
}
await new Promise(resolve => setTimeout(resolve, 500));
}
};
it.each(databases.eachSupportedId())(
'should refresh the parent location, %p',
async databaseId => {
const { knex, db } = await createDatabase(databaseId);
const refreshService = new DefaultRefreshService({ database: db });
const engine = await createPopulatedEngine({
db,
knex,
entities: [
{
kind: 'Location',
apiVersion: '1.0.0',
metadata: {
name: 'myloc',
},
},
{
kind: 'Component',
apiVersion: '1.0.0',
metadata: {
name: 'mycomp',
},
},
],
references: {
'location:default/myloc': ['component:default/mycomp'],
},
});
await engine.start();
await refreshService.refresh({
entityRef: 'component:default/mycomp',
});
await expect(
waitForRefresh(knex, 'location:default/myloc'),
).resolves.toBe(true);
await engine.stop();
},
);
it.each(databases.eachSupportedId())(
'should refresh the location further up the tree, %p',
async databaseId => {
const { knex, db } = await createDatabase(databaseId);
const refreshService = new DefaultRefreshService({ database: db });
const engine = await createPopulatedEngine({
db,
knex,
entities: [
{
kind: 'Location',
apiVersion: '1.0.0',
metadata: {
name: 'myloc',
},
},
{
kind: 'Component',
apiVersion: '1.0.0',
metadata: {
name: 'mycomp',
},
},
{
kind: 'Api',
apiVersion: '1.0.0',
metadata: {
name: 'myapi',
},
},
],
references: {
'location:default/myloc': ['component:default/mycomp'],
'component:default/mycomp': ['api:default/myapi'],
},
});
await engine.start();
await refreshService.refresh({
entityRef: 'api:default/myapi',
});
await expect(waitForRefresh(knex, 'api:default/myapi')).resolves.toBe(
true,
);
await engine.stop();
},
);
it.each(databases.eachSupportedId())(
'should refresh even when parent has no changes',
async databaseId => {
let secondRound = false;
const { knex, db } = await createDatabase(databaseId);
const refreshService = new DefaultRefreshService({ database: db });
const engine = await createPopulatedEngine({
db,
knex,
entities: [
{
kind: 'Location',
apiVersion: '1.0.0',
metadata: {
name: 'myloc',
},
},
{
kind: 'Component',
apiVersion: '1.0.0',
metadata: {
name: 'mycomp',
},
},
],
references: {
'location:default/myloc': ['component:default/mycomp'],
},
entityProcessor: entity => {
if (entity.metadata.name === 'mycomp' && secondRound) {
entity.apiVersion = '2.0.0';
}
},
});
await engine.start();
await refreshService.refresh({
entityRef: 'component:default/mycomp',
});
await expect(
waitForRefresh(knex, 'component:default/mycomp'),
).resolves.toBe(true);
secondRound = true;
await refreshService.refresh({
entityRef: 'component:default/mycomp',
});
await expect(
waitForRefresh(knex, 'component:default/mycomp'),
).resolves.toBe(true);
await engine.stop();
},
);
});
plugins/catalog-backend/src/next/DefaultRefreshService.ts
+46
View File
@@ -0,0 +1,46 @@
/*
* Copyright 2021 The Backstage Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import { DefaultProcessingDatabase } from './database/DefaultProcessingDatabase';
import { RefreshOptions, RefreshService } from './types';
export class DefaultRefreshService implements RefreshService {
private database: DefaultProcessingDatabase;
constructor(options: { database: DefaultProcessingDatabase }) {
this.database = options.database;
}
async refresh(options: RefreshOptions) {
await this.database.transaction(async tx => {
const { entityRefs } = await this.database.listAncestors(tx, {
entityRef: options.entityRef,
});
const locationAncestor = entityRefs.find(ref =>
ref.startsWith('location:'),
);
if (locationAncestor) {
await this.database.refresh(tx, {
entityRef: locationAncestor,
});
}
await this.database.refresh(tx, {
entityRef: options.entityRef,
});
});
}
}
plugins/catalog-backend/src/next/NextCatalogBuilder.ts
+5 -2
View File
@@ -77,6 +77,7 @@ import {
} from './refresh';
import { CatalogEnvironment } from '../service/CatalogBuilder';
import { createNextRouter } from './NextRouter';
import { DefaultRefreshService } from './DefaultRefreshService';
/**
* A builder that helps wire up all of the component parts of the catalog.
@@ -335,12 +336,14 @@ export class NextCatalogBuilder {
locationStore,
orchestrator,
);
const refreshService = new DefaultRefreshService({
database: processingDatabase,
});
const router = await createNextRouter({
entitiesCatalog,
locationAnalyzer,
locationService,
processingEngine,
refreshService,
logger,
config,
});
plugins/catalog-backend/src/next/NextRouter.ts
+6 -10
View File
@@ -34,17 +34,13 @@ import {
parseEntityTransformParams,
} from '../service/request';
import { disallowReadonlyMode, validateRequestBody } from '../service/util';
import {
CatalogProcessingEngine,
LocationService,
CatalogProcessingEngineRefreshOptions,
} from './types';
import { RefreshService, RefreshOptions, LocationService } from './types';
export interface NextRouterOptions {
entitiesCatalog?: EntitiesCatalog;
locationAnalyzer?: LocationAnalyzer;
locationService: LocationService;
processingEngine?: CatalogProcessingEngine;
refreshService?: RefreshService;
logger: Logger;
config: Config;
}
@@ -56,7 +52,7 @@ export async function createNextRouter(
entitiesCatalog,
locationAnalyzer,
locationService,
processingEngine,
refreshService,
config,
logger,
} = options;
@@ -70,10 +66,10 @@ export async function createNextRouter(
logger.info('Catalog is running in readonly mode');
}
if (processingEngine) {
if (refreshService) {
router.post('/refresh', async (req, res) => {
const refreshOptions: CatalogProcessingEngineRefreshOptions = req.body;
await processingEngine.refresh(refreshOptions);
const refreshOptions: RefreshOptions = req.body;
await refreshService.refresh(refreshOptions);
res.status(200).send();
});
}
plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts
+1 -1
View File
@@ -550,7 +550,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
);
}
const parentRef = rows[0].source_entity_ref;
const parentRef = rows.find(r => r.source_entity_ref)?.source_entity_ref;
if (!parentRef) {
// We've reached the top of the tree which is the entityProvider.
// In this case we refresh the entity itself.
plugins/catalog-backend/src/next/index.ts
+1 -1
View File
@@ -28,5 +28,5 @@ export type {
CatalogProcessingEngine,
LocationService,
LocationStore,
CatalogProcessingEngineRefreshOptions,
RefreshOptions,
} from './types';
plugins/catalog-backend/src/next/types.ts
+21 -3
View File
@@ -37,11 +37,29 @@ export interface LocationStore {
export interface CatalogProcessingEngine {
start(): Promise<void>;
stop(): Promise<void>;
refresh(options: CatalogProcessingEngineRefreshOptions): Promise<void>;
}
/** @public */
export type CatalogProcessingEngineRefreshOptions = { entityRef: string };
/**
* Options for requesting a refresh of entities in the catalog.
*
* @public
*/
export type RefreshOptions = {
/** The reference to a single entity that should be refreshed */
entityRef: string;
};
/**
* A service that manages refreshes of entities in the catalog.
*
* @public
*/
export interface RefreshService {
/**
* Request a refresh of entities in the catalog.
*/
refresh(options: RefreshOptions): Promise<void>;
}
export type EntityProviderMutation =
| { type: 'full'; entities: DeferredEntity[] }
plugins/catalog-backend/src/service/router.ts
+6 -10
View File
@@ -28,11 +28,7 @@ import { Logger } from 'winston';
import yn from 'yn';
import { EntitiesCatalog, LocationsCatalog } from '../catalog';
import { HigherOrderOperation, LocationAnalyzer } from '../ingestion/types';
import {
LocationService,
CatalogProcessingEngine,
CatalogProcessingEngineRefreshOptions,
} from '../next/types';
import { RefreshService, LocationService, RefreshOptions } from '../next/types';
import {
basicEntityFilter,
parseEntityFilterParams,
@@ -51,7 +47,7 @@ export interface RouterOptions {
higherOrderOperation?: HigherOrderOperation;
locationAnalyzer?: LocationAnalyzer;
locationService?: LocationService;
processingEngine?: CatalogProcessingEngine;
refreshService?: RefreshService;
logger: Logger;
config: Config;
}
@@ -65,7 +61,7 @@ export async function createRouter(
higherOrderOperation,
locationAnalyzer,
locationService,
processingEngine,
refreshService,
config,
logger,
} = options;
@@ -79,10 +75,10 @@ export async function createRouter(
logger.info('Catalog is running in readonly mode');
}
if (processingEngine) {
if (refreshService) {
router.post('/refresh', async (req, res) => {
const refreshOptions: CatalogProcessingEngineRefreshOptions = req.body;
await processingEngine.refresh(refreshOptions);
const refreshOptions: RefreshOptions = req.body;
await refreshService.refresh(refreshOptions);
res.status(200).send();
});
}