[WIP] Adds Prometheus metrics to ODIS combiner daemon (Odis Combiner to k8s)

.github/workflows/odis-combiner-container.yml

@@ -0,0 +1,44 @@
+---
+name: Build ODIS combiner image
+
+on:
+  push:
+    paths:
+      - 'dockerfiles/phone-number-privacy/Dockerfile-combiner'
+      - 'packages/phone-number-privacy/combiner/**'
+    branches:
+      - main
+  pull_request:
+    paths:
+      - 'dockerfiles/phone-number-privacy/Dockerfile-combiner'
+      - 'packages/phone-number-privacy/combiner/**'
+  workflow_dispatch:
+
+jobs:
+  odis-combiner-build-dev:
+    uses: celo-org/reusable-workflows/.github/workflows/[email protected]
+    name: Build us-west1-docker.pkg.dev/devopsre/dev-images/odis-combiner
+    if: |
+      github.ref != 'refs/heads/main'
+    with:
+      workload-id-provider: projects/1094498259535/locations/global/workloadIdentityPools/gh-social-connect/providers/github-by-repos
+      service-account: '[email protected]'
+      artifact-registry: us-west1-docker.pkg.dev/devopsre/dev-images/odis-combiner
+      tag: ${{ github.sha }}
+      context: .
+      file: dockerfiles/phone-number-privacy/Dockerfile-combiner
+      trivy: true
+
+  odis-combiner-build:
+    uses: celo-org/reusable-workflows/.github/workflows/[email protected]
+    name: Build us-west1-docker.pkg.dev/devopsre/social-connect/odis-combiner
+    if: |
+      github.ref == 'refs/heads/main'
+    with:
+      workload-id-provider: projects/1094498259535/locations/global/workloadIdentityPools/gh-social-connect-main/providers/github-by-repos
+      service-account: '[email protected]'
+      artifact-registry: us-west1-docker.pkg.dev/devopsre/social-connect/odis-combiner
+      tag: ${{ github.sha }}
+      context: .
+      file: dockerfiles/phone-number-privacy/Dockerfile-combiner
+      trivy: true

dockerfiles/phone-number-privacy/Dockerfile-combiner

@@ -0,0 +1,42 @@
+##### Gathering dependencies
+FROM scratch AS packages
+
+# Copy phone-number-privacy package and its dependency closure.
+# Assemble all dependencies into the packages folder so the second stage can select whether to
+# include all packages, or just the phone-number-privacy packages.
+WORKDIR /celo-phone-number-privacy/
+COPY packages/phone-number-privacy/combiner packages/phone-number-privacy/combiner
+COPY packages/phone-number-privacy/common packages/phone-number-privacy/common
+COPY packages/sdk/encrypted-backup packages/sdk/encrypted-backup
+COPY packages/sdk/identity packages/sdk/identity
+
+##### Main stage
+FROM node:18
+LABEL org.opencontainers.image.authors="[email protected]"
+
+WORKDIR /celo-phone-number-privacy/
+
+# Copy monorepo settings
+COPY lerna.json package.json yarn.lock ./
+
+# Makes build fail if it doesn't copy git, will be removed after build
+COPY .git .git 
+
+# Setting ONLY_PUBLISHED_DEPENDENCIES to true or any non-empty string results in only the
+# phone-number-privacy package being copied into the image, and therefore it will only build using
+# published dependencies. Setting ONLY_PUBLISHED_DEPENDENCIES to "" will copy in all dependecies.
+ARG ONLY_PUBLISHED_DEPENDENCIES=""
+ARG PACKAGE_SELECTOR=${ONLY_PUBLISHED_DEPENDENCIES:+phone-number-privacy/combiner}
+COPY --from=packages celo-phone-number-privacy/packages/${PACKAGE_SELECTOR} packages/${PACKAGE_SELECTOR}
+
+# Install dependencies and build.
+RUN yarn install --network-timeout 100000 --frozen-lockfile && yarn cache clean
+RUN yarn build
+
+RUN rm -r .git
+
+# Setup and run the combiner application.
+ENV NODE_ENV production
+WORKDIR /celo-phone-number-privacy/packages/phone-number-privacy/combiner
+EXPOSE 8080
+ENTRYPOINT ["yarn", "start:docker"]

packages/phone-number-privacy/combiner/package.json

@@ -52,6 +52,7 @@
     "lru-cache": "^10.0.1",
     "node-fetch": "^2.6.9",
     "pg": "^8.2.1",
