feat(coordinator): business metrics — jobs/tasks/workers by status

A scrape-time collector reports scimesh_tasks/jobs/workers gauges keyed by
status, sourced from cheap GROUP BY queries (StatsRepo), zero-filled across all
known statuses so the dashboard shows flat zeros instead of gaps. A failed
query yields no samples for that scrape rather than crashing it.

Metrics is now built in main so the DB-backed collector can be registered
(NewServer takes *metrics.Metrics; nil self-provisions for tests). Grafana
dashboard gains a Domain state row: tasks/jobs/workers by status and a queue-
depth stat.

coordinator/cmd/coordinator/main.go

@@ -9,6 +9,7 @@ import (
 	"syscall"
 
 	"github.com/emil28092005/SciMesh/coordinator/internal/infra"
+	"github.com/emil28092005/SciMesh/coordinator/internal/metrics"
 	"github.com/emil28092005/SciMesh/coordinator/internal/storage/blob"
 	"github.com/emil28092005/SciMesh/coordinator/internal/storage/postgres"
 	httptransport "github.com/emil28092005/SciMesh/coordinator/internal/transport/http"
@@ -108,9 +109,18 @@ func run() error {
 		}(r.name, r.fn)
 	}
 
+	// Business metrics: gauges of tasks/jobs/workers by status, sampled from the
+	// database on every Prometheus scrape.
+	statsRepo := postgres.NewStatsRepo(pool)
+	m := metrics.New()
+	m.RegisterBusiness(func(ctx context.Context) (metrics.Stats, error) {
+		tasks, jobs, workers, err := statsRepo.Counts(ctx)
+		return metrics.Stats{Tasks: tasks, Jobs: jobs, Workers: workers}, err
+	})
+
 	// pool.Ping backs /health: readiness means the database answers, not just
 	// that the process is alive.
-	api := httptransport.NewServer(useCases, log, cfg.RequestTimeout, cfg.HeartbeatInterval, cfg.MaxUploadBytes, cfg.JWTSecret, cfg.UserserviceURL, pool.Ping)
+	api := httptransport.NewServer(useCases, log, cfg.RequestTimeout, cfg.HeartbeatInterval, cfg.MaxUploadBytes, cfg.JWTSecret, cfg.UserserviceURL, m, pool.Ping)
 	err = infra.RunServer(ctx, log, cfg.Addr, api.Handler(cfg.Token, cfg.UIToken))
 
 	// Shutdown order matters, and defers alone cannot express it (they run

coordinator/internal/metrics/business.go

