diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java index d5a5e01b..c2aa5112 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java +++ b/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 { diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/Contract.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/Contract.java index 647ec8ca..985e77e5 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/Contract.java +++ b/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 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 circuitBreaker(CircuitBreakerState c) { + Map 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 dead(DeadJob d) { Map m = new LinkedHashMap<>(); m.put("id", d.id); @@ -75,15 +91,4 @@ static Map dead(DeadJob d) { m.put("dlq_retry_count", d.dlqRetryCount); return m; } - - static Map metric(TaskMetric t) { - Map 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; - } } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/CoreHandlers.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/CoreHandlers.java index fa7ffddf..2980211a 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/CoreHandlers.java +++ b/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 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 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 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 query, String key, long fallback) { - String value = query.get(key); - return value == null ? fallback : Long.parseLong(value); - } } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/Metrics.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/Metrics.java new file mode 100644 index 00000000..5858a7bf --- /dev/null +++ b/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 aggregateByTask(List rows) { + Map> byTask = new LinkedHashMap<>(); + for (TaskMetric row : rows) { + byTask.computeIfAbsent(row.taskName, k -> new ArrayList<>()).add(row); + } + Map out = new LinkedHashMap<>(); + byTask.forEach((task, taskRows) -> out.put(task, summarise(taskRows))); + return out; + } + + /** Bucket rows by {@code recordedAt / bucketMs}, oldest first. */ + static List> timeseries(List rows, long bucketMs) { + long width = bucketMs > 0 ? bucketMs : 60_000; + Map> buckets = new TreeMap<>(); + for (TaskMetric row : rows) { + long start = (row.recordedAt / width) * width; + buckets.computeIfAbsent(start, k -> new ArrayList<>()).add(row); + } + List> out = new ArrayList<>(); + buckets.forEach((start, bucketRows) -> { + Map 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 summarise(List rows) { + double[] sorted = sortedDurations(rows); + int success = countSucceeded(rows); + Map 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 rows) { + int count = 0; + for (TaskMetric row : rows) { + if (row.succeeded) { + count++; + } + } + return count; + } + + private static double[] sortedDurations(List 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]; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/MetricsHandlers.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/MetricsHandlers.java new file mode 100644 index 00000000..a46838dd --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/MetricsHandlers.java @@ -0,0 +1,31 @@ +package org.byteveda.taskito.dashboard.api; + +import java.util.Map; +import org.byteveda.taskito.Taskito; +import org.byteveda.taskito.dashboard.support.Http; + +/** + * Aggregated metrics endpoints. {@code /api/metrics} returns the per-task + * summary the SPA expects (not raw rows); {@code /api/metrics/timeseries} + * returns time-bucketed points. Both aggregate raw {@link org.byteveda.taskito.model.TaskMetric} + * rows in pure Java via {@link Metrics}. + */ +public final class MetricsHandlers { + private static final long DEFAULT_BUCKET_MS = 60_000; + + private final Taskito queue; + + public MetricsHandlers(Taskito queue) { + this.queue = queue; + } + + public Object aggregated(Map query) { + return Metrics.aggregateByTask(queue.metrics(query.get("task"), Http.longParam(query, "since", 0))); + } + + public Object timeseries(Map query) { + long since = Http.longParam(query, "since", 0); + long bucket = Http.longParam(query, "bucket", DEFAULT_BUCKET_MS); + return Metrics.timeseries(queue.metrics(query.get("task"), since), bucket); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/OpsHandlers.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/OpsHandlers.java new file mode 100644 index 00000000..7e593cc3 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/OpsHandlers.java @@ -0,0 +1,94 @@ +package org.byteveda.taskito.dashboard.api; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +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.events.EventName; +import org.byteveda.taskito.model.QueueStats; +import org.byteveda.taskito.model.WorkerInfo; + +/** + * Operational endpoints: circuit breakers, event types, the KEDA scaler payload, + * readiness, and the Prometheus exposition. The scaler and Prometheus outputs are + * computed purely from stats + workers. + */ +public final class OpsHandlers { + private static final long DEFAULT_TARGET_QUEUE_DEPTH = 10; + + private final Taskito queue; + + public OpsHandlers(Taskito queue) { + this.queue = queue; + } + + public Object circuitBreakers() { + return queue.listCircuitBreakers().stream() + .map(Contract::circuitBreaker) + .collect(Collectors.toList()); + } + + public Object eventTypes() { + List types = new ArrayList<>(); + for (EventName name : EventName.values()) { + types.add(name.name().toLowerCase(Locale.ROOT)); + } + return types; + } + + public Object scaler(Map query) { + String queueName = query.get("queue"); + long target = Http.longParam(query, "target", DEFAULT_TARGET_QUEUE_DEPTH); + QueueStats stats = queueName == null ? queue.stats() : queue.statsByQueue(queueName); + List workers = queue.listWorkers(); + long capacity = workers.stream().mapToLong(w -> w.threads).sum(); + + Map out = new LinkedHashMap<>(); + out.put("metric_name", queueName == null ? "taskito_queue_depth" : "taskito_queue_depth_" + queueName); + out.put("metric_value", stats.pending); + out.put("target_queue_depth", target); + out.put("is_active", stats.pending > 0 || stats.running > 0); + out.put("live_workers", workers.size()); + out.put("total_capacity", capacity); + if (queueName == null) { + Map perQueue = new LinkedHashMap<>(); + queue.statsAllQueues().forEach((name, s) -> perQueue.put(name, s.pending)); + out.put("per_queue", perQueue); + } + return out; + } + + public Object readiness() { + Map checks = new LinkedHashMap<>(); + checks.put("storage", true); + checks.put("workers", queue.listWorkers().size()); + Map out = new LinkedHashMap<>(); + out.put("status", "ready"); + out.put("checks", checks); + return out; + } + + /** Prometheus text exposition of the queue's job counts and worker count. */ + public String prometheus() { + QueueStats stats = queue.stats(); + StringBuilder sb = new StringBuilder(); + gauge(sb, "taskito_jobs_pending", "Jobs waiting to run", stats.pending); + gauge(sb, "taskito_jobs_running", "Jobs currently running", stats.running); + gauge(sb, "taskito_jobs_completed", "Jobs completed", stats.completed); + gauge(sb, "taskito_jobs_failed", "Jobs failed", stats.failed); + gauge(sb, "taskito_jobs_dead", "Jobs dead-lettered", stats.dead); + gauge(sb, "taskito_jobs_cancelled", "Jobs cancelled", stats.cancelled); + gauge(sb, "taskito_workers", "Registered workers", queue.listWorkers().size()); + return sb.toString(); + } + + private static void gauge(StringBuilder sb, String name, String help, long value) { + sb.append("# HELP ").append(name).append(' ').append(help).append('\n'); + sb.append("# TYPE ").append(name).append(" gauge\n"); + sb.append(name).append(' ').append(value).append('\n'); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/OverridesHandlers.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/OverridesHandlers.java new file mode 100644 index 00000000..93c4e330 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/OverridesHandlers.java @@ -0,0 +1,80 @@ +package org.byteveda.taskito.dashboard.api; + +import java.util.ArrayList; +import java.util.Map; +import java.util.TreeSet; +import org.byteveda.taskito.Taskito; +import org.byteveda.taskito.dashboard.store.OverridesStore; +import org.byteveda.taskito.model.CircuitBreakerState; +import org.byteveda.taskito.model.TaskMetric; + +/** + * Task/queue override CRUD plus the task/queue listings the overrides UI selects + * from. The Java SDK has no client-side task/queue registry, so listings are + * derived from observable state (metrics, circuit breakers, live queues, and + * existing override rows) — a best-effort superset rather than a declared registry. + * Setting a queue's {@code paused} override also pauses/resumes it live. + */ +public final class OverridesHandlers { + private final Taskito queue; + private final OverridesStore store; + + public OverridesHandlers(Taskito queue, OverridesStore store) { + this.queue = queue; + this.store = store; + } + + public Object listTasks() { + TreeSet names = new TreeSet<>(); + for (TaskMetric metric : queue.metrics(null, 0)) { + names.add(metric.taskName); + } + for (CircuitBreakerState breaker : queue.listCircuitBreakers()) { + names.add(breaker.taskName); + } + names.addAll(store.taskNames()); + return new ArrayList<>(names); + } + + public Object listQueues() { + TreeSet names = new TreeSet<>(queue.statsAllQueues().keySet()); + names.addAll(queue.listPausedQueues()); + names.addAll(store.queueNames()); + return new ArrayList<>(names); + } + + public Object getTaskOverride(String name) { + return store.getTask(name); + } + + public Object putTaskOverride(String name, Map body) { + return store.putTask(name, body); + } + + public Object deleteTaskOverride(String name) { + return Map.of("cleared", store.deleteTask(name)); + } + + public Object getQueueOverride(String name) { + return store.getQueue(name); + } + + public Object putQueueOverride(String name, Map body) { + Map row = store.putQueue(name, body); + // Reconcile the live queue to the resulting override whenever the caller + // touches `paused` — including clearing it (which resolves to not-paused), + // where reading the request value alone would leave the queue paused. + if (body.containsKey("paused")) { + if (Boolean.TRUE.equals(row.get("paused"))) { + queue.queue(name).pause(); + } else { + queue.queue(name).resume(); + } + } + return row; + } + + public Object deleteQueueOverride(String name) { + return Map.of("cleared", store.deleteQueue(name)); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/SettingsHandlers.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/SettingsHandlers.java new file mode 100644 index 00000000..244c63a4 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/SettingsHandlers.java @@ -0,0 +1,88 @@ +package org.byteveda.taskito.dashboard.api; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.byteveda.taskito.dashboard.store.SettingsAccess; +import org.byteveda.taskito.dashboard.support.DashboardError; +import org.byteveda.taskito.dashboard.support.Json; + +/** + * Generic settings KV API. Keys under the reserved prefixes ({@code auth:}, + * {@code webhooks:}) are treated as absent everywhere — never listed, read, + * written, or deleted through this surface — so auth and webhook state cannot be + * exposed or clobbered. Keys are capped at 256 chars, values at 64 KiB. + */ +public final class SettingsHandlers { + static final int MAX_KEY_LENGTH = 256; + static final int MAX_VALUE_LENGTH = 64 * 1024; + // Hide auth state and the webhook store (persisted under the "taskito.webhooks" key). + private static final List PROTECTED_PREFIXES = List.of("auth:", "webhooks:", "taskito.webhooks"); + + private final SettingsAccess settings; + + public SettingsHandlers(SettingsAccess settings) { + this.settings = settings; + } + + public Object list() { + Map out = new LinkedHashMap<>(); + settings.listSettings().forEach((key, value) -> { + if (!isProtected(key)) { + out.put(key, value); + } + }); + return out; + } + + public Object get(String key) { + if (isProtected(key)) { + return null; // read as absent → 404 + } + return settings.getSetting(key).map(value -> entry(key, value)).orElse(null); + } + + public Object put(String key, Map body) { + validateKey(key); + Object raw = body.get("value"); + String value = raw instanceof String s ? s : Json.toString(raw); + if (value.getBytes(java.nio.charset.StandardCharsets.UTF_8).length > MAX_VALUE_LENGTH) { + throw DashboardError.badRequest("value too large"); + } + settings.setSetting(key, value); + return entry(key, value); + } + + public Object delete(String key) { + if (isProtected(key)) { + throw DashboardError.notFound("not found"); + } + return Map.of("deleted", settings.deleteSetting(key)); + } + + private static Map entry(String key, String value) { + Map m = new LinkedHashMap<>(); + m.put("key", key); + m.put("value", value); + return m; + } + + private static boolean isProtected(String key) { + return PROTECTED_PREFIXES.stream().anyMatch(key::startsWith); + } + + private static void validateKey(String key) { + if (key == null || key.isEmpty() || key.length() > MAX_KEY_LENGTH) { + throw DashboardError.badRequest("invalid setting key"); + } + for (int i = 0; i < key.length(); i++) { + char c = key.charAt(i); + if (c < 32 || c == 127) { + throw DashboardError.badRequest("invalid setting key"); + } + } + if (isProtected(key)) { + throw DashboardError.badRequest("setting key is reserved"); + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/store/OverridesStore.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/store/OverridesStore.java new file mode 100644 index 00000000..f00b5783 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/store/OverridesStore.java @@ -0,0 +1,147 @@ +package org.byteveda.taskito.dashboard.store; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.byteveda.taskito.dashboard.support.DashboardError; +import org.byteveda.taskito.dashboard.support.Json; + +/** + * Per-task and per-queue runtime overrides, persisted in the settings KV at + * {@code overrides:task:} / {@code overrides:queue:}. Rows are + * normalised (every field present, unset fields {@code null}) and stamped with + * {@code updated_at} in milliseconds. A {@code null} field in a PUT patch clears + * it; absent fields are left unchanged. + */ +public final class OverridesStore { + public static final String TASK_PREFIX = "overrides:task:"; + public static final String QUEUE_PREFIX = "overrides:queue:"; + + private static final List TASK_FIELDS = + List.of("rate_limit", "max_concurrent", "max_retries", "retry_backoff", "timeout", "priority", "paused"); + private static final List QUEUE_FIELDS = List.of("rate_limit", "max_concurrent", "paused"); + + private final SettingsAccess settings; + + public OverridesStore(SettingsAccess settings) { + this.settings = settings; + } + + public Map getTask(String name) { + return read(TASK_PREFIX + name); + } + + public Map getQueue(String name) { + return read(QUEUE_PREFIX + name); + } + + public Map putTask(String name, Map patch) { + return put(TASK_PREFIX + name, "task_name", name, TASK_FIELDS, patch); + } + + public Map putQueue(String name, Map patch) { + return put(QUEUE_PREFIX + name, "queue_name", name, QUEUE_FIELDS, patch); + } + + public boolean deleteTask(String name) { + return settings.deleteSetting(TASK_PREFIX + name); + } + + public boolean deleteQueue(String name) { + return settings.deleteSetting(QUEUE_PREFIX + name); + } + + /** Task names that carry an override row. */ + public java.util.Set taskNames() { + return names(TASK_PREFIX); + } + + /** Queue names that carry an override row. */ + public java.util.Set queueNames() { + return names(QUEUE_PREFIX); + } + + private java.util.Set names(String prefix) { + java.util.Set out = new java.util.TreeSet<>(); + for (String key : settings.listSettings().keySet()) { + if (key.startsWith(prefix)) { + out.add(key.substring(prefix.length())); + } + } + return out; + } + + private Map read(String key) { + return settings.getSetting(key).map(Json::parseMap).orElse(null); + } + + private Map put( + String key, String nameKey, String name, List fields, Map patch) { + Map values = new LinkedHashMap<>(); + Map existing = read(key); + if (existing != null) { + for (String field : fields) { + if (existing.get(field) != null) { + values.put(field, existing.get(field)); + } + } + } + for (Map.Entry entry : patch.entrySet()) { + String field = entry.getKey(); + if (!fields.contains(field)) { + throw DashboardError.badRequest("unknown override field: " + field); + } + Object value = entry.getValue(); + validate(field, value); + if (value == null) { + values.remove(field); + } else { + values.put(field, value); + } + } + Map row = new LinkedHashMap<>(); + row.put(nameKey, name); + for (String field : fields) { + row.put(field, field.equals("paused") ? Boolean.TRUE.equals(values.get("paused")) : values.get(field)); + } + row.put("updated_at", System.currentTimeMillis()); + settings.setSetting(key, Json.toString(row)); + return row; + } + + private static void validate(String field, Object value) { + if (value == null) { + return; // clears the field + } + switch (field) { + case "rate_limit" -> { + if (!(value instanceof String s) || s.isEmpty() || !s.contains("/")) { + throw DashboardError.badRequest("rate_limit must look like '/'"); + } + } + case "max_concurrent", "max_retries" -> requireIntegral(field, value, 0); + case "timeout" -> requireIntegral(field, value, 1); + case "priority" -> requireIntegral(field, value, Long.MIN_VALUE); + case "retry_backoff" -> { + if (!(value instanceof Number n) || n.doubleValue() < 0) { + throw DashboardError.badRequest("retry_backoff must be a non-negative number"); + } + } + case "paused" -> { + if (!(value instanceof Boolean)) { + throw DashboardError.badRequest("paused must be a boolean"); + } + } + default -> throw DashboardError.badRequest("unknown override field: " + field); + } + } + + private static void requireIntegral(String field, Object value, long min) { + if (!(value instanceof Number number) || number.doubleValue() != Math.floor(number.doubleValue())) { + throw DashboardError.badRequest(field + " must be an integer"); + } + if (number.longValue() < min) { + throw DashboardError.badRequest(field + " must be >= " + min); + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/support/Http.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/support/Http.java index bf1bd310..134f2c92 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/support/Http.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/support/Http.java @@ -70,4 +70,30 @@ public static void respondError(HttpExchange exchange, int status, String code) public static Map errorBody(String code) { return Collections.singletonMap("error", code); } + + /** Parse a numeric query param; {@code fallback} when absent, 400 when malformed. */ + public static long longParam(Map query, String key, long fallback) { + String value = query.get(key); + if (value == null) { + return fallback; + } + try { + return Long.parseLong(value); + } catch (NumberFormatException e) { + throw DashboardError.badRequest(key + " must be a number"); + } + } + + /** Parse an integer query param; {@code fallback} when absent, 400 when malformed/overflowing. */ + public static int intParam(Map query, String key, int fallback) { + String value = query.get(key); + if (value == null) { + return fallback; + } + try { + return Integer.parseInt(value); + } catch (NumberFormatException e) { + throw DashboardError.badRequest(key + " must be a number"); + } + } } diff --git a/sdks/java/src/test/java/org/byteveda/taskito/dashboard/DashboardClient.java b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/DashboardClient.java index 81eb66ac..ba481926 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/dashboard/DashboardClient.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/DashboardClient.java @@ -41,6 +41,14 @@ HttpResponse post(String path, String json) throws Exception { return body("POST", path, json, true); } + HttpResponse put(String path, String json) throws Exception { + return body("PUT", path, json, true); + } + + HttpResponse delete(String path) throws Exception { + return body("DELETE", path, null, true); + } + HttpResponse postWithoutCsrf(String path, String json) throws Exception { return body("POST", path, json, false); } diff --git a/sdks/java/src/test/java/org/byteveda/taskito/dashboard/DashboardOpsTest.java b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/DashboardOpsTest.java new file mode 100644 index 00000000..99b99503 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/DashboardOpsTest.java @@ -0,0 +1,127 @@ +package org.byteveda.taskito.dashboard; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.file.Path; +import java.util.Collections; +import java.util.Map; +import org.byteveda.taskito.Taskito; +import org.byteveda.taskito.task.EnqueueOptions; +import org.byteveda.taskito.task.Task; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +/** Coverage of the Phase B ops endpoints: settings, overrides, metrics, scaler, probes. */ +@Timeout(30) +class DashboardOpsTest { + + @SuppressWarnings("unchecked") + private static Class> mapType() { + return (Class>) (Class) Map.class; + } + + private static Taskito seededQueue(Path dir) { + Taskito queue = Taskito.builder().sqlite(dir.resolve("t.db").toString()).open(); + Task> add = Task.of("email.send", mapType()); + queue.enqueue( + add, + Collections.singletonMap("a", 1), + EnqueueOptions.builder().queue("emails").build()); + return queue; + } + + private static HttpResponse probe(int port, String path) throws Exception { + return HttpClient.newHttpClient() + .send( + HttpRequest.newBuilder(URI.create("http://localhost:" + port + path)) + .GET() + .build(), + HttpResponse.BodyHandlers.ofString()); + } + + @Test + void settingsCrud(@TempDir Path dir) throws Exception { + try (Taskito queue = seededQueue(dir); + DashboardServer server = DashboardServer.start(queue, 0)) { + DashboardClient client = new DashboardClient(server.port()).as(DashboardClient.seedAdmin(queue)); + assertEquals( + 200, + client.put("/api/settings/theme", "{\"value\":\"dark\"}").statusCode()); + assertTrue(client.get("/api/settings/theme").body().contains("\"value\":\"dark\"")); + assertTrue(client.get("/api/settings").body().contains("theme")); + assertTrue(client.delete("/api/settings/theme").body().contains("\"deleted\":true")); + assertEquals( + 400, client.put("/api/settings/auth:x", "{\"value\":\"1\"}").statusCode()); + } + } + + @Test + void taskAndQueueOverrides(@TempDir Path dir) throws Exception { + try (Taskito queue = seededQueue(dir); + DashboardServer server = DashboardServer.start(queue, 0)) { + DashboardClient client = new DashboardClient(server.port()).as(DashboardClient.seedAdmin(queue)); + assertEquals( + 200, + client.put("/api/tasks/email.send/override", "{\"max_concurrent\":3}") + .statusCode()); + assertTrue(client.get("/api/tasks/email.send/override").body().contains("\"max_concurrent\":3")); + assertTrue(client.get("/api/tasks").body().contains("email.send")); + assertTrue(client.delete("/api/tasks/email.send/override").body().contains("\"cleared\":true")); + + // Queue override with paused=true pauses the queue live. + client.put("/api/queues/emails/override", "{\"paused\":true}"); + assertTrue(client.get("/api/queues/paused").body().contains("emails")); + } + } + + @Test + void aggregatedMetricsAndOps(@TempDir Path dir) throws Exception { + try (Taskito queue = seededQueue(dir); + DashboardServer server = DashboardServer.start(queue, 0)) { + DashboardClient client = new DashboardClient(server.port()).as(DashboardClient.seedAdmin(queue)); + // No metrics recorded yet → empty aggregate object / timeseries array. + assertEquals("{}", client.get("/api/metrics").body()); + assertEquals("[]", client.get("/api/metrics/timeseries").body()); + assertEquals("[]", client.get("/api/circuit-breakers").body()); + + String eventTypes = client.get("/api/event-types").body(); + assertTrue(eventTypes.contains("success") && eventTypes.contains("dead")); + + String scaler = client.get("/api/scaler").body(); + assertTrue(scaler.contains("\"metric_name\":\"taskito_queue_depth\"")); + assertTrue(scaler.contains("\"metric_value\":1")); + } + } + + @Test + void probesAreServed(@TempDir Path dir) throws Exception { + try (Taskito queue = seededQueue(dir); + DashboardServer server = DashboardServer.start(queue, 0)) { + int port = server.port(); + assertTrue(probe(port, "/health").body().contains("\"status\":\"ok\"")); + assertTrue(probe(port, "/readiness").body().contains("\"status\":\"ready\"")); + HttpResponse metrics = probe(port, "/metrics"); + assertEquals(200, metrics.statusCode()); + assertTrue(metrics.body().contains("taskito_jobs_pending 1")); + assertTrue(metrics.headers().firstValue("content-type").orElse("").startsWith("text/plain")); + } + } + + @Test + void malformedNumericParamsReturn400(@TempDir Path dir) throws Exception { + try (Taskito queue = seededQueue(dir); + DashboardServer server = DashboardServer.start(queue, 0)) { + DashboardClient client = new DashboardClient(server.port()).as(DashboardClient.seedAdmin(queue)); + assertEquals(400, client.get("/api/metrics?since=abc").statusCode()); + assertEquals(400, client.get("/api/dead-letters?limit=xyz").statusCode()); + assertEquals(400, client.get("/api/scaler?target=nope").statusCode()); + assertEquals(400, client.get("/api/jobs?limit=notanint").statusCode()); + } + } +} diff --git a/sdks/java/src/test/java/org/byteveda/taskito/dashboard/InMemorySettings.java b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/InMemorySettings.java new file mode 100644 index 00000000..e1c415e7 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/InMemorySettings.java @@ -0,0 +1,31 @@ +package org.byteveda.taskito.dashboard; + +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import org.byteveda.taskito.dashboard.store.SettingsAccess; + +/** In-memory {@link SettingsAccess} for unit tests. */ +public final class InMemorySettings implements SettingsAccess { + private final Map map = new HashMap<>(); + + @Override + public Optional getSetting(String key) { + return Optional.ofNullable(map.get(key)); + } + + @Override + public void setSetting(String key, String value) { + map.put(key, value); + } + + @Override + public boolean deleteSetting(String key) { + return map.remove(key) != null; + } + + @Override + public Map listSettings() { + return new HashMap<>(map); + } +} diff --git a/sdks/java/src/test/java/org/byteveda/taskito/dashboard/api/SettingsHandlersTest.java b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/api/SettingsHandlersTest.java new file mode 100644 index 00000000..c1bf4ec3 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/api/SettingsHandlersTest.java @@ -0,0 +1,69 @@ +package org.byteveda.taskito.dashboard.api; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.Map; +import org.byteveda.taskito.dashboard.InMemorySettings; +import org.byteveda.taskito.dashboard.support.DashboardError; +import org.junit.jupiter.api.Test; + +class SettingsHandlersTest { + + private final InMemorySettings settings = new InMemorySettings(); + private final SettingsHandlers handlers = new SettingsHandlers(settings); + + @SuppressWarnings("unchecked") + private static Map asMap(Object value) { + return (Map) value; + } + + @Test + void putAndGetRoundTrip() { + assertEquals("v", asMap(handlers.put("k", Map.of("value", "v"))).get("value")); + assertEquals("v", asMap(handlers.get("k")).get("value")); + assertEquals("k", asMap(handlers.get("k")).get("key")); + } + + @Test + void nonStringValueIsJsonEncoded() { + handlers.put("cfg", Map.of("value", Map.of("a", 1))); + assertEquals("{\"a\":1}", asMap(handlers.get("cfg")).get("value")); + } + + @Test + void protectedKeysAreInvisible() { + settings.setSetting("auth:users", "{}"); + settings.setSetting("webhooks:subscriptions", "[]"); + handlers.put("visible", Map.of("value", "1")); + Map listed = asMap(handlers.list()); + assertTrue(listed.containsKey("visible")); + assertTrue(!listed.containsKey("auth:users")); + assertTrue(!listed.containsKey("webhooks:subscriptions")); + assertNull(handlers.get("auth:users")); + assertThrows(DashboardError.class, () -> handlers.put("auth:x", Map.of("value", "1"))); + assertThrows(DashboardError.class, () -> handlers.delete("webhooks:subscriptions")); + } + + @Test + void deleteReportsExistence() { + handlers.put("k", Map.of("value", "v")); + assertEquals(true, asMap(handlers.delete("k")).get("deleted")); + assertEquals(false, asMap(handlers.delete("k")).get("deleted")); + } + + @Test + void rejectsInvalidKeys() { + assertThrows(DashboardError.class, () -> handlers.put("", Map.of("value", "v"))); + assertThrows(DashboardError.class, () -> handlers.put("a".repeat(257), Map.of("value", "v"))); + assertThrows(DashboardError.class, () -> handlers.put("bad\nkey", Map.of("value", "v"))); + } + + @Test + void rejectsOversizedValue() { + String big = "x".repeat(64 * 1024 + 1); + assertThrows(DashboardError.class, () -> handlers.put("k", Map.of("value", big))); + } +} diff --git a/sdks/java/src/test/java/org/byteveda/taskito/dashboard/auth/AuthStoreTest.java b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/auth/AuthStoreTest.java index 7c262a68..da52ee6f 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/dashboard/auth/AuthStoreTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/auth/AuthStoreTest.java @@ -7,42 +7,16 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; -import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Optional; -import org.byteveda.taskito.dashboard.store.SettingsAccess; +import org.byteveda.taskito.dashboard.InMemorySettings; import org.byteveda.taskito.dashboard.support.DashboardError; import org.junit.jupiter.api.Test; class AuthStoreTest { - private static final class MemSettings implements SettingsAccess { - private final Map map = new HashMap<>(); - - @Override - public Optional getSetting(String key) { - return Optional.ofNullable(map.get(key)); - } - - @Override - public void setSetting(String key, String value) { - map.put(key, value); - } - - @Override - public boolean deleteSetting(String key) { - return map.remove(key) != null; - } - - @Override - public Map listSettings() { - return new HashMap<>(map); - } - } - private AuthStore store() { - return new AuthStore(new MemSettings()); + return new AuthStore(new InMemorySettings()); } @Test diff --git a/sdks/java/src/test/java/org/byteveda/taskito/dashboard/store/OverridesStoreTest.java b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/store/OverridesStoreTest.java new file mode 100644 index 00000000..8577628e --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/store/OverridesStoreTest.java @@ -0,0 +1,79 @@ +package org.byteveda.taskito.dashboard.store; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.HashMap; +import java.util.Map; +import org.byteveda.taskito.dashboard.InMemorySettings; +import org.byteveda.taskito.dashboard.support.DashboardError; +import org.junit.jupiter.api.Test; + +class OverridesStoreTest { + + private final OverridesStore store = new OverridesStore(new InMemorySettings()); + + private static Map patch(String key, Object value) { + Map m = new HashMap<>(); + m.put(key, value); + return m; + } + + @Test + void normalisesTaskRow() { + Map row = store.putTask("email.send", patch("max_concurrent", 5)); + assertEquals("email.send", row.get("task_name")); + assertEquals(5, ((Number) row.get("max_concurrent")).intValue()); + assertEquals(false, row.get("paused")); + assertNull(row.get("timeout")); + assertTrue(row.containsKey("updated_at")); + } + + @Test + void mergesAndClearsFields() { + store.putTask("t", patch("max_concurrent", 5)); + Map merged = store.putTask("t", patch("paused", true)); + assertEquals(5, ((Number) merged.get("max_concurrent")).intValue()); + assertEquals(true, merged.get("paused")); + + Map cleared = store.putTask("t", patch("max_concurrent", null)); + assertNull(cleared.get("max_concurrent")); + assertEquals(true, cleared.get("paused")); + } + + @Test + void validatesTaskFields() { + assertThrows(DashboardError.class, () -> store.putTask("t", patch("rate_limit", "100"))); + assertThrows(DashboardError.class, () -> store.putTask("t", patch("max_concurrent", -1))); + assertThrows(DashboardError.class, () -> store.putTask("t", patch("timeout", 0))); + assertThrows(DashboardError.class, () -> store.putTask("t", patch("retry_backoff", -1))); + assertThrows(DashboardError.class, () -> store.putTask("t", patch("priority", 1.5))); + assertThrows(DashboardError.class, () -> store.putTask("t", patch("paused", "yes"))); + assertThrows(DashboardError.class, () -> store.putTask("t", patch("bogus", 1))); + } + + @Test + void acceptsValidRateLimit() { + Map row = store.putTask("t", patch("rate_limit", "100/minute")); + assertEquals("100/minute", row.get("rate_limit")); + } + + @Test + void queueRejectsTaskOnlyFields() { + assertThrows(DashboardError.class, () -> store.putQueue("q", patch("max_retries", 3))); + Map row = store.putQueue("q", patch("max_concurrent", 2)); + assertEquals("q", row.get("queue_name")); + assertEquals(2, ((Number) row.get("max_concurrent")).intValue()); + } + + @Test + void deleteAndGet() { + assertNull(store.getTask("t")); + store.putTask("t", patch("max_concurrent", 1)); + assertTrue(store.deleteTask("t")); + assertFalse(store.deleteTask("t")); + } +}