+    "prom-client": "12.0.0",
     "uuid": "^7.0.3"
   },
   "devDependencies": {

packages/phone-number-privacy/combiner/src/common/combine.ts

@@ -12,6 +12,7 @@ import { Request } from 'express'
 import * as t from 'io-ts'
 import { PerformanceObserver } from 'perf_hooks'
 import { fetchSignerResponseWithFallback, SignerResponse } from './io'
+import { Counters, Histograms, newMeter } from './metrics'
 
 export interface Signer {
   url: string
@@ -85,17 +86,23 @@ export async function thresholdCallToSigners<R extends OdisRequest>(
   await Promise.all(
     signers.map(async (signer) => {
       try {
-        const signerFetchResult = await fetchSignerResponseWithFallback(
-          signer,
-          endpoint,
-          keyVersionInfo.keyVersion,
-          request,
-          logger,
-          // @ts-ignore
-          abortSignal
+        const _meter = newMeter(Histograms.signerLatency, signer.url, request.url)
+        const signerFetchResult = await _meter(() =>
+          fetchSignerResponseWithFallback(
+            signer,
+            endpoint,
+            keyVersionInfo.keyVersion,
+            request,
+            logger,
+            // @ts-ignore
+            abortSignal
+          )
         )
 
         // used for log based metrics
+        Counters.sigResponses
+          .labels(signerFetchResult.status.toString(), signer.url, request.url)
+          .inc()
         logger.info({
           message: 'Received signerFetchResult',
           signer: signer.url,
@@ -105,6 +112,9 @@ export async function thresholdCallToSigners<R extends OdisRequest>(
         if (!signerFetchResult.ok) {
           const responseData = await signerFetchResult.json()
           // used for log based metrics
+          Counters.sigResponsesErrors
+            .labels(signerFetchResult.status.toString(), signer.url, request.url)
+            .inc()
           logger.info({
             message: 'Received signerFetchResult on unsuccessful signer response',
             res: responseData,
@@ -156,11 +166,18 @@ export async function thresholdCallToSigners<R extends OdisRequest>(
         }
       } catch (err) {
         if (isTimeoutError(err)) {
+          Counters.sigRequestErrors
+            .labels(signer.url, request.url, ErrorMessage.TIMEOUT_FROM_SIGNER)
+            .inc()
           logger.error({ signer }, ErrorMessage.TIMEOUT_FROM_SIGNER)
         } else if (isAbortError(err)) {
+          Counters.warnings.labels(request.url, WarningMessage.CANCELLED_REQUEST_TO_SIGNER).inc()
           logger.info({ signer }, WarningMessage.CANCELLED_REQUEST_TO_SIGNER)
         } else {
           // Logging the err & message simultaneously fails to log the message in some cases
+          Counters.sigRequestErrors
+            .labels(signer.url, request.url, ErrorMessage.SIGNER_REQUEST_ERROR)
+            .inc()
           logger.error({ signer }, ErrorMessage.SIGNER_REQUEST_ERROR)
           logger.error({ signer, err })
 
@@ -180,6 +197,8 @@ export async function thresholdCallToSigners<R extends OdisRequest>(
 
   if (errorCodes.size > 0) {
     if (errorCodes.size > 1) {
+      Counters.errors.labels(request.url).inc()
+      Counters.sigInconsistenciesErrors.labels(request.url).inc()
       logger.error(
         { errorCodes: JSON.stringify([...errorCodes]) },
         ErrorMessage.INCONSISTENT_SIGNER_RESPONSES

packages/phone-number-privacy/combiner/src/common/crypto-clients/bls-crypto-client.ts

@@ -1,6 +1,7 @@
 import { ErrorMessage } from '@celo/phone-number-privacy-common'
 import threshold_bls from 'blind-threshold-bls'
 import Logger from 'bunyan'
+import { Counters } from '../../common/metrics'
 import { CryptoClient, ServicePartialSignature } from './crypto-client'
 
 function flattenSigsArray(sigs: Uint8Array[]) {
@@ -75,6 +76,8 @@ export class BLSCryptographyClient extends CryptoClient {
       // We move it to the verified set so that we don't need to re-verify in the future
       this.verifiedSignatures.push(unverifiedSignature)
     } else {
+      Counters.errors.labels(unverifiedSignature.url).inc()
+      Counters.blsComputeErrors.labels(unverifiedSignature.url).inc()
       logger.error({ url: unverifiedSignature.url }, ErrorMessage.VERIFY_PARITAL_SIGNATURE_ERROR)
     }
   }

packages/phone-number-privacy/combiner/src/common/handlers.ts

@@ -13,8 +13,10 @@ import { SemanticAttributes } from '@opentelemetry/semantic-conventions'
 import Logger from 'bunyan'
 import { Request, Response } from 'express'
 import { performance, PerformanceObserver } from 'perf_hooks'
+import * as client from 'prom-client'
 import { getCombinerVersion } from '../config'
 import { OdisError } from './error'
+import { Counters, newMeter } from './metrics'
 
 const tracer = opentelemetry.trace.getTracer('combiner-tracer')
 
@@ -37,10 +39,14 @@ export function catchErrorHandler<R extends OdisRequest>(
       const logger: Logger = res.locals.logger
       logger.error(ErrorMessage.CAUGHT_ERROR_IN_ENDPOINT_HANDLER)
       logger.error(err)
+      Counters.errors.labels(req.url).inc()
+      Counters.errorsCaughtInEndpointHandler.labels(req.url).inc()
       if (!res.headersSent) {
         if (err instanceof OdisError) {
           sendFailure(err.code, err.status, res, req.url)
         } else {
+          Counters.errors.labels(req.url).inc()
+          Counters.unknownErrors.labels(req.url).inc()
           sendFailure(ErrorMessage.UNKNOWN_ERROR, 500, res, req.url)
         }
       } else {
@@ -84,50 +90,57 @@ export function tracingHandler<R extends OdisRequest>(
 }
 
 export function meteringHandler<R extends OdisRequest>(
+  histogram: client.Histogram<string>,
   handler: PromiseHandler<R>
 ): PromiseHandler<R> {
-  return async (req, res) => {
-    const logger: Logger = res.locals.logger
-
-    // used for log based metrics
-    logger.info({ req: req.body }, 'Request received')
+  return async (req, res) =>
+    newMeter(
+      histogram,
+      req.url
+    )(async () => {
+      const logger: Logger = res.locals.logger
 
-    const eventLoopLagMeasurementStart = Date.now()
-    setTimeout(() => {
-      const eventLoopLag = Date.now() - eventLoopLagMeasurementStart
-      logger.info({ eventLoopLag }, 'Measure event loop lag')
-    })
-    // TODO:(soloseng): session ID may not always exist
-    const startMark = `Begin ${req.url}/${req.body.sessionID}`
-    const endMark = `End ${req.url}/${req.body.sessionID}`
-    const entryName = `${req.url}/${req.body.sessionID} latency`
-
-    const obs = new PerformanceObserver((list) => {
-      const entry = list.getEntriesByName(entryName)[0]
-      if (entry) {
-        logger.info({ latency: entry }, 'e2e response latency measured')
+      // used for log based metrics
+      logger.info({ req: req.body }, 'Request received')
+
+      const eventLoopLagMeasurementStart = Date.now()
+      setTimeout(() => {
+        const eventLoopLag = Date.now() - eventLoopLagMeasurementStart
+        logger.info({ eventLoopLag }, 'Measure event loop lag')
+      })
+      // TODO:(soloseng): session ID may not always exist
+      const startMark = `Begin ${req.url}/${req.body.sessionID}`
+      const endMark = `End ${req.url}/${req.body.sessionID}`
+      const entryName = `${req.url}/${req.body.sessionID} latency`
+
+      const obs = new PerformanceObserver((list) => {
+        const entry = list.getEntriesByName(entryName)[0]
+        if (entry) {
+          logger.info({ latency: entry }, 'e2e response latency measured')
+        }
+      })
+      obs.observe({ entryTypes: ['measure'], buffered: false })
+
+      performance.mark(startMark)
+
+      try {
+        Counters.requests.labels(req.url).inc()
+        await handler(req, res)
+        if (res.headersSent) {
+          // used for log based metrics
+          logger.info({ res }, 'Response sent')
+          Counters.responses.labels(req.url, res.statusCode.toString()).inc()
+        }
+      } finally {
+        performance.mark(endMark)
+        performance.measure(entryName, startMark, endMark)
+
+        performance.clearMeasures(entryName)
+        performance.clearMarks(startMark)
+        performance.clearMarks(endMark)
+        obs.disconnect()
       }
     })
-    obs.observe({ entryTypes: ['measure'], buffered: false })
-
-    performance.mark(startMark)
-
-    try {
-      await handler(req, res)
-      if (res.headersSent) {
-        // used for log based metrics
-        logger.info({ res }, 'Response sent')
-      }
-    } finally {
-      performance.mark(endMark)
-      performance.measure(entryName, startMark, endMark)
-
-      performance.clearMeasures(entryName)
-      performance.clearMarks(startMark)
-      performance.clearMarks(endMark)
-      obs.disconnect()
-    }
-  }
 }
 
 export function timeoutHandler<R extends OdisRequest>(
@@ -154,6 +167,7 @@ export async function disabledHandler<R extends OdisRequest>(
   req: Request<{}, {}, R>,
   response: Response<OdisResponse<R>, Locals>
 ): Promise<void> {
+  Counters.warnings.labels(req.url, WarningMessage.API_UNAVAILABLE).inc()
   sendFailure(WarningMessage.API_UNAVAILABLE, 503, response, req.url)
 }