diff --git a/.changeset/pink-fans-yell.md b/.changeset/pink-fans-yell.md new file mode 100644 index 0000000000..ba04495852 --- /dev/null +++ b/.changeset/pink-fans-yell.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend': patch +--- + +Properly handle Date objects being returned for timestamps in the database driver diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts index 7d4d2ef8db..f7dede4adb 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts @@ -85,8 +85,8 @@ describe('DefaultCatalogProcessingEngine', () => { }, resultHash: '', state: new Map(), - nextUpdateAt: DateTime.now().toSQL(), - lastDiscoveryAt: '', + nextUpdateAt: DateTime.now(), + lastDiscoveryAt: DateTime.now(), }, ], }); @@ -148,8 +148,8 @@ describe('DefaultCatalogProcessingEngine', () => { }, resultHash: '', state: new Map(), - nextUpdateAt: DateTime.now().toSQL(), - lastDiscoveryAt: '', + nextUpdateAt: DateTime.now(), + lastDiscoveryAt: DateTime.now(), }, ], }); @@ -182,8 +182,8 @@ describe('DefaultCatalogProcessingEngine', () => { unprocessedEntity: entity, resultHash: 'the matching hash', state: new Map(), - nextUpdateAt: DateTime.now().toSQL(), - lastDiscoveryAt: '', + nextUpdateAt: DateTime.now(), + lastDiscoveryAt: DateTime.now(), }; hash.digest.mockReturnValue('the matching hash'); diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index 41e81b1777..d33094ec6d 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -22,11 +22,13 @@ import { import { serializeError } from '@backstage/errors'; import { Hash } from 'crypto'; import stableStringify from 'fast-json-stable-stringify'; -import { DateTime } from 'luxon'; import { Logger } from 'winston'; import { ProcessingDatabase, RefreshStateItem } from './database/types'; import { createCounterMetric, createSummaryMetric } from './metrics'; -import { CatalogProcessingOrchestrator } from './processing/types'; +import { + CatalogProcessingOrchestrator, + EntityProcessingResult, +} from './processing/types'; import { Stitcher } from './stitching/Stitcher'; import { startTaskPipeline } from './TaskPipeline'; import { @@ -85,21 +87,8 @@ class Connection implements EntityProviderConnection { } export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { + private readonly tracker = progressTracker(); private stopFunc?: () => void; - private readonly metrics = { - processedEntities: createCounterMetric({ - name: 'catalog_processed_entities_count', - help: 'Amount of entities processed', - }), - processingDuration: createSummaryMetric({ - name: 'catalog_processing_duration_seconds', - help: 'Processing duration', - }), - processingQueueDelay: createSummaryMetric({ - name: 'catalog_processing_queue_delay_seconds', - help: 'The amount of delay between being scheduled for processing, and the start of actually being processed', - }), - }; constructor( private readonly logger: Logger, @@ -143,16 +132,9 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { } }, processTask: async item => { - let endTimer; - try { - this.metrics.processedEntities.inc(1); - this.metrics.processingQueueDelay.observe( - -DateTime.fromSQL(item.nextUpdateAt, { zone: 'UTC' }) - .diffNow() - .as('seconds'), - ); - endTimer = this.metrics.processingDuration.startTimer(); + const track = this.tracker.processStart(item, this.logger); + try { const { id, state, @@ -166,6 +148,8 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { state, }); + track.markProcessorsCompleted(result); + for (const error of result.errors) { // TODO(freben): Try to extract the location out of the unprocessed // entity and add as meta to the log lines @@ -191,6 +175,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { // If nothing changed in our produced outputs, we cannot have any // significant effect on our surroundings; therefore, we just abort // without any updates / stitching. + track.markSuccessfulWithNoChanges(); return; } @@ -212,6 +197,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { await this.stitcher.stitch( new Set([stringifyEntityRef(unprocessedEntity)]), ); + track.markSuccessfulWithErrors(); return; } @@ -236,10 +222,10 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { ), ]); await this.stitcher.stitch(setOfThingsToStitch); + + track.markSuccessfulWithChanges(setOfThingsToStitch.size); } catch (error) { - this.logger.warn('Processing failed with:', error); - } finally { - endTimer?.(); + track.markFailed(error); } }, }); @@ -252,3 +238,76 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { } } } + +// Helps wrap the timing and logging behaviors +function progressTracker() { + const stitchedEntities = createCounterMetric({ + name: 'catalog_stitched_entities_count', + help: 'Amount of entities stitched', + }); + const processedEntities = createCounterMetric({ + name: 'catalog_processed_entities_count', + help: 'Amount of entities processed', + labelNames: ['result'], + }); + const processingDuration = createSummaryMetric({ + name: 'catalog_processing_duration_seconds', + help: 'Time spent executing the full processing flow', + labelNames: ['result'], + }); + const processorsDuration = createSummaryMetric({ + name: 'catalog_processors_duration_seconds', + help: 'Time spent executing catalog processors', + labelNames: ['result'], + }); + const processingQueueDelay = createSummaryMetric({ + name: 'catalog_processing_queue_delay_seconds', + help: 'The amount of delay between being scheduled for processing, and the start of actually being processed', + }); + + function processStart(item: RefreshStateItem, logger: Logger) { + logger.debug(`Processing ${item.entityRef}`); + + if (item.nextUpdateAt) { + processingQueueDelay.observe(-item.nextUpdateAt.diffNow().as('seconds')); + } + + const endOverallTimer = processingDuration.startTimer(); + const endProcessorsTimer = processorsDuration.startTimer(); + + function markProcessorsCompleted(result: EntityProcessingResult) { + endProcessorsTimer({ result: result.ok ? 'ok' : 'failed' }); + } + + function markSuccessfulWithNoChanges() { + endOverallTimer({ result: 'unchanged' }); + processedEntities.inc({ result: 'unchanged' }, 1); + } + + function markSuccessfulWithErrors() { + endOverallTimer({ result: 'errors' }); + processedEntities.inc({ result: 'errors' }, 1); + } + + function markSuccessfulWithChanges(stitchedCount: number) { + endOverallTimer({ result: 'changed' }); + stitchedEntities.inc(stitchedCount); + processedEntities.inc({ result: 'changed' }, 1); + } + + function markFailed(error: Error) { + processedEntities.inc({ result: 'failed' }, 1); + logger.warn(`Processing of ${item.entityRef} failed`, error); + } + + return { + markProcessorsCompleted, + markSuccessfulWithNoChanges, + markSuccessfulWithErrors, + markSuccessfulWithChanges, + markFailed, + }; + } + + return { processStart }; +} diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 7326a25a63..30457f2cbd 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -24,6 +24,7 @@ import type { Logger } from 'winston'; import { Transaction } from '../../database'; import { DeferredEntity } from '../processing/types'; import { RefreshIntervalFunction } from '../refresh'; +import { rethrowError, timestampToDateTime } from './conversion'; import { initDatabaseMetrics } from './metrics'; import { DbRefreshStateReferencesRow, @@ -471,6 +472,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { .limit(request.processBatchSize) .orderBy('next_update_at', 'asc'); + const interval = this.options.refreshInterval(); await tx('refresh_state') .whereIn( 'entity_ref', @@ -479,12 +481,8 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { .update({ next_update_at: tx.client.config.client === 'sqlite3' - ? tx.raw(`datetime('now', ?)`, [ - `${this.options.refreshInterval()} seconds`, - ]) - : tx.raw( - `now() + interval '${this.options.refreshInterval()} seconds'`, - ), + ? tx.raw(`datetime('now', ?)`, [`${interval} seconds`]) + : tx.raw(`now() + interval '${interval} seconds'`), }); return { @@ -498,8 +496,8 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { ? (JSON.parse(i.processed_entity) as Entity) : undefined, resultHash: i.result_hash || '', - nextUpdateAt: i.next_update_at, - lastDiscoveryAt: i.last_discovery_at, + nextUpdateAt: timestampToDateTime(i.next_update_at), + lastDiscoveryAt: timestampToDateTime(i.last_discovery_at), state: i.cache ? JSON.parse(i.cache) : new Map(), @@ -529,15 +527,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { return result!; } catch (e) { this.options.logger.debug(`Error during transaction, ${e}`); - - if ( - /SQLITE_CONSTRAINT: UNIQUE/.test(e.message) || - /unique constraint/.test(e.message) - ) { - throw new ConflictError(`Rejected due to a conflicting entity`, e); - } - - throw e; + throw rethrowError(e); } } } diff --git a/plugins/catalog-backend/src/next/database/conversion.test.ts b/plugins/catalog-backend/src/next/database/conversion.test.ts new file mode 100644 index 0000000000..b345348827 --- /dev/null +++ b/plugins/catalog-backend/src/next/database/conversion.test.ts @@ -0,0 +1,54 @@ +/* + * Copyright 2021 The Backstage Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { timestampToDateTime, rethrowError } from './conversion'; + +describe('timestampToDateTime', () => { + it('converts all known types', () => { + const js = new Date(Date.UTC(2021, 7, 20, 10, 11, 12)); + const sql = '2021-08-20 10:11:12'; + const iso = '2021-08-20T10:11:12Z'; + expect(timestampToDateTime(js).toISO()).toBe('2021-08-20T10:11:12.000Z'); + expect(timestampToDateTime(sql).toISO()).toBe('2021-08-20T10:11:12.000Z'); + expect(timestampToDateTime(iso).toISO()).toBe('2021-08-20T10:11:12.000Z'); + }); +}); + +describe('rethrowError', () => { + it('leaves regular errors untouched', () => { + const e = new Error('nothing special here'); + expect(() => rethrowError(e)).toThrow(e); + }); + + it('translates to conflict error when appropriate', () => { + const sqliteUnique = new Error('SQLITE_CONSTRAINT: UNIQUE blah'); + const postgresUnique = new Error('unique constraint foo'); + + expect(() => rethrowError(sqliteUnique)).toThrow( + expect.objectContaining({ + name: 'ConflictError', + message: expect.stringContaining('SQLITE_CONSTRAINT: UNIQUE blah'), + }), + ); + + expect(() => rethrowError(postgresUnique)).toThrow( + expect.objectContaining({ + name: 'ConflictError', + message: expect.stringContaining('unique constraint foo'), + }), + ); + }); +}); diff --git a/plugins/catalog-backend/src/next/database/conversion.ts b/plugins/catalog-backend/src/next/database/conversion.ts new file mode 100644 index 0000000000..54b642f3ff --- /dev/null +++ b/plugins/catalog-backend/src/next/database/conversion.ts @@ -0,0 +1,58 @@ +/* + * Copyright 2021 The Backstage Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { ConflictError, InputError } from '@backstage/errors'; +import { DateTime } from 'luxon'; + +/** + * Takes a TIMESTAMP type column and converts it to a DateTime. + * + * Some engines return the SQL string form (e.g. 'YYYY-MM-DD hh:mm:ss'), some + * return ISO string form (e.g. 'YYYY-MM-DDThh:mm:ss.SSSZ'), some return a js + * Date object. + */ +export function timestampToDateTime(input: Date | string): DateTime { + try { + if (typeof input === 'object') { + return DateTime.fromJSDate(input).toUTC(); + } + + const result = input.includes(' ') + ? DateTime.fromSQL(input, { zone: 'utc' }) + : DateTime.fromISO(input, { zone: 'utc' }); + if (!result.isValid) { + throw new TypeError('Not valid'); + } + + return result; + } catch (e) { + throw new InputError(`Failed to parse database timestamp ${input}`, e); + } +} + +/** + * Rethrows an error, possibly translating it to a more precise error type. + */ +export function rethrowError(e: any): never { + if ( + /SQLITE_CONSTRAINT: UNIQUE/.test(e.message) || + /unique constraint/.test(e.message) + ) { + throw new ConflictError(`Rejected due to a conflicting entity`, e); + } + + throw e; +} diff --git a/plugins/catalog-backend/src/next/database/tables.ts b/plugins/catalog-backend/src/next/database/tables.ts index d735607554..b40964d225 100644 --- a/plugins/catalog-backend/src/next/database/tables.ts +++ b/plugins/catalog-backend/src/next/database/tables.ts @@ -27,8 +27,8 @@ export type DbRefreshStateRow = { processed_entity?: string; result_hash?: string; cache?: string; - next_update_at: string; - last_discovery_at: string; // remove? + next_update_at: string | Date; + last_discovery_at: string | Date; // remove? errors?: string; location_key?: string; }; diff --git a/plugins/catalog-backend/src/next/database/types.ts b/plugins/catalog-backend/src/next/database/types.ts index c93b90084d..38ca59a431 100644 --- a/plugins/catalog-backend/src/next/database/types.ts +++ b/plugins/catalog-backend/src/next/database/types.ts @@ -16,6 +16,7 @@ import { Entity, EntityRelationSpec } from '@backstage/catalog-model'; import { JsonObject } from '@backstage/config'; +import { DateTime } from 'luxon'; import { Transaction } from '../../database/types'; import { DeferredEntity } from '../processing/types'; @@ -54,8 +55,8 @@ export type RefreshStateItem = { unprocessedEntity: Entity; processedEntity?: Entity; resultHash: string; - nextUpdateAt: string; - lastDiscoveryAt: string; // remove? + nextUpdateAt: DateTime; + lastDiscoveryAt: DateTime; // remove? state: Map; errors?: string; locationKey?: string;