Merge pull request #6901 from backstage/freben/db-datetime

use DateTime all the way from the database

.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

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');

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 };
+}

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<DbRefreshStateRow>('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<string, JsonObject>(),
@@ -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);
     }
   }
 }

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'),
+      }),
+    );
+  });
+});

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;
+}

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;
 };

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<string, JsonObject>;
   errors?: string;
   locationKey?: string;