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

Merge pull request #7342 from backstage/freben/catalog-entity-provider-full-upsert

Browse Source 
Make entity provider `full` updates behave as a large set replacement
This commit is contained in:
Fredrik Adelöw
2021-09-28 13:57:59 +02:00
committed by GitHub
parent 1c43d2c25c bf650f8603
commit 6084db3352
6 changed files with 224 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/breezy-foxes-wonder.md
+5
View File
@@ -0,0 +1,5 @@
---
'@backstage/plugin-catalog-backend': patch
---
When issuing a `full` update from an entity provider, entities with updates are now properly persisted.
plugins/catalog-backend/migrations/20210925102509_add_refresh_state_input_hash.js
+38
View File
@@ -0,0 +1,38 @@
/*
* 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.
*/
// @ts-check
/**
* @param {import('knex').Knex} knex
*/
exports.up = async function up(knex) {
await knex.schema.alterTable('refresh_state', table => {
table
.text('unprocessed_hash')
.nullable()
.comment('A hash of the unprocessed contents, used to detect changes');
});
};
/**
* @param {import('knex').Knex} knex
*/
exports.down = async function down(knex) {
await knex.schema.alterTable('refresh_state', table => {
table.dropColumn('unprocessed_hash');
});
};
plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts
+96
View File
@@ -30,6 +30,7 @@ import {
} from './tables';
import { createRandomRefreshInterval } from '../refresh';
import { timestampToDateTime } from './conversion';
import { generateStableHash } from './util';
describe('Default Processing Database', () => {
const defaultLogger = getVoidLogger();
@@ -309,6 +310,7 @@ describe('Default Processing Database', () => {
entity_id: id,
entity_ref: 'location:default/fakelocation',
unprocessed_entity: '{}',
unprocessed_hash: generateStableHash({} as any),
processed_entity: '{}',
errors: '[]',
next_update_at: '2021-04-01 13:37:00',
@@ -996,6 +998,100 @@ describe('Default Processing Database', () => {
},
60_000,
);
it.each(databases.eachSupportedId())(
'should support replacing modified entities during a full update, %p',
async databaseId => {
const { knex, db } = await createDatabase(databaseId);
await db.transaction(async tx => {
await db.replaceUnprocessedEntities(tx, {
type: 'full',
sourceKey: 'lols',
items: [
{
entity: {
apiVersion: '1',
kind: 'Component',
metadata: { name: 'a' },
spec: { marker: 'WILL_CHANGE' },
} as Entity,
locationKey: 'file:///tmp/a',
},
{
entity: {
apiVersion: '1',
kind: 'Component',
metadata: { name: 'b' },
spec: { marker: 'NEVER_CHANGES' },
} as Entity,
locationKey: 'file:///tmp/b',
},
],
});
});
let state = await knex<DbRefreshStateRow>('refresh_state').select();
expect(state).toEqual(
expect.arrayContaining([
expect.objectContaining({
entity_ref: 'component:default/a',
location_key: 'file:///tmp/a',
unprocessed_entity: expect.stringContaining('WILL_CHANGE'),
}),
expect.objectContaining({
entity_ref: 'component:default/b',
location_key: 'file:///tmp/b',
unprocessed_entity: expect.stringContaining('NEVER_CHANGES'),
}),
]),
);
await db.transaction(async tx => {
await db.replaceUnprocessedEntities(tx, {
type: 'full',
sourceKey: 'lols',
items: [
{
entity: {
apiVersion: '1',
kind: 'Component',
metadata: { name: 'a' },
spec: { marker: 'HAS_CHANGED' },
} as Entity,
locationKey: 'file:///tmp/a',
},
{
entity: {
apiVersion: '1',
kind: 'Component',
metadata: { name: 'b' },
spec: { marker: 'NEVER_CHANGES' },
} as Entity,
locationKey: 'file:///tmp/b',
},
],
});
});
state = await knex<DbRefreshStateRow>('refresh_state').select();
expect(state).toEqual(
expect.arrayContaining([
expect.objectContaining({
entity_ref: 'component:default/a',
location_key: 'file:///tmp/a',
unprocessed_entity: expect.stringContaining('HAS_CHANGED'),
}),
expect.objectContaining({
entity_ref: 'component:default/b',
location_key: 'file:///tmp/b',
unprocessed_entity: expect.stringContaining('NEVER_CHANGES'),
}),
]),
);
},
60_000,
);
});
describe('getProcessableEntities', () => {
plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts
+59 -16
View File
@@ -41,6 +41,7 @@ import {
ListAncestorsResult,
UpdateEntityCacheOptions,
} from './types';
import { generateStableHash } from './util';
// The number of items that are sent per batch to the database layer, when
// doing .batchInsert calls to knex. This needs to be low enough to not cause
@@ -156,7 +157,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
): Promise<void> {
const tx = txOpaque as Knex.Transaction;
const { toAdd, toRemove } = await this.createDelta(tx, options);
const { toUpsert, toRemove } = await this.createDelta(tx, options);
if (toRemove.length) {
// TODO(freben): Batch split, to not hit variable limits?
@@ -273,14 +274,27 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
);
}
if (toAdd.length) {
for (const { entity, locationKey } of toAdd) {
if (toUpsert.length) {
for (const {
deferred: { entity, locationKey },
hash,
} of toUpsert) {
const entityRef = stringifyEntityRef(entity);
try {
let ok = await this.insertUnprocessedEntity(tx, entity, locationKey);
let ok = await this.updateUnprocessedEntity(
tx,
entity,
hash,
locationKey,
);
if (!ok) {
ok = await this.updateUnprocessedEntity(tx, entity, locationKey);
ok = await this.insertUnprocessedEntity(
tx,
entity,
hash,
locationKey,
);
}
if (ok) {
@@ -448,6 +462,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
private async updateUnprocessedEntity(
tx: Knex.Transaction,
entity: Entity,
hash: string,
locationKey?: string,
): Promise<boolean> {
const entityRef = stringifyEntityRef(entity);
@@ -456,6 +471,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
const refreshResult = await tx<DbRefreshStateRow>('refresh_state')
.update({
unprocessed_entity: serializedEntity,
unprocessed_hash: hash,
location_key: locationKey,
last_discovery_at: tx.fn.now(),
// We only get to this point if a processed entity actually had any changes, or
@@ -483,6 +499,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
private async insertUnprocessedEntity(
tx: Knex.Transaction,
entity: Entity,
hash: string,
locationKey?: string,
): Promise<boolean> {
const entityRef = stringifyEntityRef(entity);
@@ -493,6 +510,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
entity_id: uuid(),
entity_ref: entityRef,
unprocessed_entity: serializedEntity,
unprocessed_hash: hash,
errors: '',
location_key: locationKey,
next_update_at: tx.fn.now(),
@@ -558,10 +576,16 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
private async createDelta(
tx: Knex.Transaction,
options: ReplaceUnprocessedEntitiesOptions,
): Promise<{ toAdd: DeferredEntity[]; toRemove: string[] }> {
): Promise<{
toUpsert: { deferred: DeferredEntity; hash: string }[];
toRemove: string[];
}> {
if (options.type === 'delta') {
return {
toAdd: options.added,
toUpsert: options.added.map(e => ({
deferred: e,
hash: generateStableHash(e.entity),
})),
toRemove: options.removed.map(e => stringifyEntityRef(e.entity)),
};
}
@@ -570,39 +594,55 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
const oldRefs = await tx<DbRefreshStateReferencesRow>(
'refresh_state_references',
)
.where({ source_key: options.sourceKey })
.leftJoin<DbRefreshStateRow>('refresh_state', {
target_entity_ref: 'entity_ref',
})
.select(['target_entity_ref', 'location_key']);
.where({ source_key: options.sourceKey })
.select({
target_entity_ref: 'refresh_state_references.target_entity_ref',
location_key: 'refresh_state.location_key',
unprocessed_hash: 'refresh_state.unprocessed_hash',
});
const items = options.items.map(deferred => ({
deferred,
ref: stringifyEntityRef(deferred.entity),
hash: generateStableHash(deferred.entity),
}));
const oldRefsSet = new Map(
oldRefs.map(r => [r.target_entity_ref, r.location_key]),
oldRefs.map(r => [
r.target_entity_ref,
{
locationKey: r.location_key,
oldEntityHash: r.unprocessed_hash,
},
]),
);
const newRefsSet = new Set(items.map(item => item.ref));
const toAdd = new Array<DeferredEntity>();
const toUpsert = new Array<{ deferred: DeferredEntity; hash: string }>();
const toRemove = oldRefs
.map(row => row.target_entity_ref)
.filter(ref => !newRefsSet.has(ref));
for (const item of items) {
if (!oldRefsSet.has(item.ref)) {
const oldRef = oldRefsSet.get(item.ref);
const upsertItem = { deferred: item.deferred, hash: item.hash };
if (!oldRef) {
// Add any entity that does not exist in the database
toAdd.push(item.deferred);
} else if (oldRefsSet.get(item.ref) !== item.deferred.locationKey) {
toUpsert.push(upsertItem);
} else if (oldRef.locationKey !== item.deferred.locationKey) {
// Remove and then re-add any entity that exists, but with a different location key
toRemove.push(item.ref);
toAdd.push(item.deferred);
toUpsert.push(upsertItem);
} else if (oldRef.oldEntityHash !== item.hash) {
// Entities with modifications should be pushed through too
toUpsert.push(upsertItem);
}
}
return { toAdd, toRemove };
return { toUpsert, toRemove };
}
/**
@@ -626,10 +666,12 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
// their entity ref.
for (const { entity, locationKey } of options.entities) {
const entityRef = stringifyEntityRef(entity);
const hash = generateStableHash(entity);
const updated = await this.updateUnprocessedEntity(
tx,
entity,
hash,
locationKey,
);
if (updated) {
@@ -640,6 +682,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
const inserted = await this.insertUnprocessedEntity(
tx,
entity,
hash,
locationKey,
);
if (inserted) {
plugins/catalog-backend/src/next/database/tables.ts
+1
View File
@@ -24,6 +24,7 @@ export type DbRefreshStateRow = {
entity_id: string;
entity_ref: string;
unprocessed_entity: string;
unprocessed_hash?: string;
processed_entity?: string;
result_hash?: string;
cache?: string;
plugins/catalog-backend/src/next/database/util.ts
+25
View File
@@ -0,0 +1,25 @@
/*
* 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 } from '@backstage/catalog-model';
import { createHash } from 'crypto';
import stableStringify from 'fast-json-stable-stringify';
export function generateStableHash(entity: Entity) {
return createHash('sha1')
.update(stableStringify({ ...entity }))
.digest('hex');
}