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

Merge pull request #19869 from backstage/freben/stitch-refactor

Browse Source 
Internal refactors as a path toward #18062
This commit is contained in:
Fredrik Adelöw
2023-09-26 10:56:55 +02:00
committed by GitHub
parent ec1ca01848 e349190374
commit 6574c6858a
27 changed files with 1335 additions and 355 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/polite-donuts-fail.md
+5
View File
@@ -0,0 +1,5 @@
---
'@backstage/plugin-catalog-backend': patch
---
Internal refactors, laying the foundation for later introducing deferred stitching (see #18062).
plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts
-6
View File
@@ -42,7 +42,6 @@ import {
import { checkLocationKeyConflict } from './operations/refreshState/checkLocationKeyConflict';
import { insertUnprocessedEntity } from './operations/refreshState/insertUnprocessedEntity';
import { updateUnprocessedEntity } from './operations/refreshState/updateUnprocessedEntity';
import { deleteOrphanedEntities } from './operations/util/deleteOrphanedEntities';
import { generateStableHash } from './util';
import { EventBroker, EventParams } from '@backstage/plugin-events-node';
import { DateTime } from 'luxon';
@@ -275,11 +274,6 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
return { entityRefs };
}
async deleteOrphanedEntities(txOpaque: Transaction): Promise<number> {
const tx = txOpaque as Knex.Transaction;
return await deleteOrphanedEntities({ tx });
}
async transaction<T>(fn: (tx: Transaction) => Promise<T>): Promise<T> {
try {
let result: T | undefined = undefined;
plugins/catalog-backend/src/stitching/buildEntitySearch.test.ts → plugins/catalog-backend/src/database/operations/stitcher/buildEntitySearch.test.ts
+1 -1
View File
@@ -14,7 +14,7 @@
* limitations under the License.
*/
import { Entity, DEFAULT_NAMESPACE } from '@backstage/catalog-model';
import { DEFAULT_NAMESPACE, Entity } from '@backstage/catalog-model';
import { buildEntitySearch, mapToRows, traverse } from './buildEntitySearch';
describe('buildEntitySearch', () => {
plugins/catalog-backend/src/stitching/buildEntitySearch.ts → plugins/catalog-backend/src/database/operations/stitcher/buildEntitySearch.ts
+2 -2
View File
@@ -14,9 +14,9 @@
* limitations under the License.
*/
import { Entity, DEFAULT_NAMESPACE } from '@backstage/catalog-model';
import { DEFAULT_NAMESPACE, Entity } from '@backstage/catalog-model';
import { InputError } from '@backstage/errors';
import { DbSearchRow } from '../database/tables';
import { DbSearchRow } from '../../tables';
// These are excluded in the generic loop, either because they do not make sense
// to index, or because they are special-case always inserted whether they are
plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts
+260
View File
@@ -0,0 +1,260 @@
/*
* Copyright 2023 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 { TestDatabases } from '@backstage/backend-test-utils';
import { applyDatabaseMigrations } from '../../migrations';
import { markForStitching } from './markForStitching';
import { DbFinalEntitiesRow, DbRefreshStateRow } from '../../tables';
jest.setTimeout(60_000);
describe('markForStitching', () => {
const databases = TestDatabases.create({
ids: ['MYSQL_8', 'POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
});
it.each(databases.eachSupportedId())(
'marks the right rows in immediate mode %p',
async databaseId => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
await knex<DbRefreshStateRow>('refresh_state').insert([
{
entity_id: '1',
entity_ref: 'k:ns/one',
unprocessed_entity: '{}',
processed_entity: '{}',
result_hash: 'old',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '2',
entity_ref: 'k:ns/two',
unprocessed_entity: '{}',
processed_entity: '{}',
result_hash: 'old',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '3',
entity_ref: 'k:ns/three',
unprocessed_entity: '{}',
processed_entity: '{}',
result_hash: 'old',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '4',
entity_ref: 'k:ns/four',
unprocessed_entity: '{}',
processed_entity: '{}',
result_hash: 'old',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
]);
await knex<DbFinalEntitiesRow>('final_entities').insert([
{
entity_id: '1',
final_entity: '{}',
hash: 'old',
stitch_ticket: 'old',
},
{
entity_id: '2',
final_entity: '{}',
hash: 'old',
stitch_ticket: 'old',
},
{
entity_id: '3',
final_entity: '{}',
hash: 'old',
stitch_ticket: 'old',
},
{
entity_id: '4',
final_entity: '{}',
hash: 'old',
stitch_ticket: 'old',
},
]);
async function result() {
return knex<DbRefreshStateRow>('refresh_state')
.leftJoin(
'final_entities',
'final_entities.entity_id',
'refresh_state.entity_id',
)
.select({
entity_id: 'refresh_state.entity_id',
next_update_at: 'refresh_state.next_update_at',
refresh_state_hash: 'refresh_state.result_hash',
final_entities_hash: 'final_entities.hash',
})
.orderBy('entity_id', 'asc');
}
// Ensure that now() isn't evaluating to the same thing
await new Promise(resolve => setTimeout(resolve, 1100));
const original = await result();
await markForStitching({
knex,
strategy: { mode: 'immediate' },
entityRefs: new Set(),
});
await expect(result()).resolves.toEqual([
{
entity_id: '1',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
{
entity_id: '2',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
{
entity_id: '3',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
{
entity_id: '4',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
]);
await markForStitching({
knex,
strategy: { mode: 'immediate' },
entityRefs: new Set(['k:ns/one']),
});
await expect(result()).resolves.toEqual([
{
entity_id: '1',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
{
entity_id: '2',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
{
entity_id: '3',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
{
entity_id: '4',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
]);
await markForStitching({
knex,
strategy: { mode: 'immediate' },
entityRefs: ['k:ns/two'],
});
await expect(result()).resolves.toEqual([
{
entity_id: '1',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
{
entity_id: '2',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
{
entity_id: '3',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
{
entity_id: '4',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
]);
await markForStitching({
knex,
strategy: { mode: 'immediate' },
entityIds: ['3', '4'],
});
await expect(result()).resolves.toEqual([
{
entity_id: '1',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
{
entity_id: '2',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
{
entity_id: '3',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
{
entity_id: '4',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
]);
// It overwrites timers
const final = await result();
for (let i = 0; i < final.length; ++i) {
expect(original[i].next_update_at).not.toEqual(final[i].next_update_at);
}
},
);
});
plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts
+82
View File
@@ -0,0 +1,82 @@
/*
* Copyright 2023 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 { Knex } from 'knex';
import splitToChunks from 'lodash/chunk';
import { StitchingStrategy } from '../../../stitching/types';
import { DbFinalEntitiesRow, DbRefreshStateRow } from '../../tables';
/**
* Marks a number of entities for deferred stitching some time in the near
* future.
*
* @remarks
*/
export async function markForStitching(options: {
knex: Knex | Knex.Transaction;
strategy: StitchingStrategy;
entityRefs?: Iterable<string>;
entityIds?: Iterable<string>;
}): Promise<void> {
// Splitting to chunks just to cover pathological cases that upset the db
const entityRefs = split(options.entityRefs);
const entityIds = split(options.entityIds);
const knex = options.knex;
for (const chunk of entityRefs) {
await knex
.table<DbFinalEntitiesRow>('final_entities')
.update({
hash: 'force-stitching',
})
.whereIn(
'entity_id',
knex<DbRefreshStateRow>('refresh_state')
.select('entity_id')
.whereIn('entity_ref', chunk),
);
await knex
.table<DbRefreshStateRow>('refresh_state')
.update({
result_hash: 'force-stitching',
next_update_at: knex.fn.now(),
})
.whereIn('entity_ref', chunk);
}
for (const chunk of entityIds) {
await knex
.table<DbFinalEntitiesRow>('final_entities')
.update({
hash: 'force-stitching',
})
.whereIn('entity_id', chunk);
await knex
.table<DbRefreshStateRow>('refresh_state')
.update({
result_hash: 'force-stitching',
next_update_at: knex.fn.now(),
})
.whereIn('entity_id', chunk);
}
}
function split(input: Iterable<string> | undefined): string[][] {
if (!input) {
return [];
}
return splitToChunks(Array.isArray(input) ? input : [...input], 200);
}
plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts
+287
View File
@@ -0,0 +1,287 @@
/*
* 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 { TestDatabases } from '@backstage/backend-test-utils';
import { Entity } from '@backstage/catalog-model';
import { applyDatabaseMigrations } from '../../migrations';
import {
DbFinalEntitiesRow,
DbRefreshStateReferencesRow,
DbRefreshStateRow,
DbRelationsRow,
DbSearchRow,
} from '../../tables';
import { performStitching } from './performStitching';
jest.setTimeout(60_000);
describe('performStitching', () => {
const databases = TestDatabases.create({
ids: ['MYSQL_8', 'POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
});
const logger = getVoidLogger();
it.each(databases.eachSupportedId())(
'runs the happy path in immediate mode for %p',
async databaseId => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
let entities: DbFinalEntitiesRow[];
let entity: Entity;
await knex<DbRefreshStateRow>('refresh_state').insert([
{
entity_id: 'my-id',
entity_ref: 'k:ns/n',
unprocessed_entity: JSON.stringify({}),
processed_entity: JSON.stringify({
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
},
spec: {
k: 'v',
},
}),
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
]);
await knex<DbRefreshStateReferencesRow>(
'refresh_state_references',
).insert([{ source_key: 'a', target_entity_ref: 'k:ns/n' }]);
await knex<DbRelationsRow>('relations').insert([
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/other',
},
// handles and ignores duplicates
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/other',
},
]);
await performStitching({
knex,
logger,
strategy: { mode: 'immediate' },
entityRef: 'k:ns/n',
});
entities = await knex<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
entity = JSON.parse(entities[0].final_entity!);
expect(entity).toEqual({
relations: [
{
type: 'looksAt',
targetRef: 'k:ns/other',
},
],
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
etag: expect.any(String),
uid: 'my-id',
},
spec: {
k: 'v',
},
});
expect(entity.metadata.etag).toEqual(entities[0].hash);
const last_updated_at = entities[0].last_updated_at;
expect(last_updated_at).not.toBeNull();
const firstHash = entities[0].hash;
const search = await knex<DbSearchRow>('search');
expect(search).toEqual(
expect.arrayContaining([
{
entity_id: 'my-id',
key: 'relations.looksat',
original_value: 'k:ns/other',
value: 'k:ns/other',
},
{
entity_id: 'my-id',
key: 'apiversion',
original_value: 'a',
value: 'a',
},
{
entity_id: 'my-id',
key: 'kind',
original_value: 'k',
value: 'k',
},
{
entity_id: 'my-id',
key: 'metadata.name',
original_value: 'n',
value: 'n',
},
{
entity_id: 'my-id',
key: 'metadata.namespace',
original_value: 'ns',
value: 'ns',
},
{
entity_id: 'my-id',
key: 'metadata.uid',
original_value: 'my-id',
value: 'my-id',
},
{
entity_id: 'my-id',
key: 'spec.k',
original_value: 'v',
value: 'v',
},
]),
);
// Re-stitch without any changes
await performStitching({
knex,
logger,
strategy: { mode: 'immediate' },
entityRef: 'k:ns/n',
});
entities = await knex<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
entity = JSON.parse(entities[0].final_entity!);
expect(entities[0].hash).toEqual(firstHash);
expect(entity.metadata.etag).toEqual(firstHash);
// Now add one more relation and re-stitch
await knex<DbRelationsRow>('relations').insert([
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/third',
},
]);
await performStitching({
knex,
logger,
strategy: { mode: 'immediate' },
entityRef: 'k:ns/n',
});
entities = await knex<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
entity = JSON.parse(entities[0].final_entity!);
expect(entity).toEqual({
relations: expect.arrayContaining([
{
type: 'looksAt',
targetRef: 'k:ns/other',
},
{
type: 'looksAt',
targetRef: 'k:ns/third',
},
]),
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
etag: expect.any(String),
uid: 'my-id',
},
spec: {
k: 'v',
},
});
expect(entities[0].hash).not.toEqual(firstHash);
expect(entities[0].hash).toEqual(entity.metadata.etag);
expect(await knex<DbSearchRow>('search')).toEqual(
expect.arrayContaining([
{
entity_id: 'my-id',
key: 'relations.looksat',
original_value: 'k:ns/other',
value: 'k:ns/other',
},
{
entity_id: 'my-id',
key: 'relations.looksat',
original_value: 'k:ns/third',
value: 'k:ns/third',
},
{
entity_id: 'my-id',
key: 'apiversion',
original_value: 'a',
value: 'a',
},
{
entity_id: 'my-id',
key: 'kind',
original_value: 'k',
value: 'k',
},
{
entity_id: 'my-id',
key: 'metadata.name',
original_value: 'n',
value: 'n',
},
{
entity_id: 'my-id',
key: 'metadata.namespace',
original_value: 'ns',
value: 'ns',
},
{
entity_id: 'my-id',
key: 'metadata.uid',
original_value: 'my-id',
value: 'my-id',
},
{
entity_id: 'my-id',
key: 'spec.k',
original_value: 'v',
value: 'v',
},
]),
);
},
);
});
plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts
+236
View File
@@ -0,0 +1,236 @@
/*
* Copyright 2023 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 { ENTITY_STATUS_CATALOG_PROCESSING_TYPE } from '@backstage/catalog-client';
import {
ANNOTATION_EDIT_URL,
ANNOTATION_VIEW_URL,
EntityRelation,
} from '@backstage/catalog-model';
import { AlphaEntity, EntityStatusItem } from '@backstage/catalog-model/alpha';
import { SerializedError } from '@backstage/errors';
import { Knex } from 'knex';
import { v4 as uuid } from 'uuid';
import { Logger } from 'winston';
import { StitchingStrategy } from '../../../stitching/types';
import {
DbFinalEntitiesRow,
DbRefreshStateRow,
DbSearchRow,
} from '../../tables';
import { buildEntitySearch } from './buildEntitySearch';
import { BATCH_SIZE, generateStableHash } from './util';
// See https://github.com/facebook/react/blob/f0cf832e1d0c8544c36aa8b310960885a11a847c/packages/react-dom-bindings/src/shared/sanitizeURL.js
const scriptProtocolPattern =
// eslint-disable-next-line no-control-regex
/^[\u0000-\u001F ]*j[\r\n\t]*a[\r\n\t]*v[\r\n\t]*a[\r\n\t]*s[\r\n\t]*c[\r\n\t]*r[\r\n\t]*i[\r\n\t]*p[\r\n\t]*t[\r\n\t]*\:/i;
/**
* Performs the act of stitching - to take all of the various outputs from the
* ingestion process, and stitching them together into the final entity JSON
* shape.
*/
export async function performStitching(options: {
knex: Knex | Knex.Transaction;
logger: Logger;
strategy: StitchingStrategy;
entityRef: string;
stitchTicket?: string;
}): Promise<'changed' | 'unchanged' | 'abandoned'> {
const { knex, logger, entityRef } = options;
const stitchTicket = options.stitchTicket ?? uuid();
const entityResult = await knex<DbRefreshStateRow>('refresh_state')
.where({ entity_ref: entityRef })
.limit(1)
.select('entity_id');
if (!entityResult.length) {
// Entity does no exist in refresh state table, no stitching required.
return 'abandoned';
}
// Insert stitching ticket that will be compared before inserting the final entity.
await knex<DbFinalEntitiesRow>('final_entities')
.insert({
entity_id: entityResult[0].entity_id,
hash: '',
stitch_ticket: stitchTicket,
})
.onConflict('entity_id')
.merge(['stitch_ticket']);
// Selecting from refresh_state and final_entities should yield exactly
// one row (except in abnormal cases where the stitch was invoked for
// something that didn't exist at all, in which case it's zero rows).
// The join with the temporary incoming_references still gives one row.
const [processedResult, relationsResult] = await Promise.all([
knex
.with('incoming_references', function incomingReferences(builder) {
return builder
.from('refresh_state_references')
.where({ target_entity_ref: entityRef })
.count({ count: '*' });
})
.select({
entityId: 'refresh_state.entity_id',
processedEntity: 'refresh_state.processed_entity',
errors: 'refresh_state.errors',
incomingReferenceCount: 'incoming_references.count',
previousHash: 'final_entities.hash',
})
.from('refresh_state')
.where({ 'refresh_state.entity_ref': entityRef })
.crossJoin(knex.raw('incoming_references'))
.leftOuterJoin('final_entities', {
'final_entities.entity_id': 'refresh_state.entity_id',
}),
knex
.distinct({
relationType: 'type',
relationTarget: 'target_entity_ref',
})
.from('relations')
.where({ source_entity_ref: entityRef })
.orderBy('relationType', 'asc')
.orderBy('relationTarget', 'asc'),
]);
// If there were no rows returned, it would mean that there was no
// matching row even in the refresh_state. This can happen for example
// if we emit a relation to something that hasn't been ingested yet.
// It's safe to ignore this stitch attempt in that case.
if (!processedResult.length) {
logger.debug(
`Unable to stitch ${entityRef}, item does not exist in refresh state table`,
);
return 'abandoned';
}
const {
entityId,
processedEntity,
errors,
incomingReferenceCount,
previousHash,
} = processedResult[0];
// If there was no processed entity in place, the target hasn't been
// through the processing steps yet. It's safe to ignore this stitch
// attempt in that case, since another stitch will be triggered when
// that processing has finished.
if (!processedEntity) {
logger.debug(
`Unable to stitch ${entityRef}, the entity has not yet been processed`,
);
return 'abandoned';
}
// Grab the processed entity and stitch all of the relevant data into
// it
const entity = JSON.parse(processedEntity) as AlphaEntity;
const isOrphan = Number(incomingReferenceCount) === 0;
let statusItems: EntityStatusItem[] = [];
if (isOrphan) {
logger.debug(`${entityRef} is an orphan`);
entity.metadata.annotations = {
...entity.metadata.annotations,
['backstage.io/orphan']: 'true',
};
}
if (errors) {
const parsedErrors = JSON.parse(errors) as SerializedError[];
if (Array.isArray(parsedErrors) && parsedErrors.length) {
statusItems = parsedErrors.map(e => ({
type: ENTITY_STATUS_CATALOG_PROCESSING_TYPE,
level: 'error',
message: `${e.name}: ${e.message}`,
error: e,
}));
}
}
// We opt to do this check here as we otherwise can't guarantee that it will be run after all processors
for (const annotation of [ANNOTATION_VIEW_URL, ANNOTATION_EDIT_URL]) {
const value = entity.metadata.annotations?.[annotation];
if (typeof value === 'string' && scriptProtocolPattern.test(value)) {
entity.metadata.annotations![annotation] =
'https://backstage.io/annotation-rejected-for-security-reasons';
}
}
// TODO: entityRef is lower case and should be uppercase in the final
// result
entity.relations = relationsResult
.filter(row => row.relationType /* exclude null row, if relevant */)
.map<EntityRelation>(row => ({
type: row.relationType!,
targetRef: row.relationTarget!,
}));
if (statusItems.length) {
entity.status = {
...entity.status,
items: [...(entity.status?.items ?? []), ...statusItems],
};
}
// If the output entity was actually not changed, just abort
const hash = generateStableHash(entity);
if (hash === previousHash) {
logger.debug(`Skipped stitching of ${entityRef}, no changes`);
return 'unchanged';
}
entity.metadata.uid = entityId;
if (!entity.metadata.etag) {
// If the original data source did not have its own etag handling,
// use the hash as a good-quality etag
entity.metadata.etag = hash;
}
// This may throw if the entity is invalid, so we call it before
// the final_entities write, even though we may end up not needing
// to write the search index.
const searchEntries = buildEntitySearch(entityId, entity);
const amountOfRowsChanged = await knex<DbFinalEntitiesRow>('final_entities')
.update({
final_entity: JSON.stringify(entity),
hash,
last_updated_at: knex.fn.now(),
})
.where('entity_id', entityId)
.where('stitch_ticket', stitchTicket)
.onConflict('entity_id')
.merge(['final_entity', 'hash', 'last_updated_at']);
if (amountOfRowsChanged === 0) {
logger.debug(`Entity ${entityRef} is already stitched, skipping write.`);
return 'abandoned';
}
// TODO(freben): Search will probably need a similar safeguard against
// race conditions like the final_entities ticket handling above.
// Otherwise, it can be the case that:
// A writes the entity ->
// B writes the entity ->
// B writes search ->
// A writes search
await knex<DbSearchRow>('search').where({ entity_id: entityId }).delete();
await knex.batchInsert('search', searchEntries, BATCH_SIZE);
return 'changed';
}
plugins/catalog-backend/src/stitching/util.ts → plugins/catalog-backend/src/database/operations/stitcher/util.ts
View File
plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts
+24 -12
View File
@@ -16,7 +16,7 @@
import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils';
import { Knex } from 'knex';
import * as uuid from 'uuid';
import { StitchingStrategy } from '../../../stitching/types';
import { applyDatabaseMigrations } from '../../migrations';
import {
DbFinalEntitiesRow,
@@ -39,14 +39,14 @@ describe('deleteOrphanedEntities', () => {
return knex;
}
async function run(knex: Knex): Promise<number> {
async function run(knex: Knex, strategy: StitchingStrategy): Promise<number> {
let result: number;
await knex.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 deleteOrphanedEntities({ tx });
result = await deleteOrphanedEntities({ knex: tx, strategy });
},
{
// If we explicitly trigger a rollback, don't fail.
@@ -58,9 +58,8 @@ describe('deleteOrphanedEntities', () => {
async function insertEntity(knex: Knex, ...entityRefs: string[]) {
for (const ref of entityRefs) {
const entityId = uuid.v4();
await knex<DbRefreshStateRow>('refresh_state').insert({
entity_id: entityId,
entity_id: `id-${ref}`,
entity_ref: ref,
unprocessed_entity: '{}',
processed_entity: '{}',
@@ -70,7 +69,7 @@ describe('deleteOrphanedEntities', () => {
result_hash: 'original',
});
await knex<DbFinalEntitiesRow>('final_entities').insert({
entity_id: entityId,
entity_id: `id-${ref}`,
hash: 'original',
stitch_ticket: '',
});
@@ -120,7 +119,7 @@ describe('deleteOrphanedEntities', () => {
}
it.each(databases.eachSupportedId())(
'works for some mixed paths, %p',
'works for some mixed paths in immediate mode, %p',
async databaseId => {
/*
In this graph, edges represent refresh state references, not entity relations:
@@ -176,18 +175,31 @@ describe('deleteOrphanedEntities', () => {
await insertRelation(knex, 'E1', 'E2');
await insertRelation(knex, 'E2', 'E3');
await insertRelation(knex, 'E10', 'E6');
await expect(run(knex)).resolves.toEqual(5);
await insertRelation(knex, 'E7', 'E6');
await expect(run(knex, { mode: 'immediate' })).resolves.toEqual(5);
await expect(refreshState(knex)).resolves.toEqual([
{ entity_ref: 'E1', result_hash: 'original' },
{ entity_ref: 'E2', result_hash: 'orphan-relation-deleted' },
{ entity_ref: 'E7', result_hash: 'original' },
{
entity_ref: 'E2',
result_hash: 'force-stitching',
},
{
entity_ref: 'E7',
result_hash: 'force-stitching',
},
{ entity_ref: 'E8', result_hash: 'original' },
{ entity_ref: 'E9', result_hash: 'original' },
]);
await expect(finalEntities(knex)).resolves.toEqual([
{ entity_ref: 'E1', hash: 'original' },
{ entity_ref: 'E2', hash: 'orphan-relation-deleted' },
{ entity_ref: 'E7', hash: 'original' },
{
entity_ref: 'E2',
hash: 'force-stitching',
},
{
entity_ref: 'E7',
hash: 'force-stitching',
},
{ entity_ref: 'E8', hash: 'original' },
{ entity_ref: 'E9', hash: 'original' },
]);
plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.ts
+14 -20
View File
@@ -16,7 +16,9 @@
import { Knex } from 'knex';
import uniq from 'lodash/uniq';
import { DbFinalEntitiesRow, DbRefreshStateRow } from '../../tables';
import { StitchingStrategy } from '../../../stitching/types';
import { DbRefreshStateRow } from '../../tables';
import { markForStitching } from '../stitcher/markForStitching';
/**
* Finds and deletes all orphaned entities, i.e. entities that do not have any
@@ -24,15 +26,16 @@ import { DbFinalEntitiesRow, DbRefreshStateRow } from '../../tables';
* that would otherwise become orphaned.
*/
export async function deleteOrphanedEntities(options: {
tx: Knex.Transaction | Knex;
knex: Knex.Transaction | Knex;
strategy: StitchingStrategy;
}): Promise<number> {
const { tx } = options;
const { knex, strategy } = options;
let total = 0;
// Limit iterations for sanity
for (let i = 0; i < 100; ++i) {
const candidates = await tx
const candidates = await knex
.with('orphans', ['entity_id', 'entity_ref'], orphans =>
orphans
.from('refresh_state')
@@ -72,26 +75,17 @@ export async function deleteOrphanedEntities(options: {
total += orphanIds.length;
// Delete the orphans themselves
await tx
await knex
.table<DbRefreshStateRow>('refresh_state')
.delete()
.whereIn('entity_id', orphanIds);
// Mark all of things that the orphans had relations to for processing and
// stitching
await tx
.table<DbFinalEntitiesRow>('final_entities')
.update({
hash: 'orphan-relation-deleted',
})
.whereIn('entity_id', orphanRelationIds);
await tx
.table<DbRefreshStateRow>('refresh_state')
.update({
result_hash: 'orphan-relation-deleted',
next_update_at: tx.fn.now(),
})
.whereIn('entity_id', orphanRelationIds);
// Mark all of the things that the orphans had relations to for stitching
await markForStitching({
knex,
strategy,
entityIds: orphanRelationIds,
});
}
return total;
plugins/catalog-backend/src/database/tables.ts
+72 -1
View File
@@ -29,17 +29,88 @@ export type DbLocationsRow = {
target: string;
};
/**
* Represents the refresh_state table.
*
* @remarks
*
* Every unique entity ref emitted by a provider or a parent entity becomes a
* row in this table, even before processing has started on it. The actual final
* data, after processing and stitching completes, is instead in the
* final_entities table.
*
* Datetime columns are both string and Date, because different database engines
* return them in different forms on the client side.
*/
export type DbRefreshStateRow = {
/**
* The unique ID of the entity. This is different to the entity ref, in that
* it gets regenerated randomly each time a row is added to the table, no
* matter what the original entity data was.
*/
entity_id: string;
/**
* The entity string ref (on lowercase kind:namespace/name form)
*/
entity_ref: string;
/**
* The JSON of the raw entity, as it was received from the entity provider.
*/
unprocessed_entity: string;
/**
* A stable hash of the unprocessed entity, used to detect changed/unchanged
* data for a given entity over time.
*/
unprocessed_hash?: string;
/**
* The JSON of the processed entity (if processing has run yet on it).
*/
processed_entity?: string;
/**
* A stable hash of the processed entity AND all other emitted things during
* processing, such as relations.
*/
result_hash?: string;
/**
* Per-entity cached data on JSON form. This is read and written by processors
* who wish to leverage this feature.
*/
cache?: string;
/**
* The next point in time that this entity is due for processing. This
* continuously gets moved forward as items are picked up for processing.
*/
next_update_at: string | Date;
last_discovery_at: string | Date; // remove?
/**
* The last time that this entity was emitted by somebody (the entity provider
* or a parent entity).
*
* @remarks
*
* Don't rely on this column more than at most as being loosely informative.
* Its semantics aren't fully settled yet.
*/
last_discovery_at: string | Date;
/**
* A JSON serialized array of errors (if any) encountered during processing.
*/
errors?: string;
/**
* A conflict detection/resolution key for the entity.
*
* @remarks
*
* The exact value semantics differs, but may for example be a URL pointing to
* where the entity was sourced from. If a "competing" provider or parent
* entity tries to emit an entity that has the same entity ref but a different
* location key, a conflict is detected (you aren't allowed to "trample" over
* a previously existing entity).
*
* Some providers may choose to emit entities with no location key set at all.
* This is a signal that it's only loosely claimed, and that any other
* competing provider/parent is allowed to overwrite and claim it as theirs
* instead.
*/
location_key?: string;
};
plugins/catalog-backend/src/database/types.ts
-2
View File
@@ -151,8 +151,6 @@ export interface ProcessingDatabase {
txOpaque: Transaction,
options: ListParentsOptions,
): Promise<ListParentsResult>;
deleteOrphanedEntities(txOpaque: Transaction): Promise<number>;
}
/**
plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts
+14 -6
View File
@@ -21,7 +21,7 @@ import waitForExpect from 'wait-for-expect';
import { DefaultProcessingDatabase } from '../database/DefaultProcessingDatabase';
import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine';
import { CatalogProcessingOrchestrator } from './types';
import { Stitcher } from '../stitching/Stitcher';
import { Stitcher } from '../stitching/types';
import { ConfigReader } from '@backstage/config';
describe('DefaultCatalogProcessingEngine', () => {
@@ -66,6 +66,7 @@ describe('DefaultCatalogProcessingEngine', () => {
config: new ConfigReader({}),
logger: getVoidLogger(),
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
createHash: () => hash,
@@ -133,6 +134,7 @@ describe('DefaultCatalogProcessingEngine', () => {
config: new ConfigReader({}),
logger: getVoidLogger(),
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
createHash: () => hash,
@@ -216,6 +218,7 @@ describe('DefaultCatalogProcessingEngine', () => {
config: new ConfigReader({}),
logger: getVoidLogger(),
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
createHash: () => hash,
@@ -292,6 +295,7 @@ describe('DefaultCatalogProcessingEngine', () => {
config: new ConfigReader({}),
logger: getVoidLogger(),
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
createHash: () => hash,
@@ -350,6 +354,7 @@ describe('DefaultCatalogProcessingEngine', () => {
config: new ConfigReader({}),
logger: getVoidLogger(),
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
createHash: () => hash,
@@ -450,10 +455,10 @@ describe('DefaultCatalogProcessingEngine', () => {
await waitForExpect(() => {
expect(stitcher.stitch).toHaveBeenCalledTimes(2);
});
expect([...stitcher.stitch.mock.calls[0][0]]).toEqual(
expect([...stitcher.stitch.mock.calls[0][0].entityRefs!]).toEqual(
expect.arrayContaining(['k:ns/me', 'k:ns/other1', 'k:ns/other2']),
);
expect([...stitcher.stitch.mock.calls[1][0]]).toEqual(
expect([...stitcher.stitch.mock.calls[1][0].entityRefs!]).toEqual(
expect.arrayContaining(['k:ns/me', 'k:ns/other1', 'k:ns/other3']),
);
await engine.stop();
@@ -464,6 +469,7 @@ describe('DefaultCatalogProcessingEngine', () => {
config: new ConfigReader({}),
logger: getVoidLogger(),
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
createHash: () => hash,
@@ -535,7 +541,7 @@ describe('DefaultCatalogProcessingEngine', () => {
await waitForExpect(() => {
expect(stitcher.stitch).toHaveBeenCalledTimes(1);
});
expect([...stitcher.stitch.mock.calls[0][0]]).toEqual(
expect([...stitcher.stitch.mock.calls[0][0].entityRefs!]).toEqual(
expect.arrayContaining(['k:ns/me']),
);
await engine.stop();
@@ -546,6 +552,7 @@ describe('DefaultCatalogProcessingEngine', () => {
config: new ConfigReader({}),
logger: getVoidLogger(),
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
createHash: () => hash,
@@ -622,7 +629,7 @@ describe('DefaultCatalogProcessingEngine', () => {
await waitForExpect(() => {
expect(stitcher.stitch).toHaveBeenCalledTimes(1);
});
expect([...stitcher.stitch.mock.calls[0][0]]).toEqual(
expect([...stitcher.stitch.mock.calls[0][0].entityRefs!]).toEqual(
expect.arrayContaining(['k:ns/me', 'k:ns/other2']),
);
await engine.stop();
@@ -633,6 +640,7 @@ describe('DefaultCatalogProcessingEngine', () => {
config: new ConfigReader({}),
logger: getVoidLogger(),
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
createHash: () => hash,
@@ -699,7 +707,7 @@ describe('DefaultCatalogProcessingEngine', () => {
await waitForExpect(() => {
expect(stitcher.stitch).toHaveBeenCalledTimes(1);
});
expect([...stitcher.stitch.mock.calls[0][0]]).toEqual(
expect([...stitcher.stitch.mock.calls[0][0].entityRefs!]).toEqual(
expect.arrayContaining(['k:ns/me', 'k:ns/other2']),
);
await engine.stop();
plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts
+32 -31
View File
@@ -22,16 +22,13 @@ import {
import { assertError, serializeError, stringifyError } from '@backstage/errors';
import { Hash } from 'crypto';
import stableStringify from 'fast-json-stable-stringify';
import { Knex } from 'knex';
import { Logger } from 'winston';
import { metrics, trace } from '@opentelemetry/api';
import { ProcessingDatabase, RefreshStateItem } from '../database/types';
import { createCounterMetric, createSummaryMetric } from '../util/metrics';
import {
CatalogProcessingEngine,
CatalogProcessingOrchestrator,
EntityProcessingResult,
} from './types';
import { Stitcher } from '../stitching/Stitcher';
import { CatalogProcessingOrchestrator, EntityProcessingResult } from './types';
import { Stitcher } from '../stitching/types';
import { startTaskPipeline } from './TaskPipeline';
import { PluginTaskScheduler } from '@backstage/backend-tasks';
import { Config } from '@backstage/config';
@@ -40,6 +37,7 @@ import {
TRACER_ID,
withActiveSpan,
} from '../util/opentelemetry';
import { deleteOrphanedEntities } from '../database/operations/util/deleteOrphanedEntities';
const CACHE_TTL = 5;
@@ -47,10 +45,17 @@ const tracer = trace.getTracer(TRACER_ID);
export type ProgressTracker = ReturnType<typeof progressTracker>;
export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
// NOTE(freben): Perhaps surprisingly, this class does not implement the
// CatalogProcessingEngine type. That type is externally visible and its name is
// the way it is for historic reasons. This class has no particular reason to
// implement that precise interface; nowadays there are several different
// engines "hiding" behind the CatalogProcessingEngine interface, of which this
// is just one.
export class DefaultCatalogProcessingEngine {
private readonly config: Config;
private readonly scheduler?: PluginTaskScheduler;
private readonly logger: Logger;
private readonly knex: Knex;
private readonly processingDatabase: ProcessingDatabase;
private readonly orchestrator: CatalogProcessingOrchestrator;
private readonly stitcher: Stitcher;
@@ -69,6 +74,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
config: Config;
scheduler?: PluginTaskScheduler;
logger: Logger;
knex: Knex;
processingDatabase: ProcessingDatabase;
orchestrator: CatalogProcessingOrchestrator;
stitcher: Stitcher;
@@ -84,6 +90,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
this.config = options.config;
this.scheduler = options.scheduler;
this.logger = options.logger;
this.knex = options.knex;
this.processingDatabase = options.processingDatabase;
this.orchestrator = options.orchestrator;
this.stitcher = options.stitcher;
@@ -255,9 +262,11 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
resultHash,
});
});
await this.stitcher.stitch(
new Set([stringifyEntityRef(unprocessedEntity)]),
);
await this.stitcher.stitch({
entityRefs: [stringifyEntityRef(unprocessedEntity)],
});
track.markSuccessfulWithErrors();
return;
}
@@ -305,9 +314,11 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
}
});
await this.stitcher.stitch(setOfThingsToStitch);
await this.stitcher.stitch({
entityRefs: setOfThingsToStitch,
});
track.markSuccessfulWithChanges(setOfThingsToStitch.size);
track.markSuccessfulWithChanges();
} catch (error) {
assertError(error);
track.markFailed(error);
@@ -318,20 +329,21 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
}
private startOrphanCleanup(): () => void {
const strategy =
const orphanStrategy =
this.config.getOptionalString('catalog.orphanStrategy') ?? 'keep';
if (strategy !== 'delete') {
if (orphanStrategy !== 'delete') {
return () => {};
}
const runOnce = async () => {
try {
await this.processingDatabase.transaction(async tx => {
const n = await this.processingDatabase.deleteOrphanedEntities(tx);
if (n > 0) {
this.logger.info(`Deleted ${n} orphaned entities`);
}
const n = await deleteOrphanedEntities({
knex: this.knex,
strategy: { mode: 'immediate' },
});
if (n > 0) {
this.logger.info(`Deleted ${n} orphaned entities`);
}
} catch (error) {
this.logger.warn(`Failed to delete orphaned entities`, error);
}
@@ -363,10 +375,6 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
// Helps wrap the timing and logging behaviors
function progressTracker() {
// prom-client metrics are deprecated in favour of OpenTelemetry metrics.
const promStitchedEntities = createCounterMetric({
name: 'catalog_stitched_entities_count',
help: 'Amount of entities stitched. DEPRECATED, use OpenTelemetry metrics instead',
});
const promProcessedEntities = createCounterMetric({
name: 'catalog_processed_entities_count',
help: 'Amount of entities processed, DEPRECATED, use OpenTelemetry metrics instead',
@@ -388,11 +396,6 @@ function progressTracker() {
});
const meter = metrics.getMeter('default');
const stitchedEntities = meter.createCounter(
'catalog.stitched.entities.count',
{ description: 'Amount of entities stitched' },
);
const processedEntities = meter.createCounter(
'catalog.processed.entities.count',
{ description: 'Amount of entities processed' },
@@ -464,13 +467,11 @@ function progressTracker() {
processedEntities.add(1, { result: 'errors' });
}
function markSuccessfulWithChanges(stitchedCount: number) {
function markSuccessfulWithChanges() {
endOverallTimer({ result: 'changed' });
promStitchedEntities.inc(stitchedCount);
promProcessedEntities.inc({ result: 'changed' }, 1);
processingDuration.record(endTime(), { result: 'changed' });
stitchedEntities.add(stitchedCount);
processedEntities.add(1, { result: 'changed' });
}
plugins/catalog-backend/src/processing/types.ts
+2 -2
View File
@@ -64,8 +64,8 @@ export interface CatalogProcessingOrchestrator {
}
/**
* Represents the engine that drives the processing loops. Some backend
* instances may choose to not call start, if they focus only on API
* Represents the engine that drives the processing and stitching loops. Some
* backend instances may choose to not call start, if they focus only on API
* interactions.
*
* @public
plugins/catalog-backend/src/service/CatalogBuilder.ts
+17 -3
View File
@@ -65,7 +65,7 @@ import { DefaultCatalogProcessingEngine } from '../processing/DefaultCatalogProc
import { DefaultLocationService } from './DefaultLocationService';
import { DefaultEntitiesCatalog } from './DefaultEntitiesCatalog';
import { DefaultCatalogProcessingOrchestrator } from '../processing/DefaultCatalogProcessingOrchestrator';
import { Stitcher } from '../stitching/Stitcher';
import { DefaultStitcher } from '../stitching/DefaultStitcher';
import {
createRandomProcessingInterval,
ProcessingIntervalFunction,
@@ -449,6 +449,11 @@ export class CatalogBuilder {
await applyDatabaseMigrations(dbClient);
}
const stitcher = DefaultStitcher.fromConfig(config, {
knex: dbClient,
logger,
});
const processingDatabase = new DefaultProcessingDatabase({
database: dbClient,
logger,
@@ -474,7 +479,6 @@ export class CatalogBuilder {
policy,
legacySingleProcessorValidation: this.legacySingleProcessorValidation,
});
const stitcher = new Stitcher(dbClient, logger);
const unauthorizedEntitiesCatalog = new DefaultEntitiesCatalog({
database: dbClient,
logger,
@@ -535,6 +539,7 @@ export class CatalogBuilder {
config,
scheduler,
logger,
knex: dbClient,
processingDatabase,
orchestrator,
stitcher,
@@ -572,7 +577,16 @@ export class CatalogBuilder {
await connectEntityProviders(providerDatabase, entityProviders);
return {
processingEngine,
processingEngine: {
async start() {
await processingEngine.start();
await stitcher.start();
},
async stop() {
await processingEngine.stop();
await stitcher.stop();
},
},
router,
};
}
plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts
+5 -5
View File
@@ -30,10 +30,10 @@ import {
DbRefreshStateRow,
DbSearchRow,
} from '../database/tables';
import { Stitcher } from '../stitching/Stitcher';
import { buildEntitySearch } from '../stitching/buildEntitySearch';
import { Stitcher } from '../stitching/types';
import { DefaultEntitiesCatalog } from './DefaultEntitiesCatalog';
import { EntitiesRequest } from '../catalog/types';
import { buildEntitySearch } from '../database/operations/stitcher/buildEntitySearch';
jest.setTimeout(60_000);
@@ -1684,9 +1684,9 @@ describe('DefaultEntitiesCatalog', () => {
{ entity_ref: 'k:default/unrelated1', result_hash: 'not-changed' },
{ entity_ref: 'k:default/unrelated2', result_hash: 'not-changed' },
]);
expect(stitch).toHaveBeenCalledWith(
new Set(['k:default/unrelated1', 'k:default/unrelated2']),
);
expect(stitch).toHaveBeenCalledWith({
entityRefs: new Set(['k:default/unrelated1', 'k:default/unrelated2']),
});
},
);
});
plugins/catalog-backend/src/service/DefaultEntitiesCatalog.ts
+4 -3
View File
@@ -49,8 +49,7 @@ import {
DbRelationsRow,
DbSearchRow,
} from '../database/tables';
import { Stitcher } from '../stitching/Stitcher';
import { Stitcher } from '../stitching/types';
import {
isQueryEntitiesCursorRequest,
@@ -606,7 +605,9 @@ export class DefaultEntitiesCatalog implements EntitiesCatalog {
.where('entity_id', uid)
.delete();
await this.stitcher.stitch(new Set(relationPeers.map(p => p.ref)));
await this.stitcher.stitch({
entityRefs: new Set(relationPeers.map(p => p.ref)),
});
}
async entityAncestry(rootRef: string): Promise<EntityAncestryResponse> {
plugins/catalog-backend/src/service/DefaultRefreshService.test.ts
+7 -2
View File
@@ -31,9 +31,9 @@ import {
import { ProcessingDatabase } from '../database/types';
import { DefaultCatalogProcessingEngine } from '../processing/DefaultCatalogProcessingEngine';
import { EntityProcessingRequest } from '../processing/types';
import { Stitcher } from '../stitching/Stitcher';
import { DefaultRefreshService } from './DefaultRefreshService';
import { ConfigReader } from '@backstage/config';
import { DefaultStitcher } from '../stitching/DefaultStitcher';
jest.setTimeout(60_000);
@@ -109,10 +109,16 @@ describe('DefaultRefreshService', () => {
}
}
const stitcher = DefaultStitcher.fromConfig(new ConfigReader({}), {
knex,
logger: defaultLogger,
});
const engine = new DefaultCatalogProcessingEngine({
config: new ConfigReader({}),
logger: defaultLogger,
processingDatabase: db,
knex: knex,
stitcher: stitcher,
orchestrator: {
async process(request: EntityProcessingRequest) {
const entityRef = stringifyEntityRef(request.entity);
@@ -151,7 +157,6 @@ describe('DefaultRefreshService', () => {
};
},
},
stitcher: new Stitcher(knex, defaultLogger),
createHash: () => createHash('sha1'),
pollingIntervalMs: 50,
});
plugins/catalog-backend/src/stitching/Stitcher.test.ts → plugins/catalog-backend/src/stitching/DefaultStitcher.test.ts
+9 -5
View File
@@ -25,7 +25,7 @@ import {
DbRelationsRow,
DbSearchRow,
} from '../database/tables';
import { Stitcher } from './Stitcher';
import { DefaultStitcher } from './DefaultStitcher';
jest.setTimeout(60_000);
@@ -41,7 +41,11 @@ describe('Stitcher', () => {
const db = await databases.init(databaseId);
await applyDatabaseMigrations(db);
const stitcher = new Stitcher(db, logger);
const stitcher = new DefaultStitcher({
knex: db,
logger,
strategy: { mode: 'immediate' },
});
let entities: DbFinalEntitiesRow[];
let entity: Entity;
@@ -85,7 +89,7 @@ describe('Stitcher', () => {
},
]);
await stitcher.stitch(new Set(['k:ns/n']));
await stitcher.stitch({ entityRefs: ['k:ns/n'] });
entities = await db<DbFinalEntitiesRow>('final_entities');
@@ -165,7 +169,7 @@ describe('Stitcher', () => {
);
// Re-stitch without any changes
await stitcher.stitch(new Set(['k:ns/n']));
await stitcher.stitch({ entityRefs: ['k:ns/n'] });
entities = await db<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
@@ -183,7 +187,7 @@ describe('Stitcher', () => {
},
]);
await stitcher.stitch(new Set(['k:ns/n']));
await stitcher.stitch({ entityRefs: ['k:ns/n'] });
entities = await db<DbFinalEntitiesRow>('final_entities');
plugins/catalog-backend/src/stitching/DefaultStitcher.ts
+124
View File
@@ -0,0 +1,124 @@
/*
* 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 { Config } from '@backstage/config';
import { Knex } from 'knex';
import splitToChunks from 'lodash/chunk';
import { DateTime } from 'luxon';
import { Logger } from 'winston';
import { performStitching } from '../database/operations/stitcher/performStitching';
import { DbRefreshStateRow } from '../database/tables';
import { progressTracker } from './progressTracker';
import { Stitcher, StitchingStrategy } from './types';
type StitchProgressTracker = ReturnType<typeof progressTracker>;
/**
* Performs the act of stitching - to take all of the various outputs from the
* ingestion process, and stitching them together into the final entity JSON
* shape.
*/
export class DefaultStitcher implements Stitcher {
private readonly knex: Knex;
private readonly logger: Logger;
private readonly strategy: StitchingStrategy;
private readonly tracker: StitchProgressTracker;
static fromConfig(
_config: Config,
options: {
knex: Knex;
logger: Logger;
},
): DefaultStitcher {
return new DefaultStitcher({
knex: options.knex,
logger: options.logger,
strategy: { mode: 'immediate' },
});
}
constructor(options: {
knex: Knex;
logger: Logger;
strategy: StitchingStrategy;
}) {
this.knex = options.knex;
this.logger = options.logger;
this.strategy = options.strategy;
this.tracker = progressTracker(options.knex, options.logger);
}
async stitch(options: {
entityRefs?: Iterable<string>;
entityIds?: Iterable<string>;
}) {
const { entityRefs, entityIds } = options;
if (entityRefs) {
for (const entityRef of entityRefs) {
await this.#stitchOne({ entityRef });
}
}
if (entityIds) {
const chunks = splitToChunks(
Array.isArray(entityIds) ? entityIds : [...entityIds],
100,
);
for (const chunk of chunks) {
const rows = await this.knex<DbRefreshStateRow>('refresh_state')
.select('entity_ref')
.whereIn('entity_id', chunk);
for (const row of rows) {
await this.#stitchOne({ entityRef: row.entity_ref });
}
}
}
}
async start() {
// Only called immediately for now
}
async stop() {
// Only called immediately for now
}
async #stitchOne(options: {
entityRef: string;
stitchTicket?: string;
stitchRequestedAt?: DateTime;
}) {
const track = this.tracker.stitchStart({
entityRef: options.entityRef,
stitchRequestedAt: options.stitchRequestedAt,
});
try {
const result = await performStitching({
knex: this.knex,
logger: this.logger,
strategy: this.strategy,
entityRef: options.entityRef,
stitchTicket: options.stitchTicket,
});
track.markComplete(result);
} catch (error) {
track.markFailed(error);
}
}
}
plugins/catalog-backend/src/stitching/Stitcher.ts
-250
View File
@@ -1,250 +0,0 @@
/*
* 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 { ENTITY_STATUS_CATALOG_PROCESSING_TYPE } from '@backstage/catalog-client';
import { AlphaEntity, EntityStatusItem } from '@backstage/catalog-model/alpha';
import {
ANNOTATION_EDIT_URL,
ANNOTATION_VIEW_URL,
EntityRelation,
} from '@backstage/catalog-model';
import { SerializedError, stringifyError } from '@backstage/errors';
import { Knex } from 'knex';
import { v4 as uuid } from 'uuid';
import { Logger } from 'winston';
import {
DbFinalEntitiesRow,
DbRefreshStateRow,
DbSearchRow,
} from '../database/tables';
import { buildEntitySearch } from './buildEntitySearch';
import { BATCH_SIZE, generateStableHash } from './util';
// See https://github.com/facebook/react/blob/f0cf832e1d0c8544c36aa8b310960885a11a847c/packages/react-dom-bindings/src/shared/sanitizeURL.js
const scriptProtocolPattern =
// eslint-disable-next-line no-control-regex
/^[\u0000-\u001F ]*j[\r\n\t]*a[\r\n\t]*v[\r\n\t]*a[\r\n\t]*s[\r\n\t]*c[\r\n\t]*r[\r\n\t]*i[\r\n\t]*p[\r\n\t]*t[\r\n\t]*\:/i;
/**
* Performs the act of stitching - to take all of the various outputs from the
* ingestion process, and stitching them together into the final entity JSON
* shape.
*/
export class Stitcher {
constructor(
private readonly database: Knex,
private readonly logger: Logger,
) {}
async stitch(entityRefs: Set<string>) {
for (const entityRef of entityRefs) {
try {
await this.stitchOne(entityRef);
} catch (error) {
this.logger.error(
`Failed to stitch ${entityRef}, ${stringifyError(error)}`,
);
}
}
}
private async stitchOne(entityRef: string): Promise<void> {
const entityResult = await this.database<DbRefreshStateRow>('refresh_state')
.where({ entity_ref: entityRef })
.limit(1)
.select('entity_id');
if (!entityResult.length) {
// Entity does no exist in refresh state table, no stitching required.
return;
}
// Insert stitching ticket that will be compared before inserting the final entity.
const ticket = uuid();
await this.database<DbFinalEntitiesRow>('final_entities')
.insert({
entity_id: entityResult[0].entity_id,
hash: '',
stitch_ticket: ticket,
})
.onConflict('entity_id')
.merge(['stitch_ticket']);
// Selecting from refresh_state and final_entities should yield exactly
// one row (except in abnormal cases where the stitch was invoked for
// something that didn't exist at all, in which case it's zero rows).
// The join with the temporary incoming_references still gives one row.
const [processedResult, relationsResult] = await Promise.all([
this.database
.with('incoming_references', function incomingReferences(builder) {
return builder
.from('refresh_state_references')
.where({ target_entity_ref: entityRef })
.count({ count: '*' });
})
.select({
entityId: 'refresh_state.entity_id',
processedEntity: 'refresh_state.processed_entity',
errors: 'refresh_state.errors',
incomingReferenceCount: 'incoming_references.count',
previousHash: 'final_entities.hash',
})
.from('refresh_state')
.where({ 'refresh_state.entity_ref': entityRef })
.crossJoin(this.database.raw('incoming_references'))
.leftOuterJoin('final_entities', {
'final_entities.entity_id': 'refresh_state.entity_id',
}),
this.database
.distinct({
relationType: 'type',
relationTarget: 'target_entity_ref',
})
.from('relations')
.where({ source_entity_ref: entityRef })
.orderBy('relationType', 'asc')
.orderBy('relationTarget', 'asc'),
]);
// If there were no rows returned, it would mean that there was no
// matching row even in the refresh_state. This can happen for example
// if we emit a relation to something that hasn't been ingested yet.
// It's safe to ignore this stitch attempt in that case.
if (!processedResult.length) {
this.logger.error(
`Unable to stitch ${entityRef}, item does not exist in refresh state table`,
);
return;
}
const {
entityId,
processedEntity,
errors,
incomingReferenceCount,
previousHash,
} = processedResult[0];
// If there was no processed entity in place, the target hasn't been
// through the processing steps yet. It's safe to ignore this stitch
// attempt in that case, since another stitch will be triggered when
// that processing has finished.
if (!processedEntity) {
this.logger.debug(
`Unable to stitch ${entityRef}, the entity has not yet been processed`,
);
return;
}
// Grab the processed entity and stitch all of the relevant data into
// it
const entity = JSON.parse(processedEntity) as AlphaEntity;
const isOrphan = Number(incomingReferenceCount) === 0;
let statusItems: EntityStatusItem[] = [];
if (isOrphan) {
this.logger.debug(`${entityRef} is an orphan`);
entity.metadata.annotations = {
...entity.metadata.annotations,
['backstage.io/orphan']: 'true',
};
}
if (errors) {
const parsedErrors = JSON.parse(errors) as SerializedError[];
if (Array.isArray(parsedErrors) && parsedErrors.length) {
statusItems = parsedErrors.map(e => ({
type: ENTITY_STATUS_CATALOG_PROCESSING_TYPE,
level: 'error',
message: `${e.name}: ${e.message}`,
error: e,
}));
}
}
// We opt to do this check here as we otherwise can't guarantee that it will be run after all processors
for (const annotation of [ANNOTATION_VIEW_URL, ANNOTATION_EDIT_URL]) {
const value = entity.metadata.annotations?.[annotation];
if (typeof value === 'string' && scriptProtocolPattern.test(value)) {
entity.metadata.annotations![annotation] =
'https://backstage.io/annotation-rejected-for-security-reasons';
}
}
// TODO: entityRef is lower case and should be uppercase in the final
// result
entity.relations = relationsResult
.filter(row => row.relationType /* exclude null row, if relevant */)
.map<EntityRelation>(row => ({
type: row.relationType!,
targetRef: row.relationTarget!,
}));
if (statusItems.length) {
entity.status = {
...entity.status,
items: [...(entity.status?.items ?? []), ...statusItems],
};
}
// If the output entity was actually not changed, just abort
const hash = generateStableHash(entity);
if (hash === previousHash) {
this.logger.debug(`Skipped stitching of ${entityRef}, no changes`);
return;
}
entity.metadata.uid = entityId;
if (!entity.metadata.etag) {
// If the original data source did not have its own etag handling,
// use the hash as a good-quality etag
entity.metadata.etag = hash;
}
// This may throw if the entity is invalid, so we call it before
// the final_entities write, even though we may end up not needing
// to write the search index.
const searchEntries = buildEntitySearch(entityId, entity);
const amountOfRowsChanged = await this.database<DbFinalEntitiesRow>(
'final_entities',
)
.update({
final_entity: JSON.stringify(entity),
hash,
last_updated_at: this.database.fn.now(),
})
.where('entity_id', entityId)
.where('stitch_ticket', ticket)
.onConflict('entity_id')
.merge(['final_entity', 'hash', 'last_updated_at']);
if (amountOfRowsChanged === 0) {
this.logger.debug(
`Entity ${entityRef} is already processed, skipping write.`,
);
return;
}
// TODO(freben): Search will probably need a similar safeguard against
// race conditions like the final_entities ticket handling above.
// Otherwise, it can be the case that:
// A writes the entity ->
// B writes the entity ->
// B writes search ->
// A writes search
await this.database<DbSearchRow>('search')
.where({ entity_id: entityId })
.delete();
await this.database.batchInsert('search', searchEntries, BATCH_SIZE);
}
}
plugins/catalog-backend/src/stitching/progressTracker.ts
+84
View File
@@ -0,0 +1,84 @@
/*
* Copyright 2023 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 { stringifyError } from '@backstage/errors';
import { metrics } from '@opentelemetry/api';
import { Knex } from 'knex';
import { DateTime } from 'luxon';
import { Logger } from 'winston';
import { createCounterMetric } from '../util/metrics';
// Helps wrap the timing and logging behaviors
export function progressTracker(_knex: Knex, logger: Logger) {
// prom-client metrics are deprecated in favour of OpenTelemetry metrics.
const promStitchedEntities = createCounterMetric({
name: 'catalog_stitched_entities_count',
help: 'Amount of entities stitched. DEPRECATED, use OpenTelemetry metrics instead',
});
const meter = metrics.getMeter('default');
const stitchedEntities = meter.createCounter(
'catalog.stitched.entities.count',
{
description: 'Amount of entities stitched',
},
);
const stitchingDuration = meter.createHistogram(
'catalog.stitching.duration',
{
description: 'Time spent executing the full stitching flow',
unit: 'seconds',
},
);
function stitchStart(item: {
entityRef: string;
stitchRequestedAt?: DateTime;
}) {
logger.debug(`Stitching ${item.entityRef}`);
const startTime = process.hrtime();
function endTime() {
const delta = process.hrtime(startTime);
return delta[0] + delta[1] / 1e9;
}
function markComplete(result: string) {
promStitchedEntities.inc(1);
stitchedEntities.add(1, { result });
stitchingDuration.record(endTime(), { result });
}
function markFailed(error: Error) {
promStitchedEntities.inc(1);
stitchedEntities.add(1, { result: 'error' });
stitchingDuration.record(endTime(), { result: 'error' });
logger.error(
`Failed to stitch ${item.entityRef}, ${stringifyError(error)}`,
);
}
return {
markComplete,
markFailed,
};
}
return { stitchStart };
}
plugins/catalog-backend/src/stitching/types.ts
+40
View File
@@ -0,0 +1,40 @@
/*
* Copyright 2023 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.
*/
/**
* Performs the act of stitching - to take all of the various outputs from the
* ingestion process, and stitching them together into the final entity JSON
* shape.
*/
export interface Stitcher {
stitch(options: {
entityRefs?: Iterable<string>;
entityIds?: Iterable<string>;
}): Promise<void>;
}
/**
* The strategies supported by the stitching process, in terms of when to
* perform stitching.
*
* @remarks
*
* In immediate mode, stitching happens "in-band" (blocking) immediately when
* each processing task finishes.
*/
export type StitchingStrategy = {
mode: 'immediate';
};
plugins/catalog-backend/src/tests/integration.test.ts
+13 -3
View File
@@ -53,7 +53,7 @@ import { CatalogProcessingEngine } from '../processing/types';
import { DefaultEntitiesCatalog } from '../service/DefaultEntitiesCatalog';
import { DefaultRefreshService } from '../service/DefaultRefreshService';
import { RefreshOptions, RefreshService } from '../service/types';
import { Stitcher } from '../stitching/Stitcher';
import { DefaultStitcher } from '../stitching/DefaultStitcher';
const voidLogger = getVoidLogger();
@@ -268,7 +268,7 @@ class TestHarness {
policy: EntityPolicies.allOf([]),
legacySingleProcessorValidation: false,
});
const stitcher = new Stitcher(db, logger);
const stitcher = DefaultStitcher.fromConfig(config, { knex: db, logger });
const catalog = new DefaultEntitiesCatalog({
database: db,
logger,
@@ -282,6 +282,7 @@ class TestHarness {
config: new ConfigReader({}),
logger,
processingDatabase,
knex: db,
orchestrator,
stitcher,
createHash: () => createHash('sha1'),
@@ -300,7 +301,16 @@ class TestHarness {
return new TestHarness(
catalog,
engine,
{
async start() {
await engine.start();
await stitcher.start();
},
async stop() {
await engine.stop();
await stitcher.stop();
},
},
refresh,
provider,
proxyProgressTracker,
plugins/catalog-backend/src/tests/performance/stitchingPerformance.test.ts
+1 -1
View File
@@ -181,6 +181,7 @@ describePerformanceTest('stitchingPerformance', () => {
const backend = await startTestBackend({
features: [
import('@backstage/plugin-catalog-backend/alpha'),
staticDatabase(knex),
createBackendModule({
moduleId: 'syntheticLoadEntities',
pluginId: 'catalog',
@@ -200,7 +201,6 @@ describePerformanceTest('stitchingPerformance', () => {
});
},
}),
staticDatabase(knex),
],
});