add pull metrics

docs/API.md

@@ -1466,6 +1466,129 @@ Every row has `id, eventType, chainId, contract, block, txHash` plus event-speci
 
 ---
 
+## Get Node Metrics
+
+### `HTTP` GET /nodeMetrics
+
+### `HTTP` POST /directCommand
+
+### `P2P` command: getNodeMetrics
+
+#### Description
+
+Returns a live per-node resource snapshot, rolled up across every C2D engine (the same aggregate the telemetry layer exports) blended with host `os` readings. Read-only, no parameters. `hasAggregate` is a freshness flag: when `false`, no engine had a fresh compute aggregate (metrics collection is disabled via `C2D_METRICS_INTERVAL_SECONDS=0`, or nothing has been sampled yet) and every scalar is a structural zero rather than a genuine reading. The snapshot is returned either way.
+
+#### Parameters
+
+| name    | type   | required  | description                     |
+| ------- | ------ | --------- | ------------------------------- |
+| command | string | POST only | command name (`getNodeMetrics`) |
+
+#### Response
+
+```json
+{
+  "collectedAt": 1730370000000,
+  "hasAggregate": true,
+  "cpu": {
+    "usagePercent": 42.5,
+    "coresAllocated": 4,
+    "hostCores": 16,
+    "throttledCount": 0,
+    "loadAverage": [1.2, 1.1, 0.9]
+  },
+  "memory": {
+    "usedBytes": 2147483648,
+    "limitBytes": 8589934592,
+    "hostFreeBytes": 12000000000,
+    "hostTotalBytes": 34359738368
+  },
+  "disk": { "usedBytes": 1073741824 },
+  "network": { "rxBytes": 12345, "txBytes": 6789 },
+  "jobs": { "running": 1, "runningFree": 0, "queued": 0, "queuedFree": 0 },
+  "gpu": [
+    {
+      "resourceId": "0",
+      "vendor": "nvidia",
+      "utilizationPercent": 55,
+      "memoryUsedBytes": 2000000000,
+      "memoryTotalBytes": 16000000000,
+      "temperatureC": 61,
+      "powerWatts": 120
+    }
+  ],
+  "env": [{ "env": "env-hash", "resource": "cpu", "total": 16, "inUse": 4 }],
+  "meta": { "sampledContainers": 1, "oldestSampleAgeSeconds": 8 }
+}
+```
+
+---
+
+## Get Node Metrics History
+
+### `HTTP` GET /nodeMetrics/history?startTime=&stopTime=
+
+### `HTTP` POST /directCommand
+
+### `P2P` command: getNodeMetricsHistory
+
+#### Description
+
+Returns ordered hourly averages of the per-node resource snapshot, persisted to SQLite by the sampler/roll-up cron jobs and retained for `NODE_METRICS_RETENTION_DAYS` (default 180). Scalars are arithmetic means over the hour's minute-samples; `sampleCount` is how many samples fed each bucket; GPU entries are averaged per `resourceId`, env entries per `env`+`resource`. Requires the node-metrics database (returns `503` when unavailable, e.g. history disabled via `NODE_METRICS_HISTORY_ENABLED=false`).
+
+When the requested range includes the current, in-progress hour, the last bucket is a **live** average computed on the fly from the raw samples collected so far this hour (before the top-of-hour roll-up has stored it). It is flagged `"partial": true` and is the only bucket that carries that flag; every completed hour is a finalized, stored average. This lets a caller see fresh data without waiting for the hourly roll-up.
+
+#### Parameters
+
+| name      | type          | required | description                                                                        |
+| --------- | ------------- | -------- | ---------------------------------------------------------------------------------- |
+| command   | string        | POST only | command name (`getNodeMetricsHistory`)                                             |
+| startTime | number/string |          | range start — epoch ms or ISO-8601. Defaults to now minus the retention window     |
+| stopTime  | number/string |          | range end — epoch ms or ISO-8601. Defaults to now                                  |
+
+`startTime` must be earlier than `stopTime` (else `400`); the range is clamped to the retention window and the row count is capped.
+
+#### Request (POST /directCommand)
+
+```json
+{
+  "command": "getNodeMetricsHistory",
+  "startTime": 1727778000000,
+  "stopTime": 1730370000000
+}
+```
+
+#### Response
+
+```json
+{
+  "startTime": 1727778000000,
+  "stopTime": 1730370000000,
+  "count": 1,
+  "buckets": [
+    {
+      "hourStart": 1730368800000,
+      "sampleCount": 60,
+      "cpu": { "usagePercent": 40.1, "coresAllocated": 4, "hostCores": 16, "throttledCount": 0 },
+      "memory": {
+        "usedBytes": 2000000000,
+        "limitBytes": 8589934592,
+        "hostFreeBytes": 12000000000,
+        "hostTotalBytes": 34359738368
+      },
+      "disk": { "usedBytes": 1073741824 },
+      "network": { "rxBytes": 12000, "txBytes": 6000 },
+      "jobs": { "running": 1, "runningFree": 0, "queued": 0, "queuedFree": 0 },
+      "gpu": [{ "resourceId": "0", "vendor": "nvidia", "utilizationPercent": 50 }],
+      "env": [{ "env": "env-hash", "resource": "cpu", "total": 16, "inUse": 4 }],
+      "meta": { "sampledContainers": 1 }
+    }
+  ]
+}
+```
+
+---
+
 # Compute
 
 For starters, you can find a list of algorithms in the [Ocean Algorithms repository](https://github.com/oceanprotocol/algo_dockers) and the docker images in the [Algo Dockerhub](https://hub.docker.com/r/oceanprotocol/algo_dockers/tags).

docs/env.md

@@ -184,6 +184,14 @@ setting it through the environment reaches both.
 - `CRON_DELETE_DB_LOGS`: Delete old logs from database Cron expression. Example: `0 0 * * *` (runs every day at midnight)
 - `CRON_CLEANUP_C2D_STORAGE`: Clear c2d expired resources/storage and delete old jobs. Example: `*/5 * * * *` (runs every 5 minutes)
 
+## Node Metrics History
+
+Powers the `getNodeMetrics` (live snapshot) and `getNodeMetricsHistory` (hourly averages) commands / REST routes. The history layer is SQLite-backed (`databases/nodeMetrics.sqlite`) so it works even with no metadata DB configured. A minute sampler writes the same per-node aggregate the live command returns into a short-lived raw buffer, an hourly roll-up at minute `:05` averages each complete hour into `node_metrics_hourly`, and a daily sweep drops rows older than the retention window. The sampler **warns and skips** (persists no row) when there is no fresh compute aggregate — i.e. `C2D_METRICS_INTERVAL_SECONDS=0` or no engine has sampled yet — so all-zero rows never skew the averages. The live `getNodeMetrics` command still returns a (zeroed) snapshot in that case.
+
+- `NODE_METRICS_HISTORY_ENABLED`: Enable/disable the node-metrics history sampler + roll-up + retention cron jobs. Defaults to enabled whenever a database is available. Set to `false` (also accepts `0`/`no`) to turn the history layer off; the live `getNodeMetrics` command is unaffected. Example: `true`
+- `NODE_METRICS_SAMPLE_CRON`: Cron expression for the minute sampler. Defaults to `* * * * *` (every minute). Example: `* * * * *`
+- `NODE_METRICS_RETENTION_DAYS`: How many days of hourly rows to keep before the daily retention sweep deletes them. Also clamps the range `getNodeMetricsHistory` will return. Defaults to `180` (~6 months). Example: `180`
+
 ## Compute
 
 - `C2D_DOWNLOAD_TIMEOUT`: Timeout (in seconds) for pulling the algorithm docker image during a C2D job. If the pull exceeds this timeout, the job fails with `PullImageFailed` instead of getting stuck. Defaults to `900` (15 minutes). Example: `900`

src/@types/commands.ts

@@ -29,6 +29,16 @@ export interface FindPeerCommand extends Command {
 export interface GetP2PPeersCommand extends Command {}
 export interface GetP2PNetworkStatsCommand extends Command {}
 
+// Live per-node resource snapshot. No params.
+export interface GetNodeMetricsCommand extends Command {}
+
+// Hourly per-node resource history. Both bounds optional; accept epoch ms (number/string) or an
+// ISO-8601 date string. Default range is the last retention window (~6 months) up to now.
+export interface GetNodeMetricsHistoryCommand extends Command {
+  startTime?: number | string
+  stopTime?: number | string
+}
+
 export interface GetAccessListCommand extends Command {
   chainId: number
   contractAddress: string

src/@types/nodeMetrics.ts

@@ -0,0 +1,118 @@
+/**
+ * Shapes for the per-node resource metrics API (`getNodeMetrics` /
+ * `getNodeMetricsHistory`).
+ *
+ * A `NodeMetricsSnapshot` is the same per-node roll-up the telemetry layer already computes
+ * (`ComputeEngineAggregate` on each engine's `lastAggregate`, plus host `os` reads), assembled
+ * by `collectNodeMetricsSnapshot()` and shared by both the live handler and the cron sampler so
+ * the live payload and the stored history have an identical shape.
+ */
+
+export interface NodeMetricsGpu {
+  resourceId: string
+  vendor?: string
+  utilizationPercent?: number
+  memoryUsedBytes?: number
+  memoryTotalBytes?: number
+  temperatureC?: number
+  powerWatts?: number
+}
+
+export interface NodeMetricsEnvResource {
+  env: string
+  resource: string
+  total: number
+  inUse: number
+}
+
+export interface NodeMetricsSnapshot {
+  // epoch ms when the snapshot was assembled
+  collectedAt: number
+  // Freshness signal: false means NO engine had a fresh compute aggregate, so every scalar
+  // below is a structural zero rather than a genuine reading. The live handler returns the
+  // snapshot regardless; the sampler uses this to warn-and-skip instead of persisting zeros.
+  hasAggregate: boolean
+  cpu: {
+    usagePercent: number
+    coresAllocated: number
+    hostCores: number
+    throttledCount: number
+    loadAverage: number[]
+  }
+  memory: {
+    usedBytes: number
+    limitBytes: number
+    hostFreeBytes: number
+    hostTotalBytes: number
+  }
+  disk: {
+    usedBytes: number
+  }
+  network: {
+    rxBytes: number
+    txBytes: number
+  }
+  jobs: {
+    running: number
+    runningFree: number
+    queued: number
+    queuedFree: number
+  }
+  gpu: NodeMetricsGpu[]
+  env: NodeMetricsEnvResource[]
+  meta: {
+    sampledContainers: number
+    oldestSampleAgeSeconds: number
+  }
+}
+
+/**
+ * One hourly bucket returned by `getNodeMetricsHistory`. Scalars are arithmetic means over the
+ * hour's samples; `sampleCount` is how many minute-samples fed the average. GPU entries are
+ * averaged per `resourceId`, env entries per `env`+`resource`.
+ */
+export interface NodeMetricsHourly {
+  // epoch ms of the floored UTC hour this bucket covers
+  hourStart: number
+  sampleCount: number
+  // true only for the live, not-yet-finalized current-hour bucket computed on the fly from raw
+  // samples (never present on a stored/rolled-up bucket). Absent/false = a completed hour.
+  partial?: boolean
+  cpu: {
+    usagePercent: number
+    coresAllocated: number
+    hostCores: number
+    throttledCount: number
+  }
+  memory: {
+    usedBytes: number
+    limitBytes: number
+    hostFreeBytes: number
+    hostTotalBytes: number
+  }
+  disk: {
+    usedBytes: number
+  }
+  network: {
+    rxBytes: number
+    txBytes: number
+  }
+  jobs: {
+    running: number
+    runningFree: number
+    queued: number
+    queuedFree: number
+  }
+  gpu: NodeMetricsGpu[]
+  env: NodeMetricsEnvResource[]
+  meta: {
+    sampledContainers: number
+  }
+}
+
+export interface NodeMetricsHistoryResult {
+  startTime: number
+  stopTime: number
+  count: number
+  buckets: NodeMetricsHourly[]
+}

src/components/core/handler/coreHandlersRegistry.ts

@@ -49,6 +49,7 @@ import {
   GetP2PNetworkStatsHandler,
   FindPeerHandler
 } from './p2p.js'
+import { GetNodeMetricsHandler, GetNodeMetricsHistoryHandler } from './nodeMetrics.js'
 import {
   CreateAuthTokenHandler,
   InvalidateAuthTokenHandler,
@@ -214,6 +215,14 @@ export class CoreHandlersRegistry {
       new GetP2PNetworkStatsHandler(node)
     )
     this.registerCoreHandler(PROTOCOL_COMMANDS.FIND_PEER, new FindPeerHandler(node))
+    this.registerCoreHandler(
+      PROTOCOL_COMMANDS.GET_NODE_METRICS,
+      new GetNodeMetricsHandler(node)
+    )
+    this.registerCoreHandler(
+      PROTOCOL_COMMANDS.GET_NODE_METRICS_HISTORY,
+      new GetNodeMetricsHistoryHandler(node)
+    )
     this.registerCoreHandler(
       PROTOCOL_COMMANDS.CREATE_AUTH_TOKEN,
       new CreateAuthTokenHandler(node)