Merge pull request #6848 from backstage/mob/catalog-instrumentation

Instrument catalog with prometheus metrics

plugins/catalog-backend/package.json

@@ -51,8 +51,10 @@
     "glob": "^7.1.6",
     "knex": "^0.95.1",
     "lodash": "^4.17.15",
+    "luxon": "^2.0.2",
     "morgan": "^1.10.0",
     "p-limit": "^3.0.2",
+    "prom-client": "^13.2.0",
     "qs": "^6.9.4",
     "uuid": "^8.0.0",
     "winston": "^3.2.1",

plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts

@@ -16,6 +16,7 @@
 
 import { getVoidLogger } from '@backstage/backend-common';
 import { Hash } from 'crypto';
+import { DateTime } from 'luxon';
 import waitForExpect from 'wait-for-expect';
 import { DefaultProcessingDatabase } from './database/DefaultProcessingDatabase';
 import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine';
@@ -84,7 +85,7 @@ describe('DefaultCatalogProcessingEngine', () => {
             },
             resultHash: '',
             state: new Map(),
-            nextUpdateAt: '',
+            nextUpdateAt: DateTime.now().toSQL(),
             lastDiscoveryAt: '',
           },
         ],
@@ -147,7 +148,7 @@ describe('DefaultCatalogProcessingEngine', () => {
             },
             resultHash: '',
             state: new Map(),
-            nextUpdateAt: '',
+            nextUpdateAt: DateTime.now().toSQL(),
             lastDiscoveryAt: '',
           },
         ],
@@ -181,7 +182,7 @@ describe('DefaultCatalogProcessingEngine', () => {
       unprocessedEntity: entity,
       resultHash: 'the matching hash',
       state: new Map(),
-      nextUpdateAt: '',
+      nextUpdateAt: DateTime.now().toSQL(),
       lastDiscoveryAt: '',
     };
 

plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts

@@ -22,8 +22,10 @@ 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 { Stitcher } from './stitching/Stitcher';
 import { startTaskPipeline } from './TaskPipeline';
@@ -84,6 +86,20 @@ class Connection implements EntityProviderConnection {
 
 export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
   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,
@@ -95,6 +111,10 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
   ) {}
 
   async start() {
+    if (this.stopFunc) {
+      throw new Error('Processing engine is already started');
+    }
+
     for (const provider of this.entityProviders) {
       await provider.connect(
         new Connection({
@@ -104,10 +124,6 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
       );
     }
 
-    if (this.stopFunc) {
-      throw new Error('Processing engine is already started');
-    }
-
     this.stopFunc = startTaskPipeline<RefreshStateItem>({
       lowWatermark: 5,
       highWatermark: 10,
@@ -127,7 +143,16 @@ 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 {
             id,
             state,
@@ -213,6 +238,8 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
           await this.stitcher.stitch(setOfThingsToStitch);
         } catch (error) {
           this.logger.warn('Processing failed with:', error);
+        } finally {
+          endTimer?.();
         }
       },
     });

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 { initDatabaseMetrics } from './metrics';
 import {
   DbRefreshStateReferencesRow,
   DbRefreshStateRow,
@@ -51,7 +52,9 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
       logger: Logger;
       refreshInterval: RefreshIntervalFunction;
     },
-  ) {}
+  ) {
+    initDatabaseMetrics(options.database);
+  }
 
   async updateProcessedEntity(
     txOpaque: Transaction,

plugins/catalog-backend/src/next/database/metrics.ts

@@ -0,0 +1,72 @@
+/*
+ * 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 { Knex } from 'knex';
+import { DbLocationsRow } from '../../database/types';
+import { createGaugeMetric } from '../metrics';
+import { DbRefreshStateRow, DbRelationsRow } from './tables';
+
+export function initDatabaseMetrics(knex: Knex) {
+  const seen = new Set<string>();
+  return {
+    entities_count: createGaugeMetric({
+      name: 'catalog_entities_count',
+      help: 'Total amount of entities in the catalog',
+      labelNames: ['kind'],
+      async collect() {
+        const result = await knex<DbRefreshStateRow>('refresh_state').select(
+          'entity_ref',
+        );
+        const results = result
+          .map(row => row.entity_ref.split(':')[0])
+          .reduce((acc, e) => acc.set(e, (acc.get(e) || 0) + 1), new Map());
+
+        results.forEach((value, key) => {
+          seen.add(key);
+          this.set({ kind: key }, value);
+        });
+
+        // Set all the entities that were not seen to 0 and delete them from the seen set.
+        seen.forEach(key => {
+          if (!results.has(key)) {
+            this.set({ kind: key }, 0);
+            seen.delete(key);
+          }
+        });
+      },
+    }),
+    registered_locations: createGaugeMetric({
+      name: 'catalog_registered_locations_count',
+      help: 'Total amount of registered locations in the catalog',
+      async collect() {
+        const total = await knex<DbLocationsRow>('locations').count({
+          count: '*',
+        });
+        this.set(Number(total[0].count));
+      },
+    }),
+    relations: createGaugeMetric({
+      name: 'catalog_relations_count',
+      help: 'Total amount of relations between entities',
+      async collect() {
+        const total = await knex<DbRelationsRow>('relations').count({
+          count: '*',
+        });
+        this.set(Number(total[0].count));
+      },
+    }),
+  };
+}

plugins/catalog-backend/src/next/metrics.ts

@@ -0,0 +1,55 @@
+/*
+ * 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 {
+  Counter,
+  CounterConfiguration,
+  Gauge,
+  GaugeConfiguration,
+  Histogram,
+  HistogramConfiguration,
+  register,
+  Summary,
+  SummaryConfiguration,
+} from 'prom-client';
+
+export function createCounterMetric<T extends string>(
+  config: CounterConfiguration<T>,
+): Counter<T> {
+  const existing = register.getSingleMetric(config.name) as Counter<T>;
+  return existing || new Counter<T>(config);
+}
+
+export function createGaugeMetric<T extends string>(
+  config: GaugeConfiguration<T>,
+): Gauge<T> {
+  const existing = register.getSingleMetric(config.name) as Gauge<T>;
+  return existing || new Gauge<T>(config);
+}
+
+export function createSummaryMetric<T extends string>(
+  config: SummaryConfiguration<T>,
+): Summary<T> {
+  const existing = register.getSingleMetric(config.name) as Summary<T>;
+  return existing || new Summary<T>(config);
+}
+
+export function createHistogramMetric<T extends string>(
+  config: HistogramConfiguration<T>,
+): Histogram<T> {
+  const existing = register.getSingleMetric(config.name) as Histogram<T>;
+  return existing || new Histogram<T>(config);
+}