Instrument catalog with prometheus

Signed-off-by: Johan Haals <johan.haals@gmail.com>
2021-08-16 11:49:06 +02:00
parent 7bcf0b9d14
commit 693a17fcdc
7 changed files with 166 additions and 11 deletions
@@ -58,9 +58,11 @@
    "example-app": "^0.2.41",
    "express": "^4.17.1",
    "express-promise-router": "^4.1.0",
+    "express-prom-bundle": "^6.3.6",
    "knex": "^0.95.1",
    "pg": "^8.3.0",
    "pg-connection-string": "^2.3.0",
+    "prom-client": "^13.2.0",
    "sqlite3": "^5.0.1",
    "winston": "^3.2.1"
  },
@@ -36,6 +36,7 @@ import {
 } from '@backstage/backend-common';
 import { Config } from '@backstage/config';
 import healthcheck from './plugins/healthcheck';
+import { metricsInit, metricsHandler } from './metrics';
 import auth from './plugins/auth';
 import catalog from './plugins/catalog';
 import codeCoverage from './plugins/codecoverage';
@@ -72,6 +73,7 @@ function makeCreateEnv(config: Config) {
 }

 async function main() {
+  metricsInit();
  const logger = getRootLogger();

  logger.info(
@@ -124,6 +126,7 @@ async function main() {
  const service = createServiceBuilder(module)
    .loadConfig(config)
    .addRouter('', await healthcheck(healthcheckEnv))
+    .addRouter('', metricsHandler())
    .addRouter('/api', apiRouter)
    .addRouter('', await app(appEnv));

@@ -0,0 +1,59 @@
+/*
+ * 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 { useHotCleanup } from '@backstage/backend-common';
+import { RequestHandler, Request } from 'express';
+import promBundle from 'express-prom-bundle';
+import prom from 'prom-client';
+import * as url from 'url';
+
+const rootRegEx = new RegExp('^/([^/]*)/.*');
+const apiRegEx = new RegExp('^/api/([^/]*)/.*');
+
+function normalizePath(req: Request): string {
+  const path = url.parse(req.originalUrl || req.url).pathname || '/';
+
+  // Capture /api/ and the plugin name
+  if (apiRegEx.test(path)) {
+    return path.replace(apiRegEx, '/api/$1');
+  }
+
+  // Only the first path segment at root level
+  return path.replace(rootRegEx, '/$1');
+}
+
+export function metricsInit(): void {
+  prom.collectDefaultMetrics({ prefix: 'backstage_' });
+}
+
+/**
+ * Adds a /metrics endpoint, register default runtime metrics and instrument the router.
+ */
+export function metricsHandler(): RequestHandler {
+  // We can only initialize the metrics once and have to clean them up between hot reloads
+  useHotCleanup(module, () => prom.register.clear());
+
+  return promBundle({
+    includeMethod: true,
+    includePath: true,
+    // Using includePath alone is problematic, as it will include path labels with high
+    // cardinality (e.g. path params). Instead we would have to template them. However, this
+    // is difficult, as every backend plugin might use different routes. Instead we only take
+    // the first directory of the path, to have at least an idea how each plugin performs:
+    normalizePath,
+    promClient: { collectDefaultMetrics: {} },
+  });
+}
@@ -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",
@@ -22,6 +22,7 @@ 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 { CatalogProcessingOrchestrator } from './processing/types';
@@ -33,6 +34,7 @@ import {
  EntityProviderConnection,
  EntityProviderMutation,
 } from './types';
+import { Counter, Histogram, Summary } from 'prom-client';

 class Connection implements EntityProviderConnection {
  readonly validateEntityEnvelope = entityEnvelopeSchemaValidator();
@@ -82,8 +84,55 @@ class Connection implements EntityProviderConnection {
  }
 }

+/*
+export class Metrics {
+  private readonly processedEntities = new Counter({
+    name: 'processed_entities',
+    help: 'Amount of entities processed',
+  });
+  private readonly processingDuration = new Histogram({
+    name: 'processing_duration',
+    help: 'Processing duration',
+  });
+  private readonly processingQueueDelay = new Histogram({
+    name: 'processing_queue_delay',
+    help: 'The amount of delay between being scheduled for processing, and the start of actually being processed',
+  });
+
+  async markProcess(item: RefreshStateItem, inner: () => Promise<void>) {
+    this.processedEntities.add(1);
+    this.processingQueueDelay.add(
+      DateTime.fromSQL(item.nextUpdateAt, { zone: 'UTC' })
+        .diffNow()
+        .as('seconds'),
+    );
+    const endTimer = this.processingDuration.startTimer();
+
+    try {
+      await inner();
+    } finally {
+      endTimer();
+    }
+  }
+}
+*/
+
 export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
  private stopFunc?: () => void;
+  private readonly metrics = {
+    processedEntities: new Counter({
+      name: 'catalog_processed_entities_count',
+      help: 'Amount of entities processed',
+    }),
+    processingDuration: new Summary({
+      name: 'catalog_processing_duration_seconds',
+      help: 'Processing duration',
+    }),
+    processingQueueDelay: new Summary({
+      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 +144,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 +157,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 +176,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 +271,8 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
          await this.stitcher.stitch(setOfThingsToStitch);
        } catch (error) {
          this.logger.warn('Processing failed with:', error);
+        } finally {
+          endTimer?.();
        }
      },
    });
@@ -468,6 +468,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
      .limit(request.processBatchSize)
      .orderBy('next_update_at', 'asc');

+    const nextRefresh = Math.random() * (150 - 100) + 100;
    await tx<DbRefreshStateRow>('refresh_state')
      .whereIn(
        'entity_ref',
@@ -476,12 +477,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', ?)`, [`${nextRefresh} seconds`])
+            : tx.raw(`now() + interval '${Number(nextRefresh)} seconds'`),
      });

    return {
@@ -9446,6 +9446,11 @@ bindings@^1.5.0:
  dependencies:
    file-uri-to-path "1.0.0"

+bintrees@1.0.1:
+  version "1.0.1"
+  resolved "https://registry.npmjs.org/bintrees/-/bintrees-1.0.1.tgz#0e655c9b9c2435eaab68bf4027226d2b55a34524"
+  integrity sha1-DmVcm5wkNeqraL9AJyJtK1WjRSQ=
+
 bl@^4.0.3, bl@^4.1.0:
  version "4.1.0"
  resolved "https://registry.npmjs.org/bl/-/bl-4.1.0.tgz#451535264182bec2fbbc83a62ab98cf11d9f7b3a"
@@ -13475,6 +13480,14 @@ expect@^26.6.2:
    jest-message-util "^26.6.2"
    jest-regex-util "^26.0.0"

+express-prom-bundle@^6.3.6:
+  version "6.3.6"
+  resolved "https://registry.npmjs.org/express-prom-bundle/-/express-prom-bundle-6.3.6.tgz#c8da1c1024edfcc54953c365991aca57ffd0cfda"
+  integrity sha512-IRsTRCEKCVCHEriQlZ1FuutjEFc89KASsveXh+1HcGEnuZKiAC4LugxrsGEIdySqYvqOYSr2SWHJ6L8/BK2SHA==
+  dependencies:
+    on-finished "^2.3.0"
+    url-value-parser "^2.0.0"
+
 express-promise-router@^4.1.0:
  version "4.1.0"
  resolved "https://registry.npmjs.org/express-promise-router/-/express-promise-router-4.1.0.tgz#79160e145c27610ba411bceb0552a36f11dbab4f"
@@ -20451,7 +20464,7 @@ oidc-token-hash@^5.0.1:
  resolved "https://registry.npmjs.org/oidc-token-hash/-/oidc-token-hash-5.0.1.tgz#ae6beec3ec20f0fd885e5400d175191d6e2f10c6"
  integrity sha512-EvoOtz6FIEBzE+9q253HsLCVRiK/0doEJ2HCvvqMQb3dHZrP3WlJKYtJ55CRTw4jmYomzH4wkPuCj/I3ZvpKxQ==

-on-finished@~2.3.0:
+on-finished@^2.3.0, on-finished@~2.3.0:
  version "2.3.0"
  resolved "https://registry.npmjs.org/on-finished/-/on-finished-2.3.0.tgz#20f1336481b083cd75337992a16971aa2d906947"
  integrity sha1-IPEzZIGwg811M3mSoWlxqi2QaUc=
@@ -22089,6 +22102,13 @@ progress@^2.0.0:
  resolved "https://registry.npmjs.org/progress/-/progress-2.0.3.tgz#7e8cf8d8f5b8f239c1bc68beb4eb78567d572ef8"
  integrity sha512-7PiHtLll5LdnKIMw100I+8xJXR5gW2QwWYkT6iJva0bXitZKa/XMrSbdmg3r2Xnaidz9Qumd0VPaMrZlF9V9sA==

+prom-client@^13.2.0:
+  version "13.2.0"
+  resolved "https://registry.npmjs.org/prom-client/-/prom-client-13.2.0.tgz#99d13357912dd400f8911b77df19f7b328a93e92"
+  integrity sha512-wGr5mlNNdRNzEhRYXgboUU2LxHWIojxscJKmtG3R8f4/KiWqyYgXTLHs0+Ted7tG3zFT7pgHJbtomzZ1L0ARaQ==
+  dependencies:
+    tdigest "^0.1.1"
+
 promise-inflight@^1.0.1:
  version "1.0.1"
  resolved "https://registry.npmjs.org/promise-inflight/-/promise-inflight-1.0.1.tgz#98472870bf228132fcbdd868129bad12c3c029e3"
@@ -25669,6 +25689,13 @@ tarn@^3.0.1:
  resolved "https://registry.npmjs.org/tarn/-/tarn-3.0.1.tgz#ebac2c6dbc6977d34d4526e0a7814200386a8aec"
  integrity sha512-6usSlV9KyHsspvwu2duKH+FMUhqJnAh6J5J/4MITl8s94iSUQTLkJggdiewKv4RyARQccnigV48Z+khiuVZDJw==

+tdigest@^0.1.1:
+  version "0.1.1"
+  resolved "https://registry.npmjs.org/tdigest/-/tdigest-0.1.1.tgz#2e3cb2c39ea449e55d1e6cd91117accca4588021"
+  integrity sha1-Ljyyw56kSeVdHmzZEReszKRYgCE=
+  dependencies:
+    bintrees "1.0.1"
+
 teeny-request@^3.11.3:
  version "3.11.3"
  resolved "https://registry.npmjs.org/teeny-request/-/teeny-request-3.11.3.tgz#335c629f7645e5d6599362df2f3230c4cbc23a55"
@@ -26847,6 +26874,11 @@ url-parse@^1.4.3, url-parse@^1.5.1:
    querystringify "^2.1.1"
    requires-port "^1.0.0"

+url-value-parser@^2.0.0:
+  version "2.0.3"
+  resolved "https://registry.npmjs.org/url-value-parser/-/url-value-parser-2.0.3.tgz#cd4b8d6754e458d65e8125260c09718d926e6e21"
+  integrity sha512-FjIX+Q9lYmDM9uYIGdMYfQW0uLbWVwN2NrL2ayAI7BTOvEwzH+VoDdNquwB9h4dFAx+u6mb0ONLa3sHD5DvyvA==
+
 url@0.10.3:
  version "0.10.3"
  resolved "https://registry.npmjs.org/url/-/url-0.10.3.tgz#021e4d9c7705f21bbf37d03ceb58767402774c64"