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

Merge pull request #5473 from backstage/mob/catalog-mutations

Browse Source 
catalog/next: Add support for entity mutations
This commit is contained in:
Fredrik Adelöw
2021-04-28 15:33:10 +02:00
committed by GitHub
parent 21e6600116 11461a20b3
commit c7de293327
37 changed files with 3220 additions and 322 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/catalog-backend/migrationsv2/20200511113813_init.js
+133
View File
@@ -0,0 +1,133 @@
/*
* Copyright 2020 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// @ts-check
/**
* @param {import('knex').Knex} knex
*/
exports.up = async function up(knex) {
return (
knex.schema
//
// locations
//
.createTable('locations', table => {
table.comment(
'Registered locations that shall be contiuously scanned for catalog item updates',
);
table
.uuid('id')
.primary()
.notNullable()
.comment('Auto-generated ID of the location');
table.string('type').notNullable().comment('The type of location');
table
.string('target')
.notNullable()
.comment('The actual target of the location');
})
//
// entities
//
.createTable('entities', table => {
table.comment('All entities currently stored in the catalog');
table.uuid('id').primary().comment('Auto-generated ID of the entity');
table
.uuid('location_id')
.references('id')
.inTable('locations')
.nullable()
.comment('The location that originated the entity');
table
.string('etag')
.notNullable()
.comment(
'An opaque string that changes for each update operation to any part of the entity, including metadata.',
);
table
.string('generation')
.notNullable()
.unsigned()
.comment(
'A positive nonzero number that indicates the current generation of data for this entity; the value is incremented each time the spec changes.',
);
table
.string('api_version')
.notNullable()
.comment('The apiVersion field of the entity');
table
.string('kind')
.notNullable()
.comment('The kind field of the entity');
table
.string('name')
.nullable()
.comment('The metadata.name field of the entity');
table
.string('namespace')
.nullable()
.comment('The metadata.namespace field of the entity');
table
.string('metadata')
.notNullable()
.comment('The entire metadata JSON blob of the entity');
table
.string('spec')
.nullable()
.comment('The entire spec JSON blob of the entity');
})
.alterTable('entities', table => {
// https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.18/#objectmeta-v1-meta
table.unique(['kind', 'name', 'namespace'], 'entities_unique_name');
})
//
// entities_search
//
.createTable('entities_search', table => {
table.comment(
'Flattened key-values from the entities, used for quick filtering',
);
table
.uuid('entity_id')
.references('id')
.inTable('entities')
.onDelete('CASCADE')
.comment('The entity that matches this key/value');
table
.string('key')
.notNullable()
.comment('A key that occurs in the entity');
table
.string('value')
.nullable()
.comment('The corresponding value to match on');
})
);
};
/**
* @param {import('knex').Knex} knex
*/
exports.down = async function down(knex) {
return knex.schema
.dropTable('entities_search')
.alterTable('entities', table => {
table.dropUnique([], 'entities_unique_name');
})
.dropTable('entities')
.dropTable('locations');
};
plugins/catalog-backend/migrationsv2/20200520140700_location_update_log_table.js
+43
View File
@@ -0,0 +1,43 @@
/*
* Copyright 2020 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// @ts-check
/**
* @param {import('knex').Knex} knex
*/
exports.up = async function up(knex) {
return knex.schema.createTable('location_update_log', table => {
table.uuid('id').primary();
table.enum('status', ['success', 'fail']).notNullable();
table.dateTime('created_at').defaultTo(knex.fn.now()).notNullable();
table.string('message');
table
.uuid('location_id')
.references('id')
.inTable('locations')
.onUpdate('CASCADE')
.onDelete('CASCADE');
table.string('entity_name').nullable();
});
};
/**
* @param {import('knex').Knex} knex
*/
exports.down = async function down(knex) {
return knex.schema.dropTableIfExists('location_update_log');
};
plugins/catalog-backend/migrationsv2/20200527114117_location_update_log_latest_view.js
+45
View File
@@ -0,0 +1,45 @@
/*
* Copyright 2020 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// @ts-check
/**
* @param {import('knex').Knex} knex
*/
exports.up = async function up(knex) {
// Get list sorted by created_at timestamp in descending order
// Grouped by location_id
return knex.schema.raw(`
CREATE VIEW location_update_log_latest AS
SELECT t1.* FROM location_update_log t1
JOIN
(
SELECT location_id, MAX(created_at) AS MAXDATE
FROM location_update_log
GROUP BY location_id
) t2
ON t1.location_id = t2.location_id
AND t1.created_at = t2.MAXDATE
ORDER BY created_at DESC;
`);
};
/**
* @param {import('knex').Knex} knex
*/
exports.down = async function down(knex) {
return knex.schema.raw(`DROP VIEW location_update_log_latest;`);
};
plugins/catalog-backend/migrationsv2/20200702153613_entities.js
+236
View File
@@ -0,0 +1,236 @@
/*
* Copyright 2020 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// @ts-check
/**
* @param {import('knex').Knex} knex
*/
exports.up = async function up(knex) {
// SQLite does not support FK and PK
if (knex.client.config.client !== 'sqlite3') {
await knex.schema.alterTable('entities_search', table => {
table.dropForeign(['entity_id']);
});
await knex.schema.alterTable('entities', table => {
table.dropPrimary('entities_pkey');
});
}
await knex.schema.alterTable('entities', table => {
table.dropUnique([], 'entities_unique_name');
});
// Setup temporary tables
await knex.schema.renameTable('entities_search', 'tmp_entities_search');
await knex.schema.renameTable('entities', 'tmp_entities');
//
// entities
//
await knex.schema
.createTable('entities', table => {
table.comment('All entities currently stored in the catalog');
table.uuid('id').primary().comment('Auto-generated ID of the entity');
table
.uuid('location_id')
.references('id')
.inTable('locations')
.nullable()
.comment('The location that originated the entity');
table
.string('etag')
.notNullable()
.comment(
'An opaque string that changes for each update operation to any part of the entity, including metadata.',
);
table
.string('generation')
.notNullable()
.unsigned()
.comment(
'A positive nonzero number that indicates the current generation of data for this entity; the value is incremented each time the spec changes.',
);
table
.string('api_version')
.notNullable()
.comment('The apiVersion field of the entity');
table
.string('kind')
.notNullable()
.comment('The kind field of the entity');
table
.string('name')
.nullable()
.comment('The metadata.name field of the entity');
table
.string('namespace')
.nullable()
.comment('The metadata.namespace field of the entity');
table
.text('metadata')
.notNullable()
.comment('The entire metadata JSON blob of the entity');
table
.text('spec')
.nullable()
.comment('The entire spec JSON blob of the entity');
})
.alterTable('entities', table => {
// https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.18/#objectmeta-v1-meta
table.unique(['kind', 'name', 'namespace'], 'entities_unique_name');
});
await knex.schema.raw(`INSERT INTO entities SELECT * FROM tmp_entities`);
//
// entities_search
//
await knex.schema.createTable('entities_search', table => {
table.comment(
'Flattened key-values from the entities, used for quick filtering',
);
table
.uuid('entity_id')
.references('id')
.inTable('entities')
.onDelete('CASCADE')
.comment('The entity that matches this key/value');
table
.string('key')
.notNullable()
.comment('A key that occurs in the entity');
table
.string('value')
.nullable()
.comment('The corresponding value to match on');
});
await knex.schema.raw(
`INSERT INTO entities_search SELECT * FROM tmp_entities_search`,
);
// Clean up
await knex.schema.dropTable('tmp_entities');
return knex.schema.dropTable('tmp_entities_search');
};
/**
* @param {import('knex').Knex} knex
*/
exports.down = async function down(knex) {
// SQLite does not support FK and PK
if (knex.client.config.client !== 'sqlite3') {
await knex.schema.alterTable('entities_search', table => {
table.dropForeign(['entity_id']);
});
await knex.schema.alterTable('entities', table => {
table.dropPrimary('entities_pkey');
});
}
await knex.schema.alterTable('entities', table => {
table.dropUnique([], 'entities_unique_name');
});
// Setup temporary tables
await knex.schema.renameTable('entities_search', 'tmp_entities_search');
await knex.schema.renameTable('entities', 'tmp_entities');
//
// entities
//
await knex.schema
.createTable('entities', table => {
table.comment('All entities currently stored in the catalog');
table.uuid('id').primary().comment('Auto-generated ID of the entity');
table
.uuid('location_id')
.references('id')
.inTable('locations')
.nullable()
.comment('The location that originated the entity');
table
.string('etag')
.notNullable()
.comment(
'An opaque string that changes for each update operation to any part of the entity, including metadata.',
);
table
.string('generation')
.notNullable()
.unsigned()
.comment(
'A positive nonzero number that indicates the current generation of data for this entity; the value is incremented each time the spec changes.',
);
table
.string('api_version')
.notNullable()
.comment('The apiVersion field of the entity');
table
.string('kind')
.notNullable()
.comment('The kind field of the entity');
table
.string('name')
.nullable()
.comment('The metadata.name field of the entity');
table
.string('namespace')
.nullable()
.comment('The metadata.namespace field of the entity');
table
.string('metadata')
.notNullable()
.comment('The entire metadata JSON blob of the entity');
table
.string('spec')
.nullable()
.comment('The entire spec JSON blob of the entity');
})
.alterTable('entities', table => {
// https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.18/#objectmeta-v1-meta
table.unique(['kind', 'name', 'namespace'], 'entities_unique_name');
});
await knex.schema.raw(`INSERT INTO entities SELECT * FROM tmp_entities`);
//
// entities_search
//
await knex.schema.createTable('entities_search', table => {
table.comment(
'Flattened key-values from the entities, used for quick filtering',
);
table
.uuid('entity_id')
.references('id')
.inTable('entities')
.onDelete('CASCADE')
.comment('The entity that matches this key/value');
table
.string('key')
.notNullable()
.comment('A key that occurs in the entity');
table
.string('value')
.nullable()
.comment('The corresponding value to match on');
});
await knex.schema.raw(
`INSERT INTO entities_search SELECT * FROM tmp_entities_search`,
);
// Clean up
await knex.schema.dropTable('tmp_entities');
return knex.schema.dropTable('tmp_entities_search');
};
plugins/catalog-backend/migrationsv2/20200721115244_location_update_log_latest_deduplicate.js
+44
View File
@@ -0,0 +1,44 @@
/*
* Copyright 2020 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// @ts-check
/**
* @param {import('knex').Knex} knex
*/
exports.up = function up(knex) {
return knex.schema.raw(`DROP VIEW location_update_log_latest;`).raw(`
CREATE VIEW location_update_log_latest AS
SELECT t1.* FROM location_update_log t1
JOIN
(
SELECT location_id, MAX(created_at) AS MAXDATE
FROM location_update_log
GROUP BY location_id
) t2
ON t1.location_id = t2.location_id
AND t1.created_at = t2.MAXDATE
GROUP BY t1.location_id, t1.id
ORDER BY created_at DESC;
`);
};
/**
* @param {import('knex').Knex} knex
*/
exports.down = function down(knex) {
knex.schema.raw(`DROP VIEW location_update_log_latest;`);
};
plugins/catalog-backend/migrationsv2/20200805163904_location_update_log_duplication_fix.js
+87
View File
@@ -0,0 +1,87 @@
/*
* Copyright 2020 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// @ts-check
/**
* @param {import('knex').Knex} knex
*/
exports.up = function up(knex) {
return knex.schema
.raw('DROP VIEW location_update_log_latest;')
.dropTable('location_update_log')
.createTable('location_update_log', table => {
table.bigIncrements('id').primary(); // instead of uuid, so we can MAX it
table.enum('status', ['success', 'fail']).notNullable();
table.dateTime('created_at').defaultTo(knex.fn.now()).notNullable();
table.string('message');
table
.uuid('location_id')
.references('id')
.inTable('locations')
.onUpdate('CASCADE')
.onDelete('CASCADE');
table.string('entity_name').nullable();
}).raw(`
CREATE VIEW location_update_log_latest AS
SELECT t1.* FROM location_update_log t1
JOIN
(
SELECT location_id, MAX(id) AS MAXID
FROM location_update_log
GROUP BY location_id
) t2
ON t1.location_id = t2.location_id
AND t1.id = t2.MAXID
GROUP BY t1.location_id, t1.id
ORDER BY created_at DESC;
`);
};
/**
* @param {import('knex').Knex} knex
*/
exports.down = function down(knex) {
return knex.schema
.raw('DROP VIEW location_update_log_latest;')
.dropTable('location_update_log')
.createTable('location_update_log', table => {
table.uuid('id').primary();
table.enum('status', ['success', 'fail']).notNullable();
table.dateTime('created_at').defaultTo(knex.fn.now()).notNullable();
table.string('message');
table
.uuid('location_id')
.references('id')
.inTable('locations')
.onUpdate('CASCADE')
.onDelete('CASCADE');
table.string('entity_name').nullable();
}).raw(`
CREATE VIEW location_update_log_latest AS
SELECT t1.* FROM location_update_log t1
JOIN
(
SELECT location_id, MAX(created_at) AS MAXDATE
FROM location_update_log
GROUP BY location_id
) t2
ON t1.location_id = t2.location_id
AND t1.created_at = t2.MAXDATE
GROUP BY t1.location_id, t1.id
ORDER BY created_at DESC;
`);
};
plugins/catalog-backend/migrationsv2/20200807120600_entitySearch.js
+41
View File
@@ -0,0 +1,41 @@
/*
* Copyright 2020 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// @ts-check
/**
* @param {import('knex').Knex} knex
*/
exports.up = async function up(knex) {
// Sqlite does not support alter column.
if (knex.client.config.client !== 'sqlite3') {
await knex.schema.alterTable('entities_search', table => {
table.text('value').nullable().alter();
});
}
};
/**
* @param {import('knex').Knex} knex
*/
exports.down = async function down(knex) {
// Sqlite does not support alter column.
if (knex.client.config.client !== 'sqlite3') {
await knex.schema.alterTable('entities_search', table => {
table.string('value').nullable().alter();
});
}
};
plugins/catalog-backend/migrationsv2/20200809202832_add_bootstrap_location.js
+42
View File
@@ -0,0 +1,42 @@
/*
* Copyright 2020 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// @ts-check
/**
* @param {import('knex').Knex} knex
*/
exports.up = async function up(knex) {
// Adds a single 'bootstrap' location that can be used to trigger work in processors.
// This is primarily here to fulfill foreign key constraints.
await knex('locations').insert({
id: require('uuid').v4(),
type: 'bootstrap',
target: 'bootstrap',
});
};
/**
* @param {import('knex').Knex} knex
*/
exports.down = async function down(knex) {
await knex('locations')
.where({
type: 'bootstrap',
target: 'bootstrap',
})
.del();
};
plugins/catalog-backend/migrationsv2/20200923104503_case_insensitivity.js
+32
View File
@@ -0,0 +1,32 @@
/*
* Copyright 2020 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// @ts-check
/**
* @param {import('knex').Knex} knex
*/
exports.up = async function up(knex) {
await knex('entities')
.where({ namespace: null })
.update({ namespace: 'default' });
await knex('entities_search').update({
key: knex.raw('LOWER(key)'),
value: knex.raw('LOWER(value)'),
});
};
exports.down = async function down() {};
plugins/catalog-backend/migrationsv2/20201005122705_add_entity_full_name.js
+60
View File
@@ -0,0 +1,60 @@
/*
* Copyright 2020 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// @ts-check
/**
* @param {import('knex').Knex} knex
*/
exports.up = async function up(knex) {
await knex.schema.alterTable('entities', table => {
table.text('full_name').nullable();
});
await knex('entities').update({
full_name: knex.raw(
"LOWER(kind) || ':' || LOWER(COALESCE(namespace, 'default')) || '/' || LOWER(name)",
),
});
// SQLite does not support alter column
if (knex.client.config.client !== 'sqlite3') {
await knex.schema.alterTable('entities', table => {
table.text('full_name').notNullable().alter();
});
}
await knex.schema.alterTable('entities', table => {
// https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.18/#objectmeta-v1-meta
table.unique(['full_name'], 'entities_unique_full_name');
table.dropUnique([], 'entities_unique_name');
});
};
/**
* @param {import('knex').Knex} knex
*/
exports.down = async function down(knex) {
await knex.schema.alterTable('entities', table => {
// https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.18/#objectmeta-v1-meta
table.dropUnique([], 'entities_unique_full_name');
table.unique(['kind', 'namespace', 'name'], 'entities_unique_name');
});
await knex.schema.alterTable('entities_search', table => {
table.dropColumn('full_name');
});
};
plugins/catalog-backend/migrationsv2/20201006130744_entity_data_column.js
+66
View File
@@ -0,0 +1,66 @@
/*
* Copyright 2020 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// @ts-check
/**
* @param {import('knex').Knex} knex
*/
exports.up = async function up(knex) {
await knex.schema.alterTable('entities', table => {
table
.text('data')
.nullable()
.comment('The entire JSON data blob of the entity');
});
await knex('entities').update({
// apiVersion and kind should not contain any JSON unsafe chars, and both
// metadata and spec are already valid serialized JSON
data: knex.raw(
`'{"apiVersion":"' || api_version || '","kind":"' || kind || '","metadata":' || metadata || COALESCE(',"spec":' || spec, '') || '}'`,
),
});
await knex.schema.alterTable('entities', table => {
table.dropColumn('metadata');
table.dropColumn('spec');
});
// SQLite does not support ALTER COLUMN.
if (knex.client.config.client !== 'sqlite3') {
await knex.schema.alterTable('entities', table => {
table.text('data').notNullable().alter();
});
}
};
/**
* @param {import('knex').Knex} knex
*/
exports.down = async function down(knex) {
await knex.schema.alterTable('entities', table => {
table
.text('metadata')
.notNullable()
.comment('The entire metadata JSON blob of the entity');
table
.text('spec')
.nullable()
.comment('The entire spec JSON blob of the entity');
table.dropColumn('data');
});
};
plugins/catalog-backend/migrationsv2/20201006203131_entity_remove_redundant_columns.js
+50
View File
@@ -0,0 +1,50 @@
/*
* Copyright 2020 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// @ts-check
/**
* @param {import('knex').Knex} knex
*/
exports.up = async function up(knex) {
await knex.schema.alterTable('entities', table => {
table.dropColumn('api_version');
table.dropColumn('kind');
table.dropColumn('name');
table.dropColumn('namespace');
});
};
/**
* @param {import('knex').Knex} knex
*/
exports.down = async function down(knex) {
await knex.schema.alterTable('entities', table => {
table
.string('api_version')
.notNullable()
.comment('The apiVersion field of the entity');
table.string('kind').notNullable().comment('The kind field of the entity');
table
.string('name')
.nullable()
.comment('The metadata.name field of the entity');
table
.string('namespace')
.nullable()
.comment('The metadata.namespace field of the entity');
});
};
plugins/catalog-backend/migrationsv2/20201007201501_index_entity_search.js
+37
View File
@@ -0,0 +1,37 @@
/*
* Copyright 2020 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// @ts-check
/**
* @param {import('knex').Knex} knex
*/
exports.up = async function up(knex) {
await knex.schema.alterTable('entities_search', table => {
table.index(['key'], 'entities_search_key');
table.index(['value'], 'entities_search_value');
});
};
/**
* @param {import('knex').Knex} knex
*/
exports.down = async function down(knex) {
await knex.schema.alterTable('entities_search', table => {
table.dropIndex('', 'entities_search_key');
table.dropIndex('', 'entities_search_value');
});
};
plugins/catalog-backend/migrationsv2/20201019130742_add_relations_table.js
+54
View File
@@ -0,0 +1,54 @@
/*
* Copyright 2020 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// @ts-check
/**
* @param {import('knex').Knex} knex
*/
exports.up = async function up(knex) {
await knex.schema.createTable('entities_relations', table => {
table.comment('All relations between entities in the catalog');
table
.uuid('originating_entity_id')
.references('id')
.inTable('entities')
.onDelete('CASCADE')
.notNullable()
.comment('The entity that provided the relation');
table
.string('source_full_name')
.notNullable()
.comment('The full name of the source entity of the relation');
table
.string('type')
.notNullable()
.comment('The type of the relation between the entities');
table
.string('target_full_name')
.notNullable()
.comment('The full name of the target entity of the relation');
table.primary(['source_full_name', 'type', 'target_full_name']);
});
};
/**
* @param {import('knex').Knex} knex
*/
exports.down = async function down(knex) {
await knex.schema.dropTable('entities_relations');
};
plugins/catalog-backend/migrationsv2/20201123205611_relations_table_uniq.js
+93
View File
@@ -0,0 +1,93 @@
/*
* Copyright 2020 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// @ts-check
/**
* @param {import('knex').Knex} knex
*/
exports.up = async function up(knex) {
if (knex.client.config.client === 'sqlite3') {
// sqlite doesn't support dropPrimary so we recreate it properly instead
await knex.schema.dropTable('entities_relations');
await knex.schema.createTable('entities_relations', table => {
table.comment('All relations between entities in the catalog');
table
.uuid('originating_entity_id')
.references('id')
.inTable('entities')
.onDelete('CASCADE')
.notNullable()
.comment('The entity that provided the relation');
table
.string('source_full_name')
.notNullable()
.comment('The full name of the source entity of the relation');
table
.string('type')
.notNullable()
.comment('The type of the relation between the entities');
table
.string('target_full_name')
.notNullable()
.comment('The full name of the target entity of the relation');
table.index('source_full_name', 'source_full_name_idx');
});
} else {
await knex.schema.alterTable('entities_relations', table => {
table.dropPrimary();
table.index('source_full_name', 'source_full_name_idx');
});
}
};
/**
* @param {import('knex').Knex} knex
*/
exports.down = async function down(knex) {
if (knex.client.config.client === 'sqlite3') {
await knex.schema.dropTable('entities_relations');
await knex.schema.createTable('entities_relations', table => {
table.comment('All relations between entities in the catalog');
table
.uuid('originating_entity_id')
.references('id')
.inTable('entities')
.onDelete('CASCADE')
.notNullable()
.comment('The entity that provided the relation');
table
.string('source_full_name')
.notNullable()
.comment('The full name of the source entity of the relation');
table
.string('type')
.notNullable()
.comment('The type of the relation between the entities');
table
.string('target_full_name')
.notNullable()
.comment('The full name of the target entity of the relation');
table.primary(['source_full_name', 'type', 'target_full_name']);
});
} else {
await knex.schema.alterTable('entities_relations', table => {
table.dropIndex([], 'source_full_name_idx');
table.primary(['source_full_name', 'type', 'target_full_name']);
});
}
};
plugins/catalog-backend/migrationsv2/20201210185851_fk_index.js
+45
View File
@@ -0,0 +1,45 @@
/*
* Copyright 2020 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// @ts-check
/**
* @param {import('knex').Knex} knex
*/
exports.up = async function up(knex) {
if (knex.client.config.client !== 'sqlite3') {
await knex.schema.alterTable('entities_relations', table => {
table.index('originating_entity_id', 'originating_entity_id_idx');
});
await knex.schema.alterTable('entities_search', table => {
table.index('entity_id', 'entity_id_idx');
});
}
};
/**
* @param {import('knex').Knex} knex
*/
exports.down = async function down(knex) {
if (knex.client.config.client !== 'sqlite3') {
await knex.schema.alterTable('entities_relations', table => {
table.dropIndex([], 'originating_entity_id_idx');
});
await knex.schema.alterTable('entities_relations', table => {
table.dropIndex([], 'entity_id_idx');
});
}
};
plugins/catalog-backend/migrationsv2/20201230103504_update_log_varchar.js
+73
View File
@@ -0,0 +1,73 @@
/*
* Copyright 2020 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// @ts-check
/**
* @param {import('knex').Knex} knex
*/
exports.up = async function up(knex) {
if (knex.client.config.client !== 'sqlite3') {
// We actually just want to widen columns, but can't do that while a
// view is dependent on them - so we just reconstruct it exactly as it was
await knex.schema
.raw('DROP VIEW location_update_log_latest;')
.alterTable('location_update_log', table => {
table.text('message').alter();
table.text('entity_name').nullable().alter();
}).raw(`
CREATE VIEW location_update_log_latest AS
SELECT t1.* FROM location_update_log t1
JOIN
(
SELECT location_id, MAX(id) AS MAXID
FROM location_update_log
GROUP BY location_id
) t2
ON t1.location_id = t2.location_id
AND t1.id = t2.MAXID
GROUP BY t1.location_id, t1.id
ORDER BY created_at DESC;
`);
}
};
/**
* @param {import('knex').Knex} knex
*/
exports.down = async function down(knex) {
if (knex.client.config.client !== 'sqlite3') {
await knex.schema
.raw('DROP VIEW location_update_log_latest;')
.alterTable('location_update_log', table => {
table.string('message').alter();
table.string('entity_name').nullable().alter();
}).raw(`
CREATE VIEW location_update_log_latest AS
SELECT t1.* FROM location_update_log t1
JOIN
(
SELECT location_id, MAX(id) AS MAXID
FROM location_update_log
GROUP BY location_id
) t2
ON t1.location_id = t2.location_id
AND t1.id = t2.MAXID
GROUP BY t1.location_id, t1.id
ORDER BY created_at DESC;
`);
}
};
plugins/catalog-backend/migrationsv2/20210209121210_locations_fk_index.js
+45
View File
@@ -0,0 +1,45 @@
/*
* Copyright 2020 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// @ts-check
/**
* @param {import('knex').Knex} knex
*/
exports.up = async function up(knex) {
if (knex.client.config.client !== 'sqlite3') {
await knex.schema.alterTable('entities', table => {
table.index('location_id', 'entity_location_id_idx');
});
await knex.schema.alterTable('location_update_log', table => {
table.index('location_id', 'update_log_location_id_idx');
});
}
};
/**
* @param {import('knex').Knex} knex
*/
exports.down = async function down(knex) {
if (knex.client.config.client !== 'sqlite3') {
await knex.schema.alterTable('entities', table => {
table.dropIndex([], 'entity_location_id_idx');
});
await knex.schema.alterTable('location_update_log', table => {
table.dropIndex([], 'update_log_location_id_idx');
});
}
};
plugins/catalog-backend/migrationsv2/20210302150147_refresh_state.js
+51 -18
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,15 @@ exports.up = async function up(knex) {
.notNullable()
.comment('JSON array containing all errors related to entity');
table
.dateTime('next_update_at')
.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')
.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');
});
@@ -94,38 +95,38 @@ exports.up = async function up(knex) {
'Holds edges between refresh state rows. Every time when an entity is processed and emits another entity, an edge will be stored to represent that fact. This is used to detect orphans and ultimately deletions.',
);
table
.text('source_special_key')
.increments('id')
.comment('Primary key to distinguish unique lines from each other');
table
.text('source_key')
.nullable()
.comment(
'When the reference source is not an entity, this is an opaque identifier for that source.',
);
table
.text('source_entity_id')
.text('source_entity_ref')
.nullable()
.references('entity_id')
.references('entity_ref')
.inTable('refresh_state')
.onDelete('CASCADE')
.comment(
'When the reference source is an entity, this is the ID of the source entity.',
'When the reference source is an entity, this is the EntityRef of the source entity.',
);
table
.text('target_entity_id')
.text('target_entity_ref')
.notNullable()
.references('entity_id')
.references('entity_ref')
.inTable('refresh_state')
.onDelete('CASCADE')
.comment('The ID of the target entity.');
.comment('The EntityRef of the target entity.');
table.index('source_key', 'refresh_state_references_source_key_idx');
table.index(
'source_special_key',
'refresh_state_references_source_special_key_idx',
'source_entity_ref',
'refresh_state_references_source_entity_ref_idx',
);
table.index(
'source_entity_id',
'refresh_state_references_source_entity_id_idx',
);
table.index(
'target_entity_id',
'refresh_state_references_target_entity_id_idx',
'target_entity_ref',
'refresh_state_references_target_entity_ref_idx',
);
});
@@ -153,6 +154,29 @@ exports.up = async function up(knex) {
table.index('source_entity_ref', 'relations_source_entity_ref_idx');
table.index('originating_entity_id', 'relations_source_entity_id_idx');
});
await knex.schema.createTable('search', table => {
table.comment(
'Flattened key-values from the entities, used for quick filtering',
);
table
.text('entity_id')
.references('entity_id')
.inTable('refresh_state')
.onDelete('CASCADE')
.comment('The entity that matches this key/value');
table
.string('key')
.notNullable()
.comment('A key that occurs in the entity');
table
.string('value')
.nullable()
.comment('The corresponding value to match on');
table.index(['entity_id'], 'search_entity_id_idx');
table.index(['key'], 'search_key_idx');
table.index(['value'], 'search_value_idx');
});
};
/**
@@ -165,6 +189,8 @@ 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');
});
@@ -175,6 +201,13 @@ exports.down = async function down(knex) {
table.index('source_entity_ref', 'relations_source_entity_ref_idx');
table.index('originating_entity_id', 'relations_source_entity_id_idx');
});
await knex.schema.alterTable('search', table => {
table.dropIndex([], 'search_entity_id_idx');
table.dropIndex([], 'search_key_idx');
table.dropIndex([], 'search_value_idx');
});
await knex.schema.dropTable('search');
await knex.schema.dropTable('final_entities');
await knex.schema.dropTable('relations');
await knex.schema.dropTable('references');
plugins/catalog-backend/package.json
+1 -4
View File
@@ -33,7 +33,6 @@
"@backstage/backend-common": "^0.6.3",
"@backstage/catalog-model": "^0.7.7",
"@backstage/config": "^0.1.4",
"@backstage/core": "^0.7.6",
"@backstage/errors": "^0.1.1",
"@backstage/integration": "^0.5.1",
"@backstage/plugin-search-backend-node": "^0.1.3",
@@ -61,8 +60,7 @@
"winston": "^3.2.1",
"yaml": "^1.9.2",
"yn": "^4.0.0",
"yup": "^0.29.3",
"zen-observable": "^0.8.15"
"yup": "^0.29.3"
},
"devDependencies": {
"@backstage/cli": "^0.6.9",
@@ -80,7 +78,6 @@
"files": [
"dist",
"migrations/**/*.{js,d.ts}",
"migrationsv2/**/*.{js,d.ts}",
"config.d.ts"
],
"configSchema": "config.d.ts"
plugins/catalog-backend/src/next/ConfigLocationProvider.test.ts
+67
View File
@@ -0,0 +1,67 @@
/*
* 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 { ConfigLocationProvider } from './ConfigLocationProvider';
import { EntityProviderConnection } from './types';
import { ConfigReader } from '@backstage/config';
import { resolvePackagePath } from '@backstage/backend-common';
import path from 'path';
describe('Config Location Provider', () => {
it('should apply mutation with the correct paths in the config', async () => {
const mockConfig = new ConfigReader({
catalog: {
locations: [
{ type: 'file', target: './lols.yaml' },
{ type: 'url', target: 'https://github.com/backstage/backstage' },
],
},
});
const mockConnection = ({
applyMutation: jest.fn(),
} as unknown) as EntityProviderConnection;
const locationProvider = new ConfigLocationProvider(mockConfig);
await locationProvider.connect(mockConnection);
expect(mockConnection.applyMutation).toHaveBeenCalledWith({
type: 'full',
entities: expect.arrayContaining([
expect.objectContaining({
spec: {
target: path.join(
resolvePackagePath('@backstage/plugin-catalog-backend'),
'./lols.yaml',
),
type: 'file',
},
}),
]),
});
expect(mockConnection.applyMutation).toHaveBeenCalledWith({
type: 'full',
entities: expect.arrayContaining([
expect.objectContaining({
spec: {
target: 'https://github.com/backstage/backstage',
type: 'url',
},
}),
]),
});
});
});
plugins/catalog-backend/src/next/ConfigLocationProvider.ts
+51
View File
@@ -0,0 +1,51 @@
/*
* 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 { EntityProviderConnection, EntityProvider } from './types';
import path from 'path';
import { Config } from '@backstage/config';
import { locationSpecToLocationEntity } from './util';
export class ConfigLocationProvider implements EntityProvider {
private connection: EntityProviderConnection | undefined;
constructor(private readonly config: Config) {}
getProviderName(): string {
return 'ConfigLocationProvider';
}
async connect(connection: EntityProviderConnection): Promise<void> {
this.connection = connection;
const locationConfigs =
this.config.getOptionalConfigArray('catalog.locations') ?? [];
const entities = locationConfigs.map(location => {
const type = location.getString('type');
const target = location.getString('target');
return locationSpecToLocationEntity({
type,
target: type === 'file' ? path.resolve(target) : target,
});
});
await this.connection.applyMutation({
type: 'full',
entities,
});
}
}
plugins/catalog-backend/src/next/DatabaseLocationProvider.ts
-71
View File
@@ -1,71 +0,0 @@
/*
* Copyright 2021 Spotify AB
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import { Observable } from '@backstage/core';
import { ENTITY_DEFAULT_NAMESPACE } from '@backstage/catalog-model';
import { EntityProvider, LocationStore, EntityMessage } from './types';
import ObservableImpl from 'zen-observable';
import {
locationSpecToLocationEntity,
locationSpecToMetadataName,
} from './util';
export class DatabaseLocationProvider implements EntityProvider {
private subscribers = new Set<
ZenObservable.SubscriptionObserver<EntityMessage>
>();
constructor(private readonly store: LocationStore) {
store.location$().subscribe({
next: locations => {
if ('all' in locations) {
this.notify({
all: locations.all.map(l => locationSpecToLocationEntity(l)),
});
} else {
this.notify({
added: locations.added.map(l => locationSpecToLocationEntity(l)),
removed: locations.removed.map(l => ({
kind: 'Location',
namespace: ENTITY_DEFAULT_NAMESPACE,
name: locationSpecToMetadataName(l),
})),
});
}
},
});
}
private notify(message: EntityMessage) {
for (const subscriber of this.subscribers) {
subscriber.next(message);
}
}
entityChange$(): Observable<EntityMessage> {
return new ObservableImpl(subscriber => {
this.store.listLocations().then(locations => {
subscriber.next({
all: locations.map(l => locationSpecToLocationEntity(l)),
});
this.subscribers.add(subscriber);
});
return () => {
this.subscribers.delete(subscriber);
};
});
}
}
plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts
+38 -35
View File
@@ -14,11 +14,11 @@
* limitations under the License.
*/
import { Subscription } from '@backstage/core';
import {
CatalogProcessingEngine,
EntityProvider,
EntityMessage,
EntityProviderConnection,
EntityProviderMutation,
ProcessingStateManager,
CatalogProcessingOrchestrator,
} from './types';
@@ -27,8 +27,35 @@ import { Logger } from 'winston';
import { stringifyEntityRef } from '@backstage/catalog-model';
import { Stitcher } from './Stitcher';
class Connection implements EntityProviderConnection {
constructor(
private readonly config: {
stateManager: ProcessingStateManager;
id: string;
},
) {}
async applyMutation(mutation: EntityProviderMutation): Promise<void> {
if (mutation.type === 'full') {
await this.config.stateManager.replaceProcessingItems({
sourceKey: this.config.id,
type: 'full',
items: mutation.entities,
});
return;
}
await this.config.stateManager.replaceProcessingItems({
sourceKey: this.config.id,
type: 'delta',
added: mutation.added,
removed: mutation.removed,
});
}
}
export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
private subscriptions: Subscription[] = [];
private running: boolean = false;
constructor(
@@ -41,11 +68,12 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
async start() {
for (const provider of this.entityProviders) {
const id = 'databaseProvider';
const subscription = provider
.entityChange$()
.subscribe({ next: m => this.onNext(id, m) });
this.subscriptions.push(subscription);
await provider.connect(
new Connection({
stateManager: this.stateManager,
id: provider.getProviderName(),
}),
);
}
this.running = true;
@@ -54,12 +82,12 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
const {
id,
entity,
state: intialState,
state: initialState,
} = await this.stateManager.getNextProcessingItem();
const result = await this.orchestrator.process({
entity,
state: intialState,
state: initialState,
});
for (const error of result.errors) {
@@ -92,30 +120,5 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
async stop() {
this.running = false;
for (const subscription of this.subscriptions) {
subscription.unsubscribe();
}
}
private async onNext(id: string, message: EntityMessage) {
if ('all' in message) {
// TODO unhandled rejection
await this.stateManager.addProcessingItems({
id,
type: 'provider',
entities: message.all,
});
}
if ('added' in message) {
await this.stateManager.addProcessingItems({
id,
type: 'provider',
entities: message.added,
});
// TODO deletions of message.removed
}
}
}
plugins/catalog-backend/src/next/DefaultLocationStore.test.ts
+147
View File
@@ -0,0 +1,147 @@
/*
* 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 { DatabaseManager } from './database/DatabaseManager';
import { DefaultLocationStore } from './DefaultLocationStore';
import { v4 } from 'uuid';
/* eslint-disable */
xdescribe('Default Location Store', () => {
const createLocationStore = async () => {
const db = await DatabaseManager.createTestDatabase();
const connection = { applyMutation: jest.fn() };
const store = new DefaultLocationStore(db);
await store.connect(connection);
return { store, connection };
};
it('should do a full sync with the locations on connect', async () => {
const { connection } = await createLocationStore();
expect(connection.applyMutation).toHaveBeenCalledWith({
type: 'full',
entities: [],
});
});
describe('listLocations', () => {
it('lists empty locations when there is no locations', async () => {
const { store } = await createLocationStore();
expect(await store.listLocations()).toEqual([]);
});
it('lists locations that are added to the db', async () => {
const { store } = await createLocationStore();
await store.createLocation({
target:
'https://github.com/backstage/demo/blob/master/catalog-info.yml',
type: 'url',
});
const listLocations = await store.listLocations();
expect(listLocations).toHaveLength(1);
expect(listLocations).toEqual(
expect.arrayContaining([
expect.objectContaining({
target:
'https://github.com/backstage/demo/blob/master/catalog-info.yml',
type: 'url',
}),
]),
);
});
});
describe('createLocation', () => {
it('throws when the location already exists', async () => {
const { store } = await createLocationStore();
const spec = {
target:
'https://github.com/backstage/demo/blob/master/catalog-info.yml',
type: 'url',
};
await store.createLocation(spec);
await expect(() => store.createLocation(spec)).rejects.toThrow(
new RegExp(`Location ${spec.type}:${spec.target} already exists`),
);
});
it('calls apply mutation when adding a new location', async () => {
const { store, connection } = await createLocationStore();
await store.createLocation({
target:
'https://github.com/backstage/demo/blob/master/catalog-info.yml',
type: 'url',
});
expect(connection.applyMutation).toHaveBeenCalledWith({
type: 'delta',
removed: [],
added: expect.arrayContaining([
expect.objectContaining({
spec: {
target:
'https://github.com/backstage/demo/blob/master/catalog-info.yml',
type: 'url',
},
}),
]),
});
});
});
describe('deleteLocation', () => {
it('throws if the location does not exist', async () => {
const { store } = await createLocationStore();
const id = v4();
await expect(() => store.deleteLocation(id)).rejects.toThrow(
new RegExp(`Found no location with ID ${id}`),
);
});
it('calls apply mutation when adding a new location', async () => {
const { store, connection } = await createLocationStore();
const location = await store.createLocation({
target:
'https://github.com/backstage/demo/blob/master/catalog-info.yml',
type: 'url',
});
await store.deleteLocation(location.id);
expect(connection.applyMutation).toHaveBeenCalledWith({
type: 'delta',
added: [],
removed: expect.arrayContaining([
expect.objectContaining({
spec: {
target:
'https://github.com/backstage/demo/blob/master/catalog-info.yml',
type: 'url',
},
}),
]),
});
});
});
});
plugins/catalog-backend/src/next/DefaultLocationStore.ts
+53 -45
View File
@@ -16,27 +16,26 @@
import { LocationSpec, Location } from '@backstage/catalog-model';
import { Database } from '../database';
import { LocationStore } from './types';
import {
LocationStore,
EntityProvider,
EntityProviderConnection,
} from './types';
import { v4 as uuidv4 } from 'uuid';
import { locationSpecToLocationEntity } from './util';
import { ConflictError } from '@backstage/errors';
import { Observable } from '@backstage/core';
import ObservableImpl from 'zen-observable';
export type LocationMessage =
| { all: Location[] }
| { added: Location[]; removed: Location[] };
export class DefaultLocationStore implements LocationStore {
private subscribers = new Set<
ZenObservable.SubscriptionObserver<LocationMessage>
>();
export class DefaultLocationStore implements LocationStore, EntityProvider {
private _connection: EntityProviderConnection | undefined;
constructor(private readonly db: Database) {}
createLocation(spec: LocationSpec): Promise<Location> {
return this.db.transaction(async tx => {
// TODO: id should really be type and target combined and not a uuid.
getProviderName(): string {
return 'DefaultLocationStore';
}
async createLocation(spec: LocationSpec): Promise<Location> {
return this.db.transaction(async tx => {
// Attempt to find a previous location matching the spec
const previousLocations = await this.listLocations();
const previousLocation = previousLocations.some(
@@ -49,13 +48,18 @@ export class DefaultLocationStore implements LocationStore {
);
}
// TODO: id should really be type and target combined and not a uuid.
const location = await this.db.addLocation(tx, {
id: uuidv4(),
type: spec.type,
target: spec.target,
});
this.notifyAddition(location);
await this.connection.applyMutation({
type: 'delta',
added: [locationSpecToLocationEntity(location)],
removed: [],
});
return location;
});
@@ -63,11 +67,17 @@ export class DefaultLocationStore implements LocationStore {
async listLocations(): Promise<Location[]> {
const dbLocations = await this.db.locations();
return dbLocations.map(item => ({
id: item.id,
target: item.target,
type: item.type,
}));
return (
dbLocations
// TODO(blam): We should create a mutation to remove this location for everyone
// eventually when it's all done and dusted
.filter(({ type }) => type !== 'bootstrap')
.map(item => ({
id: item.id,
target: item.target,
type: item.type,
}))
);
}
getLocation(id: string): Promise<Location> {
@@ -75,40 +85,38 @@ export class DefaultLocationStore implements LocationStore {
}
deleteLocation(id: string): Promise<void> {
if (!this.connection) {
throw new Error('location store is not initialized');
}
return this.db.transaction(async tx => {
const location = await this.db.location(id);
if (!location) {
throw new ConflictError(`No location found with id: ${id}`);
}
await this.db.removeLocation(tx, id);
this.notifyDeletion(location);
await this.connection.applyMutation({
type: 'delta',
added: [],
removed: [locationSpecToLocationEntity(location)],
});
});
}
private notifyAddition(location: Location) {
for (const subscriber of this.subscribers) {
subscriber.next({
added: [location],
removed: [],
});
private get connection(): EntityProviderConnection {
if (!this._connection) {
throw new Error('location store is not initialized');
}
return this._connection;
}
private notifyDeletion(location: Location) {
for (const subscriber of this.subscribers) {
subscriber.next({
added: [],
removed: [location],
});
}
}
location$(): Observable<LocationMessage> {
return new ObservableImpl<LocationMessage>(subscriber => {
this.subscribers.add(subscriber);
return () => {
this.subscribers.delete(subscriber);
};
async connect(connection: EntityProviderConnection): Promise<void> {
this._connection = connection;
const locations = await this.listLocations();
const entities = locations.map(location => {
return locationSpecToLocationEntity(location);
});
await this.connection.applyMutation({
type: 'full',
entities,
});
}
}
plugins/catalog-backend/src/next/DefaultProcessingStateManager.ts
+27 -32
View File
@@ -14,17 +14,25 @@
* limitations under the License.
*/
import { ProcessingDatabase, RefreshStateItem } from './database/types';
import { ProcessingDatabase } from './database/types';
import {
AddProcessingItemRequest,
ProccessingItem,
ProcessingItem,
ProcessingItemResult,
ProcessingStateManager,
ReplaceProcessingItemsRequest,
} from './types';
export class DefaultProcessingStateManager implements ProcessingStateManager {
constructor(private readonly db: ProcessingDatabase) {}
replaceProcessingItems(
request: ReplaceProcessingItemsRequest,
): Promise<void> {
return this.db.transaction(async tx => {
await this.db.replaceUnprocessedEntities(tx, request);
});
}
async setProcessingItemResult(result: ProcessingItemResult) {
return this.db.transaction(async tx => {
await this.db.updateProcessedEntity(tx, {
@@ -38,37 +46,24 @@ export class DefaultProcessingStateManager implements ProcessingStateManager {
});
}
async addProcessingItems(request: AddProcessingItemRequest) {
return this.db.transaction(async tx => {
await this.db.addUnprocessedEntities(tx, request);
});
}
async getNextProcessingItem(): Promise<ProccessingItem> {
const entities = await new Promise<RefreshStateItem[]>(resolve =>
this.popFromQueue(resolve),
);
const { id, state, unprocessedEntity } = entities[0];
return {
id,
entity: unprocessedEntity,
state,
};
}
async popFromQueue(resolve: (rows: RefreshStateItem[]) => void) {
const entities = await this.db.transaction(async tx => {
return this.db.getProcessableEntities(tx, {
processBatchSize: 1,
async getNextProcessingItem(): Promise<ProcessingItem> {
for (;;) {
const { items } = await this.db.transaction(async tx => {
return this.db.getProcessableEntities(tx, {
processBatchSize: 1,
});
});
});
// No entities require refresh, wait and try again.
if (!entities.items.length) {
setTimeout(() => this.popFromQueue(resolve), 1000);
return;
if (items.length) {
const { id, state, unprocessedEntity } = items[0];
return {
id,
entity: unprocessedEntity,
state,
};
}
await new Promise<void>(resolve => setTimeout(resolve, 1000));
}
resolve(entities.items);
}
}
plugins/catalog-backend/src/next/NextCatalogBuilder.ts
+3 -6
View File
@@ -53,7 +53,6 @@ import {
MicrosoftGraphOrgReaderProcessor,
PlaceholderProcessor,
PlaceholderResolver,
StaticLocationProcessor,
UrlReaderProcessor,
} from '../ingestion';
import { RepoLocationAnalyzer } from '../ingestion/LocationAnalyzer';
@@ -67,13 +66,13 @@ import { LocationAnalyzer } from '../ingestion/types';
import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine';
import { DefaultCatalogProcessingOrchestrator } from './DefaultCatalogProcessingOrchestrator';
import { DefaultProcessingDatabase } from './database/DefaultProcessingDatabase';
import { DatabaseLocationProvider } from '../next/DatabaseLocationProvider';
import { DefaultLocationStore } from './DefaultLocationStore';
import { DefaultProcessingStateManager } from './DefaultProcessingStateManager';
import { CatalogProcessingEngine } from '../next/types';
import { NextEntitiesCatalog } from './NextEntitiesCatalog';
import { Stitcher } from './Stitcher';
import { CommonDatabase } from '../database/CommonDatabase';
import { ConfigLocationProvider } from './ConfigLocationProvider';
export type CatalogEnvironment = {
logger: Logger;
@@ -272,11 +271,11 @@ export class NextCatalogBuilder {
const entitiesCatalog = new NextEntitiesCatalog(dbClient);
const locationStore = new DefaultLocationStore(db);
const dbLocationProvider = new DatabaseLocationProvider(locationStore);
const stitcher = new Stitcher(dbClient, logger);
const configLocationProvider = new ConfigLocationProvider(config);
const processingEngine = new DefaultCatalogProcessingEngine(
logger,
[dbLocationProvider], // entityproviders
[locationStore, configLocationProvider],
stateManager,
orchestrator,
stitcher,
@@ -324,7 +323,6 @@ export class NextCatalogBuilder {
// These are always there no matter what
const processors: CatalogProcessor[] = [
StaticLocationProcessor.fromConfig(config),
new PlaceholderProcessor({ resolvers: placeholderResolvers, reader }),
new BuiltinKindsEntityProcessor(),
];
@@ -340,7 +338,6 @@ export class NextCatalogBuilder {
MicrosoftGraphOrgReaderProcessor.fromConfig(config, { logger }),
new UrlReaderProcessor({ reader, logger }),
CodeOwnersProcessor.fromConfig(config, { logger, reader }),
// new LocationEntityProcessor({ integrations }),
new AnnotateLocationEntityProcessor({ integrations }),
);
}
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
+100 -39
View File
@@ -14,18 +14,20 @@
* 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 {
DbRefreshStateReferences,
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, 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
// errors in the underlying engine due to exceeding query limits, but large
// enough to get the speed benefits.
const BATCH_SIZE = 50;
export type DbFinalEntitiesRow = {
entity_id: string;
@@ -49,64 +51,123 @@ export class Stitcher {
for (const entityRef of entityRefs) {
await this.transaction(async txOpaque => {
const tx = txOpaque as Knex.Transaction;
const [result] = await tx<DbRefreshStateRow>('refresh_state')
.select('entity_id', 'processed_entity')
.where({ entity_ref: entityRef });
if (!result) {
// 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;
errors: string;
incomingReferenceCount: string | number;
previousEtag?: string;
relationType?: string;
relationTarget?: string;
}> = await tx
.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',
previousEtag: 'final_entities.etag',
relationType: 'relations.type',
relationTarget: 'relations.target_entity_ref',
})
.from('refresh_state')
.leftJoin('incoming_references', {})
.leftOuterJoin('final_entities', {
'final_entities.entity_id': 'refresh_state.entity_id',
})
.leftOuterJoin('relations', {
'relations.source_entity_ref': 'refresh_state.entity_ref',
})
.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`,
);
return;
} else if (!result.processed_entity) {
}
const {
entityId,
processedEntity,
// 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`,
);
return;
}
const entity: Entity = JSON.parse(result.processed_entity);
// 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;
const entityId = entity?.metadata?.uid;
if (!entityId) {
this.logger.error(`missing ID in entity ${JSON.stringify(entity)}`);
return;
}
const [reference_count_result] = await tx<DbRefreshStateReferences>(
'refresh_state_references',
)
.where({ target_entity_id: entity.metadata.uid })
.count({ reference_count: 'target_entity_id' });
if (Number(reference_count_result.reference_count) === 0) {
this.logger.debug(`${entityRef} is orphan`);
if (isOrphan) {
this.logger.debug(`${entityRef} is an orphan`);
entity.metadata.annotations = {
...entity.metadata.annotations,
['backstage.io/orphan']: 'true',
};
}
const relationResults = await tx<DbRelationsRow>('relations')
.where({ source_entity_ref: entityRef })
.select();
// TODO: entityRef is lower case and should be uppercase in the final result
entity.relations = result
.filter(row => row.relationType)
.map(row => ({
type: row.relationType!,
target: parseEntityRef(row.relationTarget!),
}));
// TODO: entityRef is lower case and should be uppercase in the final result.
entity.relations = relationResults.map(relation => ({
type: relation.type,
target: parseEntityRef(relation.target_entity_ref),
}));
entity.metadata.generation = 1;
// If the output entity was actually not changed, just abort
const etag = generateEntityEtag(entity);
if (etag === previousEtag) {
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')
.insert({
finalized_entity: JSON.stringify(entity),
entity_id: entityId,
finalized_entity: JSON.stringify(entity),
etag,
})
.onConflict('entity_id')
.merge(['finalized_entity', 'etag']);
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/database/DatabaseManager.ts
+99
View File
@@ -0,0 +1,99 @@
/*
* 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 { getVoidLogger, resolvePackagePath } from '@backstage/backend-common';
import knexFactory, { Knex } from 'knex';
import { v4 as uuidv4 } from 'uuid';
import { Logger } from 'winston';
import { CommonDatabase } from '../../database/CommonDatabase';
import { Database } from '../../database/types';
export type CreateDatabaseOptions = {
logger: Logger;
};
const defaultOptions: CreateDatabaseOptions = {
logger: getVoidLogger(),
};
export class DatabaseManager {
public static async createDatabase(
knex: Knex,
options: Partial<CreateDatabaseOptions> = {},
): Promise<Database> {
const migrationsDir = resolvePackagePath(
'@backstage/plugin-catalog-backend',
'migrationsv2',
);
await knex.migrate.latest({
directory: migrationsDir,
});
const { logger } = { ...defaultOptions, ...options };
return new CommonDatabase(knex, logger);
}
public static async createInMemoryDatabase(): Promise<Database> {
const knex = await this.createInMemoryDatabaseConnection();
return await this.createDatabase(knex);
}
public static async createInMemoryDatabaseConnection(): Promise<Knex> {
const knex = knexFactory({
client: 'sqlite3',
connection: ':memory:',
useNullAsDefault: true,
});
knex.client.pool.on('createSuccess', (_eventId: any, resource: any) => {
resource.run('PRAGMA foreign_keys = ON', () => {});
});
return knex;
}
public static async createTestDatabase(): Promise<Database> {
const knex = await this.createTestDatabaseConnection();
return await this.createDatabase(knex);
}
public static async createTestDatabaseConnection(): Promise<Knex> {
const config: Knex.Config<any> = {
client: 'sqlite3',
connection: ':memory:',
useNullAsDefault: true,
};
let knex = knexFactory(config);
if (typeof config.connection !== 'string') {
const tempDbName = `d${uuidv4().replace(/-/g, '')}`;
await knex.raw(`CREATE DATABASE ${tempDbName};`);
knex = knexFactory({
...config,
connection: {
...config.connection,
database: tempDbName,
},
});
}
knex.client.pool.on('createSuccess', (_eventId: any, resource: any) => {
resource.run('PRAGMA foreign_keys = ON', () => {});
});
return knex;
}
}
plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts
+517
View File
@@ -0,0 +1,517 @@
/*
* 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 { DefaultProcessingDatabase } from './DefaultProcessingDatabase';
import { DatabaseManager } from './DatabaseManager';
import { Knex } from 'knex';
import {
DbRefreshStateReferencesRow,
DbRefreshStateRow,
DbRelationsRow,
DefaultProcessingDatabase,
} from './DefaultProcessingDatabase';
import { Entity, EntityRelationSpec } from '@backstage/catalog-model';
import * as uuid from 'uuid';
import { getVoidLogger } from '@backstage/backend-common';
describe('Default Processing Database', () => {
let db: Knex;
let processingDatabase: DefaultProcessingDatabase;
const logger = getVoidLogger();
beforeEach(async () => {
db = await DatabaseManager.createTestDatabaseConnection();
await DatabaseManager.createDatabase(db);
processingDatabase = new DefaultProcessingDatabase(db, logger);
});
describe('replaceUnprocessedEntities', () => {
const insertRefRow = async (ref: DbRefreshStateReferencesRow) => {
return db<DbRefreshStateReferencesRow>('refresh_state_references').insert(
ref,
);
};
const insertRefreshStateRow = async (ref: DbRefreshStateRow) => {
await db<DbRefreshStateRow>('refresh_state').insert(ref);
};
const createLocations = async (entityRefs: string[]) => {
for (const ref of entityRefs) {
await insertRefreshStateRow({
entity_id: uuid.v4(),
entity_ref: ref,
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '',
next_update_at: 'now()',
last_discovery_at: 'now()',
});
}
};
it('replaces all existing state correctly for simple dependency chains', async () => {
/*
config -> location:default/root -> location:default/root-1 -> location:default/root-2
database -> location:default/second -> location:default/root-2
*/
await createLocations([
'location:default/root',
'location:default/root-1',
'location:default/root-2',
'location:default/second',
]);
await insertRefRow({
source_key: 'config',
target_entity_ref: 'location:default/root',
});
await insertRefRow({
source_key: 'database',
target_entity_ref: 'location:default/second',
});
await insertRefRow({
source_entity_ref: 'location:default/root',
target_entity_ref: 'location:default/root-1',
});
await insertRefRow({
source_entity_ref: 'location:default/root-1',
target_entity_ref: 'location:default/root-2',
});
await insertRefRow({
source_entity_ref: 'location:default/second',
target_entity_ref: 'location:default/root-2',
});
await processingDatabase.transaction(async tx => {
await processingDatabase.replaceUnprocessedEntities(tx, {
type: 'full',
sourceKey: 'config',
items: [
{
apiVersion: '1.0.0',
metadata: {
name: 'new-root',
},
kind: 'Location',
} as Entity,
],
});
});
const currentRefreshState = await db<DbRefreshStateRow>(
'refresh_state',
).select();
const currentRefRowState = await db<DbRefreshStateReferencesRow>(
'refresh_state_references',
).select();
for (const ref of ['location:default/root', 'location:default/root-1']) {
expect(currentRefreshState.some(t => t.entity_ref === ref)).toBeFalsy();
}
expect(
currentRefreshState.some(
t => t.entity_ref === 'location:default/new-root',
),
).toBeTruthy();
expect(
currentRefRowState.some(
t =>
t.source_entity_ref === 'location:default/root' &&
t.target_entity_ref === 'location:default/root-1',
),
).toBeFalsy();
expect(
currentRefRowState.some(
t =>
t.source_entity_ref === 'location:default/root-1' &&
t.target_entity_ref === 'location:default/root-2',
),
).toBeFalsy();
expect(
currentRefRowState.some(
t =>
t.target_entity_ref === 'location:default/root-1' &&
t.source_key === 'config',
),
).toBeFalsy();
expect(
currentRefRowState.some(
t =>
t.target_entity_ref === 'location:default/new-root' &&
t.source_key === 'config',
),
).toBeTruthy();
});
it('should work for more complex chains', async () => {
/*
config -> location:default/root -> location:default/root-1 -> location:default/root-2
config -> location:default/root -> location:default/root-1a -> location:default/root-2
*/
await createLocations([
'location:default/root',
'location:default/root-1',
'location:default/root-2',
'location:default/root-1a',
]);
await insertRefRow({
source_key: 'config',
target_entity_ref: 'location:default/root',
});
await insertRefRow({
source_entity_ref: 'location:default/root',
target_entity_ref: 'location:default/root-1',
});
await insertRefRow({
source_entity_ref: 'location:default/root',
target_entity_ref: 'location:default/root-1a',
});
await insertRefRow({
source_entity_ref: 'location:default/root-1',
target_entity_ref: 'location:default/root-2',
});
await insertRefRow({
source_entity_ref: 'location:default/root-1a',
target_entity_ref: 'location:default/root-2',
});
await processingDatabase.transaction(async tx => {
await processingDatabase.replaceUnprocessedEntities(tx, {
type: 'full',
sourceKey: 'config',
items: [
{
apiVersion: '1.0.0',
metadata: {
name: 'new-root',
},
kind: 'Location',
} as Entity,
],
});
});
const currentRefreshState = await db<DbRefreshStateRow>(
'refresh_state',
).select();
const currentRefRowState = await db<DbRefreshStateReferencesRow>(
'refresh_state_references',
).select();
const deletedRefs = [
'location:default/root',
'location:default/root-1',
'location:default/root-1a',
'location:default/root-2',
];
for (const ref of deletedRefs) {
expect(currentRefreshState.some(t => t.entity_ref === ref)).toBeFalsy();
}
expect(
currentRefreshState.some(
t => t.entity_ref === 'location:default/new-root',
),
).toBeTruthy();
expect(
currentRefRowState.some(
t =>
t.source_key === 'config' &&
t.target_entity_ref === 'location:default/new-root',
),
).toBeTruthy();
expect(
currentRefRowState.some(
t =>
t.source_key === 'config' &&
t.target_entity_ref === 'location:default/root',
),
).toBeFalsy();
expect(
currentRefRowState.some(
t =>
t.source_entity_ref === 'location:default/root' &&
t.target_entity_ref === 'location:default/root-1',
),
).toBeFalsy();
expect(
currentRefRowState.some(
t =>
t.source_entity_ref === 'location:default/root' &&
t.target_entity_ref === 'location:default/root-1a',
),
).toBeFalsy();
expect(
currentRefRowState.some(
t =>
t.source_entity_ref === 'location:default/root-1' &&
t.target_entity_ref === 'location:default/root-2',
),
).toBeFalsy();
expect(
currentRefRowState.some(
t =>
t.source_entity_ref === 'location:default/root-1a' &&
t.target_entity_ref === 'location:default/root-2',
),
).toBeFalsy();
});
it('should add new locations using the delta options', async () => {
await processingDatabase.transaction(async tx => {
await processingDatabase.replaceUnprocessedEntities(tx, {
type: 'delta',
sourceKey: 'lols',
removed: [],
added: [
{
apiVersion: '1.0.0',
metadata: {
name: 'new-root',
},
kind: 'Location',
} as Entity,
],
});
});
const currentRefreshState = await db<DbRefreshStateRow>(
'refresh_state',
).select();
const currentRefRowState = await db<DbRefreshStateReferencesRow>(
'refresh_state_references',
).select();
expect(
currentRefreshState.some(
t => t.entity_ref === 'location:default/new-root',
),
).toBeTruthy();
expect(
currentRefRowState.some(
t =>
t.source_key === 'lols' &&
t.target_entity_ref === 'location:default/new-root',
),
).toBeTruthy();
});
it('should not remove locations that are referenced elsewhere', async () => {
/*
config-1 -> location:default/root
config-2 -> location:default/root
*/
await createLocations(['location:default/root']);
await insertRefRow({
source_key: 'config-1',
target_entity_ref: 'location:default/root',
});
await insertRefRow({
source_key: 'config-2',
target_entity_ref: 'location:default/root',
});
await processingDatabase.transaction(async tx => {
await processingDatabase.replaceUnprocessedEntities(tx, {
type: 'full',
sourceKey: 'config-1',
items: [],
});
});
const currentRefreshState = await db<DbRefreshStateRow>(
'refresh_state',
).select();
const currentRefRowState = await db<DbRefreshStateReferencesRow>(
'refresh_state_references',
).select();
expect(currentRefRowState).toEqual([
expect.objectContaining({
source_key: 'config-2',
target_entity_ref: 'location:default/root',
}),
]);
expect(currentRefreshState).toEqual([
expect.objectContaining({
entity_ref: 'location:default/root',
}),
]);
});
it('should remove old locations using the delta options', async () => {
await createLocations(['location:default/new-root']);
await insertRefRow({
source_key: 'lols',
target_entity_ref: 'location:default/new-root',
});
await processingDatabase.transaction(async tx => {
await processingDatabase.replaceUnprocessedEntities(tx, {
type: 'delta',
sourceKey: 'lols',
added: [],
removed: [
{
apiVersion: '1.0.0',
metadata: {
name: 'new-root',
},
kind: 'Location',
} as Entity,
],
});
});
const currentRefreshState = await db<DbRefreshStateRow>(
'refresh_state',
).select();
const currentRefRowState = await db<DbRefreshStateReferencesRow>(
'refresh_state_references',
).select();
expect(
currentRefreshState.some(
t => t.entity_ref === 'location:default/new-root',
),
).toBeFalsy();
expect(
currentRefRowState.some(
t =>
t.source_key === 'lols' &&
t.target_entity_ref === 'location:default/new-root',
),
).toBeFalsy();
});
});
describe('updateProcessedEntity', () => {
it('should throw if the entity does not exist', async () => {
await processingDatabase.transaction(async tx => {
await expect(
processingDatabase.updateProcessedEntity(tx, {
id: '9',
processedEntity: {
apiVersion: '1.0.0',
metadata: {
name: 'new-root',
},
kind: 'Location',
} as Entity,
deferredEntities: [],
relations: [],
}),
).rejects.toThrow('Processing state not found for 9');
});
});
it('should update a processed entity', async () => {
await db<DbRefreshStateRow>('refresh_state').insert({
entity_id: '321',
entity_ref: 'location:default/new-root',
unprocessed_entity: '',
errors: '',
next_update_at: 'now()',
last_discovery_at: 'now()',
});
const deferredEntity = {
apiVersion: '1.0.0',
metadata: {
name: 'deferred',
},
kind: 'Location',
} as Entity;
const relation: EntityRelationSpec = {
source: {
kind: 'Component',
namespace: 'Default',
name: 'foo',
},
target: {
kind: 'Component',
namespace: 'Default',
name: 'foo',
},
type: 'url',
};
await processingDatabase.transaction(async tx => {
await processingDatabase.updateProcessedEntity(tx, {
id: '321',
processedEntity: {
apiVersion: '1.0.0',
metadata: {
name: 'new-root',
},
kind: 'Location',
} as Entity,
deferredEntities: [deferredEntity],
relations: [relation],
});
});
const deferredResult = await db<DbRefreshStateRow>('refresh_state')
.where({ entity_ref: 'location:default/deferred' })
.select();
expect(deferredResult.length).toBe(1);
const [relations] = await db<DbRelationsRow>('relations')
.where({ originating_entity_id: '321' })
.select();
expect(relations).toEqual({
originating_entity_id: '321',
source_entity_ref: 'component:default/foo',
type: 'url',
target_entity_ref: 'component:default/foo',
});
});
});
});
plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts
+237 -47
View File
@@ -15,6 +15,7 @@
*/
import { ConflictError, NotFoundError } from '@backstage/errors';
import { stringifyEntityRef, Entity } from '@backstage/catalog-model';
import { Knex } from 'knex';
import { Transaction } from '../../database';
import lodash from 'lodash';
@@ -24,20 +25,23 @@ import {
AddUnprocessedEntitiesOptions,
UpdateProcessedEntityOptions,
GetProcessableEntitiesResult,
ReplaceUnprocessedEntitiesOptions,
RefreshStateItem,
} from './types';
import type { Logger } from 'winston';
import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
import { v4 as uuid } from 'uuid';
import { JsonObject } from '@backstage/config';
export type DbRefreshStateRow = {
entity_id: string;
entity_ref: string;
unprocessed_entity: string;
processed_entity: string;
cache: string;
processed_entity?: string;
cache?: string;
next_update_at: string;
last_discovery_at: string; // remove?
errors: string;
errors?: string;
};
export type DbRelationsRow = {
@@ -47,10 +51,10 @@ export type DbRelationsRow = {
type: string;
};
export type DbRefreshStateReferences = {
source_special_key?: string;
source_entity_id?: string;
target_entity_id: string;
export type DbRefreshStateReferencesRow = {
source_key?: string;
source_entity_ref?: string;
target_entity_ref: string;
};
// The number of items that are sent per batch to the database layer, when
@@ -86,7 +90,6 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
errors,
})
.where('entity_id', id);
if (refreshResult === 0) {
throw new NotFoundError(`Processing state not found for ${id}`);
}
@@ -94,8 +97,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
// Schedule all deferred entities for future processing.
await this.addUnprocessedEntities(tx, {
entities: deferredEntities,
id,
type: 'entity',
entityRef: stringifyEntityRef(processedEntity),
});
// Update fragments
@@ -128,49 +130,230 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
);
}
private async createDelta(
tx: Knex.Transaction,
options: ReplaceUnprocessedEntitiesOptions,
): Promise<{ toAdd: Entity[]; toRemove: string[] }> {
if (options.type === 'delta') {
return {
toAdd: options.added,
toRemove: options.removed.map(e => stringifyEntityRef(e)),
};
}
const oldRefs = await tx<DbRefreshStateReferencesRow>(
'refresh_state_references',
)
.where({ source_key: options.sourceKey })
.select('target_entity_ref')
.then(rows => rows.map(r => r.target_entity_ref));
const items = options.items.map(entity => ({
entity,
ref: stringifyEntityRef(entity),
}));
const oldRefsSet = new Set(oldRefs);
const newRefsSet = new Set(items.map(item => item.ref));
const toAdd = items.filter(item => !oldRefsSet.has(item.ref));
const toRemove = oldRefs.filter(ref => !newRefsSet.has(ref));
return { toAdd: toAdd.map(({ entity }) => entity), toRemove };
}
async replaceUnprocessedEntities(
txOpaque: Transaction,
options: ReplaceUnprocessedEntitiesOptions,
): Promise<void> {
const tx = txOpaque as Knex.Transaction;
const { toAdd, toRemove } = await this.createDelta(tx, options);
if (toRemove.length) {
// TODO(freben): Batch split, to not hit variable limits?
/*
WITH RECURSIVE
-- All the nodes that can be reached downwards from our root
descendants(root_id, entity_ref) AS (
SELECT id, target_entity_ref
FROM refresh_state_references
WHERE source_key = "R1" AND target_entity_ref = "A"
UNION
SELECT descendants.root_id, target_entity_ref
FROM descendants
JOIN refresh_state_references ON source_entity_ref = descendants.entity_ref
),
-- All the nodes that can be reached upwards from the descendants
ancestors(root_id, via_entity_ref, to_entity_ref) AS (
SELECT NULL, entity_ref, entity_ref
FROM descendants
UNION
SELECT
CASE WHEN source_key IS NOT NULL THEN id ELSE NULL END,
source_entity_ref,
ancestors.to_entity_ref
FROM ancestors
JOIN refresh_state_references ON target_entity_ref = ancestors.via_entity_ref
)
-- Start out with all of the descendants
SELECT descendants.entity_ref
FROM descendants
-- Expand with all ancestors that point to those, but aren't the current root
LEFT OUTER JOIN ancestors
ON ancestors.to_entity_ref = descendants.entity_ref
AND ancestors.root_id IS NOT NULL
AND ancestors.root_id != descendants.root_id
-- Exclude all lines that had such a foreign ancestor
WHERE ancestors.root_id IS NULL;
*/
const removedCount = await tx<DbRefreshStateRow>('refresh_state')
.whereIn('entity_ref', function orphanedEntityRefs(orphans) {
return (
orphans
// All the nodes that can be reached downwards from our root
.withRecursive('descendants', function descendants(outer) {
return outer
.select({ root_id: 'id', entity_ref: 'target_entity_ref' })
.from('refresh_state_references')
.where('source_key', options.sourceKey)
.whereIn('target_entity_ref', toRemove)
.union(function recursive(inner) {
return inner
.select({
root_id: 'descendants.root_id',
entity_ref:
'refresh_state_references.target_entity_ref',
})
.from('descendants')
.join('refresh_state_references', {
'descendants.entity_ref':
'refresh_state_references.source_entity_ref',
});
});
})
// All the nodes that can be reached upwards from the descendants
.withRecursive('ancestors', function ancestors(outer) {
return outer
.select({
root_id: tx.raw('NULL', []),
via_entity_ref: 'entity_ref',
to_entity_ref: 'entity_ref',
})
.from('descendants')
.union(function recursive(inner) {
return inner
.select({
root_id: tx.raw(
'CASE WHEN source_key IS NOT NULL THEN id ELSE NULL END',
[],
),
via_entity_ref: 'source_entity_ref',
to_entity_ref: 'ancestors.to_entity_ref',
})
.from('ancestors')
.join('refresh_state_references', {
target_entity_ref: 'ancestors.via_entity_ref',
});
});
})
// Start out with all of the descendants
.select('descendants.entity_ref')
.from('descendants')
// Expand with all ancestors that point to those, but aren't the current root
.leftOuterJoin('ancestors', function keepaliveRoots() {
this.on(
'ancestors.to_entity_ref',
'=',
'descendants.entity_ref',
);
this.andOnNotNull('ancestors.root_id');
this.andOn('ancestors.root_id', '!=', 'descendants.root_id');
})
.whereNull('ancestors.root_id')
);
})
.delete();
await tx<DbRefreshStateReferencesRow>('refresh_state_references')
.where('source_key', '=', options.sourceKey)
.whereIn('target_entity_ref', toRemove)
.delete();
this.logger.debug(
`removed, ${removedCount} entities: ${JSON.stringify(toRemove)}`,
);
}
if (toAdd.length) {
const state: Knex.DbRecord<DbRefreshStateRow>[] = toAdd.map(entity => ({
entity_id: uuid(),
entity_ref: stringifyEntityRef(entity),
unprocessed_entity: JSON.stringify(entity),
errors: '',
next_update_at: tx.fn.now(),
last_discovery_at: tx.fn.now(),
}));
const stateReferences: DbRefreshStateReferencesRow[] = toAdd.map(
entity => ({
source_key: options.sourceKey,
target_entity_ref: stringifyEntityRef(entity),
}),
);
// TODO(freben): Concurrency? If we did these one by one, a .onConflict().merge would have made sense
await tx.batchInsert('refresh_state', state, BATCH_SIZE);
await tx.batchInsert(
'refresh_state_references',
stateReferences,
BATCH_SIZE,
);
}
}
async addUnprocessedEntities(
txOpaque: Transaction,
options: AddUnprocessedEntitiesOptions,
): Promise<void> {
const tx = txOpaque as Knex.Transaction;
const entityIds = new Array<string>();
for (const entity of options.entities) {
const entityRef = stringifyEntityRef(entity);
await tx<DbRefreshStateRow>('refresh_state')
.insert({
const stateRows = options.entities.map(
entity =>
({
entity_id: uuid(),
entity_ref: entityRef,
entity_ref: stringifyEntityRef(entity),
unprocessed_entity: JSON.stringify(entity),
errors: '',
next_update_at: tx.fn.now(),
last_discovery_at: tx.fn.now(),
})
} as Knex.DbRecord<DbRefreshStateRow>),
);
const stateReferenceRows = stateRows.map(
stateRow =>
({
source_entity_ref: options.entityRef,
target_entity_ref: stateRow.entity_ref,
} as Knex.DbRecord<DbRefreshStateReferencesRow>),
);
// Upsert all of the unprocessed entities into the refresh_state table, by
// their entity ref.
// TODO(freben): Can this be batched somehow?
for (const row of stateRows) {
await tx<DbRefreshStateRow>('refresh_state')
.insert(row)
.onConflict('entity_ref')
.merge(['unprocessed_entity', 'last_discovery_at']);
const [{ entity_id: entityId }] = await tx<DbRefreshStateRow>(
'refresh_state',
).where({ entity_ref: entityRef });
entityIds.push(entityId);
}
const key =
options.type === 'provider'
? { source_special_key: options.id }
: { source_entity_id: options.id };
// copied from update refs
await tx<DbRefreshStateReferences>('refresh_state_references')
.where(key)
// Replace all references for the originating entity before creating new ones
await tx<DbRefreshStateReferencesRow>('refresh_state_references')
.where({ source_entity_ref: options.entityRef })
.delete();
const referenceRows: DbRefreshStateReferences[] = entityIds.map(
entityId => ({
...key,
target_entity_id: entityId,
}),
await tx.batchInsert(
'refresh_state_references',
stateReferenceRows,
BATCH_SIZE,
);
await tx.batchInsert('refresh_state_references', referenceRows, BATCH_SIZE);
}
async getProcessableEntities(
@@ -198,16 +381,23 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
});
return {
items: items.map(i => ({
id: i.entity_id,
entityRef: i.entity_ref,
unprocessedEntity: JSON.parse(i.unprocessed_entity) as Entity,
processedEntity: JSON.parse(i.processed_entity) as Entity,
nextUpdateAt: i.next_update_at,
lastDiscoveryAt: i.last_discovery_at,
state: JSON.parse(i.cache),
errors: i.errors,
})),
items: items.map(
i =>
({
id: i.entity_id,
entityRef: i.entity_ref,
unprocessedEntity: JSON.parse(i.unprocessed_entity) as Entity,
processedEntity: i.processed_entity
? (JSON.parse(i.processed_entity) as Entity)
: undefined,
nextUpdateAt: i.next_update_at,
lastDiscoveryAt: i.last_discovery_at,
state: i.cache
? JSON.parse(i.cache)
: new Map<string, JsonObject>(),
errors: i.errors,
} as RefreshStateItem),
),
};
}
plugins/catalog-backend/src/next/database/types.ts
+20 -4
View File
@@ -19,8 +19,7 @@ import { JsonObject } from '@backstage/config';
import { Transaction } from '../../database/types';
export type AddUnprocessedEntitiesOptions = {
type: 'entity' | 'provider';
id: string;
entityRef: string;
entities: Entity[];
};
@@ -39,17 +38,30 @@ export type RefreshStateItem = {
id: string;
entityRef: string;
unprocessedEntity: Entity;
processedEntity: Entity;
processedEntity?: Entity;
nextUpdateAt: string;
lastDiscoveryAt: string; // remove?
state: Map<string, JsonObject>;
errors: string;
errors?: string;
};
export type GetProcessableEntitiesResult = {
items: RefreshStateItem[];
};
export type ReplaceUnprocessedEntitiesOptions =
| {
sourceKey: string;
items: Entity[];
type: 'full';
}
| {
sourceKey: string;
added: Entity[];
removed: Entity[];
type: 'delta';
};
export interface ProcessingDatabase {
transaction<T>(fn: (tx: Transaction) => Promise<T>): Promise<T>;
@@ -58,6 +70,10 @@ export interface ProcessingDatabase {
options: AddUnprocessedEntitiesOptions,
): Promise<void>;
replaceUnprocessedEntities(
txOpaque: Transaction,
options: ReplaceUnprocessedEntitiesOptions,
): Promise<void>;
getProcessableEntities(
txOpaque: Transaction,
request: { processBatchSize: number },
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
+27 -21
View File
@@ -13,15 +13,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import {
Entity,
EntityName,
LocationSpec,
Location,
EntityRelationSpec,
} from '@backstage/catalog-model';
import { JsonObject } from '@backstage/config';
import { Observable } from '@backstage/core'; // << nooo
export interface LocationEntity {
apiVersion: 'backstage.io/v1alpha1';
@@ -45,20 +44,11 @@ export interface LocationService {
deleteLocation(id: string): Promise<void>;
}
export type EntityMessage =
| { all: Entity[] }
| { added: Entity[]; removed: EntityName[] };
export interface LocationStore {
// extends EntityProvider
createLocation(spec: LocationSpec): Promise<Location>;
listLocations(): Promise<Location[]>;
getLocation(id: string): Promise<Location>;
deleteLocation(id: string): Promise<void>;
location$(): Observable<
{ all: Location[] } | { added: Location[]; removed: Location[] }
>;
}
export interface CatalogProcessingEngine {
@@ -66,8 +56,17 @@ export interface CatalogProcessingEngine {
stop(): Promise<void>;
}
export type EntityProviderMutation =
| { type: 'full'; entities: Entity[] }
| { type: 'delta'; added: Entity[]; removed: Entity[] };
export interface EntityProviderConnection {
applyMutation(mutation: EntityProviderMutation): Promise<void>;
}
export interface EntityProvider {
entityChange$(): Observable<EntityMessage>;
getProviderName(): string;
connect(connection: EntityProviderConnection): Promise<void>;
}
export type EntityProcessingRequest = {
@@ -103,20 +102,27 @@ export type ProcessingItemResult = {
deferredEntities: Entity[];
};
export type AddProcessingItemRequest = {
type: 'entity' | 'provider';
id: string;
entities: Entity[];
};
export type ProccessingItem = {
export type ProcessingItem = {
id: string;
entity: Entity;
state: Map<string, JsonObject>;
};
export type ReplaceProcessingItemsRequest =
| {
sourceKey: string;
items: Entity[];
type: 'full';
}
| {
sourceKey: string;
added: Entity[];
removed: Entity[];
type: 'delta';
};
export interface ProcessingStateManager {
setProcessingItemResult(result: ProcessingItemResult): Promise<void>;
getNextProcessingItem(): Promise<ProccessingItem>;
addProcessingItems(request: AddProcessingItemRequest): Promise<void>;
getNextProcessingItem(): Promise<ProcessingItem>;
replaceProcessingItems(request: ReplaceProcessingItemsRequest): Promise<void>;
}