feat: telemetry - loop duration & delay

Signed-off-by: Miroslav Bajtoš <oss@bajtos.net>

observer/bin/spark-observer.js

@@ -3,7 +3,7 @@ import * as SparkImpactEvaluator from '@filecoin-station/spark-impact-evaluator'
 import { ethers } from 'ethers'
 import * as Sentry from '@sentry/node'
 import timers from 'node:timers/promises'
-import { InfluxDB } from '@influxdata/influxdb-client'
+import { createInflux } from '../lib/telemetry.js'
 import assert from 'node:assert/strict'
 
 import { RPC_URL, rpcHeaders } from '../lib/config.js'
@@ -26,11 +26,7 @@ const provider = new ethers.JsonRpcProvider(fetchRequest, null, { polling: true
 
 const ieContract = new ethers.Contract(SparkImpactEvaluator.ADDRESS, SparkImpactEvaluator.ABI, provider)
 
-const influx = new InfluxDB({
-  url: 'https://eu-central-1-1.aws.cloud2.influxdata.com',
-  // spark-stats-observer-read
-  token: INFLUXDB_TOKEN
-})
+const { influx, recordTelemetry } = createInflux(INFLUXDB_TOKEN)
 
 const influxQueryApi = influx.getQueryApi('Filecoin Station')
 
@@ -47,7 +43,14 @@ const loop = async (name, fn, interval) => {
     }
     const dt = Date.now() - start
     console.log(`Loop "${name}" took ${dt}ms`)
-    await timers.setTimeout(interval - dt)
+    recordTelemetry(`loop_${name.replaceAll(' ', '_')}`, point => {
+      point.intField('interval_ms', interval)
+      point.intField('duration_ms', dt)
+      point.intField('delay_ms', interval - dt)
+    })
+    if (dt < interval) {
+      await timers.setTimeout(interval - dt)
+    }
   }
 }
 

observer/lib/telemetry.js

@@ -0,0 +1,37 @@
+import { InfluxDB, Point } from '@influxdata/influxdb-client'
+import createDebug from 'debug'
+
+const debug = createDebug('spark:observer:telemetry')
+
+export const createInflux = token => {
+  const influx = new InfluxDB({
+    url: 'https://eu-central-1-1.aws.cloud2.influxdata.com',
+    // bucket permissions: spark-evaluate:read spark-observer:write
+    token
+  })
+  const writeClient = influx.getWriteApi(
+    'Filecoin Station', // org
+    'spark-observer', // bucket
+    'ms' // precision
+  )
+  setInterval(() => {
+    writeClient.flush().catch(console.error)
+  }, 10_000).unref()
+
+  return {
+    influx,
+
+    /**
+     * @param {string} name
+     * @param {(p: Point) => void} fn
+     */
+    recordTelemetry: (name, fn) => {
+      const point = new Point(name)
+      fn(point)
+      writeClient.writePoint(point)
+      debug('%s %o', name, point)
+    }
+  }
+}
+
+export { Point }