feat(java): dashboard settings, overrides, metrics, ops

sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java

@@ -5,14 +5,20 @@
 import java.io.IOException;
 import java.io.OutputStream;
 import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.security.MessageDigest;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Executors;
 import org.byteveda.taskito.Taskito;
 import org.byteveda.taskito.dashboard.api.CoreHandlers;
+import org.byteveda.taskito.dashboard.api.MetricsHandlers;
+import org.byteveda.taskito.dashboard.api.OpsHandlers;
+import org.byteveda.taskito.dashboard.api.OverridesHandlers;
+import org.byteveda.taskito.dashboard.api.SettingsHandlers;
 import org.byteveda.taskito.dashboard.auth.AuthHandlers;
 import org.byteveda.taskito.dashboard.auth.AuthStore;
 import org.byteveda.taskito.dashboard.auth.Cookies;
@@ -21,6 +27,7 @@
 import org.byteveda.taskito.dashboard.auth.TokenAuth;
 import org.byteveda.taskito.dashboard.routing.Req;
 import org.byteveda.taskito.dashboard.routing.Router;
+import org.byteveda.taskito.dashboard.store.OverridesStore;
 import org.byteveda.taskito.dashboard.store.SettingsAccess;
 import org.byteveda.taskito.dashboard.support.DashboardError;
 import org.byteveda.taskito.dashboard.support.Http;