@@ -0,0 +1,66 @@
+package metrics
+
+import (
+	"context"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// Stats is a point-in-time snapshot of the coordinator's domain state: counts of
+// tasks, jobs, and workers keyed by their status. Maps are expected to be
+// zero-filled by the provider so every known status is always present, giving
+// the dashboard flat zero lines instead of gaps.
+type Stats struct {
+	Tasks   map[string]int
+	Jobs    map[string]int
+	Workers map[string]int
+}
+
+// StatsFunc returns the current snapshot. It is called on every scrape, so it
+// must be a cheap aggregate query.
+type StatsFunc func(context.Context) (Stats, error)
+
+// RegisterBusiness registers a collector that reports domain-state gauges
+// (scimesh_tasks/jobs/workers by status) sourced from collect on each scrape.
+// Deriving the gauges at scrape time keeps them fresh without a background
+// goroutine, and a failed query simply yields no samples for that scrape.
+func (m *Metrics) RegisterBusiness(collect StatsFunc) {
+	m.reg.MustRegister(&businessCollector{
+		collect: collect,
+		tasks:   prometheus.NewDesc("scimesh_tasks", "Tasks by status.", []string{"status"}, nil),
+		jobs:    prometheus.NewDesc("scimesh_jobs", "Jobs by status.", []string{"status"}, nil),
+		workers: prometheus.NewDesc("scimesh_workers", "Workers by status.", []string{"status"}, nil),
+	})
+}
+
+type businessCollector struct {
+	collect              StatsFunc
+	tasks, jobs, workers *prometheus.Desc
+}
+
+func (c *businessCollector) Describe(ch chan<- *prometheus.Desc) {
+	ch <- c.tasks
+	ch <- c.jobs
+	ch <- c.workers
+}
+
+func (c *businessCollector) Collect(ch chan<- prometheus.Metric) {
+	// A bounded query so one slow scrape cannot stall Prometheus.
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+
+	s, err := c.collect(ctx)
+	if err != nil {
+		return // no samples this scrape; Prometheus keeps the last value
+	}
+	emit(ch, c.tasks, s.Tasks)
+	emit(ch, c.jobs, s.Jobs)
+	emit(ch, c.workers, s.Workers)
+}
+
+func emit(ch chan<- prometheus.Metric, desc *prometheus.Desc, counts map[string]int) {
+	for status, n := range counts {
+		ch <- prometheus.MustNewConstMetric(desc, prometheus.GaugeValue, float64(n), status)
+	}
+}

coordinator/internal/metrics/business_test.go

@@ -0,0 +1,51 @@
+package metrics
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+func scrape(t *testing.T, m *Metrics) string {
+	t.Helper()
+	rec := httptest.NewRecorder()
+	req, _ := http.NewRequestWithContext(context.Background(), http.MethodGet, "/metrics", nil)
+	m.Handler().ServeHTTP(rec, req)
+	return rec.Body.String()
+}
+
+func TestBusinessCollectorEmitsGauges(t *testing.T) {
+	m := New()
+	m.RegisterBusiness(func(context.Context) (Stats, error) {
+		return Stats{
+			Tasks:   map[string]int{"pending": 3, "running": 1, "completed": 0},
+			Jobs:    map[string]int{"running": 2},
+			Workers: map[string]int{"online": 4},
+		}, nil
+	})
+
+	body := scrape(t, m)
+	for _, want := range []string{
+		`scimesh_tasks{status="pending"} 3`,
+		`scimesh_tasks{status="completed"} 0`,
+		`scimesh_jobs{status="running"} 2`,
+		`scimesh_workers{status="online"} 4`,
+	} {
+		if !strings.Contains(body, want) {
+			t.Errorf("metrics missing %q\n%s", want, body)
+		}
+	}
+}
+
+func TestBusinessCollectorSkipsOnError(t *testing.T) {
+	m := New()
+	m.RegisterBusiness(func(context.Context) (Stats, error) {
+		return Stats{}, errors.New("db down")
+	})
+	if strings.Contains(scrape(t, m), "scimesh_tasks") {
+		t.Error("a failed snapshot must emit no business samples")
+	}
+}

coordinator/internal/storage/postgres/stats_repo.go

@@ -0,0 +1,65 @@
+package postgres
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/emil28092005/SciMesh/coordinator/internal/domain"
+)
+
+// Known statuses per entity, so counts are zero-filled and every status is
+// always present in the metrics (a flat 0 line beats a gap on the dashboard).
+var (
+	taskStatuses   = []string{string(domain.TaskPending), string(domain.TaskLeased), string(domain.TaskRunning), string(domain.TaskCompleted), string(domain.TaskFailed), string(domain.TaskCancelled)}
+	jobStatuses    = []string{string(domain.JobPending), string(domain.JobRunning), string(domain.JobReducing), string(domain.JobCompleted), string(domain.JobFailed), string(domain.JobCancelled)}
+	workerStatuses = []string{string(domain.WorkerOnline), string(domain.WorkerBusy), string(domain.WorkerOffline)}
+)
+
+// StatsRepo answers the aggregate status counts the business metrics report. It
+// runs one cheap GROUP BY per entity; the collector calls this on every scrape.
+type StatsRepo struct {
+	pool *pgxpool.Pool
+}
+
+func NewStatsRepo(pool *pgxpool.Pool) *StatsRepo {
+	return &StatsRepo{pool: pool}
+}
+
+// Counts returns status->count maps for tasks, jobs, and workers, each
+// zero-filled across its known statuses.
+func (r *StatsRepo) Counts(ctx context.Context) (tasks, jobs, workers map[string]int, err error) {
+	if tasks, err = r.countByStatus(ctx, "tasks", taskStatuses); err != nil {
+		return nil, nil, nil, err
+	}
+	if jobs, err = r.countByStatus(ctx, "jobs", jobStatuses); err != nil {
+		return nil, nil, nil, err
+	}
+	if workers, err = r.countByStatus(ctx, "workers", workerStatuses); err != nil {
+		return nil, nil, nil, err
+	}
+	return tasks, jobs, workers, nil
+}
+
+func (r *StatsRepo) countByStatus(ctx context.Context, table string, known []string) (map[string]int, error) {
+	out := make(map[string]int, len(known))
+	for _, s := range known {
+		out[s] = 0 // zero-fill
+	}
+	// table is a fixed internal constant, never user input — safe to format.
+	rows, err := r.pool.Query(ctx, fmt.Sprintf("SELECT status, count(*) FROM %s GROUP BY status", table))
+	if err != nil {
+		return nil, fmt.Errorf("count %s by status: %w", table, err)
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var status string
+		var n int
+		if err := rows.Scan(&status, &n); err != nil {
+			return nil, err
+		}
+		out[status] = n // an unknown status still shows up, which is a useful signal
+	}
+	return out, rows.Err()
+}

coordinator/internal/transport/http/server.go

