fix: refactor metrics durable object

packages/gateway/package.json

@@ -20,6 +20,7 @@
     "multiformats": "^9.5.2",
     "nanoid": "^3.1.30",
     "p-any": "^4.0.0",
+    "p-map": "^5.3.0",
     "p-settle": "^5.0.0",
     "toucan-js": "^2.5.0"
   },

packages/gateway/src/constants.js

@@ -1 +1,3 @@
 export const METRICS_CACHE_MAX_AGE = 10 * 60 // in seconds (10 minutes)
+export const CIDS_TRACKER_ID = 'cids'
+export const GENERIC_METRICS_ID = 'generic-metrics'

packages/gateway/src/durable-objects/gateway-metrics.js

@@ -0,0 +1,97 @@
+/**
+ * @typedef {Object} GatewayMetrics
+ * @property {number} totalResponseTime total response time of the requests
+ * @property {number} totalSuccessfulRequests total number of successful requests
+ * @property {number} totalFailedRequests total number of requests failed
+ * @property {number} totalWinnerRequests number of performed requests where winner
+ * @property {Record<string, number>} responseTimeHistogram
+ *
+ * @typedef {Object} ResponseStats
+ * @property {boolean} ok request was successful
+ * @property {number} [responseTime] number of milliseconds to get response
+ * @property {boolean} [winner]
+ */
+
+const GATEWAY_METRICS_ID = 'gateway_metrics'
+
+/**
+ * Durable Object for keeping Metrics state of a gateway.
+ */
+export class GatewayMetrics0 {
+  constructor(state) {
+    this.state = state
+
+    // `blockConcurrencyWhile()` ensures no requests are delivered until initialization completes.
+    this.state.blockConcurrencyWhile(async () => {
+      // Get state and initialize if not existing
+      /** @type {GatewayMetrics} */
+      this.gatewayMetrics =
+        (await this.state.storage.get(GATEWAY_METRICS_ID)) ||
+        createMetricsTracker()
+    })
+  }
+
+  // Handle HTTP requests from clients.
+  async fetch(request) {
+    // Apply requested action.
+    let url = new URL(request.url)
+    switch (url.pathname) {
+      case '/update':
+        const data = await request.json()
+        // Updated Metrics
+        this._updateMetrics(data)
+        // Save updated Metrics
+        await this.state.storage.put(GATEWAY_METRICS_ID, this.gatewayMetrics)
+        return new Response()
+      case '/metrics':
+        return new Response(JSON.stringify(this.gatewayMetrics))
+      default:
+        return new Response('Not found', { status: 404 })
+    }
+  }
+
+  /**
+   * @param {ResponseStats} stats
+   */
+  _updateMetrics(stats) {
+    if (!stats.ok) {
+      // Update failed request count
+      this.gatewayMetrics.totalFailedRequests += 1
+      return
+    }
+
+    // Update request count and response time sum
+    this.gatewayMetrics.totalSuccessfulRequests += 1
+    this.gatewayMetrics.totalResponseTime += stats.responseTime
+
+    // Update faster count if appropriate
+    if (stats.winner) {
+      this.gatewayMetrics.totalWinnerRequests += 1
+    }
+
+    // Update histogram
+    const gwHistogram = {
+      ...this.gatewayMetrics.responseTimeHistogram,
+    }
+
+    const histogramCandidate =
+      histogram.find((h) => stats.responseTime <= h) ||
+      histogram[histogram.length - 1]
+    gwHistogram[histogramCandidate] += 1
+    this.gatewayMetrics.responseTimeHistogram = gwHistogram
+  }
+}
+
+export const histogram = [300, 500, 750, 1000, 1500, 2000, 3000, 5000, 10000]
+
+function createMetricsTracker() {
+  const h = histogram.map((h) => [h, 0])
+
+  return {
+    totalResponseTime: 0,
+    totalSuccessfulRequests: 0,
+    totalFailedRequests: 0,
+    totalWinnerRequests: 0,
+    responseTimeHistogram: Object.fromEntries(h),
+  }
+}

packages/gateway/src/durable-objects/generic-metrics.js

@@ -0,0 +1,40 @@
+// Key to track total time for winner gateway to respond
+const TOTAL_WINNER_RESPONSE_TIME_ID = 'totalWinnerResponseTime'
+
+/**
+ * Durable Object for keeping generic Metrics of gateway.nft.storage
+ */
+export class GenericMetrics0 {
+  constructor(state) {
+    this.state = state
+
+    // `blockConcurrencyWhile()` ensures no requests are delivered until initialization completes.
+    this.state.blockConcurrencyWhile(async () => {
+      // Total response time
+      this.totalWinnerResponseTime =
+        (await this.state.storage.get(TOTAL_WINNER_RESPONSE_TIME_ID)) || 0
+    })
+  }
+
+  // Handle HTTP requests from clients.
+  async fetch(request) {
+    // Apply requested action.
+    let url = new URL(request.url)
+    switch (url.pathname) {
+      case '/update':
+        const data = await request.json()
+        // Updated Metrics
+        this.totalWinnerResponseTime += data.responseTime
+        // Save updated Metrics
+        await this.state.storage.put(
+          TOTAL_WINNER_RESPONSE_TIME_ID,
+          this.totalWinnerResponseTime
+        )
+        return new Response()
+      case '/metrics':
+        return new Response(this.totalWinnerResponseTime)
+      default:
+        return new Response('Not found', { status: 404 })
+    }
+  }
+}

packages/gateway/src/env.js

@@ -14,12 +14,14 @@ import { Logging } from './logs.js'
  * @property {string} [SENTRY_DSN]
  * @property {string} [LOGTAIL_TOKEN]
  * @property {number} [REQUEST_TIMEOUT]
- * @property {Object} METRICS
+ * @property {Object} GATEWAYMETRICS
+ * @property {Object} GENERICMETRICS
  * @property {Object} CIDSTRACKER
  *
  * @typedef {Object} EnvTransformed
  * @property {Array<string>} ipfsGateways
- * @property {Object} metricsDurable
+ * @property {Object} gatewayMetricsDurable
+ * @property {Object} genericMetricsDurable
  * @property {Object} cidsTrackerDurable
  * @property {number} REQUEST_TIMEOUT
  * @property {Toucan} [sentry]
@@ -36,7 +38,8 @@ import { Logging } from './logs.js'
 export function envAll(request, env, ctx) {
   env.sentry = getSentry(request, env)
   env.ipfsGateways = JSON.parse(env.IPFS_GATEWAYS)
-  env.metricsDurable = env.METRICS
+  env.gatewayMetricsDurable = env.GATEWAYMETRICS
+  env.genericMetricsDurable = env.GENERICMETRICS
   env.cidsTrackerDurable = env.CIDSTRACKER
   env.REQUEST_TIMEOUT = env.REQUEST_TIMEOUT || 20000
   env.log = new Logging(request, env, ctx)