From dcabfcd0c302cbcd7bbde08ef5b0f2027b649514 Mon Sep 17 00:00:00 2001 From: Efremenko Arhip Date: Sun, 26 Jul 2026 22:16:35 +0300 Subject: [PATCH] =?UTF-8?q?feat(coordinator):=20business=20metrics=20?= =?UTF-8?q?=E2=80=94=20jobs/tasks/workers=20by=20status?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 12 +++- coordinator/internal/metrics/business.go | 66 +++++++++++++++++++ coordinator/internal/metrics/business_test.go | 51 ++++++++++++++ .../internal/storage/postgres/stats_repo.go | 65 ++++++++++++++++++ coordinator/internal/transport/http/server.go | 7 +- .../internal/transport/http/server_test.go | 2 +- .../grafana/dashboards/coordinator.json | 66 +++++++++++++++++++ 7 files changed, 265 insertions(+), 4 deletions(-) create mode 100644 coordinator/internal/metrics/business.go create mode 100644 coordinator/internal/metrics/business_test.go create mode 100644 coordinator/internal/storage/postgres/stats_repo.go diff --git a/coordinator/cmd/coordinator/main.go b/coordinator/cmd/coordinator/main.go index a2275f8..f9f26f6 100644 --- a/coordinator/cmd/coordinator/main.go +++ b/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 diff --git a/coordinator/internal/metrics/business.go b/coordinator/internal/metrics/business.go new file mode 100644 index 0000000..1a87348 --- /dev/null +++ b/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) + } +} diff --git a/coordinator/internal/metrics/business_test.go b/coordinator/internal/metrics/business_test.go new file mode 100644 index 0000000..d127ce8 --- /dev/null +++ b/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") + } +} diff --git a/coordinator/internal/storage/postgres/stats_repo.go b/coordinator/internal/storage/postgres/stats_repo.go new file mode 100644 index 0000000..9be7e87 --- /dev/null +++ b/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() +} diff --git a/coordinator/internal/transport/http/server.go b/coordinator/internal/transport/http/server.go index 79d080a..6439da1 100644 --- a/coordinator/internal/transport/http/server.go +++ b/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, } } diff --git a/coordinator/internal/transport/http/server_test.go b/coordinator/internal/transport/http/server_test.go index 220083c..96dc777 100644 --- a/coordinator/internal/transport/http/server_test.go +++ b/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()} diff --git a/coordinator/monitoring/grafana/dashboards/coordinator.json b/coordinator/monitoring/grafana/dashboards/coordinator.json index 65cd5ff..e52c98e 100644 --- a/coordinator/monitoring/grafana/dashboards/coordinator.json +++ b/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}}" + } + ] } ] }