metrics: graph the plugin workers

Nothing displayed the worker metrics, and the panels that look like they did are about something else: Workers Connected, Worker Slots and Worker Events in the Admin / Maintenance row read SeaweedFS_admin_*, which the older maintenance queue feeds. A cluster running plugin workers - Go or Rust - reads zero there while they are connected and busy. A Plugin Workers row graphs what the workers themselves publish: how many are connected, jobs and their failures, detection and proposal rates, job duration, slot usage, stream events, and what the Lance jobs reclaimed. The panel worth having is Objects Seen vs Skipped, since a sweep with nothing to do and a sweep that could read nothing report the same number of proposals. Also a commented scrape target in the sample Prometheus config. It is 9328 rather than 9327: the sample compose already gives 9327 to the S3 gateway, so the port the worker's own usage text suggests collides with it on a single host. Claude-Session: https://claude.ai/code/session_01Rkp1Mw5E89Jp6dzJFYiMrm
seaweed-worker: fix the metrics address, the count, and a dead field
2026-10-14 02:25:50 +00:00 · 2026-08-20 10:44:32 -07:00 · 2026-08-20 08:43:00 -07:00 · 2026-08-19 23:16:40 -07:00
18 changed files with 1876 additions and 34 deletions
@@ -11,4 +11,8 @@ scrape_configs:
          - 'master:9324'
          - 'volume:9325'
          - 'filer:9326'
-          - 's3:9327'
+          - 's3:9327'
+          # A plugin worker publishes its own metrics when started with
+          # -metricsPort (Go) or --metrics-port (Rust). 9328 continues the
+          # series, since 9327 is already the S3 gateway's.
+          # - 'worker:9328'
@@ -9263,6 +9263,959 @@
        }
      ]
    },
