diff --git a/coordinator/cmd/coordinator/main.go b/coordinator/cmd/coordinator/main.go index c1e6a69..965fece 100644 --- a/coordinator/cmd/coordinator/main.go +++ b/coordinator/cmd/coordinator/main.go @@ -76,6 +76,7 @@ type storageDeps struct { workerRepo usecase.WorkerRepository artifactRepo usecase.ArtifactRepository uiReadRepo usecase.UIReadRepository + adminReadRepo usecase.AdminReadRepository taskResultRepo usecase.TaskResultRepository statsRepo interface { Counts(ctx context.Context) (tasks, jobs, workers map[string]int, err error) @@ -172,6 +173,16 @@ func runWithConfig(cfg infra.Config) error { GetTaskInput: usecase.NewGetTaskInput(taskRepo, artifactRepo, blobStore), Dashboard: usecase.NewDashboard(uiReadRepo, catalog), PreviewArtifact: usecase.NewPreviewArtifact(uiReadRepo, blobStore), + Admin: usecase.NewAdmin(deps.adminReadRepo, uiReadRepo, usecase.AdminNodeInfo{ + Version: version, + StartedAt: clk.Now(), + Binary: executablePath(), + Addr: cfg.Addr, + DataDir: cfg.StorageDir, + DBEngine: cfg.DatabaseEngine, + PublicURL: cfg.PublicCoordinatorURL, + Userservice: cfg.UserserviceURL, + }, deps.ready, clk.Now), } // Background reapers are tracked so shutdown can wait for them. Without this @@ -225,6 +236,16 @@ func runWithConfig(cfg infra.Config) error { return err } +// executablePath resolves the running binary for the admin console's node +// information, falling back to the invocation name. +func executablePath() string { + path, err := os.Executable() + if err != nil || path == "" { + return os.Args[0] + } + return path +} + // openSQLite opens the embedded database and builds the sqlite repositories. func openSQLite(ctx context.Context, cfg infra.Config, log *slog.Logger) (*storageDeps, error) { if err := os.MkdirAll(cfg.StorageDir, 0o750); err != nil { @@ -242,6 +263,7 @@ func openSQLite(ctx context.Context, cfg infra.Config, log *slog.Logger) (*stora workerRepo: sqlite.NewWorkerRepo(db), artifactRepo: sqlite.NewArtifactRepo(db), uiReadRepo: sqlite.NewUIReadRepo(db), + adminReadRepo: sqlite.NewAdminReadRepo(db), taskResultRepo: sqlite.NewTaskResultRepo(db), statsRepo: sqlite.NewStatsRepo(db), ready: func(ctx context.Context) error { return db.PingContext(ctx) }, @@ -264,6 +286,7 @@ func openPostgres(ctx context.Context, cfg infra.Config, log *slog.Logger) (*sto workerRepo: postgres.NewWorkerRepo(pool), artifactRepo: postgres.NewArtifactRepo(pool), uiReadRepo: postgres.NewUIReadRepo(pool), + adminReadRepo: postgres.NewAdminReadRepo(pool), taskResultRepo: postgres.NewTaskResultRepo(pool), statsRepo: postgres.NewStatsRepo(pool), ready: func(ctx context.Context) error { return pool.Ping(ctx) }, diff --git a/coordinator/internal/storage/postgres/admin_read_repo.go b/coordinator/internal/storage/postgres/admin_read_repo.go new file mode 100644 index 0000000..7ca32dc --- /dev/null +++ b/coordinator/internal/storage/postgres/admin_read_repo.go @@ -0,0 +1,198 @@ +package postgres + +import ( + "context" + "fmt" + "time" + + sq "github.com/Masterminds/squirrel" + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/emil28092005/SciMesh/coordinator/internal/domain" + "github.com/emil28092005/SciMesh/coordinator/internal/usecase" +) + +// AdminReadRepo backs the coordinator admin console: paginated jobs, status +// counters, metrics buckets and storage figures. Read-only. +type AdminReadRepo struct{ pool *pgxpool.Pool } + +func NewAdminReadRepo(pool *pgxpool.Pool) *AdminReadRepo { return &AdminReadRepo{pool: pool} } + +var _ usecase.AdminReadRepository = (*AdminReadRepo)(nil) + +func (r *AdminReadRepo) ListJobsPaginated(ctx context.Context, status string, limit, offset int) ([]domain.Job, int, error) { + if limit < 1 || limit > 100 || offset < 0 { + return nil, 0, domain.ErrInvalidInput + } + countQ := psql.Select("COUNT(*)").From("jobs") + listQ := psql.Select(jobColumns...).From("jobs") + if status != "" { + countQ = countQ.Where(sq.Eq{"status": status}) + listQ = listQ.Where(sq.Eq{"status": status}) + } + countSQL, args, err := countQ.ToSql() + if err != nil { + return nil, 0, err + } + var total int + if err := conn(ctx, r.pool).QueryRow(ctx, countSQL, args...).Scan(&total); err != nil { + return nil, 0, fmt.Errorf("count jobs: %w", err) + } + listSQL, args, err := listQ.OrderBy("created_at DESC", "id DESC").Limit(uint64(limit)).Offset(uint64(offset)).ToSql() + if err != nil { + return nil, 0, err + } + rows, err := conn(ctx, r.pool).Query(ctx, listSQL, args...) + if err != nil { + return nil, 0, fmt.Errorf("list jobs paginated: %w", err) + } + defer rows.Close() + jobs := make([]domain.Job, 0) + for rows.Next() { + var j domain.Job + var statusRaw string + if err := rows.Scan( + &j.ID, &j.Workload, &j.InputURI, &j.Parameters, &statusRaw, &j.CreatedAt, &j.CompletedAt, + &j.InputArtifactID, &j.ResultArtifactID, &j.ErrorCode, &j.ErrorMessage, &j.ReducerStartedAt, + &j.OwnerID, + ); err != nil { + return nil, 0, err + } + j.Status = domain.JobStatus(statusRaw) + jobs = append(jobs, j) + } + return jobs, total, rows.Err() +} + +func (r *AdminReadRepo) CountJobsByStatus(ctx context.Context) (map[string]int, error) { + rows, err := conn(ctx, r.pool).Query(ctx, "SELECT status, COUNT(*) FROM jobs GROUP BY status") + if err != nil { + return nil, fmt.Errorf("count jobs by status: %w", err) + } + defer rows.Close() + out := make(map[string]int) + for rows.Next() { + var status string + var count int + if err := rows.Scan(&status, &count); err != nil { + return nil, err + } + out[status] = count + } + return out, rows.Err() +} + +func (r *AdminReadRepo) TaskCountsByJobs(ctx context.Context, jobIDs []uuid.UUID) (map[uuid.UUID]map[string]int, error) { + out := make(map[uuid.UUID]map[string]int, len(jobIDs)) + if len(jobIDs) == 0 { + return out, nil + } + sql, args, err := psql.Select("job_id", "status", "COUNT(*)").From("tasks"). + Where(sq.Eq{"job_id": jobIDs}).GroupBy("job_id", "status").ToSql() + if err != nil { + return nil, err + } + rows, err := conn(ctx, r.pool).Query(ctx, sql, args...) + if err != nil { + return nil, fmt.Errorf("task counts by jobs: %w", err) + } + defer rows.Close() + for rows.Next() { + var jobID uuid.UUID + var status string + var count int + if err := rows.Scan(&jobID, &status, &count); err != nil { + return nil, err + } + if out[jobID] == nil { + out[jobID] = make(map[string]int) + } + out[jobID][status] = count + } + return out, rows.Err() +} + +func (r *AdminReadRepo) JobCountsByDay(ctx context.Context, since time.Time) (map[string]int, error) { + rows, err := conn(ctx, r.pool).Query(ctx, + "SELECT to_char(date_trunc('day', created_at AT TIME ZONE 'UTC'), 'YYYY-MM-DD') AS day, COUNT(*) FROM jobs WHERE created_at >= $1 GROUP BY 1", + since.UTC()) + if err != nil { + return nil, fmt.Errorf("job counts by day: %w", err) + } + defer rows.Close() + out := make(map[string]int) + for rows.Next() { + var day string + var count int + if err := rows.Scan(&day, &count); err != nil { + return nil, err + } + out[day] = count + } + return out, rows.Err() +} + +func (r *AdminReadRepo) JobCountsByWorkload(ctx context.Context) (map[string]int, error) { + rows, err := conn(ctx, r.pool).Query(ctx, "SELECT workload, COUNT(*) FROM jobs GROUP BY workload") + if err != nil { + return nil, fmt.Errorf("job counts by workload: %w", err) + } + defer rows.Close() + out := make(map[string]int) + for rows.Next() { + var workload string + var count int + if err := rows.Scan(&workload, &count); err != nil { + return nil, err + } + out[workload] = count + } + return out, rows.Err() +} + +func (r *AdminReadRepo) TaskStats(ctx context.Context) (int64, int64, float64, error) { + var completed, failed int64 + var avgSeconds *float64 + err := conn(ctx, r.pool).QueryRow(ctx, ` + SELECT + COALESCE(SUM(CASE WHEN status = 'completed' THEN 1 ELSE 0 END), 0), + COALESCE(SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END), 0), + AVG(CASE WHEN status = 'completed' AND started_at IS NOT NULL + THEN EXTRACT(EPOCH FROM completed_at - started_at) END)::float8 + FROM tasks`).Scan(&completed, &failed, &avgSeconds) + if err != nil { + return 0, 0, 0, fmt.Errorf("task stats: %w", err) + } + var avg float64 + if avgSeconds != nil { + avg = *avgSeconds + } + return completed, failed, avg, nil +} + +func (r *AdminReadRepo) ArtifactSizeByKind(ctx context.Context) (map[string]int64, error) { + rows, err := conn(ctx, r.pool).Query(ctx, "SELECT kind, COALESCE(SUM(size_bytes), 0) FROM artifacts GROUP BY kind") + if err != nil { + return nil, fmt.Errorf("artifact sizes: %w", err) + } + defer rows.Close() + out := make(map[string]int64) + for rows.Next() { + var kind string + var size int64 + if err := rows.Scan(&kind, &size); err != nil { + return nil, err + } + out[kind] = size + } + return out, rows.Err() +} + +func (r *AdminReadRepo) DatabaseSizeBytes(ctx context.Context) (int64, error) { + var size int64 + if err := conn(ctx, r.pool).QueryRow(ctx, "SELECT pg_database_size(current_database())").Scan(&size); err != nil { + return 0, fmt.Errorf("database size: %w", err) + } + return size, nil +} diff --git a/coordinator/internal/storage/sqlite/admin_read_repo.go b/coordinator/internal/storage/sqlite/admin_read_repo.go new file mode 100644 index 0000000..dd29cbf --- /dev/null +++ b/coordinator/internal/storage/sqlite/admin_read_repo.go @@ -0,0 +1,191 @@ +package sqlite + +import ( + "context" + "database/sql" + "fmt" + "strings" + "time" + + "github.com/google/uuid" + + "github.com/emil28092005/SciMesh/coordinator/internal/domain" + "github.com/emil28092005/SciMesh/coordinator/internal/usecase" +) + +// AdminReadRepo backs the coordinator admin console: paginated jobs, status +// counters, metrics buckets and storage figures. Read-only. +type AdminReadRepo struct{ db *sql.DB } + +func NewAdminReadRepo(db *sql.DB) *AdminReadRepo { return &AdminReadRepo{db: db} } + +var _ usecase.AdminReadRepository = (*AdminReadRepo)(nil) + +func (r *AdminReadRepo) ListJobsPaginated(ctx context.Context, status string, limit, offset int) ([]domain.Job, int, error) { + if limit < 1 || limit > 100 || offset < 0 { + return nil, 0, domain.ErrInvalidInput + } + where := "" + args := []any{} + if status != "" { + where = " WHERE status = ?" + args = append(args, status) + } + var total int + if err := conn(ctx, r.db).QueryRowContext(ctx, "SELECT COUNT(*) FROM jobs"+where, args...).Scan(&total); err != nil { + return nil, 0, fmt.Errorf("count jobs: %w", err) + } + rows, err := conn(ctx, r.db).QueryContext(ctx, + "SELECT "+jobColumns+" FROM jobs"+where+" ORDER BY created_at DESC, id DESC LIMIT ? OFFSET ?", + append(args, limit, offset)...) + if err != nil { + return nil, 0, fmt.Errorf("list jobs paginated: %w", err) + } + defer func() { _ = rows.Close() }() + jobs := make([]domain.Job, 0) + for rows.Next() { + job, err := scanJob(rows) + if err != nil { + return nil, 0, err + } + jobs = append(jobs, *job) + } + return jobs, total, rows.Err() +} + +func (r *AdminReadRepo) CountJobsByStatus(ctx context.Context) (map[string]int, error) { + rows, err := conn(ctx, r.db).QueryContext(ctx, "SELECT status, COUNT(*) FROM jobs GROUP BY status") + if err != nil { + return nil, fmt.Errorf("count jobs by status: %w", err) + } + defer func() { _ = rows.Close() }() + out := make(map[string]int) + for rows.Next() { + var status string + var count int + if err := rows.Scan(&status, &count); err != nil { + return nil, err + } + out[status] = count + } + return out, rows.Err() +} + +func (r *AdminReadRepo) TaskCountsByJobs(ctx context.Context, jobIDs []uuid.UUID) (map[uuid.UUID]map[string]int, error) { + out := make(map[uuid.UUID]map[string]int, len(jobIDs)) + if len(jobIDs) == 0 { + return out, nil + } + placeholders := make([]string, 0, len(jobIDs)) + args := make([]any, 0, len(jobIDs)) + for _, id := range jobIDs { + placeholders = append(placeholders, "?") + args = append(args, id.String()) + } + rows, err := conn(ctx, r.db).QueryContext(ctx, + "SELECT job_id, status, COUNT(*) FROM tasks WHERE job_id IN ("+strings.Join(placeholders, ", ")+") GROUP BY job_id, status", + args...) + if err != nil { + return nil, fmt.Errorf("task counts by jobs: %w", err) + } + defer func() { _ = rows.Close() }() + for rows.Next() { + var jobRaw, status string + var count int + if err := rows.Scan(&jobRaw, &status, &count); err != nil { + return nil, err + } + jobID, err := uuid.Parse(jobRaw) + if err != nil { + return nil, fmt.Errorf("task counts: parse job id: %w", err) + } + if out[jobID] == nil { + out[jobID] = make(map[string]int) + } + out[jobID][status] = count + } + return out, rows.Err() +} + +func (r *AdminReadRepo) JobCountsByDay(ctx context.Context, since time.Time) (map[string]int, error) { + // created_at is unix nanos; the bucket is the UTC calendar day. + rows, err := conn(ctx, r.db).QueryContext(ctx, + "SELECT strftime('%Y-%m-%d', created_at / 1000000000, 'unixepoch') AS day, COUNT(*) FROM jobs WHERE created_at >= ? GROUP BY day", + since.UTC().UnixNano()) + if err != nil { + return nil, fmt.Errorf("job counts by day: %w", err) + } + defer func() { _ = rows.Close() }() + out := make(map[string]int) + for rows.Next() { + var day string + var count int + if err := rows.Scan(&day, &count); err != nil { + return nil, err + } + out[day] = count + } + return out, rows.Err() +} + +func (r *AdminReadRepo) JobCountsByWorkload(ctx context.Context) (map[string]int, error) { + rows, err := conn(ctx, r.db).QueryContext(ctx, "SELECT workload, COUNT(*) FROM jobs GROUP BY workload") + if err != nil { + return nil, fmt.Errorf("job counts by workload: %w", err) + } + defer func() { _ = rows.Close() }() + out := make(map[string]int) + for rows.Next() { + var workload string + var count int + if err := rows.Scan(&workload, &count); err != nil { + return nil, err + } + out[workload] = count + } + return out, rows.Err() +} + +func (r *AdminReadRepo) TaskStats(ctx context.Context) (int64, int64, float64, error) { + var completed, failed int64 + var avgNanos sql.NullFloat64 + err := conn(ctx, r.db).QueryRowContext(ctx, ` + SELECT + COALESCE(SUM(CASE WHEN status = 'completed' THEN 1 ELSE 0 END), 0), + COALESCE(SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END), 0), + AVG(CASE WHEN status = 'completed' AND started_at IS NOT NULL THEN completed_at - started_at END) + FROM tasks`).Scan(&completed, &failed, &avgNanos) + if err != nil { + return 0, 0, 0, fmt.Errorf("task stats: %w", err) + } + return completed, failed, avgNanos.Float64 / 1e9, nil +} + +func (r *AdminReadRepo) ArtifactSizeByKind(ctx context.Context) (map[string]int64, error) { + rows, err := conn(ctx, r.db).QueryContext(ctx, "SELECT kind, COALESCE(SUM(size_bytes), 0) FROM artifacts GROUP BY kind") + if err != nil { + return nil, fmt.Errorf("artifact sizes: %w", err) + } + defer func() { _ = rows.Close() }() + out := make(map[string]int64) + for rows.Next() { + var kind string + var size int64 + if err := rows.Scan(&kind, &size); err != nil { + return nil, err + } + out[kind] = size + } + return out, rows.Err() +} + +func (r *AdminReadRepo) DatabaseSizeBytes(ctx context.Context) (int64, error) { + var pageCount, pageSize int64 + if err := conn(ctx, r.db).QueryRowContext(ctx, "PRAGMA page_count").Scan(&pageCount); err != nil { + return 0, fmt.Errorf("page count: %w", err) + } + if err := conn(ctx, r.db).QueryRowContext(ctx, "PRAGMA page_size").Scan(&pageSize); err != nil { + return 0, fmt.Errorf("page size: %w", err) + } + return pageCount * pageSize, nil +} diff --git a/coordinator/internal/storage/sqlite/admin_read_repo_test.go b/coordinator/internal/storage/sqlite/admin_read_repo_test.go new file mode 100644 index 0000000..7c3e2b9 --- /dev/null +++ b/coordinator/internal/storage/sqlite/admin_read_repo_test.go @@ -0,0 +1,171 @@ +package sqlite + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/emil28092005/SciMesh/coordinator/internal/domain" +) + +func TestAdminListJobsPaginatedAndCounts(t *testing.T) { + db := newTestDB(t) + ctx := context.Background() + jobRepo := NewJobRepo(db) + adminRepo := NewAdminReadRepo(db) + + jobs := make([]*domain.Job, 5) + for i := range jobs { + jobs[i] = seedJob(t, db, 2) + } + // Two completed, two running, one pending. + if err := jobRepo.UpdateStatus(ctx, jobs[0].ID, domain.JobCompleted, nil); err != nil { + t.Fatal(err) + } + if err := jobRepo.UpdateStatus(ctx, jobs[1].ID, domain.JobCompleted, nil); err != nil { + t.Fatal(err) + } + if err := jobRepo.UpdateStatus(ctx, jobs[2].ID, domain.JobRunning, nil); err != nil { + t.Fatal(err) + } + if err := jobRepo.UpdateStatus(ctx, jobs[3].ID, domain.JobRunning, nil); err != nil { + t.Fatal(err) + } + + all, total, err := adminRepo.ListJobsPaginated(ctx, "", 100, 0) + if err != nil { + t.Fatal(err) + } + if total != 5 || len(all) != 5 { + t.Errorf("all: total=%d len=%d, want 5/5", total, len(all)) + } + completed, total, err := adminRepo.ListJobsPaginated(ctx, "completed", 100, 0) + if err != nil { + t.Fatal(err) + } + if total != 2 || len(completed) != 2 { + t.Errorf("completed: total=%d len=%d, want 2/2", total, len(completed)) + } + page, total, err := adminRepo.ListJobsPaginated(ctx, "", 2, 2) + if err != nil { + t.Fatal(err) + } + if total != 5 || len(page) != 2 { + t.Errorf("page: total=%d len=%d, want 5/2", total, len(page)) + } + + counts, err := adminRepo.CountJobsByStatus(ctx) + if err != nil { + t.Fatal(err) + } + if counts["completed"] != 2 || counts["running"] != 2 || counts["pending"] != 1 { + t.Errorf("counts = %v, want completed=2 running=2 pending=1", counts) + } +} + +func TestAdminTaskCountsByJobs(t *testing.T) { + db := newTestDB(t) + ctx := context.Background() + job := seedJob(t, db, 3) + if _, err := db.ExecContext(ctx, "UPDATE tasks SET status = 'completed', result_artifact_id = ? WHERE chunk_index = 0 AND job_id = ?", uuid.NewString(), job.ID.String()); err != nil { + t.Fatal(err) + } + if _, err := db.ExecContext(ctx, "UPDATE tasks SET status = 'failed' WHERE chunk_index = 1 AND job_id = ?", job.ID.String()); err != nil { + t.Fatal(err) + } + + counts, err := NewAdminReadRepo(db).TaskCountsByJobs(ctx, []uuid.UUID{job.ID}) + if err != nil { + t.Fatal(err) + } + got := counts[job.ID] + if got["completed"] != 1 || got["failed"] != 1 || got["pending"] != 1 { + t.Errorf("task counts = %v, want completed=1 failed=1 pending=1", got) + } +} + +func TestAdminJobCountsByDay(t *testing.T) { + db := newTestDB(t) + ctx := context.Background() + job := seedJob(t, db, 1) + // Move the seed job to two days ago; create two more today. + old := fixedTime().Add(-48 * time.Hour) + if _, err := db.ExecContext(ctx, "UPDATE jobs SET created_at = ? WHERE id = ?", old.UnixNano(), job.ID.String()); err != nil { + t.Fatal(err) + } + seedJob(t, db, 1) + seedJob(t, db, 1) + + repo := NewAdminReadRepo(db) + counts, err := repo.JobCountsByDay(ctx, fixedTime().Add(-6*24*time.Hour)) + if err != nil { + t.Fatal(err) + } + today := fixedTime().UTC().Format("2006-01-02") + oldDay := old.UTC().Format("2006-01-02") + if counts[today] != 2 { + t.Errorf("today count = %d, want 2 (got %v)", counts[today], counts) + } + if counts[oldDay] != 1 { + t.Errorf("old day count = %d, want 1 (got %v)", counts[oldDay], counts) + } +} + +func TestAdminTaskStatsAndStorage(t *testing.T) { + db := newTestDB(t) + ctx := context.Background() + repo := NewAdminReadRepo(db) + + // One completed task with a known duration, one failed. + job := seedJob(t, db, 2) + start := fixedTime().Add(-2 * time.Minute) + done := fixedTime().Add(-90 * time.Second) + queries := []string{ + "UPDATE tasks SET status='completed', result_artifact_id=?, started_at=?, completed_at=? WHERE job_id=? AND chunk_index=0", + "UPDATE tasks SET status='failed' WHERE job_id=? AND chunk_index=1", + } + for i, q := range queries { + args := []any{uuid.NewString(), start.UnixNano(), done.UnixNano(), job.ID.String()} + if i == 1 { + args = []any{job.ID.String()} + } + if _, err := db.ExecContext(ctx, q, args...); err != nil { + t.Fatal(err) + } + } + + completed, failed, avg, err := repo.TaskStats(ctx) + if err != nil { + t.Fatal(err) + } + if completed != 1 || failed != 1 { + t.Errorf("stats = completed %d failed %d, want 1/1", completed, failed) + } + if avg < 29 || avg > 31 { + t.Errorf("avg duration = %.1fs, want ~30s", avg) + } + + // Artifact sizes by kind. + for _, kind := range []string{"input", "shard", "final_result"} { + if _, err := db.ExecContext(ctx, "INSERT INTO artifacts (id, job_id, kind, filename, storage_key, content_type, size_bytes, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + uuid.NewString(), job.ID.String(), kind, kind+".csv", "key-"+kind, "text/csv", int64(len(kind)*1000), fixedTime().UnixNano()); err != nil { + t.Fatal(err) + } + } + sizes, err := repo.ArtifactSizeByKind(ctx) + if err != nil { + t.Fatal(err) + } + if sizes["input"] != 5000 || sizes["shard"] != 5000 || sizes["final_result"] != 12000 { + t.Errorf("sizes = %v", sizes) + } + dbBytes, err := repo.DatabaseSizeBytes(ctx) + if err != nil { + t.Fatal(err) + } + if dbBytes <= 0 { + t.Errorf("database size = %d, want > 0", dbBytes) + } +} diff --git a/coordinator/internal/transport/http/server.go b/coordinator/internal/transport/http/server.go index 1dced45..d01edd0 100644 --- a/coordinator/internal/transport/http/server.go +++ b/coordinator/internal/transport/http/server.go @@ -35,6 +35,7 @@ type UseCases struct { GetTaskInput *usecase.GetTaskInput Dashboard *usecase.Dashboard PreviewArtifact *usecase.PreviewArtifact + Admin *usecase.Admin } type Server struct { @@ -177,6 +178,10 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler { // Admin panel: session + admin role. ui.Handle("GET /ui/admin", chain(http.HandlerFunc(s.handleUIAdmin), gate, requireAdmin)) ui.Handle("POST /ui/admin/user-action", chain(http.HandlerFunc(s.handleUIAdminUserAction), gate, requireAdmin)) + // Admin console APIs: session + admin role, bounded read models. + ui.Handle("GET /ui/admin/api/system", chain(http.HandlerFunc(s.handleUIAdminSystemJSON), gate, requireAdmin)) + ui.Handle("GET /ui/admin/api/jobs", chain(http.HandlerFunc(s.handleUIAdminJobsJSON), gate, requireAdmin)) + ui.Handle("GET /ui/admin/api/metrics", chain(http.HandlerFunc(s.handleUIAdminMetricsJSON), gate, requireAdmin)) } else { for _, rt := range app { ui.HandleFunc(rt.pattern, rt.handler) diff --git a/coordinator/internal/transport/http/templates/admin.html b/coordinator/internal/transport/http/templates/admin.html index 37f704a..ba0f603 100644 --- a/coordinator/internal/transport/http/templates/admin.html +++ b/coordinator/internal/transport/http/templates/admin.html @@ -2,45 +2,389 @@ - - - Admin · SciMesh - + + +SciMesh · Coordinator Admin + -
-
-

Admin panel

User & run control

-
← DashboardProfile
-
-

Signed in as {{.Role}}. Promote or verify a user by their id, and control every job from the dashboard.

+
- {{if .Msg}}
{{.Msg}}
{{end}} - {{if .Error}}
{{.Error}}
{{end}} + -
-

Manage a user

-

Paste the user id (the JWT sub / the value shown at registration). Actions are applied immediately.

-
- - -

Promote makes them an admin; Verify marks them a trusted contributor (their workers skip quorum).

-
- - - - +
+
+

System

Cluster state and node information

+
admin console
+
+
+ + +
+
+
Version
—
loading…
+
Uptime
—
loading…
+
Active jobs
—
loading…
+
Workers online
—
loading…
+
+
+
+

Storage usage

—
+
+
+ Datasets · — + Artifacts · — + Database · —
- -
+
+
+

Health

—
+
+
Database
—
+
Userservice
—
+
Reducer
—
+
+
+
+
Node information
+
+
+
Binary
—
+
Listen address
—
+
Data directory
—
+
Database engine
—
+
Public URL
—
+
+
+
-
-

Jobs & tasks

-

As an admin you already see every user's jobs on the dashboard, with per-task status and job cancellation. A regular user sees only their own.

-
Open the dashboard →
-
-
+ +
+
+
+ + + +
JobWorkloadOwnerStatusProgressSubmitted
+
—
+
+
+ + +
+

Workers

milestone M2
Trust management lands with milestone M2Worker list, trust dropdown and heartbeat overview are wired next. The dashboard already shows the live fleet.
+
+ + +
+
Quick user action
+
+
+
+
+ + + + +
+ {{if .Msg}}
✓ {{.Msg}}
{{end}} + {{if .Error}}
✗ {{.Error}}
{{end}} +
+
+
Accounts and worker keys
+
User list and key management land with milestone M2The userservice gains admin list endpoints; the console gets tables, role selects and key revoke.
+
+ + +
+
Workload enable/disable lands with milestone M2The catalog is already served to the job form; persisted on/off switches arrive with the settings migration.
+
+ + +
+
+
Jobs · 7 days
—
created in the last week
+
Shards completed
—
across all workers
+
Avg shard time
—
completed shards only
+
Failure rate
—
—
+
+
+
+

Jobs per day

last 7 days
+
+
+
+

Jobs by workload

all time
+
+
+
+
+ + +
+
Cluster settings land with milestone M2Worker token reveal, public URL and storage settings follow in the next milestone.
+
+ + + + + + {{end}} diff --git a/coordinator/internal/transport/http/ui_admin.go b/coordinator/internal/transport/http/ui_admin.go index 6287b91..c20f9ba 100644 --- a/coordinator/internal/transport/http/ui_admin.go +++ b/coordinator/internal/transport/http/ui_admin.go @@ -2,9 +2,11 @@ package http import ( "context" + "encoding/json" "io" "net/http" "net/url" + "strconv" "strings" "github.com/google/uuid" @@ -112,3 +114,78 @@ func (s *Server) callUserserviceAuthed(ctx context.Context, method, path, bearer } return resp.StatusCode, body, nil } + +// handleUIAdminSystemJSON serves the admin "System" page: process info, +// storage figures and health. Admin-only via the route chain. +func (s *Server) handleUIAdminSystemJSON(w http.ResponseWriter, r *http.Request) { + ctx, cancel := s.reqCtx(r) + defer cancel() + view, err := s.uc.Admin.System(ctx) + if err != nil { + s.writeError(w, r, err) + return + } + writeJSON(w, http.StatusOK, view) +} + +// handleUIAdminJobsJSON serves one page of the admin jobs table. The owner +// emails are resolved from the userservice when it is reachable; the resolver +// failing is not fatal (cards fall back to short ids). +func (s *Server) handleUIAdminJobsJSON(w http.ResponseWriter, r *http.Request) { + ctx, cancel := s.reqCtx(r) + defer cancel() + page, _ := strconv.Atoi(r.URL.Query().Get("page")) + perPage, _ := strconv.Atoi(r.URL.Query().Get("per_page")) + status := strings.TrimSpace(r.URL.Query().Get("status")) + view, err := s.uc.Admin.Jobs(ctx, status, page, perPage, s.adminOwnerEmails(r)) + if err != nil { + s.writeError(w, r, err) + return + } + writeJSON(w, http.StatusOK, view) +} + +// handleUIAdminMetricsJSON serves the admin "Metrics" page. +func (s *Server) handleUIAdminMetricsJSON(w http.ResponseWriter, r *http.Request) { + ctx, cancel := s.reqCtx(r) + defer cancel() + view, err := s.uc.Admin.Metrics(ctx) + if err != nil { + s.writeError(w, r, err) + return + } + writeJSON(w, http.StatusOK, view) +} + +// adminOwnerEmails resolves job owner ids to emails through the userservice, +// which is the only place email addresses live. It never blocks the page on +// failure: an empty map leaves the admin jobs table on short ids. +func (s *Server) adminOwnerEmails(r *http.Request) map[uuid.UUID]string { + if s.userserviceURL == "" { + return nil + } + c, err := r.Cookie(sessionCookie) + if err != nil { + return nil + } + status, body, err := s.callUserserviceAuthed(r.Context(), http.MethodGet, "/users", c.Value) + if err != nil || status != http.StatusOK { + return nil + } + var users []struct { + ID string `json:"id"` + Email string `json:"email"` + } + if err := json.Unmarshal(body, &users); err != nil { + return nil + } + out := make(map[uuid.UUID]string, len(users)) + for _, user := range users { + id, err := uuid.Parse(user.ID) + if err != nil { + continue + } + out[id] = user.Email + } + return out +} diff --git a/coordinator/internal/usecase/admin.go b/coordinator/internal/usecase/admin.go new file mode 100644 index 0000000..9613a0e --- /dev/null +++ b/coordinator/internal/usecase/admin.go @@ -0,0 +1,341 @@ +package usecase + +import ( + "context" + "sort" + "strings" + "time" + + "github.com/google/uuid" + + "github.com/emil28092005/SciMesh/coordinator/internal/domain" +) + +// AdminReadRepository is the bounded read projection behind the coordinator +// admin console. Like UIReadRepository it exposes no storage paths or +// credentials; unlike it, every method is admin-scoped (no owner filter). +type AdminReadRepository interface { + // ListJobsPaginated returns one page of jobs filtered by stored status; + // an empty status returns all. total counts the filtered set (for the + // pager). + ListJobsPaginated(ctx context.Context, status string, limit, offset int) (jobs []domain.Job, total int, err error) + // CountJobsByStatus powers the status tabs: every stored status, all jobs. + CountJobsByStatus(ctx context.Context) (map[string]int, error) + // TaskCountsByJobs aggregates task statuses per job for progress bars. + TaskCountsByJobs(ctx context.Context, jobIDs []uuid.UUID) (map[uuid.UUID]map[string]int, error) + // JobCountsByDay buckets jobs created since `since` by UTC day + // ("2006-01-02"). + JobCountsByDay(ctx context.Context, since time.Time) (map[string]int, error) + // JobCountsByWorkload counts all jobs per workload name. + JobCountsByWorkload(ctx context.Context) (map[string]int, error) + // TaskStats totals shard execution: completed/failed counts and the mean + // run duration of completed shards (seconds; 0 when nothing completed). + TaskStats(ctx context.Context) (completed, failed int64, avgSeconds float64, err error) + // ArtifactSizeByKind sums stored bytes per artifact kind. + ArtifactSizeByKind(ctx context.Context) (map[string]int64, error) + // DatabaseSizeBytes reports the engine's own size figure (sqlite pages, + // pg_database_size); 0 when the engine cannot say. + DatabaseSizeBytes(ctx context.Context) (int64, error) +} + +// AdminNodeInfo describes the running coordinator process to the admin +// console. It is static for the process lifetime and assembled at startup. +type AdminNodeInfo struct { + Version string + StartedAt time.Time + Binary string + Addr string + DataDir string + DBEngine string + PublicURL string + Userservice string // base URL; empty when the UI runs without user auth +} + +type AdminStorageView struct { + DatasetsBytes int64 `json:"datasets_bytes"` + ArtifactsBytes int64 `json:"artifacts_bytes"` + DatabaseBytes int64 `json:"database_bytes"` +} + +type AdminHealthView struct { + Database string `json:"database"` // connected | error + Reducer string `json:"reducer"` // idle | active + Userservice string `json:"userservice"` // embedded | external | disabled +} + +type AdminNodeView struct { + Binary string `json:"binary"` + Addr string `json:"addr"` + DataDir string `json:"data_dir"` + DBEngine string `json:"db_engine"` + PublicURL string `json:"public_url"` +} + +type AdminSystemView struct { + Version string `json:"version"` + StartedAt time.Time `json:"started_at"` + UptimeSeconds int64 `json:"uptime_seconds"` + ActiveJobs int `json:"active_jobs"` + RunningJobs int `json:"running_jobs"` + WaitingJobs int `json:"waiting_jobs"` + WorkersOnline int `json:"workers_online"` + WorkersBusy int `json:"workers_busy"` + WorkersTotal int `json:"workers_total"` + Storage AdminStorageView `json:"storage"` + Health AdminHealthView `json:"health"` + Node AdminNodeView `json:"node"` +} + +// AdminJobCard is one row of the admin jobs table. Owner is a display string +// resolved by the caller (email when the userservice is reachable, a short id +// or "cluster token" otherwise). +type AdminJobCard struct { + ID string `json:"id"` + Workload string `json:"workload"` + Status string `json:"status"` + OwnerID string `json:"owner_id,omitempty"` + Owner string `json:"owner"` + Total int `json:"total"` + Completed int `json:"completed"` + Failed int `json:"failed"` + CreatedAt time.Time `json:"created_at"` + CompletedAt *time.Time `json:"completed_at,omitempty"` +} + +type AdminJobsView struct { + Jobs []AdminJobCard `json:"jobs"` + Total int `json:"total"` + Page int `json:"page"` + PerPage int `json:"per_page"` + // Counts holds every stored status for the filter tabs (all jobs, not + // just the current filter). + Counts map[string]int `json:"counts"` +} + +type AdminDayCount struct { + Day string `json:"day"` + Count int `json:"count"` +} + +type AdminWorkloadCount struct { + Workload string `json:"workload"` + Count int `json:"count"` +} + +type AdminMetricsView struct { + JobsLast7Days int `json:"jobs_last_7_days"` + JobsByDay []AdminDayCount `json:"jobs_by_day"` + JobsByWorkload []AdminWorkloadCount `json:"jobs_by_workload"` + ShardsCompleted int64 `json:"shards_completed"` + ShardsFailed int64 `json:"shards_failed"` + AvgShardSeconds float64 `json:"avg_shard_seconds"` + FailureRate float64 `json:"failure_rate"` +} + +// Admin answers the coordinator admin console from the bounded read model +// plus process info supplied at startup. +type Admin struct { + read AdminReadRepository + uiRead UIReadRepository + node AdminNodeInfo + ready func(context.Context) error + now func() time.Time +} + +func NewAdmin(read AdminReadRepository, uiRead UIReadRepository, node AdminNodeInfo, ready func(context.Context) error, now func() time.Time) *Admin { + if now == nil { + now = time.Now + } + return &Admin{read: read, uiRead: uiRead, node: node, ready: ready, now: now} +} + +func (a *Admin) System(ctx context.Context) (AdminSystemView, error) { + counts, err := a.read.CountJobsByStatus(ctx) + if err != nil { + return AdminSystemView{}, err + } + workers, err := a.uiRead.ListWorkers(ctx, 100) + if err != nil { + return AdminSystemView{}, err + } + sizes, err := a.read.ArtifactSizeByKind(ctx) + if err != nil { + return AdminSystemView{}, err + } + dbSize, err := a.read.DatabaseSizeBytes(ctx) + if err != nil { + return AdminSystemView{}, err + } + + out := AdminSystemView{ + Version: a.node.Version, + StartedAt: a.node.StartedAt, + WaitingJobs: counts[string(domain.JobPending)], + RunningJobs: counts[string(domain.JobRunning)] + counts[string(domain.JobReducing)], + } + out.ActiveJobs = out.WaitingJobs + out.RunningJobs + out.UptimeSeconds = int64(a.now().Sub(a.node.StartedAt).Seconds()) + if out.UptimeSeconds < 0 { + out.UptimeSeconds = 0 + } + for _, w := range workers { + out.WorkersTotal++ + switch w.Status { + case domain.WorkerOnline: + out.WorkersOnline++ + case domain.WorkerBusy: + out.WorkersOnline++ + out.WorkersBusy++ + } + } + for kind, size := range sizes { + if kind == string(domain.ArtifactInput) { + out.Storage.DatasetsBytes += size + } else { + out.Storage.ArtifactsBytes += size + } + } + out.Storage.DatabaseBytes = dbSize + + out.Health.Database = "connected" + if a.ready != nil { + if err := a.ready(ctx); err != nil { + out.Health.Database = "error" + } + } + out.Health.Reducer = "idle" + if counts[string(domain.JobReducing)] > 0 { + out.Health.Reducer = "active" + } + out.Health.Userservice = "disabled" + if a.node.Userservice != "" { + out.Health.Userservice = "external" + // The embedded userservice always binds the loopback interface. + if strings.Contains(a.node.Userservice, "127.0.0.1") || strings.Contains(a.node.Userservice, "localhost") { + out.Health.Userservice = "embedded" + } + } + out.Node = AdminNodeView{ + Binary: a.node.Binary, + Addr: a.node.Addr, + DataDir: a.node.DataDir, + DBEngine: a.node.DBEngine, + PublicURL: a.node.PublicURL, + } + return out, nil +} + +// Jobs returns one page of the admin jobs table. The owner emails map may be +// nil; cards then fall back to a short id or "cluster token". +func (a *Admin) Jobs(ctx context.Context, status string, page, perPage int, ownerEmails map[uuid.UUID]string) (AdminJobsView, error) { + if page < 1 { + page = 1 + } + if perPage < 1 || perPage > 100 { + perPage = 20 + } + jobs, total, err := a.read.ListJobsPaginated(ctx, status, perPage, (page-1)*perPage) + if err != nil { + return AdminJobsView{}, err + } + counts, err := a.read.CountJobsByStatus(ctx) + if err != nil { + return AdminJobsView{}, err + } + jobIDs := make([]uuid.UUID, 0, len(jobs)) + for _, job := range jobs { + jobIDs = append(jobIDs, job.ID) + } + taskCounts, err := a.read.TaskCountsByJobs(ctx, jobIDs) + if err != nil { + return AdminJobsView{}, err + } + out := AdminJobsView{ + Jobs: make([]AdminJobCard, 0, len(jobs)), + Total: total, + Page: page, + PerPage: perPage, + Counts: counts, + } + for _, job := range jobs { + tc := taskCounts[job.ID] + card := AdminJobCard{ + ID: job.ID.String(), + Workload: job.Workload, + CreatedAt: job.CreatedAt, + CompletedAt: job.CompletedAt, + Owner: "cluster token", + } + var pending, leased, cancelled int + for status, n := range tc { + card.Total += n + switch domain.TaskStatus(status) { + case domain.TaskCompleted: + card.Completed = n + case domain.TaskFailed: + card.Failed = n + case domain.TaskPending: + pending = n + case domain.TaskLeased, domain.TaskRunning: + leased += n + case domain.TaskCancelled: + cancelled = n + } + } + // Derive the status exactly like the operator dashboard does, so the + // two views never disagree about the same job. + progress := domain.JobProgress{Job: job, Total: card.Total, Pending: pending, Leased: leased, Done: card.Completed, Failed: card.Failed, Cancelled: cancelled} + card.Status = string(progress.DeriveStatus()) + if job.OwnerID != nil { + card.OwnerID = job.OwnerID.String() + card.Owner = "user " + shortID(job.OwnerID.String()) + if email, ok := ownerEmails[*job.OwnerID]; ok && email != "" { + card.Owner = email + } + } + out.Jobs = append(out.Jobs, card) + } + return out, nil +} + +func (a *Admin) Metrics(ctx context.Context) (AdminMetricsView, error) { + since := a.now().Add(-6 * 24 * time.Hour).Truncate(24 * time.Hour) + byDay, err := a.read.JobCountsByDay(ctx, since) + if err != nil { + return AdminMetricsView{}, err + } + byWorkload, err := a.read.JobCountsByWorkload(ctx) + if err != nil { + return AdminMetricsView{}, err + } + completed, failed, avg, err := a.read.TaskStats(ctx) + if err != nil { + return AdminMetricsView{}, err + } + out := AdminMetricsView{ + JobsByDay: make([]AdminDayCount, 0, 7), + JobsByWorkload: make([]AdminWorkloadCount, 0, len(byWorkload)), + ShardsCompleted: completed, + ShardsFailed: failed, + AvgShardSeconds: avg, + } + if completed+failed > 0 { + out.FailureRate = float64(failed) / float64(completed+failed) + } + for i := 0; i < 7; i++ { + day := since.Add(time.Duration(i) * 24 * time.Hour).UTC().Format("2006-01-02") + count := byDay[day] + out.JobsByDay = append(out.JobsByDay, AdminDayCount{Day: day, Count: count}) + out.JobsLast7Days += count + } + for workload, count := range byWorkload { + out.JobsByWorkload = append(out.JobsByWorkload, AdminWorkloadCount{Workload: workload, Count: count}) + } + sort.Slice(out.JobsByWorkload, func(i, j int) bool { + if out.JobsByWorkload[i].Count != out.JobsByWorkload[j].Count { + return out.JobsByWorkload[i].Count > out.JobsByWorkload[j].Count + } + return out.JobsByWorkload[i].Workload < out.JobsByWorkload[j].Workload + }) + return out, nil +} diff --git a/coordinator/internal/usecase/admin_test.go b/coordinator/internal/usecase/admin_test.go new file mode 100644 index 0000000..9b85cf5 --- /dev/null +++ b/coordinator/internal/usecase/admin_test.go @@ -0,0 +1,223 @@ +package usecase + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/emil28092005/SciMesh/coordinator/internal/domain" +) + +type fakeAdminRead struct { + jobs []domain.Job + taskCounts map[uuid.UUID]map[string]int + sizes map[string]int64 + byDay map[string]int + byWorkload map[string]int + completed int64 + failed int64 + avg float64 + dbSize int64 +} + +func (f *fakeAdminRead) ListJobsPaginated(ctx context.Context, status string, limit, offset int) ([]domain.Job, int, error) { + var out []domain.Job + for _, j := range f.jobs { + if status == "" || string(j.Status) == status { + out = append(out, j) + } + } + total := len(out) + if offset >= len(out) { + return nil, total, nil + } + if offset+limit < len(out) { + out = out[offset : offset+limit] + } else { + out = out[offset:] + } + return out, total, nil +} + +func (f *fakeAdminRead) CountJobsByStatus(ctx context.Context) (map[string]int, error) { + out := map[string]int{} + for _, j := range f.jobs { + out[string(j.Status)]++ + } + return out, nil +} + +func (f *fakeAdminRead) TaskCountsByJobs(ctx context.Context, jobIDs []uuid.UUID) (map[uuid.UUID]map[string]int, error) { + return f.taskCounts, nil +} + +func (f *fakeAdminRead) JobCountsByDay(ctx context.Context, since time.Time) (map[string]int, error) { + return f.byDay, nil +} +func (f *fakeAdminRead) JobCountsByWorkload(ctx context.Context) (map[string]int, error) { + return f.byWorkload, nil +} +func (f *fakeAdminRead) TaskStats(ctx context.Context) (int64, int64, float64, error) { + return f.completed, f.failed, f.avg, nil +} +func (f *fakeAdminRead) ArtifactSizeByKind(ctx context.Context) (map[string]int64, error) { + return f.sizes, nil +} +func (f *fakeAdminRead) DatabaseSizeBytes(ctx context.Context) (int64, error) { return f.dbSize, nil } + +type fakeUIRead struct { + UIReadRepository // embedded: only ListWorkers is exercised + workers []domain.Worker +} + +func (f *fakeUIRead) ListWorkers(ctx context.Context, limit int) ([]domain.Worker, error) { + return f.workers, nil +} + +func adminFixture() *Admin { + return NewAdmin( + &fakeAdminRead{}, + &fakeUIRead{}, + AdminNodeInfo{ + Version: "1.1.0-alpha.1", StartedAt: time.Unix(1_000_000, 0).UTC(), + Binary: "/usr/local/bin/coordinator", Addr: ":8080", DataDir: "/var/lib/scimesh", + DBEngine: "sqlite", PublicURL: "http://192.168.1.10:8080", Userservice: "http://127.0.0.1:41273", + }, + func(context.Context) error { return nil }, + func() time.Time { return time.Unix(1_000_000+3600*3, 0).UTC() }, + ) +} + +func TestAdminSystemAssemblesKpis(t *testing.T) { + owner := uuid.New() + a := adminFixture() + a.read = &fakeAdminRead{ + jobs: []domain.Job{ + {ID: uuid.New(), Status: domain.JobRunning, Workload: "similarity-search"}, + {ID: uuid.New(), Status: domain.JobPending, Workload: "similarity-search", OwnerID: &owner}, + {ID: uuid.New(), Status: domain.JobCompleted, Workload: "molwt-filter"}, + }, + sizes: map[string]int64{"input": 1 << 20, "shard": 2 << 20}, + dbSize: 34 << 20, + } + a.uiRead = &fakeUIRead{workers: []domain.Worker{ + {Status: domain.WorkerOnline}, + {Status: domain.WorkerBusy}, + {Status: domain.WorkerOffline}, + }} + + v, err := a.System(context.Background()) + if err != nil { + t.Fatal(err) + } + if v.Version != "1.1.0-alpha.1" { + t.Errorf("version = %q", v.Version) + } + if v.ActiveJobs != 2 || v.WaitingJobs != 1 || v.RunningJobs != 1 { + t.Errorf("jobs: active=%d waiting=%d running=%d, want 2/1/1", v.ActiveJobs, v.WaitingJobs, v.RunningJobs) + } + if v.WorkersOnline != 2 || v.WorkersBusy != 1 || v.WorkersTotal != 3 { + t.Errorf("workers: online=%d busy=%d total=%d, want 2/1/3", v.WorkersOnline, v.WorkersBusy, v.WorkersTotal) + } + if v.UptimeSeconds != 10800 { + t.Errorf("uptime = %d, want 10800", v.UptimeSeconds) + } + if v.Storage.DatasetsBytes != 1<<20 || v.Storage.ArtifactsBytes != 2<<20 || v.Storage.DatabaseBytes != 34<<20 { + t.Errorf("storage = %+v", v.Storage) + } + if v.Health.Database != "connected" || v.Health.Userservice != "embedded" || v.Health.Reducer != "idle" { + t.Errorf("health = %+v", v.Health) + } + if v.Node.Binary != "/usr/local/bin/coordinator" || v.Node.DBEngine != "sqlite" { + t.Errorf("node = %+v", v.Node) + } +} + +func TestAdminSystemReportsUnhealthyDatabase(t *testing.T) { + a := adminFixture() + a.ready = func(context.Context) error { return errors.New("connection refused") } + v, err := a.System(context.Background()) + if err != nil { + t.Fatal(err) + } + if v.Health.Database != "error" { + t.Errorf("database health = %q, want error", v.Health.Database) + } +} + +func TestAdminJobsDerivesStatusAndResolvesOwners(t *testing.T) { + jobID := uuid.New() + owner := uuid.New() + a := adminFixture() + a.read = &fakeAdminRead{ + jobs: []domain.Job{{ID: jobID, Status: domain.JobPending, Workload: "similarity-graph", OwnerID: &owner, CreatedAt: time.Unix(100, 0)}}, + taskCounts: map[uuid.UUID]map[string]int{ + jobID: {"completed": 5, "failed": 1, "running": 2}, + }, + } + + view, err := a.Jobs(context.Background(), "", 1, 20, map[uuid.UUID]string{owner: "alice@lab.org"}) + if err != nil { + t.Fatal(err) + } + if len(view.Jobs) != 1 { + t.Fatalf("jobs = %d, want 1", len(view.Jobs)) + } + card := view.Jobs[0] + if card.Owner != "alice@lab.org" || card.OwnerID != owner.String() { + t.Errorf("owner = %q (%s)", card.Owner, card.OwnerID) + } + if card.Total != 8 || card.Completed != 5 || card.Failed != 1 { + t.Errorf("progress: total=%d completed=%d failed=%d", card.Total, card.Completed, card.Failed) + } + if card.Status != "running" { + t.Errorf("derived status = %q, want running (5 completed / 8 with 1 failed)", card.Status) + } + if view.Counts["pending"] != 1 { + t.Errorf("counts = %v", view.Counts) + } +} + +func TestAdminJobsFallsBackWithoutOwners(t *testing.T) { + jobID := uuid.New() + a := adminFixture() + a.read = &fakeAdminRead{jobs: []domain.Job{{ID: jobID, Status: domain.JobPending, Workload: "x", CreatedAt: time.Unix(100, 0)}}} + view, err := a.Jobs(context.Background(), "", 1, 20, nil) + if err != nil { + t.Fatal(err) + } + if view.Jobs[0].Owner != "cluster token" { + t.Errorf("owner fallback = %q, want cluster token", view.Jobs[0].Owner) + } +} + +func TestAdminMetricsBuckets(t *testing.T) { + now := time.Unix(1_000_000+3600*3, 0).UTC() + since := now.Add(-6 * 24 * time.Hour).Truncate(24 * time.Hour) + day := func(offset int) string { return since.Add(time.Duration(offset) * 24 * time.Hour).Format("2006-01-02") } + a := adminFixture() + a.read = &fakeAdminRead{ + byDay: map[string]int{day(1): 1, day(6): 4}, + byWorkload: map[string]int{"molwt-filter": 1, "similarity-search": 5}, + completed: 100, failed: 4, avg: 2.5, + } + v, err := a.Metrics(context.Background()) + if err != nil { + t.Fatal(err) + } + if len(v.JobsByDay) != 7 || v.JobsLast7Days != 5 { + t.Errorf("by day: %d entries, total %d (want 7 / 5)", len(v.JobsByDay), v.JobsLast7Days) + } + if v.JobsByDay[6].Count != 4 || v.JobsByDay[1].Count != 1 { + t.Errorf("by day = %+v", v.JobsByDay) + } + if v.JobsByWorkload[0].Workload != "similarity-search" || v.JobsByWorkload[0].Count != 5 { + t.Errorf("by workload = %+v", v.JobsByWorkload) + } + if v.FailureRate != 4.0/104.0 || v.AvgShardSeconds != 2.5 { + t.Errorf("rate=%.4f avg=%.2f", v.FailureRate, v.AvgShardSeconds) + } +}