@@ -47,9 +54,13 @@ public final class DashboardServer implements AutoCloseable {
     private final Taskito queue;
     private final Path staticDir;
     private final boolean secureCookies;
+    private static final String METRICS_TOKEN_ENV = "TASKITO_DASHBOARD_METRICS_TOKEN";
+
     private final AuthStore authStore;
     private final TokenAuth tokenAuth;
     private final AuthHandlers authHandlers;
+    private final OpsHandlers ops;
+    private final String metricsToken;
     private final Router router;
 
     private DashboardServer(HttpServer server, Taskito queue, Path staticDir, boolean secureCookies, String token) {
@@ -60,6 +71,8 @@ private DashboardServer(HttpServer server, Taskito queue, Path staticDir, boolea
         this.authStore = new AuthStore(SettingsAccess.of(queue));
         this.tokenAuth = token != null ? new TokenAuth(token) : null;
         this.authHandlers = new AuthHandlers(authStore);
+        this.ops = new OpsHandlers(queue);
+        this.metricsToken = System.getenv(METRICS_TOKEN_ENV);
         this.router = buildRouter();
     }
 
@@ -116,6 +129,10 @@ private void dispatch(HttpExchange exchange) throws IOException {
             String path = exchange.getRequestURI().getPath();
             if (path.equals("/health")) {
                 Http.respondJson(exchange, 200, Map.of("status", "ok"));
+            } else if (path.equals("/readiness")) {
+                serveReadiness(exchange);
+            } else if (path.equals("/metrics")) {
+                serveMetrics(exchange);
             } else if (path.startsWith("/api/")) {
                 handleApi(exchange, path);
             } else {
@@ -181,6 +198,10 @@ private void handleTokenMode(HttpExchange exchange, String path, String method,
 
     private Router buildRouter() {
         CoreHandlers core = new CoreHandlers(queue);
+        SettingsAccess settings = SettingsAccess.of(queue);
+        SettingsHandlers settingsApi = new SettingsHandlers(settings);
+        OverridesHandlers overrides = new OverridesHandlers(queue, new OverridesStore(settings));
+        MetricsHandlers metrics = new MetricsHandlers(queue);
         long ttl = AuthStore.DEFAULT_SESSION_TTL_SECONDS;
         Router r = new Router();
 
@@ -206,9 +227,33 @@ private Router buildRouter() {
         r.get("/api/jobs", req -> core.listJobs(req.query()));
         r.get("/api/jobs/([^/]+)", req -> core.job(req.param(0)));
         r.get("/api/dead-letters", req -> core.listDead(req.query()));
-        r.get("/api/metrics", req -> core.listMetrics(req.query()));
         r.get("/api/workers", req -> core.listWorkers());
 
+        // Metrics (aggregated — the SPA contract, not raw rows)
+        r.get("/api/metrics", req -> metrics.aggregated(req.query()));
+        r.get("/api/metrics/timeseries", req -> metrics.timeseries(req.query()));
+
+        // Ops
+        r.get("/api/circuit-breakers", req -> ops.circuitBreakers());
+        r.get("/api/event-types", req -> ops.eventTypes());
+        r.get("/api/scaler", req -> ops.scaler(req.query()));
+
+        // Settings KV
+        r.get("/api/settings", req -> settingsApi.list());
+        r.get("/api/settings/(.+)", req -> settingsApi.get(req.param(0)));
+        r.put("/api/settings/(.+)", req -> settingsApi.put(req.param(0), req.jsonBody()));
+        r.delete("/api/settings/(.+)", req -> settingsApi.delete(req.param(0)));
+
+        // Tasks / queues + overrides
+        r.get("/api/tasks", req -> overrides.listTasks());
+        r.get("/api/queues", req -> overrides.listQueues());
+        r.get("/api/tasks/([^/]+)/override", req -> overrides.getTaskOverride(req.param(0)));
+        r.put("/api/tasks/([^/]+)/override", req -> overrides.putTaskOverride(req.param(0), req.jsonBody()));
+        r.delete("/api/tasks/([^/]+)/override", req -> overrides.deleteTaskOverride(req.param(0)));
+        r.get("/api/queues/([^/]+)/override", req -> overrides.getQueueOverride(req.param(0)));
+        r.put("/api/queues/([^/]+)/override", req -> overrides.putQueueOverride(req.param(0), req.jsonBody()));
+        r.delete("/api/queues/([^/]+)/override", req -> overrides.deleteQueueOverride(req.param(0)));
+
         // Action
         r.post("/api/jobs/([^/]+)/cancel", req -> core.cancel(req.param(0)));
         r.post("/api/dead-letters/([^/]+)/retry", req -> core.retryDead(req.param(0)));
@@ -238,6 +283,43 @@ private Object logout(Req req) {
         return out;
     }
 
+    // ---- probes ------------------------------------------------------------
+
+    private void serveReadiness(HttpExchange exchange) throws IOException {
+        if (!metricsTokenOk(exchange)) {
+            Http.respondError(exchange, 401, "unauthorized");
+            return;
+        }
+        Http.respondJson(exchange, 200, ops.readiness());
+    }
+
+    private void serveMetrics(HttpExchange exchange) throws IOException {
+        if (!metricsTokenOk(exchange)) {
+            Http.respondError(exchange, 401, "unauthorized");
+            return;
+        }
+        byte[] out = ops.prometheus().getBytes(StandardCharsets.UTF_8);
+        exchange.getResponseHeaders().set("Content-Type", "text/plain; version=0.0.4; charset=utf-8");
+        exchange.sendResponseHeaders(200, out.length);
+        try (OutputStream stream = exchange.getResponseBody()) {
+            stream.write(out);
+        }
+    }
+
+    /** Open when no metrics token is configured; otherwise require a bearer match. */
+    private boolean metricsTokenOk(HttpExchange exchange) {
+        if (metricsToken == null || metricsToken.isEmpty()) {
+            return true;
+        }
+        String authorization = exchange.getRequestHeaders().getFirst("Authorization");
+        if (authorization == null || !authorization.startsWith("Bearer ")) {
+            return false;
+        }
+        String presented = authorization.substring("Bearer ".length()).trim();
+        return MessageDigest.isEqual(
+                metricsToken.getBytes(StandardCharsets.UTF_8), presented.getBytes(StandardCharsets.UTF_8));
+    }
+
     // ---- static + helpers --------------------------------------------------
 
     private void serveStatic(HttpExchange exchange, String path) throws IOException {

sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/Contract.java

@@ -2,10 +2,10 @@
 
 import java.util.LinkedHashMap;
 import java.util.Map;
+import org.byteveda.taskito.model.CircuitBreakerState;
 import org.byteveda.taskito.model.DeadJob;
 import org.byteveda.taskito.model.Job;
 import org.byteveda.taskito.model.QueueStats;
-import org.byteveda.taskito.model.TaskMetric;
 import org.byteveda.taskito.model.WorkerInfo;
 
 /**
@@ -58,10 +58,26 @@ static Map<String, Object> worker(WorkerInfo w) {
         m.put("hostname", w.hostname);
         m.put("pid", w.pid);
         m.put("pool_type", w.poolType);
+        m.put("threads", w.threads);
         m.put("tags", w.tags);
         return m;
     }
 
+    static Map<String, Object> circuitBreaker(CircuitBreakerState c) {
+        Map<String, Object> m = new LinkedHashMap<>();
+        m.put("task_name", c.taskName);
+        m.put("state", c.state);
+        m.put("failure_count", c.failureCount);
+        m.put("threshold", c.threshold);
+        m.put("window_ms", c.windowMs);
+        m.put("cooldown_ms", c.cooldownMs);
+        m.put("opened_at", c.openedAt);
+        m.put("last_failure_at", c.lastFailureAt);
+        m.put("half_open_max_probes", c.halfOpenMaxProbes);
+        m.put("half_open_success_rate", c.halfOpenSuccessRate);
+        return m;
+    }
+
     static Map<String, Object> dead(DeadJob d) {
         Map<String, Object> m = new LinkedHashMap<>();
         m.put("id", d.id);
@@ -75,15 +91,4 @@ static Map<String, Object> dead(DeadJob d) {
         m.put("dlq_retry_count", d.dlqRetryCount);
         return m;
     }
-
-    static Map<String, Object> metric(TaskMetric t) {
-        Map<String, Object> m = new LinkedHashMap<>();
-        m.put("task_name", t.taskName);
-        m.put("job_id", t.jobId);
-        m.put("wall_time_ns", t.wallTimeNs);
-        m.put("memory_bytes", t.memoryBytes);
-        m.put("succeeded", t.succeeded);
-        m.put("recorded_at", t.recordedAt);
-        return m;
-    }
 }

sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/CoreHandlers.java

@@ -4,6 +4,7 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 import org.byteveda.taskito.Taskito;
+import org.byteveda.taskito.dashboard.support.Http;
 import org.byteveda.taskito.model.JobFilter;
 import org.byteveda.taskito.model.JobStatus;
 
@@ -47,10 +48,10 @@ public Object listJobs(Map<String, String> query) {
             filter.task(query.get("task"));
         }
         if (query.containsKey("limit")) {
-            filter.limit(Integer.parseInt(query.get("limit")));
+            filter.limit(Http.intParam(query, "limit", 0));
         }
         if (query.containsKey("offset")) {
-            filter.offset(Integer.parseInt(query.get("offset")));
+            filter.offset(Http.intParam(query, "offset", 0));
         }
         return queue.listJobs(filter.build()).stream().map(Contract::job).collect(Collectors.toList());
     }
@@ -60,18 +61,11 @@ public Object job(String id) {
     }
 
     public Object listDead(Map<String, String> query) {
-        long limit = longParam(query, "limit", DEFAULT_LIMIT);
-        long offset = longParam(query, "offset", 0);
+        long limit = Http.longParam(query, "limit", DEFAULT_LIMIT);
+        long offset = Http.longParam(query, "offset", 0);
         return queue.listDead(limit, offset).stream().map(Contract::dead).collect(Collectors.toList());
     }
 
-    public Object listMetrics(Map<String, String> query) {
-        long since = longParam(query, "since", 0);
-        return queue.metrics(query.get("task"), since).stream()
-                .map(Contract::metric)
-                .collect(Collectors.toList());
-    }
-
     public Object listWorkers() {
         return queue.listWorkers().stream().map(Contract::worker).collect(Collectors.toList());
     }
@@ -93,9 +87,4 @@ public Object resume(String name) {
         queue.queue(name).resume();
         return Map.of("ok", true);
     }
-
-    private static long longParam(Map<String, String> query, String key, long fallback) {
-        String value = query.get(key);
-        return value == null ? fallback : Long.parseLong(value);
-    }
 }

sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/Metrics.java

@@ -0,0 +1,109 @@
+package org.byteveda.taskito.dashboard.api;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import org.byteveda.taskito.model.TaskMetric;
+
+/**
+ * Pure aggregation of raw {@link TaskMetric} rows into the dashboard's
+ * per-task summary and time-bucketed series. Durations are wall-time
+ * nanoseconds rendered as milliseconds; percentiles use the nearest-rank method.
+ */
+final class Metrics {
+    private Metrics() {}
+
+    /** Group by task → {count, success/failure counts, avg/min/max, p50/p95/p99} (all ms). */
+    static Map<String, Object> aggregateByTask(List<TaskMetric> rows) {
+        Map<String, List<TaskMetric>> byTask = new LinkedHashMap<>();
+        for (TaskMetric row : rows) {
+            byTask.computeIfAbsent(row.taskName, k -> new ArrayList<>()).add(row);
+        }
+        Map<String, Object> out = new LinkedHashMap<>();
+        byTask.forEach((task, taskRows) -> out.put(task, summarise(taskRows)));
+        return out;
+    }
+
+    /** Bucket rows by {@code recordedAt / bucketMs}, oldest first. */
+    static List<Map<String, Object>> timeseries(List<TaskMetric> rows, long bucketMs) {
+        long width = bucketMs > 0 ? bucketMs : 60_000;
+        Map<Long, List<TaskMetric>> buckets = new TreeMap<>();
+        for (TaskMetric row : rows) {
+            long start = (row.recordedAt / width) * width;
+            buckets.computeIfAbsent(start, k -> new ArrayList<>()).add(row);
+        }
+        List<Map<String, Object>> out = new ArrayList<>();
+        buckets.forEach((start, bucketRows) -> {
+            Map<String, Object> bucket = new LinkedHashMap<>();
+            bucket.put("timestamp", start);
+            bucket.put("count", bucketRows.size());
+            bucket.put("success", countSucceeded(bucketRows));
+            bucket.put("failure", bucketRows.size() - countSucceeded(bucketRows));
+            double[] sorted = sortedDurations(bucketRows);
+            bucket.put("avg_ms", average(sorted));
+            bucket.put("p50_ms", percentile(sorted, 50));
+            bucket.put("p95_ms", percentile(sorted, 95));
+            bucket.put("p99_ms", percentile(sorted, 99));
+            out.add(bucket);
+        });
+        return out;
+    }
+
+    private static Map<String, Object> summarise(List<TaskMetric> rows) {
+        double[] sorted = sortedDurations(rows);
+        int success = countSucceeded(rows);
+        Map<String, Object> m = new LinkedHashMap<>();
+        m.put("count", rows.size());
+        m.put("success_count", success);
+        m.put("failure_count", rows.size() - success);
+        m.put("avg_ms", average(sorted));
+        m.put("p50_ms", percentile(sorted, 50));
+        m.put("p95_ms", percentile(sorted, 95));
+        m.put("p99_ms", percentile(sorted, 99));
+        m.put("min_ms", sorted.length == 0 ? 0.0 : sorted[0]);
+        m.put("max_ms", sorted.length == 0 ? 0.0 : sorted[sorted.length - 1]);
+        return m;
+    }
+
+    private static int countSucceeded(List<TaskMetric> rows) {
+        int count = 0;
+        for (TaskMetric row : rows) {
+            if (row.succeeded) {
+                count++;
+            }
+        }
+        return count;
+    }
+
+    private static double[] sortedDurations(List<TaskMetric> rows) {
+        double[] durations = new double[rows.size()];
+        for (int i = 0; i < rows.size(); i++) {
+            durations[i] = rows.get(i).wallTimeNs / 1_000_000.0;
+        }
+        java.util.Arrays.sort(durations);
+        return durations;
+    }
+
+    private static double average(double[] sorted) {
+        if (sorted.length == 0) {
+            return 0.0;
+        }
+        double sum = 0;
+        for (double value : sorted) {
+            sum += value;
+        }
+        return sum / sorted.length;
+    }
+
+    /** Nearest-rank percentile: index = ceil(p/100 * n) - 1. */
+    private static double percentile(double[] sorted, int p) {
+        if (sorted.length == 0) {
+            return 0.0;
+        }
+        int index = (int) Math.ceil(p / 100.0 * sorted.length) - 1;
+        index = Math.max(0, Math.min(sorted.length - 1, index));
+        return sorted[index];
+    }
+}