From d6dde4664f38a5a9cd43f7cd65023d4edeb8534d Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Mon, 21 Nov 2022 14:45:56 -0800 Subject: [PATCH] Fixes for migrations and routes Signed-off-by: Damon Kaswell --- .../migrations/20221116073152_init.js | 19 +++++++---------- .../IncrementalIngestionDatabaseManager.ts | 12 ++++++----- .../src/database/migrations.ts | 2 ++ .../src/database/tables.ts | 17 +++++++++++++++ .../src/router/paths.ts | 19 +++++++++++++++++ .../src/{ => router}/routes.ts | 21 ++++++++++--------- .../src/service/IncrementalCatalogBuilder.ts | 2 +- 7 files changed, 65 insertions(+), 27 deletions(-) create mode 100644 plugins/incremental-ingestion-backend/src/database/tables.ts create mode 100644 plugins/incremental-ingestion-backend/src/router/paths.ts rename plugins/incremental-ingestion-backend/src/{ => router}/routes.ts (90%) diff --git a/plugins/incremental-ingestion-backend/migrations/20221116073152_init.js b/plugins/incremental-ingestion-backend/migrations/20221116073152_init.js index edbe041afe..7fab18b186 100644 --- a/plugins/incremental-ingestion-backend/migrations/20221116073152_init.js +++ b/plugins/incremental-ingestion-backend/migrations/20221116073152_init.js @@ -18,15 +18,6 @@ * @param { import("knex").Knex } knex */ exports.up = async function up(knex) { - /** - * Create an index on the final_entities table for the provider name - * annotation - */ - - await knex.raw( - `CREATE INDEX IF NOT EXISTS increment_ingestion_provider_name_idx ON public.final_entities ((final_entity::json #>> '{metadata, annotations, backstage.io/incremental-provider-name}'));`, - ); - /** * Sets up the ingestions table */ @@ -57,6 +48,7 @@ exports.up = async function up(knex) { table .timestamp('next_action_at') + .notNullable() .defaultTo(knex.fn.now()) .comment('the moment in time at which point ingestion can begin again'); @@ -66,11 +58,13 @@ exports.up = async function up(knex) { table .integer('attempts') + .notNullable() .defaultTo(0) .comment('how many attempts have been made to burst without success'); table .timestamp('created_at') + .notNullable() .defaultTo(knex.fn.now()) .comment('when did this ingestion actually begin'); @@ -124,10 +118,14 @@ exports.up = async function up(knex) { table .integer('sequence') + .notNullable() .defaultTo(0) .comment('what is the order of this mark'); - table.timestamp('created_at').defaultTo(knex.fn.now()); + table + .timestamp('created_at') + .notNullable() + .defaultTo(knex.fn.now()); }); await knex.schema.alterTable('ingestion_marks', t => { @@ -177,5 +175,4 @@ exports.down = async function down(knex) { await knex.schema.dropTable('ingestion_mark_entities'); await knex.schema.dropTable('ingestion_marks'); await knex.schema.dropTable('ingestions'); - await knex.raw(`DROP INDEX increment_ingestion_provider_name_idx;`); }; diff --git a/plugins/incremental-ingestion-backend/src/database/IncrementalIngestionDatabaseManager.ts b/plugins/incremental-ingestion-backend/src/database/IncrementalIngestionDatabaseManager.ts index 6a8a69b886..891f94b537 100644 --- a/plugins/incremental-ingestion-backend/src/database/IncrementalIngestionDatabaseManager.ts +++ b/plugins/incremental-ingestion-backend/src/database/IncrementalIngestionDatabaseManager.ts @@ -300,10 +300,7 @@ export class IncrementalIngestionDatabaseManager { 'refresh_state.entity_id', 'final_entities.entity_id', ) - .whereRaw( - `((final_entity::json #>> '{metadata, annotations, ${INCREMENTAL_ENTITY_PROVIDER_ANNOTATION}}')) = ?`, - [provider], - ) + .join('search', 'search.entity_id', 'final_entities.entity_id') .whereNotIn( 'entity_ref', tx('ingestion_marks') @@ -314,7 +311,12 @@ export class IncrementalIngestionDatabaseManager { ) .select('ingestion_mark_entities.ref') .where('ingestion_marks.ingestion_id', ingestionId), - ); + ) + .andWhere( + 'search.key', + `metadata.annotations.${INCREMENTAL_ENTITY_PROVIDER_ANNOTATION}`, + ) + .andWhere('search.value', provider); return removed.map(entity => { return { entity: JSON.parse(entity.entity) }; }); diff --git a/plugins/incremental-ingestion-backend/src/database/migrations.ts b/plugins/incremental-ingestion-backend/src/database/migrations.ts index 45c8964054..58ccd7c8c5 100644 --- a/plugins/incremental-ingestion-backend/src/database/migrations.ts +++ b/plugins/incremental-ingestion-backend/src/database/migrations.ts @@ -16,6 +16,7 @@ import { resolvePackagePath } from '@backstage/backend-common'; import { Knex } from 'knex'; +import { DB_MIGRATIONS_TABLE } from './tables'; export async function applyDatabaseMigrations(knex: Knex): Promise { const migrationsDir = resolvePackagePath( @@ -25,5 +26,6 @@ export async function applyDatabaseMigrations(knex: Knex): Promise { await knex.migrate.latest({ directory: migrationsDir, + tableName: DB_MIGRATIONS_TABLE, }); } diff --git a/plugins/incremental-ingestion-backend/src/database/tables.ts b/plugins/incremental-ingestion-backend/src/database/tables.ts new file mode 100644 index 0000000000..73542abb6b --- /dev/null +++ b/plugins/incremental-ingestion-backend/src/database/tables.ts @@ -0,0 +1,17 @@ +/* + * Copyright 2021 The Backstage Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +export const DB_MIGRATIONS_TABLE = 'plugin_incremental_ingestion_backend__knex_migrations'; diff --git a/plugins/incremental-ingestion-backend/src/router/paths.ts b/plugins/incremental-ingestion-backend/src/router/paths.ts new file mode 100644 index 0000000000..1be540fc10 --- /dev/null +++ b/plugins/incremental-ingestion-backend/src/router/paths.ts @@ -0,0 +1,19 @@ +/* + * Copyright 2022 The Backstage Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +export const PROVIDER_CLEANUP = '/providers/cleanup'; +export const PROVIDER_HEALTH = '/providers/health'; +export const PROVIDER_BASE_PATH = '/providers/:provider'; diff --git a/plugins/incremental-ingestion-backend/src/routes.ts b/plugins/incremental-ingestion-backend/src/router/routes.ts similarity index 90% rename from plugins/incremental-ingestion-backend/src/routes.ts rename to plugins/incremental-ingestion-backend/src/router/routes.ts index 2f51ef6499..8607a4d302 100644 --- a/plugins/incremental-ingestion-backend/src/routes.ts +++ b/plugins/incremental-ingestion-backend/src/router/routes.ts @@ -17,7 +17,8 @@ import { errorHandler } from '@backstage/backend-common'; import express from 'express'; import Router from 'express-promise-router'; import { Logger } from 'winston'; -import { IncrementalIngestionDatabaseManager } from './'; +import { IncrementalIngestionDatabaseManager } from '..'; +import { PROVIDER_BASE_PATH, PROVIDER_CLEANUP, PROVIDER_HEALTH } from './paths'; /** @public */ export const createIncrementalProviderRouter = async ( @@ -28,7 +29,7 @@ export const createIncrementalProviderRouter = async ( router.use(express.json()); // Get the overall health of all incremental providers - router.get('/health', async (_, res) => { + router.get(PROVIDER_HEALTH, async (_, res) => { const records = await manager.healthcheck(); const providers = records.map(record => record.provider_name); const duplicates = [ @@ -43,13 +44,13 @@ export const createIncrementalProviderRouter = async ( }); // Clean up and pause all providers - router.delete('/cleanup', async (_, res) => { + router.post(PROVIDER_CLEANUP, async (_, res) => { const result = await manager.cleanupProviders(); res.json(result); }); // Get basic status of the provider - router.get('/:provider', async (req, res) => { + router.get(PROVIDER_BASE_PATH, async (req, res) => { const { provider } = req.params; const record = await manager.getCurrentIngestionRecord(provider); if (record) { @@ -84,7 +85,7 @@ export const createIncrementalProviderRouter = async ( }); // Trigger the provider's next action - router.put('/:provider', async (req, res) => { + router.put(PROVIDER_BASE_PATH, async (req, res) => { const { provider } = req.params; const record = await manager.getCurrentIngestionRecord(provider); if (record) { @@ -111,7 +112,7 @@ export const createIncrementalProviderRouter = async ( }); // Start a brand-new ingestion cycle for the provider - router.post('/:provider', async (req, res) => { + router.post(PROVIDER_BASE_PATH, async (req, res) => { const { provider } = req.params; const record = await manager.getCurrentIngestionRecord(provider); @@ -144,7 +145,7 @@ export const createIncrementalProviderRouter = async ( }); // Stop the provider and pause it for 24 hours - router.post('/:provider/cancel', async (req, res) => { + router.post(`${PROVIDER_BASE_PATH}/cancel`, async (req, res) => { const { provider } = req.params; const record = await manager.getCurrentIngestionRecord(provider); if (record) { @@ -178,14 +179,14 @@ export const createIncrementalProviderRouter = async ( }); // Wipe out all ingestion records for the provider and pause for 24 hours - router.delete('/:provider', async (req, res) => { + router.delete(PROVIDER_BASE_PATH, async (req, res) => { const { provider } = req.params; const result = await manager.purgeAndResetProvider(provider); res.json(result); }); // Get the ingestion marks for the current cycle - router.get('/:provider/marks', async (req, res) => { + router.get(`${PROVIDER_BASE_PATH}/marks`, async (req, res) => { const { provider } = req.params; const record = await manager.getCurrentIngestionRecord(provider); if (record) { @@ -213,7 +214,7 @@ export const createIncrementalProviderRouter = async ( } }); - router.delete('/:provider/marks', async (req, res) => { + router.delete(`${PROVIDER_BASE_PATH}/marks`, async (req, res) => { const { provider } = req.params; const deletions = await manager.clearFinishedIngestions(provider); diff --git a/plugins/incremental-ingestion-backend/src/service/IncrementalCatalogBuilder.ts b/plugins/incremental-ingestion-backend/src/service/IncrementalCatalogBuilder.ts index 7bf39e160f..587c1d5834 100644 --- a/plugins/incremental-ingestion-backend/src/service/IncrementalCatalogBuilder.ts +++ b/plugins/incremental-ingestion-backend/src/service/IncrementalCatalogBuilder.ts @@ -24,7 +24,7 @@ import { Knex } from 'knex'; import { IncrementalIngestionEngine } from '../engine/IncrementalIngestionEngine'; import { applyDatabaseMigrations } from '../database/migrations'; import { IncrementalIngestionDatabaseManager } from '../database/IncrementalIngestionDatabaseManager'; -import { createIncrementalProviderRouter } from '../routes'; +import { createIncrementalProviderRouter } from '../router/routes'; class Deferred implements Promise { #resolve?: (value: T) => void;