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

Consolidate migrations and use default schema

Browse Source 
Signed-off-by: Damon Kaswell <damon.kaswell1@hp.com>
This commit is contained in:
Damon Kaswell
2022-11-17 10:54:05 -08:00
committed by Fredrik Adelöw
parent 8a235456e9
commit 6608e08317
6 changed files with 91 additions and 240 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/incremental-ingestion-backend/migrations/20220906161750_add-ingestion-indexes.js
-70
View File
@@ -1,70 +0,0 @@
/*
* 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.
*/
/**
* @param { import("knex").Knex } knex
*/
exports.up = async function up(knex) {
const schema = () => knex.schema.withSchema('ingestion');
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}'));`,
);
await knex.raw(`DROP VIEW IF EXISTS ingestion.current_entities`);
await schema().alterTable('ingestions', t => {
t.primary('id');
t.index('provider_name', 'ingestion_provider_name_idx');
});
await schema().alterTable('ingestion_marks', t => {
t.primary('id');
t.index('ingestion_id', 'ingestion_mark_ingestion_id_idx');
});
await schema().alterTable('ingestion_mark_entities', t => {
t.primary('id');
t.index('ingestion_mark_id', 'ingestion_mark_entity_ingestion_mark_id_idx');
});
};
/**
* @param { import("knex").Knex } knex
*/
exports.down = async function down(knex) {
const schema = () => knex.schema.withSchema('ingestion');
await schema().alterTable('ingestions', t => {
t.dropIndex('provider_name', 'ingestion_provider_name_idx');
t.dropPrimary('id');
});
await schema().alterTable('ingestion_marks', t => {
t.dropIndex('ingestion_id', 'ingestion_mark_ingestion_id_idx');
t.dropPrimary('id');
});
await schema().alterTable('ingestions_mark_entities', t => {
t.dropIndex(
'ingestion_mark_id',
'ingestion_mark_entity_ingestion_mark_id_idx',
);
t.dropPrimary('id');
});
await knex.raw(`DROP INDEX increment_ingestion_provider_name_idx;`);
};
plugins/incremental-ingestion-backend/migrations/20221019897897_add-unique-active-ingestion.js
-100
View File
@@ -1,100 +0,0 @@
/*
* 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.
*/
const { v4: uuidv4 } = require('uuid');
/**
* @param { import("knex").Knex } knex
*/
exports.up = async function up(knex) {
const schema = () => knex.schema.withSchema('ingestion');
await knex.transaction(async tx => {
const providers = await tx('ingestion.ingestions').distinct(
'provider_name',
);
for (const provider of providers) {
const valid = await tx('ingestion.ingestions')
.where('provider_name', provider)
.andWhere('rest_completed_at', null)
.first();
const invalid = [];
const rows = await tx('ingestion.ingestions')
.where('provider_name', provider)
.andWhere('rest_completed_at', null);
for (const row of rows) {
if (row.id !== valid.id) {
invalid.push(row.id);
}
}
if (invalid.length > 0) {
await tx('ingestion.ingestions').delete().whereIn('id', invalid);
await tx('ingestion.ingestion_mark_entities')
.delete()
.whereIn(
'ingestion_mark_id',
tx('ingestion.ingestion_marks')
.select('id')
.whereIn('ingestion_id', invalid),
);
await tx('ingestion.ingestion_marks')
.delete()
.whereIn('ingestion_id', invalid);
}
}
});
await schema().alterTable('ingestions', t => {
t.string('completion_ticket');
});
await knex.transaction(async tx => {
await tx('ingestion.ingestions')
.update('completion_ticket', 'open')
.where('rest_completed_at', null);
const rows = await tx('ingestion.ingestions').whereNot(
'rest_completed_at',
null,
);
for (const row of rows) {
await tx('ingestion.ingestions')
.update('completion_ticket', uuidv4())
.where('id', row.id);
}
});
await schema().alterTable('ingestions', t => {
t.string('completion_ticket').notNullable().alter();
t.unique(['provider_name', 'completion_ticket'], {
indexName: 'ingestion_composite_index',
deferrable: 'deferred',
});
});
};
/**
* @param { import("knex").Knex } knex
*/
exports.down = async function down(knex) {
const schema = () => knex.schema.withSchema('ingestion');
await schema().alterTable('ingestions', t => {
t.dropUnique(['provider_name', 'completion_ticket']);
});
};
plugins/incremental-ingestion-backend/migrations/20220613125155_init.js → plugins/incremental-ingestion-backend/migrations/20221116073152_init.js
+44 -14
View File
@@ -18,19 +18,19 @@
* @param { import("knex").Knex } knex
*/
exports.up = async function up(knex) {
await knex.raw(`
CREATE VIEW ingestion.current_entities as SELECT
FORMAT('%s:%s/%s',
LOWER(final_entity::json #>> '{kind}'),
LOWER(final_entity::json #>> '{metadata, namespace}'),
LOWER(final_entity::json #>> '{metadata, name}')) as ref,
final_entity::json #>> '{metadata, annotations, backstage.io/incremental-provider-name}' as provider_name,
final_entity FROM public.final_entities;
`);
/**
* Create an index on the final_entities table for the provider name
* annotation
*/
const schema = () => knex.schema.withSchema('ingestion');
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}'));`,
);
await schema().createTable('ingestions', table => {
/**
* Sets up the ingestions table
*/
await knex.schema.createTable('ingestions', table => {
table.comment('Tracks ingestion streams for very large data sets');
table
@@ -81,9 +81,26 @@ CREATE VIEW ingestion.current_entities as SELECT
table
.timestamp('rest_completed_at')
.comment('when did the rest period actually end');
table
.string('completion_ticket')
.notNullable()
.comment('indicates whether the ticket is still open or stamped complete')
});
await schema().createTable('ingestion_marks', table => {
await knex.schema.alterTable('ingestions', t => {
t.primary('id');
t.index('provider_name', 'ingestion_provider_name_idx');
t.unique(['provider_name', 'completion_ticket'], {
indexName: 'ingestion_composite_index',
deferrable: 'deferred',
});
});
/**
* Sets up the ingestion_marks table
*/
await knex.schema.createTable('ingestion_marks', table => {
table.comment('tracks each step of an iterative ingestion');
table
@@ -110,6 +127,14 @@ CREATE VIEW ingestion.current_entities as SELECT
table.timestamp('created_at').defaultTo(knex.fn.now());
});
await schema().alterTable('ingestion_marks', t => {
t.primary('id');
t.index('ingestion_id', 'ingestion_mark_ingestion_id_idx');
});
/**
* Set up the ingestion_mark_entities table
*/
await schema().createTable('ingestion_mark_entities', table => {
table.comment(
'tracks the entities recorded in each step of an iterative ingestion',
@@ -132,15 +157,20 @@ CREATE VIEW ingestion.current_entities as SELECT
.notNullable()
.comment('the entity reference of the marked entity');
});
await schema().alterTable('ingestion_mark_entities', t => {
t.primary('id');
t.index('ingestion_mark_id', 'ingestion_mark_entity_ingestion_mark_id_idx');
});
};
/**
* @param { import("knex").Knex } knex
*/
exports.down = async function down(knex) {
const schema = () => knex.schema.withSchema('ingestion');
await schema().dropView('current_entities');
const schema = () => knex.schema.withSchema('public');
await schema().dropTable('ingestion_mark_entities');
await schema().dropTable('ingestion_marks');
await schema().dropTable('ingestions');
await knex.raw(`DROP INDEX increment_ingestion_provider_name_idx;`);
};
plugins/incremental-ingestion-backend/src/database/IncrementalIngestionDatabaseManager.ts
+43 -49
View File
@@ -43,7 +43,7 @@ export class IncrementalIngestionDatabaseManager {
async updateIngestionRecordById(options: IngestionRecordUpdate) {
await this.client.transaction(async tx => {
const { ingestionId, update } = options;
await tx('ingestion.ingestions').where('id', ingestionId).update(update);
await tx('ingestions').where('id', ingestionId).update(update);
});
}
@@ -57,7 +57,7 @@ export class IncrementalIngestionDatabaseManager {
update: Partial<IngestionUpsertIFace>,
) {
await this.client.transaction(async tx => {
await tx('ingestion.ingestions')
await tx('ingestions')
.where('provider_name', provider)
.andWhere('completion_ticket', 'open')
.update(update);
@@ -65,12 +65,12 @@ export class IncrementalIngestionDatabaseManager {
}
/**
* Performs an insert into the `ingestion.ingestions` table with the supplied values.
* Performs an insert into the `ingestions` table with the supplied values.
* @param record - IngestionUpsertIFace
*/
async insertIngestionRecord(record: IngestionUpsertIFace) {
await this.client.transaction(async tx => {
await tx('ingestion.ingestions').insert(record);
await tx('ingestions').insert(record);
});
}
@@ -87,7 +87,7 @@ export class IncrementalIngestionDatabaseManager {
let deleted = 0;
for (const chunk of chunks) {
const chunkDeleted = await tx('ingestion.ingestion_mark_entities')
const chunkDeleted = await tx('ingestion_mark_entities')
.delete()
.whereIn(
'id',
@@ -106,7 +106,7 @@ export class IncrementalIngestionDatabaseManager {
*/
async getCurrentIngestionRecord(provider: string) {
return await this.client.transaction(async tx => {
const record = await tx<IngestionRecord>('ingestion.ingestions')
const record = await tx<IngestionRecord>('ingestions')
.where('provider_name', provider)
.andWhere('completion_ticket', 'open')
.first();
@@ -122,32 +122,32 @@ export class IncrementalIngestionDatabaseManager {
*/
async clearFinishedIngestions(provider: string) {
return await this.client.transaction(async tx => {
const markEntitiesDeleted = await tx('ingestion.ingestion_mark_entities')
const markEntitiesDeleted = await tx('ingestion_mark_entities')
.delete()
.whereIn(
'ingestion_mark_id',
tx('ingestion.ingestion_marks')
tx('ingestion_marks')
.select('id')
.whereIn(
'ingestion_id',
tx('ingestion.ingestions')
tx('ingestions')
.select('id')
.where('provider_name', provider)
.andWhereNot('completion_ticket', 'open'),
),
);
const marksDeleted = await tx('ingestion.ingestion_marks')
const marksDeleted = await tx('ingestion_marks')
.delete()
.whereIn(
'ingestion_id',
tx('ingestion.ingestions')
tx('ingestions')
.select('id')
.where('provider_name', provider)
.andWhereNot('completion_ticket', 'open'),
);
const ingestionsDeleted = await tx('ingestion.ingestions')
const ingestionsDeleted = await tx('ingestions')
.delete()
.where('provider_name', provider)
.andWhereNot('completion_ticket', 'open');
@@ -171,24 +171,20 @@ export class IncrementalIngestionDatabaseManager {
*/
async clearDuplicateIngestions(ingestionId: string, provider: string) {
await this.client.transaction(async tx => {
const invalid = await tx<IngestionRecord>('ingestion.ingestions')
const invalid = await tx<IngestionRecord>('ingestions')
.where('provider_name', provider)
.andWhere('rest_completed_at', null)
.andWhereNot('id', ingestionId);
if (invalid.length > 0) {
await tx('ingestion.ingestions').delete().whereIn('id', invalid);
await tx('ingestion.ingestion_mark_entities')
await tx('ingestions').delete().whereIn('id', invalid);
await tx('ingestion_mark_entities')
.delete()
.whereIn(
'ingestion_mark_id',
tx('ingestion.ingestion_marks')
.select('id')
.whereIn('ingestion_id', invalid),
tx('ingestion_marks').select('id').whereIn('ingestion_id', invalid),
);
await tx('ingestion.ingestion_marks')
.delete()
.whereIn('ingestion_id', invalid);
await tx('ingestion_marks').delete().whereIn('ingestion_id', invalid);
}
});
}
@@ -201,13 +197,13 @@ export class IncrementalIngestionDatabaseManager {
*/
async purgeAndResetProvider(provider: string) {
return await this.client.transaction(async tx => {
const ingestionIDs: { id: string }[] = await tx('ingestion.ingestions')
const ingestionIDs: { id: string }[] = await tx('ingestions')
.select('id')
.where('provider_name', provider);
const markIDs: { id: string }[] =
ingestionIDs.length > 0
? await tx('ingestion.ingestion_marks')
? await tx('ingestion_marks')
.select('id')
.whereIn(
'ingestion_id',
@@ -217,7 +213,7 @@ export class IncrementalIngestionDatabaseManager {
const markEntityIDs: { id: string }[] =
markIDs.length > 0
? await tx('ingestion.ingestion_mark_entities')
? await tx('ingestion_mark_entities')
.select('id')
.whereIn(
'ingestion_mark_id',
@@ -232,7 +228,7 @@ export class IncrementalIngestionDatabaseManager {
const marksDeleted =
markIDs.length > 0
? await tx('ingestion.ingestion_marks')
? await tx('ingestion_marks')
.delete()
.whereIn(
'ingestion_id',
@@ -240,7 +236,7 @@ export class IncrementalIngestionDatabaseManager {
)
: 0;
const ingestionsDeleted = await tx('ingestion.ingestions')
const ingestionsDeleted = await tx('ingestions')
.delete()
.where('provider_name', provider);
@@ -310,14 +306,14 @@ export class IncrementalIngestionDatabaseManager {
)
.whereNotIn(
'entity_ref',
tx('ingestion.ingestion_marks')
tx('ingestion_marks')
.join(
'ingestion.ingestion_mark_entities',
'ingestion.ingestion_marks.id',
'ingestion.ingestion_mark_entities.ingestion_mark_id',
'ingestion_mark_entities',
'ingestion_marks.id',
'ingestion_mark_entities.ingestion_mark_id',
)
.select('ingestion.ingestion_mark_entities.ref')
.where('ingestion.ingestion_marks.ingestion_id', ingestionId),
.select('ingestion_mark_entities.ref')
.where('ingestion_marks.ingestion_id', ingestionId),
);
return removed.map(entity => {
return { entity: JSON.parse(entity.entity) };
@@ -332,7 +328,7 @@ export class IncrementalIngestionDatabaseManager {
async healthcheck() {
return await this.client.transaction(async tx => {
const records = await tx<{ id: string; provider_name: string }>(
'ingestion.ingestions',
'ingestions',
)
.distinct('id', 'provider_name')
.where('rest_completed_at', null);
@@ -352,9 +348,9 @@ export class IncrementalIngestionDatabaseManager {
/**
* Purges the following tables:
* * `ingestion.ingestions`
* * `ingestion.ingestion_marks`
* * `ingestion.ingestion_mark_entities`
* * `ingestions`
* * `ingestion_marks`
* * `ingestion_mark_entities`
*
* This function leaves the ingestions table with all providers in a paused state.
* @returns Results from cleaning up all ingestion tables.
@@ -362,7 +358,7 @@ export class IncrementalIngestionDatabaseManager {
async cleanupProviders() {
const providers = await this.listProviders();
const ingestionsDeleted = await this.purgeTable('ingestion.ingestions');
const ingestionsDeleted = await this.purgeTable('ingestions');
const next_action_at = new Date();
next_action_at.setTime(next_action_at.getTime() + 24 * 60 * 60 * 1000);
@@ -379,11 +375,9 @@ export class IncrementalIngestionDatabaseManager {
});
}
const ingestionMarksDeleted = await this.purgeTable(
'ingestion.ingestion_marks',
);
const ingestionMarksDeleted = await this.purgeTable('ingestion_marks');
const markEntitiesDeleted = await this.purgeTable(
'ingestion.ingestion_mark_entities',
'ingestion_mark_entities',
);
return { ingestionsDeleted, ingestionMarksDeleted, markEntitiesDeleted };
@@ -512,13 +506,13 @@ export class IncrementalIngestionDatabaseManager {
}
/**
* Returns the last record from `ingestion.ingestion_marks` for the supplied ingestionId.
* Returns the last record from `ingestion_marks` for the supplied ingestionId.
* @param ingestionId - string
* @returns MarkRecord | undefined
*/
async getLastMark(ingestionId: string) {
return await this.client.transaction(async tx => {
const mark = await tx<MarkRecord>('ingestion.ingestion_marks')
const mark = await tx<MarkRecord>('ingestion_marks')
.where('ingestion_id', ingestionId)
.orderBy('sequence', 'desc')
.first();
@@ -528,7 +522,7 @@ export class IncrementalIngestionDatabaseManager {
async getAllMarks(ingestionId: string) {
return await this.client.transaction(async tx => {
const marks = await tx<MarkRecord>('ingestion.ingestion_marks')
const marks = await tx<MarkRecord>('ingestion_marks')
.where('ingestion_id', ingestionId)
.orderBy('sequence', 'desc');
return marks;
@@ -536,23 +530,23 @@ export class IncrementalIngestionDatabaseManager {
}
/**
* Performs an insert into the `ingestion.ingestion_marks` table with the supplied values.
* Performs an insert into the `ingestion_marks` table with the supplied values.
* @param options - MarkRecordInsert
*/
async createMark(options: MarkRecordInsert) {
const { record } = options;
await this.client.transaction(async tx => {
await tx('ingestion.ingestion_marks').insert(record);
await tx('ingestion_marks').insert(record);
});
}
/**
* Performs an upsert to the `ingestion.ingestion_mark_entities` table for all deferred entities.
* Performs an upsert to the `ingestion_mark_entities` table for all deferred entities.
* @param markId - string
* @param entities - DeferredEntity[]
*/
async createMarkEntities(markId: string, entities: DeferredEntity[]) {
await this.client.transaction(async tx => {
await tx('ingestion.ingestion_mark_entities').insert(
await tx('ingestion_mark_entities').insert(
entities.map(entity => ({
id: v4(),
ingestion_mark_id: markId,
@@ -580,7 +574,7 @@ export class IncrementalIngestionDatabaseManager {
async listProviders() {
return await this.client.transaction(async tx => {
const providers = await tx<{ provider_name: string }>(
'ingestion.ingestions',
'ingestions',
).distinct('provider_name');
return providers.map(entry => entry.provider_name);
});
plugins/incremental-ingestion-backend/src/database/migrations.ts
+1 -4
View File
@@ -13,20 +13,17 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import { resolvePackagePath } from '@backstage/backend-common';
import { Knex } from 'knex';
/** @public */
export async function applyDatabaseMigrations(knex: Knex): Promise<void> {
const migrationsDir = resolvePackagePath(
'@backstage/plugin-incremental-ingestion-backend',
'migrations',
);
await knex.raw('CREATE SCHEMA IF NOT EXISTS ingestion;');
await knex.migrate.latest({
schemaName: 'ingestion',
directory: migrationsDir,
});
}
plugins/incremental-ingestion-backend/src/types.ts
+3 -3
View File
@@ -164,7 +164,7 @@ export interface IterationEngineOptions {
}
/**
* The shape of data inserted into or updated in the `ingestion.ingestions` table.
* The shape of data inserted into or updated in the `ingestions` table.
*
* @public
*/
@@ -234,7 +234,7 @@ export interface IngestionRecordUpdate {
}
/**
* The expected response from the `ingestion.ingestion_marks` table.
* The expected response from the `ingestion_marks` table.
*
* @public
*/
@@ -247,7 +247,7 @@ export interface MarkRecord {
}
/**
* The expected response from the `ingestion.ingestions` table.
* The expected response from the `ingestions` table.
*
* @public
*/