@@ -59,7 +59,10 @@ type Server struct {
 }
 
 func NewServer(uc UseCases, log *slog.Logger, requestTimeout, heartbeatInterval time.Duration,
-	maxUploadBytes int64, jwtSecret, userserviceURL string, ready func(context.Context) error) *Server {
+	maxUploadBytes int64, jwtSecret, userserviceURL string, m *metrics.Metrics, ready func(context.Context) error) *Server {
+	if m == nil {
+		m = metrics.New()
+	}
 	return &Server{
 		uc:                uc,
 		log:               log,
@@ -69,7 +72,7 @@ func NewServer(uc UseCases, log *slog.Logger, requestTimeout, heartbeatInterval
 		verifier:          tokenpkg.NewVerifier(jwtSecret),
 		userserviceURL:    strings.TrimRight(userserviceURL, "/"),
 		httpClient:        &http.Client{Timeout: 10 * time.Second},
-		metrics:           metrics.New(),
+		metrics:           m,
 		ready:             ready,
 	}
 }

coordinator/internal/transport/http/server_test.go

@@ -68,7 +68,7 @@ func newEnvWithUIToken(t *testing.T, ready func(context.Context) error, configur
 	if err != nil {
 		t.Fatalf("register test worker: %v", err)
 	}
-	srv := coordhttp.NewServer(uc, slog.New(slog.NewTextHandler(io.Discard, nil)), 5*time.Second, 15*time.Second, 1<<30, "", "", ready)
+	srv := coordhttp.NewServer(uc, slog.New(slog.NewTextHandler(io.Discard, nil)), 5*time.Second, 15*time.Second, 1<<30, "", "", nil, ready)
 	ts := httptest.NewServer(srv.Handler(token, configuredUIToken))
 	t.Cleanup(ts.Close)
 	return &env{ts: ts, blobs: blobs, workerID: worker.ID.String()}

coordinator/monitoring/grafana/dashboards/coordinator.json

@@ -83,6 +83,72 @@
           "legendFormat": "rss"
         }
       ]
+    },
+    {
+      "type": "row",
+      "title": "Domain state",
+      "collapsed": false,
+      "gridPos": { "h": 1, "w": 24, "x": 0, "y": 16 }
+    },
+    {
+      "type": "timeseries",
+      "title": "Tasks by status",
+      "datasource": { "type": "prometheus", "uid": "prometheus" },
+      "gridPos": { "h": 8, "w": 12, "x": 0, "y": 17 },
+      "fieldConfig": { "defaults": { "unit": "short", "custom": { "stacking": { "mode": "normal" }, "fillOpacity": 30 } }, "overrides": [] },
+      "targets": [
+        {
+          "refId": "A",
+          "datasource": { "type": "prometheus", "uid": "prometheus" },
+          "expr": "scimesh_tasks",
+          "legendFormat": "{{status}}"
+        }
+      ]
+    },
+    {
+      "type": "timeseries",
+      "title": "Jobs by status",
+      "datasource": { "type": "prometheus", "uid": "prometheus" },
+      "gridPos": { "h": 8, "w": 12, "x": 12, "y": 17 },
+      "fieldConfig": { "defaults": { "unit": "short", "custom": { "stacking": { "mode": "normal" }, "fillOpacity": 30 } }, "overrides": [] },
+      "targets": [
+        {
+          "refId": "A",
+          "datasource": { "type": "prometheus", "uid": "prometheus" },
+          "expr": "scimesh_jobs",
+          "legendFormat": "{{status}}"
+        }
+      ]
+    },
+    {
+      "type": "stat",
+      "title": "Queue depth (pending tasks)",
+      "datasource": { "type": "prometheus", "uid": "prometheus" },
+      "gridPos": { "h": 8, "w": 6, "x": 0, "y": 25 },
+      "fieldConfig": { "defaults": { "unit": "short", "color": { "mode": "thresholds" }, "thresholds": { "steps": [ { "color": "green", "value": null }, { "color": "yellow", "value": 50 }, { "color": "red", "value": 500 } ] } }, "overrides": [] },
+      "targets": [
+        {
+          "refId": "A",
+          "datasource": { "type": "prometheus", "uid": "prometheus" },
+          "expr": "sum(scimesh_tasks{status=\"pending\"})",
+          "legendFormat": "pending"
+        }
+      ]
+    },
+    {
+      "type": "timeseries",
+      "title": "Workers by status",
+      "datasource": { "type": "prometheus", "uid": "prometheus" },
+      "gridPos": { "h": 8, "w": 18, "x": 6, "y": 25 },
+      "fieldConfig": { "defaults": { "unit": "short", "custom": { "stacking": { "mode": "normal" }, "fillOpacity": 30 } }, "overrides": [] },
+      "targets": [
+        {
+          "refId": "A",
+          "datasource": { "type": "prometheus", "uid": "prometheus" },
+          "expr": "scimesh_workers",
+          "legendFormat": "{{status}}"
+        }
+      ]
     }
   ]
 }