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

update the stitcher including search

Browse Source 
Signed-off-by: Fredrik Adelöw <freben@gmail.com>
This commit is contained in:
Fredrik Adelöw
2021-04-28 09:18:36 +02:00
committed by blam
parent a60c2d9f8f
commit 35b3246116
6 changed files with 592 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/catalog-backend/migrationsv2/20210302150147_refresh_state.js
+4 -3
View File
@@ -33,7 +33,6 @@ exports.up = async function up(knex) {
);
table
.text('entity_ref')
.unique()
.notNullable()
.comment('A reference to the entity that the refresh state is tied to');
table
@@ -59,13 +58,14 @@ exports.up = async function up(knex) {
.notNullable()
.comment('JSON array containing all errors related to entity');
table
.dateTime('next_update_at') // TOOD: timezone or change to epoch-millis or similar
.dateTime('next_update_at') // TODO: timezone or change to epoch-millis or similar
.notNullable()
.comment('Timestamp of when entity should be updated');
table
.dateTime('last_discovery_at') // TOOD: timezone or change to epoch-millis or similar
.dateTime('last_discovery_at') // TODO: timezone or change to epoch-millis or similar
.notNullable()
.comment('The last timestamp of which this entity was discovered');
table.unique(['entity_ref'], 'refresh_state_entity_ref_uniq');
table.index('entity_id', 'refresh_state_entity_id_idx');
table.index('entity_ref', 'refresh_state_entity_ref_idx');
table.index('next_update_at', 'refresh_state_next_update_at_idx');
@@ -188,6 +188,7 @@ exports.down = async function down(knex) {
table.dropIndex([], 'refresh_state_references_target_entity_id_idx');
});
await knex.schema.alterTable('refresh_state', table => {
table.dropUnique([], 'refresh_state_entity_ref_uniq');
table.dropIndex([], 'refresh_state_entity_id_idx');
table.dropIndex([], 'refresh_state_entity_ref_idx');
table.dropIndex([], 'refresh_state_next_update_at_idx');
plugins/catalog-backend/src/next/Stitcher.test.ts
+205
View File
@@ -0,0 +1,205 @@
/*
* Copyright 2021 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import { getVoidLogger } from '@backstage/backend-common';
import { Knex } from 'knex';
import { DatabaseManager } from './database/DatabaseManager';
import {
DbRefreshStateReferencesRow,
DbRefreshStateRow,
DbRelationsRow,
} from './database/DefaultProcessingDatabase';
import { DbSearchRow } from './search';
import { DbFinalEntitiesRow, Stitcher } from './Stitcher';
describe('Stitcher', () => {
let db: Knex;
const logger = getVoidLogger();
beforeEach(async () => {
db = await DatabaseManager.createTestDatabaseConnection();
await DatabaseManager.createDatabase(db);
});
it('runs the happy path', async () => {
const stitcher = new Stitcher(db, logger);
await db.transaction(async tx => {
await tx<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: tx.fn.now(),
last_discovery_at: tx.fn.now(),
},
]);
await tx<DbRefreshStateReferencesRow>('refresh_state_references').insert([
{ source_key: 'a', target_entity_ref: 'k:ns/n' },
]);
await tx<DbRelationsRow>('relations').insert([
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/other',
},
]);
});
await stitcher.stitch(new Set(['k:ns/n']));
let firstEtag: string;
await db.transaction(async tx => {
const entities = await tx<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
const entity = JSON.parse(entities[0].finalized_entity);
expect(entity).toEqual({
relations: [
{
type: 'looksAt',
target: {
kind: 'k',
namespace: 'ns',
name: 'other',
},
},
],
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
etag: expect.any(String),
generation: 1,
uid: 'my-id',
},
spec: {
k: 'v',
},
});
firstEtag = entity.metadata.etag;
const search = await tx<DbSearchRow>('search');
expect(search).toEqual(
expect.arrayContaining([
{ entity_id: 'my-id', key: 'relations', value: 'looksat:k:ns/other' },
{ entity_id: 'my-id', key: 'apiversion', value: 'a' },
{ entity_id: 'my-id', key: 'kind', value: 'k' },
{ entity_id: 'my-id', key: 'metadata.name', value: 'n' },
{ entity_id: 'my-id', key: 'metadata.namespace', value: 'ns' },
{ entity_id: 'my-id', key: 'metadata.uid', value: 'my-id' },
{ entity_id: 'my-id', key: 'spec.k', value: 'v' },
]),
);
});
// Re-stitch without any changes
await stitcher.stitch(new Set(['k:ns/n']));
await db.transaction(async tx => {
const entities = await tx<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
const entity = JSON.parse(entities[0].finalized_entity);
expect(entities[0].etag).toEqual(firstEtag);
expect(entity.metadata.etag).toEqual(firstEtag);
});
// Now add one more relation and re-stitch
await db.transaction(async tx => {
await tx<DbRelationsRow>('relations').insert([
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/third',
},
]);
});
await stitcher.stitch(new Set(['k:ns/n']));
await db.transaction(async tx => {
const entities = await tx<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
const entity = JSON.parse(entities[0].finalized_entity);
expect(entity).toEqual({
relations: expect.arrayContaining([
{
type: 'looksAt',
target: {
kind: 'k',
namespace: 'ns',
name: 'other',
},
},
{
type: 'looksAt',
target: {
kind: 'k',
namespace: 'ns',
name: 'third',
},
},
]),
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
etag: expect.any(String),
generation: 1,
uid: 'my-id',
},
spec: {
k: 'v',
},
});
expect(entities[0].etag).not.toEqual(firstEtag);
expect(entities[0].etag).toEqual(entity.metadata.etag);
const search = await tx<DbSearchRow>('search');
expect(search).toEqual(
expect.arrayContaining([
{ entity_id: 'my-id', key: 'relations', value: 'looksat:k:ns/other' },
{ entity_id: 'my-id', key: 'relations', value: 'looksat:k:ns/third' },
{ entity_id: 'my-id', key: 'apiversion', value: 'a' },
{ entity_id: 'my-id', key: 'kind', value: 'k' },
{ entity_id: 'my-id', key: 'metadata.name', value: 'n' },
{ entity_id: 'my-id', key: 'metadata.namespace', value: 'ns' },
{ entity_id: 'my-id', key: 'metadata.uid', value: 'my-id' },
{ entity_id: 'my-id', key: 'spec.k', value: 'v' },
]),
);
});
});
});
plugins/catalog-backend/src/next/Stitcher.ts
+31 -28
View File
@@ -14,20 +14,14 @@
* limitations under the License.
*/
import { Entity, parseEntityRef } from '@backstage/catalog-model';
import { ConflictError } from '@backstage/errors';
import { createHash } from 'crypto';
import stableStringify from 'fast-json-stable-stringify';
import { Knex } from 'knex';
import { Logger } from 'winston';
import { Transaction } from '../database';
import { ConflictError } from '@backstage/errors';
import {
DbRefreshStateReferencesRow,
DbRefreshStateRow,
DbRelationsRow,
} from './database/DefaultProcessingDatabase';
import { Entity, parseEntityRef } from '@backstage/catalog-model';
import { createHash } from 'crypto';
import stableStringify from 'fast-json-stable-stringify';
import { buildEntitySearch } from '../database/search';
import { DbEntitiesSearchRow } from '../database/types';
import { buildEntitySearch, DbSearchRow } from './search';
// 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
@@ -58,6 +52,13 @@ export class Stitcher {
await this.transaction(async txOpaque => {
const tx = txOpaque as Knex.Transaction;
// 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.
// The only result set "expanding" join is the one with relations, so
// the output should be at least one row (if zero or one relations were
// found), or at most the same number of rows as relations.
const result: Array<{
entityId: string;
processedEntity?: string;
@@ -90,8 +91,14 @@ export class Stitcher {
.leftOuterJoin('relations', {
'relations.source_entity_ref': 'refresh_state.entity_ref',
})
.where({ 'refresh_state.entity_ref': entityRef });
.where({ 'refresh_state.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 (!result.length) {
this.logger.debug(
`Unable to stitch ${entityRef}, item does not exist in refresh state table`,
@@ -102,11 +109,15 @@ export class Stitcher {
const {
entityId,
processedEntity,
errors,
// errors,
incomingReferenceCount,
previousEtag,
} = result[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`,
@@ -114,6 +125,7 @@ export class Stitcher {
return;
}
// Grab the processed entity and stitch all of the relevant data into it
const entity = JSON.parse(processedEntity) as Entity;
const isOrphan = Number(incomingReferenceCount) === 0;
@@ -132,15 +144,16 @@ export class Stitcher {
type: row.relationType!,
target: parseEntityRef(row.relationTarget!),
}));
entity.metadata.generation = 1;
// If the output entity was actually not changed, just abort
const etag = generateEntityEtag(entity);
if (etag === previousEtag) {
console.log(`Skipped stitching of ${entityRef}, no changes`);
this.logger.debug(`Skipped stitching of ${entityRef}, no changes`);
return;
}
entity.metadata.uid = entityId;
entity.metadata.generation = 1;
entity.metadata.etag = etag;
await tx<DbFinalEntitiesRow>('final_entities')
@@ -152,19 +165,9 @@ export class Stitcher {
.onConflict('entity_id')
.merge(['finalized_entity', 'etag']);
try {
const entries = buildEntitySearch(entityId, entity);
await tx<DbEntitiesSearchRow>('search')
.where({ entity_id: entityId })
.delete();
await tx.batchInsert('search', entries, BATCH_SIZE);
} catch (e) {
this.logger.debug(
`Failed to write search entries for ${entityId}`,
e,
);
// intentionally ignored
}
const searchEntries = buildEntitySearch(entityId, entity);
await tx<DbSearchRow>('search').where({ entity_id: entityId }).delete();
await tx.batchInsert('search', searchEntries, BATCH_SIZE);
});
}
}
plugins/catalog-backend/src/next/search.test.ts
+160
View File
@@ -0,0 +1,160 @@
/*
* Copyright 2020 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import { Entity, ENTITY_DEFAULT_NAMESPACE } from '@backstage/catalog-model';
import { buildEntitySearch, mapToRows, traverse } from './search';
describe('search', () => {
describe('traverse', () => {
it('expands lists of strings to several rows', () => {
const input = { a: ['b', 'c', 'd'] };
const output = traverse(input);
expect(output).toEqual([
{ key: 'a', value: 'b' },
{ key: 'a.b', value: true },
{ key: 'a', value: 'c' },
{ key: 'a.c', value: true },
{ key: 'a', value: 'd' },
{ key: 'a.d', value: true },
]);
});
it('expands objects', () => {
const input = { a: { b: { c: 'd' }, e: 'f' } };
const output = traverse(input);
expect(output).toEqual([
{ key: 'a.b.c', value: 'd' },
{ key: 'a.e', value: 'f' },
]);
});
it('expands list of objects', () => {
const input = { root: { list: [{ a: 1 }, { a: 2 }] } };
const output = traverse(input);
expect(output).toEqual([
{ key: 'root.list.a', value: 1 },
{ key: 'root.list.a', value: 2 },
]);
});
it('skips over special keys', () => {
const input = {
state: { x: 1 },
relations: [{ y: 2 }],
a: 'a',
metadata: {
b: 'b',
name: 'name',
namespace: 'namespace',
uid: 'uid',
etag: 'etag',
generation: 'generation',
c: 'c',
},
d: 'd',
};
const output = traverse(input);
expect(output).toEqual([
{ key: 'a', value: 'a' },
{ key: 'metadata.b', value: 'b' },
{ key: 'metadata.c', value: 'c' },
{ key: 'd', value: 'd' },
]);
});
});
describe('mapToRows', () => {
it('converts base types to strings or null', () => {
const input = [
{ key: 'a', value: true },
{ key: 'b', value: false },
{ key: 'c', value: 7 },
{ key: 'd', value: 'string' },
{ key: 'e', value: null },
{ key: 'f', value: undefined },
];
const output = mapToRows(input, 'eid');
expect(output).toEqual([
{ entity_id: 'eid', key: 'a', value: 'true' },
{ entity_id: 'eid', key: 'b', value: 'false' },
{ entity_id: 'eid', key: 'c', value: '7' },
{ entity_id: 'eid', key: 'd', value: 'string' },
{ entity_id: 'eid', key: 'e', value: null },
{ entity_id: 'eid', key: 'f', value: null },
]);
});
it('emits lowercase version of keys and values', () => {
const input = [{ key: 'fOo', value: 'BaR' }];
const output = mapToRows(input, 'eid');
expect(output).toEqual([{ entity_id: 'eid', key: 'foo', value: 'bar' }]);
});
it('skips very large values', () => {
const input = [{ key: 'foo', value: 'a'.repeat(10000) }];
const output = mapToRows(input, 'eid');
expect(output).toEqual([]);
});
});
describe('buildEntitySearch', () => {
it('adds special keys even if missing', () => {
const input: Entity = {
apiVersion: 'a',
kind: 'b',
metadata: { name: 'n' },
};
expect(buildEntitySearch('eid', input)).toEqual([
{ entity_id: 'eid', key: 'apiversion', value: 'a' },
{ entity_id: 'eid', key: 'kind', value: 'b' },
{ entity_id: 'eid', key: 'metadata.name', value: 'n' },
{ entity_id: 'eid', key: 'metadata.namespace', value: null },
{ entity_id: 'eid', key: 'metadata.uid', value: null },
{
entity_id: 'eid',
key: 'metadata.namespace',
value: ENTITY_DEFAULT_NAMESPACE,
},
]);
});
it('adds relations', () => {
const input: Entity = {
relations: [
{ type: 't1', target: { kind: 'k', namespace: 'ns', name: 'a' } },
{ type: 't2', target: { kind: 'k', namespace: 'ns', name: 'b' } },
],
apiVersion: 'a',
kind: 'b',
metadata: { name: 'n' },
};
expect(buildEntitySearch('eid', input)).toEqual([
{ entity_id: 'eid', key: 'apiversion', value: 'a' },
{ entity_id: 'eid', key: 'kind', value: 'b' },
{ entity_id: 'eid', key: 'metadata.name', value: 'n' },
{ entity_id: 'eid', key: 'metadata.namespace', value: null },
{ entity_id: 'eid', key: 'metadata.uid', value: null },
{
entity_id: 'eid',
key: 'metadata.namespace',
value: ENTITY_DEFAULT_NAMESPACE,
},
{ entity_id: 'eid', key: 'relations', value: 't1:k:ns/a' },
{ entity_id: 'eid', key: 'relations', value: 't2:k:ns/b' },
]);
});
});
});
plugins/catalog-backend/src/next/search.ts
+191
View File
@@ -0,0 +1,191 @@
/*
* Copyright 2020 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import {
Entity,
ENTITY_DEFAULT_NAMESPACE,
stringifyEntityRef,
} from '@backstage/catalog-model';
export type DbSearchRow = {
entity_id: string;
key: string;
value: string | null;
};
// 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
// null or not
const SPECIAL_KEYS = [
'state',
'relations',
'metadata.name',
'metadata.namespace',
'metadata.uid',
'metadata.etag',
'metadata.generation',
];
// The maximum length allowed for search values. These columns are indexed, and
// database engines do not like to index on massive values. For example,
// postgres will balk after 8191 byte line sizes.
const MAX_VALUE_LENGTH = 200;
type Kv = {
key: string;
value: unknown;
};
// Helper for traversing through a nested structure and outputting a list of
// path->value entries of the leaves.
//
// For example, this yaml structure
//
// a: 1
// b:
// c: null
// e: [f, g]
// h:
// - i: 1
// j: k
// - i: 2
// j: l
//
// will result in
//
// "a", 1
// "b.c", null
// "b.e": "f"
// "b.e.f": true
// "b.e": "g"
// "b.e.g": true
// "h.i": 1
// "h.j": "k"
// "h.i": 2
// "h.j": "l"
export function traverse(root: unknown): Kv[] {
const output: Kv[] = [];
function visit(path: string, current: unknown) {
if (SPECIAL_KEYS.includes(path)) {
return;
}
// empty or scalar
if (
current === undefined ||
current === null ||
['string', 'number', 'boolean'].includes(typeof current)
) {
output.push({ key: path, value: current });
return;
}
// unknown
if (typeof current !== 'object') {
return;
}
// array
if (Array.isArray(current)) {
for (const item of current) {
// NOTE(freben): The reason that these are output in two different ways,
// is to support use cases where you want to express that MORE than one
// tag is present in a list. Since the EntityFilters structure is a
// record, you can't have several entries of the same key. Therefore
// you will have to match on
//
// { "a.b": ["true"], "a.c": ["true"] }
//
// rather than
//
// { "a": ["b", "c"] }
//
// because the latter means EITHER b or c has to be present.
visit(path, item);
if (typeof item === 'string') {
output.push({ key: `${path}.${item}`, value: true });
}
}
return;
}
// object
for (const [key, value] of Object.entries(current!)) {
visit(path ? `${path}.${key}` : key, value);
}
}
visit('', root);
return output;
}
// Translates a number of raw data rows to search table rows
export function mapToRows(input: Kv[], entityId: string): DbSearchRow[] {
const result: DbSearchRow[] = [];
for (const { key: rawKey, value: rawValue } of input) {
const key = rawKey.toLocaleLowerCase('en-US');
if (rawValue === undefined || rawValue === null) {
result.push({ entity_id: entityId, key, value: null });
} else {
const value = String(rawValue).toLocaleLowerCase('en-US');
if (value.length <= MAX_VALUE_LENGTH) {
result.push({ entity_id: entityId, key, value });
}
}
}
return result;
}
/**
* Generates all of the search rows that are relevant for this entity.
*
* @param entityId The uid of the entity
* @param entity The entity
* @returns A list of entity search rows
*/
export function buildEntitySearch(
entityId: string,
entity: Entity,
): DbSearchRow[] {
// Visit the base structure recursively
const raw = traverse(entity);
// Start with some special keys that are always present because you want to
// be able to easily search for null specifically
raw.push({ key: 'metadata.name', value: entity.metadata.name });
raw.push({ key: 'metadata.namespace', value: entity.metadata.namespace });
raw.push({ key: 'metadata.uid', value: entity.metadata.uid });
// Namespace not specified has the default value "default", so we want to
// match on that as well
if (!entity.metadata.namespace) {
raw.push({ key: 'metadata.namespace', value: ENTITY_DEFAULT_NAMESPACE });
}
// Visit relations
for (const relation of entity.relations ?? []) {
raw.push({
key: 'relations',
value: `${relation.type}:${stringifyEntityRef(relation.target)}`,
});
}
return mapToRows(raw, entityId);
}
plugins/catalog-backend/src/next/types.ts
+1
View File
@@ -120,6 +120,7 @@ export type ReplaceProcessingItemsRequest =
removed: Entity[];
type: 'delta';
};
export interface ProcessingStateManager {
setProcessingItemResult(result: ProcessingItemResult): Promise<void>;
getNextProcessingItem(): Promise<ProcessingItem>;