+    {
+      "type": "row",
+      "title": "Plugin Workers",
+      "collapsed": true,
+      "id": 229,
+      "gridPos": {
+        "h": 1,
+        "w": 24,
+        "x": 0,
+        "y": 27
+      },
+      "panels": [
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "title": "Plugin Workers Connected",
+          "description": "Workers with a live control stream. Unlike the Admin/Maintenance row, this counts plugin workers, which is what a Rust or Go plugin worker actually is.",
+          "type": "stat",
+          "id": 219,
+          "gridPos": {
+            "h": 4,
+            "w": 6,
+            "x": 0,
+            "y": 28
+          },
+          "fieldConfig": {
+            "defaults": {
+              "color": {
+                "mode": "thresholds"
+              },
+              "mappings": [],
+              "unit": "short",
+              "thresholds": {
+                "mode": "absolute",
+                "steps": [
+                  {
+                    "color": "green",
+                    "value": null
+                  }
+                ]
+              }
+            },
+            "overrides": []
+          },
+          "options": {
+            "colorMode": "value",
+            "graphMode": "area",
+            "justifyMode": "auto",
+            "orientation": "auto",
+            "reduceOptions": {
+              "calcs": [
+                "lastNotNull"
+              ],
+              "fields": "",
+              "values": false
+            },
+            "textMode": "auto"
+          },
+          "targets": [
+            {
+              "datasource": {
+                "type": "prometheus",
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "editorMode": "code",
+              "expr": "count(SeaweedFS_worker_connected{cluster=~\"$cluster\"} == 1) or vector(0)",
+              "instant": true,
+              "range": false,
+              "refId": "A"
+            }
+          ]
+        },
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "title": "Worker Job Failures (1h)",
+          "description": "Jobs that ended in failure across all workers in the last hour.",
+          "type": "stat",
+          "id": 220,
+          "gridPos": {
+            "h": 4,
+            "w": 6,
+            "x": 6,
+            "y": 28
+          },
+          "fieldConfig": {
+            "defaults": {
+              "color": {
+                "mode": "thresholds"
+              },
+              "mappings": [],
+              "unit": "short",
+              "thresholds": {
+                "mode": "absolute",
+                "steps": [
+                  {
+                    "color": "green",
+                    "value": null
+                  }
+                ]
+              }
+            },
+            "overrides": []
+          },
+          "options": {
+            "colorMode": "value",
+            "graphMode": "area",
+            "justifyMode": "auto",
+            "orientation": "auto",
+            "reduceOptions": {
+              "calcs": [
+                "lastNotNull"
+              ],
+              "fields": "",
+              "values": false
+            },
+            "textMode": "auto"
+          },
+          "targets": [
+            {
+              "datasource": {
+                "type": "prometheus",
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "editorMode": "code",
+              "expr": "sum(increase(SeaweedFS_worker_jobs_total{cluster=~\"$cluster\",result=\"failed\"}[1h])) or vector(0)",
+              "instant": true,
+              "range": false,
+              "refId": "A"
+            }
+          ]
+        },
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "title": "Objects Seen vs Skipped",
+          "description": "A sweep that proposes nothing because there was nothing to do and one that proposes nothing because it could read nothing produce the same proposal count. Skips are the difference, and are the thing to alert on.",
+          "type": "timeseries",
+          "id": 221,
+          "gridPos": {
+            "h": 8,
+            "w": 12,
+            "x": 12,
+            "y": 32
+          },
+          "fieldConfig": {
+            "defaults": {
+              "color": {
+                "mode": "palette-classic"
+              },
+              "custom": {
+                "axisBorderShow": false,
+                "axisCenteredZero": false,
+                "axisColorMode": "text",
+                "axisLabel": "",
+                "axisPlacement": "auto",
+                "barAlignment": 0,
+                "drawStyle": "line",
+                "fillOpacity": 10,
+                "gradientMode": "none",
+                "hideFrom": {
+                  "legend": false,
+                  "tooltip": false,
+                  "viz": false
+                },
+                "insertNulls": false,
+                "lineInterpolation": "linear",
+                "lineWidth": 1,
+                "pointSize": 4,
+                "scaleDistribution": {
+                  "type": "linear"
+                },
+                "showPoints": "never",
+                "spanNulls": false,
+                "stacking": {
+                  "group": "A",
+                  "mode": "none"
+                },
+                "thresholdsStyle": {
+                  "mode": "off"
+                }
+              },
+              "mappings": [],
+              "unit": "ops",
+              "thresholds": {
+                "mode": "absolute",
+                "steps": [
+                  {
+                    "color": "green",
+                    "value": null
+                  }
+                ]
+              }
+            },
+            "overrides": []
+          },
+          "options": {
+            "legend": {
+              "calcs": [
+                "lastNotNull",
+                "max"
+              ],
+              "displayMode": "table",
+              "placement": "bottom",
+              "showLegend": true
+            },
+            "tooltip": {
+              "mode": "multi",
+              "sort": "desc"
+            }
+          },
+          "targets": [
+            {
+              "datasource": {
+                "type": "prometheus",
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "editorMode": "code",
+              "expr": "sum by (job_type) (rate(SeaweedFS_worker_objects_seen_total{cluster=~\"$cluster\"}[$__rate_interval]))",
+              "range": true,
+              "refId": "A",
+              "legendFormat": "seen {{job_type}}"
+            },
+            {
+              "datasource": {
+                "type": "prometheus",
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "editorMode": "code",
+              "expr": "sum by (job_type, reason) (rate(SeaweedFS_worker_objects_skipped_total{cluster=~\"$cluster\"}[$__rate_interval]))",
+              "range": true,
+              "refId": "B",
+              "legendFormat": "skipped {{job_type}} ({{reason}})"
+            }
+          ]
+        },
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "title": "Detections and Proposals",
+          "description": "Detection sweeps by outcome, and the work they handed to admin.",
+          "type": "timeseries",
+          "id": 222,
+          "gridPos": {
+            "h": 8,
+            "w": 12,
+            "x": 0,
+            "y": 32
+          },
+          "fieldConfig": {
+            "defaults": {
+              "color": {
+                "mode": "palette-classic"
+              },
+              "custom": {
+                "axisBorderShow": false,
+                "axisCenteredZero": false,
+                "axisColorMode": "text",
+                "axisLabel": "",
+                "axisPlacement": "auto",
+                "barAlignment": 0,
+                "drawStyle": "line",
+                "fillOpacity": 10,
+                "gradientMode": "none",
+                "hideFrom": {
+                  "legend": false,
+                  "tooltip": false,
+                  "viz": false
+                },
+                "insertNulls": false,
+                "lineInterpolation": "linear",
+                "lineWidth": 1,
+                "pointSize": 4,
+                "scaleDistribution": {
+                  "type": "linear"
+                },
+                "showPoints": "never",
+                "spanNulls": false,
+                "stacking": {
+                  "group": "A",
+                  "mode": "none"
+                },
+                "thresholdsStyle": {
+                  "mode": "off"
+                }
+              },
+              "mappings": [],
+              "unit": "ops",
+              "thresholds": {
+                "mode": "absolute",
+                "steps": [
+                  {
+                    "color": "green",
+                    "value": null
+                  }
+                ]
+              }
+            },
+            "overrides": []
+          },
+          "options": {
+            "legend": {
+              "calcs": [
+                "lastNotNull",
+                "max"
+              ],
+              "displayMode": "table",
+              "placement": "bottom",
+              "showLegend": true
+            },
+            "tooltip": {
+              "mode": "multi",
+              "sort": "desc"
+            }
+          },
+          "targets": [
+            {
+              "datasource": {
+                "type": "prometheus",
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "editorMode": "code",
+              "expr": "sum by (job_type, result) (rate(SeaweedFS_worker_detections_total{cluster=~\"$cluster\"}[$__rate_interval]))",
+              "range": true,
+              "refId": "A",
+              "legendFormat": "detections {{job_type}} ({{result}})"
+            },
+            {
+              "datasource": {
+                "type": "prometheus",
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "editorMode": "code",
+              "expr": "sum by (job_type) (rate(SeaweedFS_worker_proposals_total{cluster=~\"$cluster\"}[$__rate_interval]))",
+              "range": true,
+              "refId": "B",
+              "legendFormat": "proposals {{job_type}}"
+            }
+          ]
+        },
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "title": "Jobs Executed",
+          "description": "Jobs run by these workers, by outcome.",
+          "type": "timeseries",
+          "id": 223,
+          "gridPos": {
+            "h": 8,
+            "w": 12,
+            "x": 0,
+            "y": 40
+          },
+          "fieldConfig": {
+            "defaults": {
+              "color": {
+                "mode": "palette-classic"
+              },
+              "custom": {
+                "axisBorderShow": false,
+                "axisCenteredZero": false,
+                "axisColorMode": "text",
+                "axisLabel": "",
+                "axisPlacement": "auto",
+                "barAlignment": 0,
+                "drawStyle": "line",
+                "fillOpacity": 10,
+                "gradientMode": "none",
+                "hideFrom": {
+                  "legend": false,
+                  "tooltip": false,
+                  "viz": false
+                },
+                "insertNulls": false,
+                "lineInterpolation": "linear",
+                "lineWidth": 1,
+                "pointSize": 4,
+                "scaleDistribution": {
+                  "type": "linear"
+                },
+                "showPoints": "never",
+                "spanNulls": false,
+                "stacking": {
+                  "group": "A",
+                  "mode": "none"
+                },
+                "thresholdsStyle": {
+                  "mode": "off"
+                }
+              },
+              "mappings": [],
+              "unit": "ops",
+              "thresholds": {
+                "mode": "absolute",
+                "steps": [
+                  {
+                    "color": "green",
+                    "value": null
+                  }
+                ]
+              }
+            },
+            "overrides": []
+          },
+          "options": {
+            "legend": {
+              "calcs": [
+                "lastNotNull",
+                "max"
+              ],
+              "displayMode": "table",
+              "placement": "bottom",
+              "showLegend": true
+            },
+            "tooltip": {
+              "mode": "multi",
+              "sort": "desc"
+            }
+          },
+          "targets": [
+            {
+              "datasource": {
+                "type": "prometheus",
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "editorMode": "code",
+              "expr": "sum by (job_type, result) (rate(SeaweedFS_worker_jobs_total{cluster=~\"$cluster\"}[$__rate_interval]))",
+              "range": true,
+              "refId": "A",
+              "legendFormat": "{{job_type}} ({{result}})"
+            }
+          ]
+        },
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "title": "Job Duration p99",
+          "description": "How long a job takes, per job type.",
+          "type": "timeseries",
+          "id": 224,
+          "gridPos": {
+            "h": 8,
+            "w": 12,
+            "x": 12,
+            "y": 40
+          },
+          "fieldConfig": {
+            "defaults": {
+              "color": {
+                "mode": "palette-classic"
+              },
+              "custom": {
+                "axisBorderShow": false,
+                "axisCenteredZero": false,
+                "axisColorMode": "text",
+                "axisLabel": "",
+                "axisPlacement": "auto",
+                "barAlignment": 0,
+                "drawStyle": "line",
+                "fillOpacity": 10,
+                "gradientMode": "none",
+                "hideFrom": {
+                  "legend": false,
+                  "tooltip": false,
+                  "viz": false
+                },
+                "insertNulls": false,
+                "lineInterpolation": "linear",
+                "lineWidth": 1,
+                "pointSize": 4,
+                "scaleDistribution": {
+                  "type": "linear"
+                },
+                "showPoints": "never",
+                "spanNulls": false,
+                "stacking": {
+                  "group": "A",
+                  "mode": "none"
+                },
+                "thresholdsStyle": {
+                  "mode": "off"
+                }
+              },
+              "mappings": [],
+              "unit": "s",
+              "thresholds": {
+                "mode": "absolute",
+                "steps": [
+                  {
+                    "color": "green",
+                    "value": null
+                  }
+                ]
+              }
+            },
+            "overrides": []
+          },
+          "options": {
+            "legend": {
+              "calcs": [
+                "lastNotNull",
+                "max"
+              ],
+              "displayMode": "table",
+              "placement": "bottom",
+              "showLegend": true
+            },
+            "tooltip": {
+              "mode": "multi",
+              "sort": "desc"
+            }
+          },
+          "targets": [
+            {
+              "datasource": {
+                "type": "prometheus",
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "editorMode": "code",
+              "expr": "histogram_quantile(0.99, sum(rate(SeaweedFS_worker_job_seconds_bucket{cluster=~\"$cluster\"}[$__rate_interval])) by (le, job_type))",
+              "range": true,
+              "refId": "A",
+              "legendFormat": "{{job_type}}"
+            }
+          ]
+        },
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "title": "Worker Slots",
+          "description": "Capacity a worker advertised and how much of it is in use. Held slots are what admin schedules against.",
+          "type": "timeseries",
+          "id": 225,
+          "gridPos": {
+            "h": 8,
+            "w": 12,
+            "x": 0,
+            "y": 48
+          },
+          "fieldConfig": {
+            "defaults": {
+              "color": {
+                "mode": "palette-classic"
+              },
+              "custom": {
+                "axisBorderShow": false,
+                "axisCenteredZero": false,
+                "axisColorMode": "text",
+                "axisLabel": "",
+                "axisPlacement": "auto",
+                "barAlignment": 0,
+                "drawStyle": "line",
+                "fillOpacity": 10,
+                "gradientMode": "none",
+                "hideFrom": {
+                  "legend": false,
+                  "tooltip": false,
+                  "viz": false
+                },
+                "insertNulls": false,
+                "lineInterpolation": "linear",
+                "lineWidth": 1,
+                "pointSize": 4,
+                "scaleDistribution": {
+                  "type": "linear"
+                },
+                "showPoints": "never",
+                "spanNulls": false,
+                "stacking": {
+                  "group": "A",
+                  "mode": "none"
+                },
+                "thresholdsStyle": {
+                  "mode": "off"
+                }
+              },
+              "mappings": [],
+              "unit": "short",
+              "thresholds": {
+                "mode": "absolute",
+                "steps": [
+                  {
+                    "color": "green",
+                    "value": null
+                  }
+                ]
+              }
+            },
+            "overrides": []
+          },
+          "options": {
+            "legend": {
+              "calcs": [
+                "lastNotNull",
+                "max"
+              ],
+              "displayMode": "table",
+              "placement": "bottom",
+              "showLegend": true
+            },
+            "tooltip": {
+              "mode": "multi",
+              "sort": "desc"
+            }
+          },
+          "targets": [
+            {
+              "datasource": {
+                "type": "prometheus",
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "editorMode": "code",
+              "expr": "sum by (lane) (SeaweedFS_worker_slots_used{cluster=~\"$cluster\"})",
+              "range": true,
+              "refId": "A",
+              "legendFormat": "used {{lane}}"
+            },
+            {
+              "datasource": {
+                "type": "prometheus",
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "editorMode": "code",
+              "expr": "sum by (lane) (SeaweedFS_worker_slots_total{cluster=~\"$cluster\"})",
+              "range": true,
+              "refId": "B",
+              "legendFormat": "total {{lane}}"
+            }
+          ]
+        },
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "title": "Control Stream Events",
+          "description": "Connects, closes and failures. A worker that reconnects steadily is usually two workers sharing one id, evicting each other.",
+          "type": "timeseries",
+          "id": 226,
+          "gridPos": {
+            "h": 8,
+            "w": 12,
+            "x": 12,
+            "y": 48
+          },
+          "fieldConfig": {
+            "defaults": {
+              "color": {
+                "mode": "palette-classic"
+              },
+              "custom": {
+                "axisBorderShow": false,
+                "axisCenteredZero": false,
+                "axisColorMode": "text",
+                "axisLabel": "",
+                "axisPlacement": "auto",
+                "barAlignment": 0,
+                "drawStyle": "line",
+                "fillOpacity": 10,
+                "gradientMode": "none",
+                "hideFrom": {
+                  "legend": false,
+                  "tooltip": false,
+                  "viz": false
+                },
+                "insertNulls": false,
+                "lineInterpolation": "linear",
+                "lineWidth": 1,
+                "pointSize": 4,
+                "scaleDistribution": {
+                  "type": "linear"
+                },
+                "showPoints": "never",
+                "spanNulls": false,
+                "stacking": {
+                  "group": "A",
+                  "mode": "none"
+                },
+                "thresholdsStyle": {
+                  "mode": "off"
+                }
+              },
+              "mappings": [],
+              "unit": "ops",
+              "thresholds": {
+                "mode": "absolute",
+                "steps": [
+                  {
+                    "color": "green",
+                    "value": null
+                  }
+                ]
+              }
+            },
+            "overrides": []
+          },
+          "options": {
+            "legend": {
+              "calcs": [
+                "lastNotNull",
+                "max"
+              ],
+              "displayMode": "table",
+              "placement": "bottom",
+              "showLegend": true
+            },
+            "tooltip": {
+              "mode": "multi",
+              "sort": "desc"
+            }
+          },
+          "targets": [
+            {
+              "datasource": {
+                "type": "prometheus",
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "editorMode": "code",
+              "expr": "sum by (event) (rate(SeaweedFS_worker_stream_events_total{cluster=~\"$cluster\"}[$__rate_interval]))",
+              "range": true,
+              "refId": "A",
+              "legendFormat": "{{event}}"
+            }
+          ]
+        },
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "title": "Lance Maintenance Reclaimed",
+          "description": "What the Lance jobs actually removed. A job that runs every minute and reclaims nothing is a different thing from a job that never runs.",
+          "type": "timeseries",
+          "id": 227,
+          "gridPos": {
+            "h": 8,
+            "w": 12,
+            "x": 0,
+            "y": 56
+          },
+          "fieldConfig": {
+            "defaults": {
+              "color": {
+                "mode": "palette-classic"
+              },
+              "custom": {
+                "axisBorderShow": false,
+                "axisCenteredZero": false,
+                "axisColorMode": "text",
+                "axisLabel": "",
+                "axisPlacement": "auto",
+                "barAlignment": 0,
+                "drawStyle": "line",
+                "fillOpacity": 10,
+                "gradientMode": "none",
+                "hideFrom": {
+                  "legend": false,
+                  "tooltip": false,
+                  "viz": false
+                },
+                "insertNulls": false,
+                "lineInterpolation": "linear",
+                "lineWidth": 1,
+                "pointSize": 4,
+                "scaleDistribution": {
+                  "type": "linear"
+                },
+                "showPoints": "never",
+                "spanNulls": false,
+                "stacking": {
+                  "group": "A",
+                  "mode": "none"
+                },
+                "thresholdsStyle": {
+                  "mode": "off"
+                }
+              },
+              "mappings": [],
+              "unit": "ops",
+              "thresholds": {
+                "mode": "absolute",
+                "steps": [
+                  {
+                    "color": "green",
+                    "value": null
+                  }
+                ]
+              }
+            },
+            "overrides": []
+          },
+          "options": {
+            "legend": {
+              "calcs": [
+                "lastNotNull",
+                "max"
+              ],
+              "displayMode": "table",
+              "placement": "bottom",
+              "showLegend": true
+            },
+            "tooltip": {
+              "mode": "multi",
+              "sort": "desc"
+            }
+          },
+          "targets": [
+            {
+              "datasource": {
+                "type": "prometheus",
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "editorMode": "code",
+              "expr": "sum(rate(SeaweedFS_worker_lance_fragments_removed_total{cluster=~\"$cluster\"}[$__rate_interval]))",
+              "range": true,
+              "refId": "A",
+              "legendFormat": "fragments merged"
+            },
+            {
+              "datasource": {
+                "type": "prometheus",
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "editorMode": "code",
+              "expr": "sum(rate(SeaweedFS_worker_lance_rows_indexed_total{cluster=~\"$cluster\"}[$__rate_interval]))",
+              "range": true,
+              "refId": "B",
+              "legendFormat": "rows indexed"
+            },
+            {
+              "datasource": {
+                "type": "prometheus",
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "editorMode": "code",
+              "expr": "sum(rate(SeaweedFS_worker_lance_versions_removed_total{cluster=~\"$cluster\"}[$__rate_interval]))",
+              "range": true,
+              "refId": "C",
+              "legendFormat": "versions removed"
+            }
+          ]
+        },
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "title": "Storage Reclaimed",
+          "description": "Bytes freed by version cleanup. Its own panel because bytes and counts do not belong on one axis.",
+          "type": "timeseries",
+          "id": 228,
+          "gridPos": {
+            "h": 8,
+            "w": 12,
+            "x": 12,
+            "y": 56
+          },
+          "fieldConfig": {
+            "defaults": {
+              "color": {
+                "mode": "palette-classic"
+              },
+              "custom": {
+                "axisBorderShow": false,
+                "axisCenteredZero": false,
+                "axisColorMode": "text",
+                "axisLabel": "",
+                "axisPlacement": "auto",
+                "barAlignment": 0,
+                "drawStyle": "line",
+                "fillOpacity": 10,
+                "gradientMode": "none",
+                "hideFrom": {
+                  "legend": false,
+                  "tooltip": false,
+                  "viz": false
+                },
+                "insertNulls": false,
+                "lineInterpolation": "linear",
+                "lineWidth": 1,
+                "pointSize": 4,
+                "scaleDistribution": {
+                  "type": "linear"
+                },
+                "showPoints": "never",
+                "spanNulls": false,
+                "stacking": {
+                  "group": "A",
+                  "mode": "none"
+                },
+                "thresholdsStyle": {
+                  "mode": "off"
+                }
+              },
+              "mappings": [],
+              "unit": "Bps",
+              "thresholds": {
+                "mode": "absolute",
+                "steps": [
+                  {
+                    "color": "green",
+                    "value": null
+                  }
+                ]
+              }
+            },
+            "overrides": []
+          },
+          "options": {
+            "legend": {
+              "calcs": [
+                "lastNotNull",
+                "max"
+              ],
+              "displayMode": "table",
+              "placement": "bottom",
+              "showLegend": true
+            },
+            "tooltip": {
+              "mode": "multi",
+              "sort": "desc"
+            }
+          },
+          "targets": [
+            {
+              "datasource": {
+                "type": "prometheus",
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "editorMode": "code",
+              "expr": "sum(rate(SeaweedFS_worker_lance_bytes_reclaimed_total{cluster=~\"$cluster\"}[$__rate_interval]))",
+              "range": true,
+              "refId": "A",
+              "legendFormat": "reclaimed"
+            }
+          ]
+        }
+      ]
+    },
    {
      "type": "row",
      "title": "Go Runtime",
@@ -9272,7 +10225,7 @@
        "h": 1,
        "w": 24,
        "x": 0,
-        "y": 27
+        "y": 28
      },
      "panels": [
        {
@@ -867,6 +867,8 @@ dependencies = [
 "http 1.5.0",
 "http-body 1.1.0",
 "http-body-util",
+ "hyper",
+ "hyper-util",
 "itoa",
 "matchit",
 "memchr",
@@ -875,10 +877,15 @@ dependencies = [
 "pin-project-lite",
 "rustversion",
 "serde",
+ "serde_json",
+ "serde_path_to_error",
+ "serde_urlencoded",
 "sync_wrapper",
+ "tokio",
 "tower 0.5.3",
 "tower-layer",
 "tower-service",
+ "tracing",
 ]

 [[package]]
@@ -899,6 +906,7 @@ dependencies = [
 "sync_wrapper",
 "tower-layer",
 "tower-service",
+ "tracing",
 ]

 [[package]]
@@ -2120,7 +2128,7 @@ version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "10d60334b3b2e7c9d91ef8150abfb6fa4c1c39ebbcf4a81c2e346aad939fee3e"
 dependencies = [
- "thiserror",
+ "thiserror 2.0.20",
 ]

 [[package]]
@@ -3107,7 +3115,7 @@ dependencies = [
 "jni-sys",
 "log",
 "simd_cesu8",
- "thiserror",
+ "thiserror 2.0.20",
 "walkdir",
 "windows-link",
 ]
@@ -4162,7 +4170,7 @@ dependencies = [
 "serde",
 "serde_json",
 "serde_urlencoded",
- "thiserror",
+ "thiserror 2.0.20",
 "tokio",
 "tracing",
 "url",
@@ -4541,6 +4549,20 @@ dependencies = [
 "unicode-ident",
 ]

+[[package]]
+name = "prometheus"
+version = "0.13.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3d33c28a30771f7f96db69893f78b857f7450d7e0237e9c8fc6427a81bae7ed1"
+dependencies = [
+ "cfg-if",
+ "fnv",
+ "lazy_static",
+ "memchr",
+ "parking_lot",
+ "thiserror 1.0.69",
+]
+
 [[package]]
 name = "prost"
 version = "0.13.5"
@@ -4690,7 +4712,7 @@ dependencies = [
 "rustc-hash",
 "rustls",
 "socket2 0.6.5",
- "thiserror",
+ "thiserror 2.0.20",
 "tokio",
 "tracing",
 "web-time",
@@ -4713,7 +4735,7 @@ dependencies = [
 "rustls",
 "rustls-pki-types",
 "slab",
- "thiserror",
+ "thiserror 2.0.20",
 "tinyvec",
 "tracing",
 "web-time",
@@ -4913,7 +4935,7 @@ checksum = "a4e608c6638b9c18977b00b475ac1f28d14e84b27d8d42f70e0bf1e3dec127ac"
 dependencies = [
 "getrandom 0.2.17",
 "libredox",
- "thiserror",
+ "thiserror 2.0.20",
 ]

 [[package]]
@@ -5353,6 +5375,8 @@ version = "0.1.0"
 dependencies = [
 "anyhow",
 "async-trait",
+ "axum",
+ "prometheus",
 "prost 0.13.5",
 "prost-types 0.13.5",
 "tokio",
@@ -5440,6 +5464,17 @@ dependencies = [
 "zmij",
 ]

+[[package]]
+name = "serde_path_to_error"
+version = "0.1.20"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457"
+dependencies = [
+ "itoa",
+ "serde",
+ "serde_core",
+]
+
 [[package]]
 name = "serde_repr"
 version = "0.1.21"
@@ -5784,13 +5819,33 @@ dependencies = [
 "windows-sys 0.61.2",
 ]

+[[package]]
+name = "thiserror"
+version = "1.0.69"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52"
+dependencies = [
+ "thiserror-impl 1.0.69",
+]
+
 [[package]]
 name = "thiserror"
 version = "2.0.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ec86235f5fcc2a73650310756d2ac5b138a5780bbbdfae3eeccec992c435ba4f"
 dependencies = [
- "thiserror-impl",
+ "thiserror-impl 2.0.20",
+]
+
+[[package]]
+name = "thiserror-impl"
+version = "1.0.69"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
 ]

 [[package]]
@@ -6030,6 +6085,7 @@ dependencies = [
 "tokio",
 "tower-layer",
 "tower-service",
+ "tracing",
 ]

 [[package]]
@@ -6073,6 +6129,7 @@ version = "0.1.44"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100"
 dependencies = [
+ "log",
 "pin-project-lite",
 "tracing-attributes",
 "tracing-core",
@@ -6422,6 +6479,7 @@ dependencies = [
 "lance",
 "lance-index",
 "lance-linalg",
+ "prometheus",
 "reqwest 0.12.28",
 "seaweed-worker-core",
 "serde",
@@ -19,6 +19,9 @@ prost-types = "0.13"
 tokio = { version = "1", features = ["full"] }
 tokio-stream = "0.1"
 tonic = { version = "0.12", features = ["tls"] }
+# Already in the tree via tonic; named here so the metrics server can use them.
+axum = "0.7"
+prometheus = { version = "0.13", default-features = false }
 tonic-build = "0.12"
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
@@ -20,6 +20,36 @@ The admin's *HTTP* address is what an operator has; the gRPC port is derived
 from it the way the Go side does. Dialling the HTTP port fails as "frame with
 invalid size", which reads like a protocol bug rather than a wrong port.

+## Metrics
+
+    cargo run -p weed-lance-worker -- --admin 127.0.0.1:23646 --metrics-port 9328
+
+Serves `/health`, `/ready` and `/metrics` on that port, the same three the Go
+worker serves under `weed worker -metricsPort`, so one scrape config covers
+workers in either language. Off by default, and bound to loopback unless
+`--metrics-ip` says otherwise, because the endpoint is unauthenticated. 9328
+continues the series the other components use (master 9324, volume 9325, filer
+9326, s3 9327); an IPv6 address works with or without brackets.
+
+Grafana: the "Plugin Workers" row of `other/metrics/grafana_seaweedfs.json`
+graphs these. Its panels filter on `$cluster`, which comes from the scrape job's
+labels, so scrape the worker the way the rest of the cluster is scraped or the
+row stays empty.
+
+Names are `SeaweedFS_worker_*`, matching the Go side's convention. The pair
+worth alerting on is `objects_seen_total` and `objects_skipped_total`: a sweep
+that proposes nothing and a sweep that could read nothing look identical from
+`proposals_total` alone.
+
+    SeaweedFS_worker_connected 1
+    SeaweedFS_worker_objects_seen_total{job_type="lance_compact"} 7
+    SeaweedFS_worker_proposals_total{job_type="lance_compact"} 2
+    SeaweedFS_worker_jobs_total{job_type="lance_compact",result="ok"} 2
+    SeaweedFS_worker_lance_fragments_removed_total 25
+
+`/ready` follows the control stream: a worker whose admin has gone away is
+running but is not going to do anything.
+
 ## Credentials

 The worker holds none. It asks the namespace to describe a table with
@@ -15,6 +15,8 @@ prost-types.workspace = true
 tokio.workspace = true
 tokio-stream.workspace = true
 tonic.workspace = true
+axum.workspace = true
+prometheus.workspace = true
 tracing.workspace = true

 [build-dependencies]
@@ -9,6 +9,7 @@
 pub mod address;
 pub mod config;
 pub mod config_form;
+pub mod metrics;
 pub mod registry;
 pub mod senders;
 pub mod stream;
@@ -19,6 +20,7 @@ pub mod pb {
 }

 pub use config::{TlsOptions, WorkerOptions};
+pub use metrics::Metrics;
 pub use registry::{JobHandler, Preview, PreviewProvider, Registry};
 pub use senders::{DetectionSender, ExecutionSender};
 pub use stream::run;
@@ -0,0 +1,395 @@
+//! What a worker will say about itself when nobody is watching the logs.
+//!
+//! The Go plugin worker serves `/health`, `/ready` and `/metrics` on an optional
+//! port (`weed worker -metricsPort`); this is the same contract, so one scrape
+//! config covers workers in either language. Names follow the Go side's
+//! convention, `SeaweedFS_<subsystem>_<name>`, under a `worker` subsystem that
+//! nothing else uses.
+//!
+//! Nothing here is Lance-specific. A worker for another format registers its own
+//! collectors on the same registry and gets the same endpoint.
+
+use std::net::SocketAddr;
+use std::sync::Arc;
+
+use anyhow::{Context, Result};
+use prometheus::{
+    Encoder, HistogramOpts, HistogramVec, IntCounter, IntCounterVec, IntGauge, IntGaugeVec, Opts,
+    Registry, TextEncoder,
+};
+use std::sync::atomic::{AtomicBool, Ordering};
+use tracing::{info, warn};
+
+const NAMESPACE: &str = "SeaweedFS";
+const SUBSYSTEM: &str = "worker";
+
+/// The metrics one worker process publishes, and the registry they live on.
+///
+/// Cloneable because the stream loop, the job handlers and the HTTP server all
+/// hold it; everything inside is already shared.
+#[derive(Clone)]
+pub struct Metrics {
+    registry: Registry,
+    connected: IntGauge,
+    connects: IntCounterVec,
+    slots_used: IntGaugeVec,
+    slots_total: IntGaugeVec,
+    detections: IntCounterVec,
+    detection_seconds: HistogramVec,
+    proposals: IntCounterVec,
+    objects_seen: IntCounterVec,
+    objects_skipped: IntCounterVec,
+    jobs: IntCounterVec,
+    job_seconds: HistogramVec,
+    previews: IntCounterVec,
+    ready: Arc<AtomicBool>,
+}
+
+impl Metrics {
+    pub fn new(worker_id: &str, worker_version: &str) -> Result<Self> {
+        let registry = Registry::new();
+
+        // Build info as a constant 1, the way the Go side does it, so a scrape
+        // can tell which worker and which build answered.
+        let build = IntGaugeVec::new(
+            Opts::new("build_info", "Worker build information.")
+                .namespace(NAMESPACE)
+                .subsystem(SUBSYSTEM),
+            &["worker_id", "version"],
+        )?;
+        build.with_label_values(&[worker_id, worker_version]).set(1);
+        registry.register(Box::new(build))?;
+
+        let connected = IntGauge::with_opts(
+            Opts::new("connected", "1 while the admin control stream is up.")
+                .namespace(NAMESPACE)
+                .subsystem(SUBSYSTEM),
+        )?;
+        let connects = IntCounterVec::new(
+            Opts::new(
+                "stream_events_total",
+                "Control stream lifecycle events by outcome (connected, closed, failed, shutdown).",
+            )
+            .namespace(NAMESPACE)
+            .subsystem(SUBSYSTEM),
+            &["event"],
+        )?;
+        let slots_used = IntGaugeVec::new(
+            Opts::new("slots_used", "Slots currently held, by lane.")
+                .namespace(NAMESPACE)
+                .subsystem(SUBSYSTEM),
+            &["lane"],
+        )?;
+        let slots_total = IntGaugeVec::new(
+            Opts::new("slots_total", "Slots this worker advertises, by lane.")
+                .namespace(NAMESPACE)
+                .subsystem(SUBSYSTEM),
+            &["lane"],
+        )?;
+        let detections = IntCounterVec::new(
+            Opts::new(
+                "detections_total",
+                "Detection sweeps by job type and outcome.",
+            )
+            .namespace(NAMESPACE)
+            .subsystem(SUBSYSTEM),
+            &["job_type", "result"],
+        )?;
+        let detection_seconds = HistogramVec::new(
+            HistogramOpts::new("detection_seconds", "How long a detection sweep took.")
+                .namespace(NAMESPACE)
+                .subsystem(SUBSYSTEM)
+                .buckets(prometheus::exponential_buckets(0.01, 2.0, 14)?),
+            &["job_type"],
+        )?;
+        let proposals = IntCounterVec::new(
+            Opts::new("proposals_total", "Jobs proposed to admin, by job type.")
+                .namespace(NAMESPACE)
+                .subsystem(SUBSYSTEM),
+            &["job_type"],
+        )?;
+        let objects_seen = IntCounterVec::new(
+            Opts::new(
+                "objects_seen_total",
+                "Objects a detection sweep read, by job type.",
+            )
+            .namespace(NAMESPACE)
+            .subsystem(SUBSYSTEM),
+            &["job_type"],
+        )?;
+        let objects_skipped = IntCounterVec::new(
+            Opts::new(
+                "objects_skipped_total",
+                "Objects a sweep could not read, by job type and reason. A sweep that \
+                 proposes nothing looks the same as one that could read nothing; this is \
+                 what tells them apart.",
+            )
+            .namespace(NAMESPACE)
+            .subsystem(SUBSYSTEM),
+            &["job_type", "reason"],
+        )?;
+        let jobs = IntCounterVec::new(
+            Opts::new("jobs_total", "Jobs executed by job type and outcome.")
+                .namespace(NAMESPACE)
+                .subsystem(SUBSYSTEM),
+            &["job_type", "result"],
+        )?;
+        let job_seconds = HistogramVec::new(
+            HistogramOpts::new("job_seconds", "How long a job took to run.")
+                .namespace(NAMESPACE)
+                .subsystem(SUBSYSTEM)
+                .buckets(prometheus::exponential_buckets(0.05, 2.0, 16)?),
+            &["job_type"],
+        )?;
+        let previews = IntCounterVec::new(
+            Opts::new(
+                "previews_total",
+                "Object previews served to admin, by outcome.",
+            )
+            .namespace(NAMESPACE)
+            .subsystem(SUBSYSTEM),
+            &["result"],
+        )?;
+
+        for collector in [
+            Box::new(connected.clone()) as Box<dyn prometheus::core::Collector>,
+            Box::new(connects.clone()),
+            Box::new(slots_used.clone()),
+            Box::new(slots_total.clone()),
+            Box::new(detections.clone()),
+            Box::new(detection_seconds.clone()),
+            Box::new(proposals.clone()),
+            Box::new(objects_seen.clone()),
+            Box::new(objects_skipped.clone()),
+            Box::new(jobs.clone()),
+            Box::new(job_seconds.clone()),
+            Box::new(previews.clone()),
+        ] {
+            registry.register(collector)?;
+        }
+
+        Ok(Self {
+            registry,
+            connected,
+            connects,
+            slots_used,
+            slots_total,
+            detections,
+            detection_seconds,
+            proposals,
+            objects_seen,
+            objects_skipped,
+            jobs,
+            job_seconds,
+            previews,
+            ready: Arc::new(AtomicBool::new(false)),
+        })
+    }
+
+    /// The registry, so a worker can add collectors of its own.
+    pub fn registry(&self) -> &Registry {
+        &self.registry
+    }
+
+    pub fn stream_connected(&self) {
+        self.connected.set(1);
+        self.ready.store(true, Ordering::Relaxed);
+        self.connects.with_label_values(&["connected"]).inc();
+    }
+
+    /// `event` is why the stream ended: closed, failed, or shutdown.
+    pub fn stream_ended(&self, event: &str) {
+        self.connected.set(0);
+        self.ready.store(false, Ordering::Relaxed);
+        self.connects.with_label_values(&[event]).inc();
+    }
+
+    pub fn set_slots(&self, lane: &str, used: i64, total: i64) {
+        self.slots_used.with_label_values(&[lane]).set(used);
+        self.slots_total.with_label_values(&[lane]).set(total);
+    }
+
+    pub fn detection_finished(&self, job_type: &str, result: &str, seconds: f64, proposals: usize) {
+        self.detections
+            .with_label_values(&[job_type, result])
+            .inc_by(1);
+        self.detection_seconds
+            .with_label_values(&[job_type])
+            .observe(seconds);
+        self.proposals
+            .with_label_values(&[job_type])
+            .inc_by(proposals as u64);
+    }
+
+    pub fn object_seen(&self, job_type: &str) {
+        self.objects_seen.with_label_values(&[job_type]).inc();
+    }
+
+    pub fn object_skipped(&self, job_type: &str, reason: &str) {
+        self.objects_skipped
+            .with_label_values(&[job_type, reason])
+            .inc();
+    }
+
+    pub fn job_finished(&self, job_type: &str, result: &str, seconds: f64) {
+        self.jobs.with_label_values(&[job_type, result]).inc();
+        self.job_seconds
+            .with_label_values(&[job_type])
+            .observe(seconds);
+    }
+
+    pub fn preview_finished(&self, result: &str) {
+        self.previews.with_label_values(&[result]).inc();
+    }
+
+    /// A counter this worker's own jobs can raise, e.g. fragments removed.
+    /// Registered lazily so a format's numbers live beside the generic ones
+    /// without core having to know what they are.
+    pub fn counter(&self, name: &str, help: &str) -> Result<IntCounter> {
+        let counter = IntCounter::with_opts(
+            Opts::new(name, help)
+                .namespace(NAMESPACE)
+                .subsystem(SUBSYSTEM),
+        )?;
+        self.registry.register(Box::new(counter.clone()))?;
+        Ok(counter)
+    }
+
+    fn gather(&self) -> Result<String> {
+        let mut buffer = Vec::new();
+        TextEncoder::new().encode(&self.registry.gather(), &mut buffer)?;
+        Ok(String::from_utf8(buffer)?)
+    }
+
+    fn is_ready(&self) -> bool {
+        self.ready.load(Ordering::Relaxed)
+    }
+}
+
+/// Serves the metrics endpoints until the process stops. Failing to bind is
+/// logged rather than fatal: a worker that cannot publish metrics should still
+/// do its work.
+pub async fn serve(metrics: Metrics, addr: SocketAddr) -> Result<()> {
+    use axum::extract::State;
+    use axum::http::StatusCode;
+    use axum::routing::get;
+    use axum::Router;
+
+    let app = Router::new()
+        .route("/health", get(|| async { StatusCode::OK }))
+        .route(
+            "/ready",
+            get(|State(metrics): State<Metrics>| async move {
+                if metrics.is_ready() {
+                    StatusCode::OK
+                } else {
+                    StatusCode::SERVICE_UNAVAILABLE
+                }
+            }),
+        )
+        .route(
+            "/metrics",
+            get(|State(metrics): State<Metrics>| async move {
+                match metrics.gather() {
+                    Ok(body) => (StatusCode::OK, body),
+                    Err(err) => (StatusCode::INTERNAL_SERVER_ERROR, format!("{err:#}")),
+                }
+            }),
+        )
+        .with_state(metrics);
+
+    let listener = tokio::net::TcpListener::bind(addr)
+        .await
+        .with_context(|| format!("bind the metrics server to {addr}"))?;
+    info!("serving worker metrics on http://{addr}/metrics");
+    axum::serve(listener, app)
+        .await
+        .context("metrics server stopped")
+}
+
+/// Starts the metrics server in the background, warning rather than failing when
+/// it cannot start.
+pub fn spawn(metrics: Metrics, addr: SocketAddr) {
+    tokio::spawn(async move {
+        if let Err(err) = serve(metrics, addr).await {
+            warn!("worker metrics server: {err:#}");
+        }
+    });
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn metrics() -> Metrics {
+        Metrics::new("worker-1", "0.1.0").expect("build metrics")
+    }
+
+    // A scrape has to be able to tell these apart: a worker that swept and found
+    // nothing to do, and one that could not read anything it swept.
+    #[test]
+    fn a_sweep_reports_what_it_read_and_what_it_could_not() {
+        let metrics = metrics();
+        metrics.object_seen("lance_compact");
+        metrics.object_seen("lance_compact");
+        metrics.object_skipped("lance_compact", "open");
+        metrics.detection_finished("lance_compact", "ok", 0.5, 0);
+
+        let text = metrics.gather().expect("gather");
+        assert!(text.contains("SeaweedFS_worker_objects_seen_total{job_type=\"lance_compact\"} 2"));
+        assert!(text.contains(
+            "SeaweedFS_worker_objects_skipped_total{job_type=\"lance_compact\",reason=\"open\"} 1"
+        ));
+        assert!(text.contains(
+            "SeaweedFS_worker_detections_total{job_type=\"lance_compact\",result=\"ok\"} 1"
+        ));
+        assert!(text.contains("SeaweedFS_worker_proposals_total{job_type=\"lance_compact\"} 0"));
+    }
+
+    // /ready follows the stream, because a worker with no admin behind it is
+    // running but not doing anything.
+    #[test]
+    fn readiness_follows_the_stream() {
+        let metrics = metrics();
+        assert!(!metrics.is_ready(), "not ready before the stream is up");
+
+        metrics.stream_connected();
+        assert!(metrics.is_ready());
+        assert!(metrics
+            .gather()
+            .unwrap()
+            .contains("SeaweedFS_worker_connected 1"));
+
+        metrics.stream_ended("closed");
+        assert!(!metrics.is_ready());
+        assert!(metrics
+            .gather()
+            .unwrap()
+            .contains("SeaweedFS_worker_connected 0"));
+        assert!(metrics
+            .gather()
+            .unwrap()
+            .contains("SeaweedFS_worker_stream_events_total{event=\"closed\"} 1"));
+    }
+
+    #[test]
+    fn build_info_names_the_worker() {
+        let text = metrics().gather().expect("gather");
+        assert!(text
+            .contains("SeaweedFS_worker_build_info{version=\"0.1.0\",worker_id=\"worker-1\"} 1"));
+    }
+
+    // A format's own numbers land on the same registry, so one endpoint serves
+    // both and a second worker implementation needs no new plumbing.
+    #[test]
+    fn a_worker_can_add_counters_of_its_own() {
+        let metrics = metrics();
+        let counter = metrics
+            .counter("lance_fragments_removed_total", "Fragments merged away.")
+            .expect("register");
+        counter.inc_by(16);
+
+        let text = metrics.gather().expect("gather");
+        assert!(text.contains("SeaweedFS_worker_lance_fragments_removed_total 16"));
+    }
+}
@@ -73,3 +73,131 @@ impl ExecutionSender for StreamSender {
        self.send(Body::JobCompleted(completed))
    }
 }
+
+/// Wraps a sender so what passes through it is counted. The handlers report
+/// their results to admin and nowhere else, so this is where a scrape can learn
+/// what happened without every handler having to know about metrics.
+pub struct MeteredSender<'a> {
+    inner: &'a StreamSender,
+    proposals: std::sync::atomic::AtomicUsize,
+    failed: std::sync::atomic::AtomicBool,
+}
+
+impl<'a> MeteredSender<'a> {
+    pub fn new(inner: &'a StreamSender) -> Self {
+        Self {
+            inner,
+            proposals: std::sync::atomic::AtomicUsize::new(0),
+            failed: std::sync::atomic::AtomicBool::new(false),
+        }
+    }
+
+    /// How many proposals went out, for the detection counter.
+    pub fn proposals(&self) -> usize {
+        self.proposals.load(std::sync::atomic::Ordering::Relaxed)
+    }
+
+    /// Whether the handler reported a failure of its own. A handler that fails
+    /// by returning an error is counted by the caller; this catches the one that
+    /// reports failure and returns Ok.
+    pub fn reported_failure(&self) -> bool {
+        self.failed.load(std::sync::atomic::Ordering::Relaxed)
+    }
+}
+
+impl DetectionSender for MeteredSender<'_> {
+    fn send_proposals(&self, proposals: DetectionProposals) -> Result<()> {
+        // Counted after the send, not before: a stream that closed mid-sweep
+        // would otherwise leave proposals_total claiming work admin never saw.
+        let count = proposals.proposals.len();
+        self.inner.send_proposals(proposals)?;
+        self.proposals
+            .fetch_add(count, std::sync::atomic::Ordering::Relaxed);
+        Ok(())
+    }
+
+    fn send_complete(&self, complete: DetectionComplete) -> Result<()> {
+        if !complete.success {
+            self.failed
+                .store(true, std::sync::atomic::Ordering::Relaxed);
+        }
+        self.inner.send_complete(complete)
+    }
+
+    fn send_activity(&self, activity: ActivityEvent) -> Result<()> {
+        self.inner.send_activity(activity)
+    }
+
+    fn send_observations(&self, observations: WorkerObservations) -> Result<()> {
+        self.inner.send_observations(observations)
+    }
+}
+
+impl ExecutionSender for MeteredSender<'_> {
+    fn send_progress(&self, progress: JobProgressUpdate) -> Result<()> {
+        self.inner.send_progress(progress)
+    }
+
+    fn send_completed(&self, completed: JobCompleted) -> Result<()> {
+        if !completed.success {
+            self.failed
+                .store(true, std::sync::atomic::Ordering::Relaxed);
+        }
+        self.inner.send_completed(completed)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::pb::JobProposal;
+
+    fn proposals(n: usize) -> DetectionProposals {
+        DetectionProposals {
+            proposals: (0..n).map(|_| JobProposal::default()).collect(),
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn proposals_are_counted_once_they_are_sent() {
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let stream = StreamSender::new("worker-1".to_string(), tx);
+        let metered = MeteredSender::new(&stream);
+
+        metered.send_proposals(proposals(3)).expect("send");
+        assert_eq!(metered.proposals(), 3);
+        assert!(rx.try_recv().is_ok(), "the proposals reached the stream");
+    }
+
+    // Admin never saw these, so counting them would report work that was not
+    // handed over.
+    #[test]
+    fn proposals_are_not_counted_when_the_stream_is_gone() {
+        let (tx, rx) = mpsc::unbounded_channel();
+        let stream = StreamSender::new("worker-1".to_string(), tx);
+        let metered = MeteredSender::new(&stream);
+        drop(rx);
+
+        assert!(metered.send_proposals(proposals(3)).is_err());
+        assert_eq!(metered.proposals(), 0);
+    }
+
+    // A handler can report failure and still return Ok; the outcome has to come
+    // from what it said, not only from what it returned.
+    #[test]
+    fn a_reported_failure_is_remembered() {
+        let (tx, _rx) = mpsc::unbounded_channel();
+        let stream = StreamSender::new("worker-1".to_string(), tx);
+        let metered = MeteredSender::new(&stream);
+        assert!(!metered.reported_failure());
+
+        metered
+            .send_complete(DetectionComplete {
+                success: false,
+                ..Default::default()
+            })
+            .expect("send");
+        assert!(metered.reported_failure());
+    }
+}
@@ -1,5 +1,5 @@
 use std::sync::Arc;
-use std::time::Duration;
+use std::time::{Duration, Instant};

 use anyhow::{anyhow, Context, Result};
 use tokio::sync::{mpsc, Semaphore};
@@ -8,6 +8,7 @@ use tonic::transport::{Certificate, Channel, ClientTlsConfig, Identity};
 use tracing::{info, warn};

 use crate::config::WorkerOptions;
+use crate::metrics::Metrics;
 use crate::pb::{
    admin_to_worker_message::Body as AdminBody,
    plugin_control_service_client::PluginControlServiceClient,
@@ -16,7 +17,7 @@ use crate::pb::{
    RunningWork, WorkerHeartbeat, WorkerHello,
 };
 use crate::registry::Registry;
-use crate::senders::StreamSender;
+use crate::senders::{MeteredSender, StreamSender};

 /// The protocol version this worker speaks, sent in WorkerHello.
 const PROTOCOL_VERSION: &str = "1";
@@ -25,6 +26,17 @@ const PROTOCOL_VERSION: &str = "1";
 /// reconnecting on failure. The stream is the only channel: everything admin
 /// asks for and everything the worker reports flows through it.
 pub async fn run(options: WorkerOptions, registry: Registry) -> Result<()> {
+    let metrics = Metrics::new(&options.worker_id, &options.worker_version)?;
+    run_with_metrics(options, registry, metrics).await
+}
+
+/// Runs with metrics a caller has already made, so it can serve them and add
+/// collectors of its own before the stream starts.
+pub async fn run_with_metrics(
+    options: WorkerOptions,
+    registry: Registry,
+    metrics: Metrics,
+) -> Result<()> {
    if registry.is_empty() {
        return Err(anyhow!("no job handlers registered"));
    }
@@ -36,20 +48,31 @@ pub async fn run(options: WorkerOptions, registry: Registry) -> Result<()> {
        ));
    }
    let slots = Slots::new(&options);
+    metrics.set_slots("detection", 0, slots.detection_total as i64);
+    metrics.set_slots("execution", 0, slots.execution_total as i64);
    loop {
-        match serve_once(&options, &registry, &slots).await {
-            Err(err) => warn!("worker stream ended: {err:#}"),
+        match serve_once(&options, &registry, &slots, &metrics).await {
+            Err(err) => {
+                metrics.stream_ended("failed");
+                warn!("worker stream ended: {err:#}")
+            }
            // Admin asked this worker to stop, so stop. Reconnecting here would
            // make shutdown impossible: the worker would log back in.
-            Ok(Outcome::ShutdownRequested) => return Ok(()),
+            Ok(Outcome::ShutdownRequested) => {
+                metrics.stream_ended("shutdown");
+                return Ok(());
+            }
            // Admin closing a healthy stream is not an error, but reconnecting
            // in silence hides the reason - two workers sharing an id evict
            // each other and produce nothing but a login every few seconds.
-            Ok(Outcome::StreamClosed) => warn!(
-                "admin closed the stream; reconnecting in {:?}. If this repeats, check for \
-                 another worker using the id {}",
-                options.reconnect_delay, options.worker_id
-            ),
+            Ok(Outcome::StreamClosed) => {
+                metrics.stream_ended("closed");
+                warn!(
+                    "admin closed the stream; reconnecting in {:?}. If this repeats, check for \
+                     another worker using the id {}",
+                    options.reconnect_delay, options.worker_id
+                )
+            }
        }
        tokio::time::sleep(options.reconnect_delay).await;
    }
@@ -131,6 +154,7 @@ async fn serve_once(
    options: &WorkerOptions,
    registry: &Registry,
    slots: &Slots,
+    metrics: &Metrics,
 ) -> Result<Outcome> {
    // Operators give the admin's HTTP address; the gRPC port is derived, the
    // same way the Go worker does it.
@@ -157,7 +181,12 @@ async fn serve_once(
        .await?
        .into_inner();

-    let heartbeat = spawn_heartbeat(sender.clone(), options.clone(), slots.clone());
+    let heartbeat = spawn_heartbeat(
+        sender.clone(),
+        options.clone(),
+        slots.clone(),
+        metrics.clone(),
+    );

    while let Some(message) = inbound.message().await? {
        let request_id = message.request_id.clone();
@@ -166,6 +195,7 @@ async fn serve_once(
                if !hello.accepted {
                    return Err(anyhow!("admin rejected this worker: {}", hello.message));
                }
+                metrics.stream_connected();
                info!(
                    "connected to admin at {} ({})",
                    options.admin_address, grpc_address
@@ -194,15 +224,28 @@ async fn serve_once(
                spawn_preview(
                    registry.clone(),
                    sender.clone(),
+                    metrics.clone(),
                    request_id.clone(),
                    request,
                );
            }
            Some(AdminBody::RunDetectionRequest(request)) => {
-                spawn_detection(registry.clone(), sender.clone(), slots.clone(), request);
+                spawn_detection(
+                    registry.clone(),
+                    sender.clone(),
+                    slots.clone(),
+                    metrics.clone(),
+                    request,
+                );
            }
            Some(AdminBody::ExecuteJobRequest(request)) => {
-                spawn_execution(registry.clone(), sender.clone(), slots.clone(), request);
+                spawn_execution(
+                    registry.clone(),
+                    sender.clone(),
+                    slots.clone(),
+                    metrics.clone(),
+                    request,
+                );
            }
            Some(AdminBody::CancelRequest(request)) => {
                // Cancellation needs a per-request handle to be honoured; until
@@ -230,6 +273,7 @@ async fn serve_once(
 fn spawn_preview(
    registry: Registry,
    sender: StreamSender,
+    metrics: Metrics,
    request_id: String,
    request: RequestObjectPreview,
 ) {
@@ -263,6 +307,7 @@ fn spawn_preview(
                },
            },
        };
+        metrics.preview_finished(if response.success { "ok" } else { "failed" });
        let _ = sender.send(WorkerBody::ObjectPreviewResponse(response));
    });
 }
@@ -271,6 +316,7 @@ fn spawn_heartbeat(
    sender: StreamSender,
    options: WorkerOptions,
    slots: Slots,
+    metrics: Metrics,
 ) -> tokio::task::JoinHandle<()> {
    // The handle has to be the heartbeat's own, or aborting it aborts nothing
    // and every reconnect leaves another ticker running.
@@ -278,6 +324,18 @@ fn spawn_heartbeat(
        let mut ticker = tokio::time::interval(options.heartbeat_interval);
        loop {
            ticker.tick().await;
+            // The heartbeat already computes this for admin; publish the same
+            // numbers so a scrape and the admin UI cannot disagree.
+            metrics.set_slots(
+                "detection",
+                slots.detection_used() as i64,
+                slots.detection_total as i64,
+            );
+            metrics.set_slots(
+                "execution",
+                slots.execution_used() as i64,
+                slots.execution_total as i64,
+            );
            let beat = WorkerHeartbeat {
                worker_id: options.worker_id.clone(),
                running_work: Vec::<RunningWork>::new(),
@@ -299,6 +357,7 @@ fn spawn_detection(
    registry: Registry,
    sender: StreamSender,
    slots: Slots,
+    metrics: Metrics,
    request: RunDetectionRequest,
 ) {
    tokio::spawn(async move {
@@ -308,7 +367,10 @@ fn spawn_detection(
        // Held until the sweep finishes, so the worker keeps to the capacity it
        // advertised and the heartbeat reports the truth while it works.
        let _permit = slots.detection.acquire().await;
-        if let Err(err) = handler.detect(&request, &sender).await {
+        let metered = MeteredSender::new(&sender);
+        let started = Instant::now();
+        let outcome = handler.detect(&request, &metered).await;
+        let result = if let Err(err) = &outcome {
            warn!("detection for {} failed: {err:#}", request.job_type);
            let _ = sender.send(WorkerBody::DetectionComplete(
                crate::pb::DetectionComplete {
@@ -319,7 +381,18 @@ fn spawn_detection(
                    total_proposals: 0,
                },
            ));
-        }
+            "failed"
+        } else if metered.reported_failure() {
+            "failed"
+        } else {
+            "ok"
+        };
+        metrics.detection_finished(
+            &request.job_type,
+            result,
+            started.elapsed().as_secs_f64(),
+            metered.proposals(),
+        );
    });
 }

@@ -327,6 +400,7 @@ fn spawn_execution(
    registry: Registry,
    sender: StreamSender,
    slots: Slots,
+    metrics: Metrics,
    request: ExecuteJobRequest,
 ) {
    tokio::spawn(async move {
@@ -344,17 +418,26 @@ fn spawn_execution(
        let Some(handler) = registry.get(&job_type) else {
            return;
        };
-        if let Err(err) = handler.execute(&request, &sender).await {
+        let metered = MeteredSender::new(&sender);
+        let started = Instant::now();
+        let outcome = handler.execute(&request, &metered).await;
+        let result = if let Err(err) = &outcome {
            warn!("job {job_id} failed: {err:#}");
            let _ = sender.send(WorkerBody::JobCompleted(JobCompleted {
                request_id: request.request_id.clone(),
                job_id,
-                job_type,
+                job_type: job_type.clone(),
                success: false,
                error_message: format!("{err:#}"),
                ..Default::default()
            }));
-        }
+            "failed"
+        } else if metered.reported_failure() {
+            "failed"
+        } else {
+            "ok"
+        };
+        metrics.job_finished(&job_type, result, started.elapsed().as_secs_f64());
    });
 }

@@ -13,6 +13,7 @@ path = "src/main.rs"

 [dependencies]
 seaweed-worker-core = { path = "../core" }
+prometheus.workspace = true
 # Only the S3 backend: the other object stores lance enables by default are
 # build time this worker never spends.
 lance = { version = "10", default-features = false, features = ["aws"] }
@@ -32,6 +32,7 @@ const MAX_MIN_VERSIONS: i64 = 1000;
 pub struct CleanupVersionsHandler {
    namespace_url: String,
    fallback: dataset::FallbackOptions,
+    metrics: Option<crate::metrics::LanceMetrics>,
 }

 impl CleanupVersionsHandler {
@@ -39,9 +40,15 @@ impl CleanupVersionsHandler {
        Self {
            namespace_url,
            fallback: dataset::FallbackOptions::new(),
+            metrics: None,
        }
    }

+    pub fn with_metrics(mut self, metrics: Option<crate::metrics::LanceMetrics>) -> Self {
+        self.metrics = metrics;
+        self
+    }
+
    pub fn with_fallback(mut self, fallback: dataset::FallbackOptions) -> Self {
        self.fallback = fallback;
        self
@@ -147,13 +154,22 @@ impl JobHandler for CleanupVersionsHandler {
            let table = match dataset::open(&client, &id, &self.fallback).await {
                Ok(table) => table,
                Err(err) => {
+                    if let Some(counters) = &self.metrics {
+                        counters.worker.object_skipped(JOB_TYPE, "open");
+                    }
                    warn!("skipping {encoded}: {err:#}");
                    continue;
                }
            };
+            if let Some(counters) = &self.metrics {
+                counters.worker.object_seen(JOB_TYPE);
+            }
            let stats = match table.stats().await {
                Ok(stats) => stats,
                Err(err) => {
+                    if let Some(counters) = &self.metrics {
+                        counters.worker.object_skipped(JOB_TYPE, "stats");
+                    }
                    warn!("skipping {encoded}: reading its stats failed: {err:#}");
                    continue;
                }
@@ -281,6 +297,11 @@ impl JobHandler for CleanupVersionsHandler {
            .await
            .with_context(|| format!("clean up versions of {}", table.location))?;

+        if let Some(counters) = &self.metrics {
+            counters.versions_removed.inc_by(stats.old_versions as u64);
+            counters.bytes_reclaimed.inc_by(stats.bytes_removed as u64);
+        }
+
        let mut output: HashMap<String, ConfigValue> = HashMap::new();
        output.insert(
            "old_versions_removed".to_string(),
@@ -35,6 +35,7 @@ const MIN_FRAGMENTS_CEILING: i64 = 4096;
 pub struct CompactHandler {
    namespace_url: String,
    fallback: dataset::FallbackOptions,
+    metrics: Option<crate::metrics::LanceMetrics>,
 }

 impl CompactHandler {
@@ -42,9 +43,15 @@ impl CompactHandler {
        Self {
            namespace_url,
            fallback: dataset::FallbackOptions::new(),
+            metrics: None,
        }
    }

+    pub fn with_metrics(mut self, metrics: Option<crate::metrics::LanceMetrics>) -> Self {
+        self.metrics = metrics;
+        self
+    }
+
    /// Storage options to use where the namespace vends none.
    pub fn with_fallback(mut self, fallback: dataset::FallbackOptions) -> Self {
        self.fallback = fallback;
@@ -143,6 +150,9 @@ impl JobHandler for CompactHandler {
                Err(err) => {
                    // A table that cannot be opened is the next run's problem,
                    // not a reason to abandon the whole sweep.
+                    if let Some(counters) = &self.metrics {
+                        counters.worker.object_skipped(JOB_TYPE, "open");
+                    }
                    warn!("skipping {encoded}: {err:#}");
                    continue;
                }
@@ -150,9 +160,15 @@ impl JobHandler for CompactHandler {
            // One unreadable table must not end the sweep: the tables already
            // read would lose their proposals, and admin would get no
            // completion for this request at all.
+            if let Some(counters) = &self.metrics {
+                counters.worker.object_seen(JOB_TYPE);
+            }
            let stats = match table.stats().await {
                Ok(stats) => stats,
                Err(err) => {
+                    if let Some(counters) = &self.metrics {
+                        counters.worker.object_skipped(JOB_TYPE, "stats");
+                    }
                    warn!("skipping {encoded}: reading its stats failed: {err:#}");
                    continue;
                }
@@ -264,6 +280,12 @@ impl JobHandler for CompactHandler {
            .await
            .with_context(|| format!("compact {}", table.location))?;

+        if let Some(counters) = &self.metrics {
+            counters
+                .fragments_removed
+                .inc_by(metrics.fragments_removed as u64);
+        }
+
        let after = table.stats().await?;
        let mut output: HashMap<String, ConfigValue> = HashMap::new();
        output.insert(
@@ -35,6 +35,7 @@ const MAX_UNINDEXED_CEILING: i64 = 100_000_000;
 pub struct OptimizeIndicesHandler {
    namespace_url: String,
    fallback: dataset::FallbackOptions,
+    metrics: Option<crate::metrics::LanceMetrics>,
 }

 impl OptimizeIndicesHandler {
@@ -42,9 +43,15 @@ impl OptimizeIndicesHandler {
        Self {
            namespace_url,
            fallback: dataset::FallbackOptions::new(),
+            metrics: None,
        }
    }

+    pub fn with_metrics(mut self, metrics: Option<crate::metrics::LanceMetrics>) -> Self {
+        self.metrics = metrics;
+        self
+    }
+
    pub fn with_fallback(mut self, fallback: dataset::FallbackOptions) -> Self {
        self.fallback = fallback;
        self
@@ -149,14 +156,23 @@ impl JobHandler for OptimizeIndicesHandler {
            let table = match dataset::open(&client, &id, &self.fallback).await {
                Ok(table) => table,
                Err(err) => {
+                    if let Some(counters) = &self.metrics {
+                        counters.worker.object_skipped(JOB_TYPE, "open");
+                    }
                    warn!("skipping {encoded}: {err:#}");
                    continue;
                }
            };
+            if let Some(counters) = &self.metrics {
+                counters.worker.object_seen(JOB_TYPE);
+            }
            let unindexed = match unindexed_rows(&table).await {
                Ok(Some(unindexed)) => unindexed,
                Ok(None) => continue,
                Err(err) => {
+                    if let Some(counters) = &self.metrics {
+                        counters.worker.object_skipped(JOB_TYPE, "index_stats");
+                    }
                    warn!("skipping {encoded}: reading its index stats failed: {err:#}");
                    continue;
                }
@@ -232,6 +248,9 @@ impl JobHandler for OptimizeIndicesHandler {
            .with_context(|| format!("optimize indices of {}", table.location))?;

        let after = unindexed_rows(&table).await?.unwrap_or(0);
+        if let Some(counters) = &self.metrics {
+            counters.rows_indexed.inc_by(before.saturating_sub(after));
+        }
        let mut output: HashMap<String, ConfigValue> = HashMap::new();
        output.insert(
            "unindexed_rows_before".to_string(),
@@ -41,16 +41,24 @@ pub(crate) fn table_id(parameters: &HashMap<String, ConfigValue>) -> Option<Vec<
 pub fn handlers(
    namespace_url: String,
    fallback: crate::dataset::FallbackOptions,
+    metrics: Option<crate::metrics::LanceMetrics>,
 ) -> Vec<Arc<dyn JobHandler>> {
    vec![
        Arc::new(
-            compact::CompactHandler::new(namespace_url.clone()).with_fallback(fallback.clone()),
+            compact::CompactHandler::new(namespace_url.clone())
+                .with_fallback(fallback.clone())
+                .with_metrics(metrics.clone()),
        ),
        Arc::new(
            indices::OptimizeIndicesHandler::new(namespace_url.clone())
-                .with_fallback(fallback.clone()),
+                .with_fallback(fallback.clone())
+                .with_metrics(metrics.clone()),
+        ),
+        Arc::new(
+            cleanup::CleanupVersionsHandler::new(namespace_url)
+                .with_fallback(fallback)
+                .with_metrics(metrics),
        ),
-        Arc::new(cleanup::CleanupVersionsHandler::new(namespace_url).with_fallback(fallback)),
    ]
 }

@@ -8,6 +8,7 @@
 pub mod catalog;
 pub mod dataset;
 pub mod jobs;
+pub mod metrics;
 pub mod preview;

 pub use jobs::handlers;
@@ -1,11 +1,13 @@
 use std::time::Duration;

-use anyhow::Result;
+use anyhow::{Context, Result};
 use clap::Parser;
+use std::net::{IpAddr, SocketAddr};
 use std::sync::Arc;

-use seaweed_worker_core::{Registry, TlsOptions, WorkerOptions};
+use seaweed_worker_core::{Metrics, Registry, TlsOptions, WorkerOptions};
 use weed_lance_worker::handlers;
+use weed_lance_worker::metrics::LanceMetrics;

 /// Mirrors `weed worker`'s flags, because this is the same contract from another
 /// language and an operator should not have to learn a second set of names.
@@ -61,6 +63,16 @@ struct Args {
    /// this worker dials.
    #[arg(long)]
    tls_server_name: Option<String>,
+
+    /// Serve /health, /ready and /metrics on this port, the way
+    /// `weed worker -metricsPort` does. Zero, the default, serves nothing.
+    #[arg(long, default_value = "0", env = "WEED_METRICS_PORT")]
+    metrics_port: u16,
+
+    /// Address the metrics server binds. Loopback by default, since the
+    /// endpoint is unauthenticated.
+    #[arg(long, default_value = "127.0.0.1", env = "WEED_METRICS_IP")]
+    metrics_ip: String,
 }

 impl Args {
@@ -89,6 +101,19 @@ impl Args {
    }
 }

+/// Builds the metrics bind address. Parsing the host separately is what makes an
+/// IPv6 literal work: "::1" and 9327 joined with a colon is not an address, and
+/// formatting them that way turns `--metrics-ip ::1` into a startup failure.
+/// Brackets are accepted too, since that is how the same address is written in a
+/// URL and an operator will reasonably try it.
+fn metrics_address(ip: &str, port: u16) -> Result<SocketAddr> {
+    let host = ip.trim().trim_start_matches('[').trim_end_matches(']');
+    let parsed: IpAddr = host
+        .parse()
+        .with_context(|| format!("parse the metrics address {ip}"))?;
+    Ok(SocketAddr::new(parsed, port))
+}
+
 #[tokio::main]
 async fn main() -> Result<()> {
    tracing_subscriber::fmt()
@@ -116,12 +141,54 @@ async fn main() -> Result<()> {
        fallback.insert("aws_secret_access_key".to_string(), secret);
    }

+    let metrics = Metrics::new(&options.worker_id, &options.worker_version)?;
+    let lance_metrics = LanceMetrics::new(&metrics)?;
+    if args.metrics_port > 0 {
+        seaweed_worker_core::metrics::spawn(
+            metrics.clone(),
+            metrics_address(&args.metrics_ip, args.metrics_port)?,
+        );
+    }
+
    let mut registry = Registry::new().with_preview(Arc::new(
        weed_lance_worker::preview::LancePreview::new(args.namespace.clone(), fallback.clone()),
    ));
-    for handler in handlers(args.namespace, fallback) {
+    for handler in handlers(args.namespace, fallback, Some(lance_metrics)) {
        registry = registry.register(handler);
    }

-    seaweed_worker_core::run(options, registry).await
+    seaweed_worker_core::stream::run_with_metrics(options, registry, metrics).await
+}
+
+#[cfg(test)]
+mod tests {
+    use super::metrics_address;
+
+    #[test]
+    fn metrics_address_takes_ipv4_ipv6_and_brackets() {
+        assert_eq!(
+            metrics_address("127.0.0.1", 9327).unwrap().to_string(),
+            "127.0.0.1:9327"
+        );
+        // Joining these with a colon gives "::1:9327", which is not an address.
+        assert_eq!(
+            metrics_address("::1", 9327).unwrap().to_string(),
+            "[::1]:9327"
+        );
+        assert_eq!(
+            metrics_address("[::1]", 9327).unwrap().to_string(),
+            "[::1]:9327"
+        );
+        assert_eq!(
+            metrics_address("0.0.0.0", 9327).unwrap().to_string(),
+            "0.0.0.0:9327"
+        );
+    }
+
+    #[test]
+    fn metrics_address_rejects_a_hostname() {
+        // Binding takes an address, not a name; saying so beats a confusing
+        // failure inside the server.
+        assert!(metrics_address("localhost", 9327).is_err());
+    }
 }
@@ -0,0 +1,45 @@
+//! What the Lance jobs reclaimed, in numbers a scrape can add up.
+//!
+//! The generic worker metrics say a job ran and how long it took. These say what
+//! it did: a compaction sweep that runs every minute and removes nothing is a
+//! different thing from one that never runs, and only these tell them apart.
+
+use anyhow::Result;
+use prometheus::IntCounter;
+use seaweed_worker_core::Metrics;
+
+/// Counters the Lance jobs raise, registered on the worker's own registry so
+/// they are served from the same endpoint.
+#[derive(Clone)]
+pub struct LanceMetrics {
+    /// The worker's own metrics, so a job can also record what it read and what
+    /// it had to skip - the pair that tells "nothing needed doing" apart from
+    /// "nothing could be read".
+    pub worker: Metrics,
+    pub fragments_removed: IntCounter,
+    pub rows_indexed: IntCounter,
+    pub versions_removed: IntCounter,
+    pub bytes_reclaimed: IntCounter,
+}
+
+impl LanceMetrics {
+    pub fn new(metrics: &Metrics) -> Result<Self> {
+        Ok(Self {
+            worker: metrics.clone(),
+            fragments_removed: metrics.counter(
+                "lance_fragments_removed_total",
+                "Fragments merged away by compaction.",
+            )?,
+            rows_indexed: metrics.counter(
+                "lance_rows_indexed_total",
+                "Rows brought under an index that did not cover them.",
+            )?,
+            versions_removed: metrics.counter(
+                "lance_versions_removed_total",
+                "Dataset versions removed by cleanup.",
+            )?,
+            bytes_reclaimed: metrics
+                .counter("lance_bytes_reclaimed_total", "Bytes freed by cleanup.")?,
+        })
+    }
+}