feat(cron): send instance=github_action label with retrieval_duration…

…_seconds metric (#2020)

* chore(cron): createPromClientRetrievalMetricsLogger logs metricPushed with request url and response status

* add test to ensure createPromClientRetrievalMetricsLogger sends http request to pushgateway

* lint

* feat(cron): createPromClientRetrievalMetricsLogger can have metricLabels passed (e.g. for instance label)

* feat(cron): cron-nft-ttr includes instance=github_action label which is sent to pushgateway

.github/workflows/cron-nft-ttr.yml

@@ -36,6 +36,7 @@ jobs:
             --gateway https://dweb.link \
             --metricsPushGateway $PUSHGATEWAY_URL \
             --metricsPushGatewayJobName $PUSHGATEWAY_JOBNAME \
+            --metricsLabelsJson '{"instance": "github_action"}' \
 
       - name: Heartbeat
         if: ${{ success() }}

packages/cron/src/bin/nft-ttr.js

@@ -64,7 +64,7 @@ export function createMeasureSecretsFromEnv(env) {
 /**
  * @param {unknown} sadeOptions
  * @param {Pick<MeasureTtrOptions['secrets'], 'metricsPushGatewayAuthorization'>} secrets
- * @returns {Omit<MeasureTtrOptions, "secrets">}
+ * @returns {Omit<MeasureTtrOptions, "secrets"> & { metricsLabels: Record<string,string> }}
  */
 export function createMeasureOptionsFromSade(sadeOptions, secrets) {
   // build gateways
@@ -119,6 +119,13 @@ export function createMeasureOptionsFromSade(sadeOptions, secrets) {
     sadeOptions.metricsPushGatewayJobName
   assert.ok(typeof metricsPushGatewayJobName === 'string')
 
+  const metricsLabelsJson = hasOwnProperty(sadeOptions, 'metricsLabelsJson')
+    ? String(sadeOptions.metricsLabelsJson)
+    : undefined
+  const metricsLabels = metricsLabelsJson
+    ? /** @type {Record<string,string>} */ (JSON.parse(metricsLabelsJson))
+    : {}
+
   // build pushRetrieveMetrics
   const promClientRegistry = new promClient.Registry()
 
@@ -129,6 +136,7 @@ export function createMeasureOptionsFromSade(sadeOptions, secrets) {
         promClientRegistry,
         createRetrievalDurationMetric(promClientRegistry),
         metricsPushGatewayJobName,
+        metricsLabels,
         metricsPushGateway,
         secrets.metricsPushGatewayAuthorization
       )
@@ -148,6 +156,7 @@ export function createMeasureOptionsFromSade(sadeOptions, secrets) {
     metricsPushGatewayJobName,
     minImageSizeBytes,
     pushRetrieveMetrics,
+    metricsLabels,
   }
   return options
 }

packages/cron/src/bin/nft-ttr.spec.js

@@ -15,6 +15,7 @@ test('createMeasureOptionsFromSade', (t) => {
     metricsPushGatewayJobName: 'nft-ttr',
     gateway: sampleGateways,
     logConfigAndExit: true,
+    metricsLabelsJson: '{"instance": "github_action"}',
   }
   const secrets = {
     metricsPushGatewayAuthorization: '',
@@ -27,6 +28,7 @@ test('createMeasureOptionsFromSade', (t) => {
   t.is(options.metricsPushGateway?.toString(), sampleSade.metricsPushGateway)
   t.is(options.metricsPushGatewayJobName, sampleSade.metricsPushGatewayJobName)
   t.is(options.logConfigAndExit, true)
+  t.is(options.metricsLabels.instance, 'github_action')
 
   // ensure single gateway is parsed correctly
   const options2 = createMeasureOptionsFromSade(

packages/cron/src/jobs/measureNftTimeToRetrievability.js

@@ -297,6 +297,7 @@ export function createStubbedRetrievalMetricsLogger() {
  * @param {import('prom-client').Registry} registry
  * @param {import('../lib/metrics.js').RetrievalDurationMetric} metric
  * @param {string} metricsPushGatewayJobName
+ * @param {Record<string,string>} metricLabels
  * @param {URL} pushGatewayUrl
  * @param {HttpAuthorizationHeaderValue} pushGatewayAuthorization
  * @returns {RetrievalMetricsLogger}
@@ -305,6 +306,7 @@ export function createPromClientRetrievalMetricsLogger(
   registry,
   metric,
   metricsPushGatewayJobName,
+  metricLabels,
   pushGatewayUrl,
   pushGatewayAuthorization
 ) {
@@ -323,8 +325,16 @@ export function createPromClientRetrievalMetricsLogger(
     metric.observe(value, {})
     const pushAddArgs = {
       jobName: metricsPushGatewayJobName,
+      groupings: metricLabels,
     }
-    await pushgateway.pushAdd(pushAddArgs)
+    const pushAddResult = await pushgateway.pushAdd(pushAddArgs)
+    const pushAddResponse = /** @type {import('http').IncomingMessage} */ (
+      pushAddResult.resp
+    )
+    // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
+    const pushAddRequest = /** @type {import('http').ClientRequest} */ (
+      /** @type {any} */ (pushAddResponse).req
+    )
     options.console.debug({
       type: 'metricPushed',
       metric: {
@@ -334,6 +344,16 @@ export function createPromClientRetrievalMetricsLogger(
         value,
       },
       pushgateway: pushAddArgs,
+      request: {
+        url: new URL(
+          `${pushAddRequest.protocol}//${String(
+            pushAddRequest.getHeader('host')
+          )}${pushAddRequest.path}`
+        ).toString(),
+      },
+      response: {
+        status: pushAddResponse.statusCode,
+      },
     })
   }
   return push

packages/cron/src/jobs/measureNftTimeToRetrievability.spec.js

@@ -1,4 +1,5 @@
 import {
+  createPromClientRetrievalMetricsLogger,
   createStubbedImageFetcher,
   createStubbedRetrievalMetricsLogger,
   createStubStoreFunction,
@@ -8,6 +9,11 @@ import { test } from '../lib/testing.js'
 import { createTestImages } from '../bin/nft-ttr.js'
 import all from 'it-all'
 import { Writable } from 'node:stream'
+import { Registry } from 'prom-client'
+import { createRetrievalDurationMetric } from '../lib/metrics.js'
+import { withHttpServer } from '../lib/http.js'
+import { Milliseconds } from '../lib/time.js'
+import { Console } from 'node:console'
 
 test('measureNftTimeToRetrievability', async (t) => {
   /** this is meant to be a test that doesn't use the network (e.g. inject stubs) */
@@ -71,3 +77,55 @@ test('measureNftTimeToRetrievability', async (t) => {
   const finish = results.find((log) => log.type === 'finish')
   t.assert(finish)
 })
+
+test('createPromClientRetrievalMetricsLogger', async (t) => {
+  const registry = new Registry()
+  const metric = createRetrievalDurationMetric(registry)
+  const metricsPushGatewayJobName =
+    'test-job-createPromClientRetrievalMetricsLogger'
+  const metricLabels = {
+    instance: 'instance-createPromClientRetrievalMetricsLogger',
+  }
+  const pushGatewayAuthorization = 'bearer fake-auth'
+  /** @type {import('http').IncomingMessage[]} */
+  const fakePushGatewayRequests = []
+  /** @type {import('http').RequestListener} */
+  const fakePushGateway = (req, res) => {
+    fakePushGatewayRequests.push(req)
+    res.writeHead(200)
+    res.end()
+  }
+  const silentConsole = new Console(new Writable())
+  /** @type {import('./measureNftTimeToRetrievability.js').RetrieveLog} */
+  const fakeRetrieve = {
+    type: 'retrieve',
+    image: 'fake-image',
+    gateway: new URL('https://example.com/fake-gateway'),
+    url: new URL('https://example.com/fake-gateway/fake-image'),
+    contentLength: 1,
+    startTime: new Date(),
+    duration: new Milliseconds(1000),
+  }
+  await withHttpServer(fakePushGateway, async (pushGatewayUrl) => {
+    const metricsLogger = createPromClientRetrievalMetricsLogger(
+      registry,
+      metric,
+      metricsPushGatewayJobName,
+      metricLabels,
+      pushGatewayUrl,
+      pushGatewayAuthorization
+    )
+    await metricsLogger({ console: silentConsole }, fakeRetrieve)
+  })
+  t.is(fakePushGatewayRequests.length, 1)
+  const [firstRequest] = fakePushGatewayRequests
+  t.assert(
+    firstRequest.url?.startsWith(`/metrics/job/${metricsPushGatewayJobName}`)
+  )
+  for (const [label, value] of Object.entries(metricLabels)) {
+    t.assert(
+      firstRequest.url?.includes([label, value].join('/')),
+      `expected metric push request url to contain label '${label}'`
+    )
+  }
+})

packages/cron/src/lib/http.js

@@ -0,0 +1,43 @@
+import * as nodeHttp from 'http'
+
+/**
+ * Spin up an http server on an unused port, do some work with it, then shut it down.
+ * @param {import('http').RequestListener} listener
+ * @param {(baseUrl: URL) => Promise<void>} useServer
+ * @returns
+ */
+export async function withHttpServer(listener, useServer) {
+  const httpServer = nodeHttp.createServer(listener)
+  // listen on unused port
+  await new Promise((resolve) => {
+    httpServer.listen(0, () => {
+      resolve(true)
+    })
+  })
+  const baseUrl = addressUrl(httpServer.address())
+  if (!baseUrl) {
+    throw new Error(`failed to determine baseUrl from server`)
+  }
+  try {
+    await useServer(baseUrl)
+  } finally {
+    await new Promise((resolve) => {
+      httpServer.close(resolve)
+    })
+  }
+}
+
+/**
+ * Given return type of node http Server#address, return a URL descriving the server address
+ * @param {string|null|import('net').AddressInfo} addressInfo
+ * @returns {URL}
+ */
+export function addressUrl(addressInfo) {
+  if (addressInfo === null)
+    throw new TypeError('addressInfo is unexpectedly null')
+  if (typeof addressInfo === 'string') return new URL(addressInfo)
+  const { address, port } = addressInfo
+  const host = address === '::' ? '127.0.0.1' : address
+  const urlString = `http://${host}:${port}`
+  return new URL(urlString)
+}