From ec5c96a290a34266c2dc55e0f0a22e502ea87384 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sat, 3 Oct 2026 13:13:21 +0100 Subject: [PATCH] Keep scheduler diagnostics private and failure-aware Use one bounded, authenticated snapshot and local queries instead of secret-bearing argv and repeated requests. Preserve configuration precedence and rollback state, and judge successful polls rather than a universal queue-depth threshold. Exercise copied commands with HTTP, transport, schema and credential controls. Contract-Neutral: Documentation and copied-command tests only; scheduler runtime and API contracts are unchanged. Change-source: pulse-maintainer --- docs/api/SCHEDULER_HEALTH.md | 70 +++++- docs/monitoring/ADAPTIVE_POLLING.md | 9 +- docs/operations/ADAPTIVE_POLLING_ROLLOUT.md | 82 +++++-- scripts/tests/test_scheduler_polling_docs.py | 228 +++++++++++++++++++ 4 files changed, 358 insertions(+), 31 deletions(-) create mode 100644 scripts/tests/test_scheduler_polling_docs.py diff --git a/docs/api/SCHEDULER_HEALTH.md b/docs/api/SCHEDULER_HEALTH.md index 28381da1c9..c25ecbb4b4 100644 --- a/docs/api/SCHEDULER_HEALTH.md +++ b/docs/api/SCHEDULER_HEALTH.md @@ -1,7 +1,8 @@ # đŸŠē Scheduler Health API **Endpoint**: `GET /api/monitoring/scheduler/health` -**Auth**: Required (`Authorization: Bearer `, `X-API-Token`, or session cookie) +**Auth**: Required. Use a signed-in Pulse browser or an API token in a private +header file; do not copy a session cookie or token into a command. Returns a real-time snapshot of the adaptive scheduler, including queue state, circuit breakers, and dead-letter tasks. @@ -64,7 +65,7 @@ Returns a real-time snapshot of the adaptive scheduler, including queue state, c "firstFailureAt": "2025-10-20T13:05:20Z" }, "breaker": { - "state": "half_open", // closed, open, half_open + "state": "half_open", "retryAt": "2025-10-20T13:06:15Z", "failureCount": 3, "since": "2025-10-20T12:58:10Z", @@ -99,20 +100,73 @@ The authoritative source for per-instance health. ## 🛠ī¸ Common Queries (jq) +For a one-off check, open `/api/monitoring/scheduler/health` in your signed-in +browser on that same Pulse instance. For curl, first prepare the private header +file from [API authentication](../API.md#api-token-recommended), using a token +with only the access needed for your monitoring task. These commands require +curl 7.76 or later and jq. They make one read-only request and keep its response +in a new private directory: + +```bash +( + set -eu + umask 077 + snapshot_dir="$(mktemp -d "${TMPDIR:-/tmp}/pulse-scheduler.XXXXXX")" + snapshot="$snapshot_dir/health.json" + status="$(curl --disable --fail-with-body --connect-timeout 5 --max-time 15 \ + --header "@$HOME/.config/pulse/api-header" --output "$snapshot" \ + --write-out '%{http_code}' http://127.0.0.1:7655/api/monitoring/scheduler/health)" + if [ "$status" != 200 ]; then + printf 'Scheduler snapshot unavailable (HTTP %s); no health conclusion.\n' "$status" >&2 + exit 1 + fi + jq -e 'type == "object" and (.updatedAt | type == "string") and + (.enabled | type == "boolean") and (.deadLetter.count | type == "number") and + (.queue.depth | type == "number") and (.instances | type == "array")' \ + "$snapshot" >/dev/null + printf 'Saved scheduler snapshot: %s\n' "$snapshot" +) +``` + +The loopback URL is only for curl running on the Pulse host. Elsewhere, use your +Pulse HTTPS URL with certificate verification enabled, never `--insecure`. +Keep `--disable` first to ignore curl configuration that could enable trace +output. HTTP errors, transport failures and unexpected response shapes stop the +recipe; none is a healthy or empty scheduler result. A 401/403 concerns access, +not polling health. Do not use verbose/trace output or share the header file. + +Use the printed filename in the following local queries. They make no further +requests. Keep the raw response private: it includes infrastructure names, +addresses and error text; redact these before sharing selected observations. +Check `enabled` first: a disabled scheduler's empty queue is not a health pass. +`updatedAt` is the snapshot time; per-instance `pollStatus.lastSuccess` records +the last successful poll. Compare freshness with the effective poll intervals, +not a universal queue-depth threshold. + +**Summarise the Snapshot:** +```bash +snapshot='/absolute/path/to/health.json' # use the printed filename +jq '{updatedAt, enabled, queueDepth: .queue.depth, deadLetterCount: .deadLetter.count, + instanceCount: (.instances | length)}' "$snapshot" +``` + **Find Failing Instances:** ```bash -curl -s http://HOST:7655/api/monitoring/scheduler/health | \ -jq '.instances[] | select(.pollStatus.consecutiveFailures > 0) | {key, failures: .pollStatus.consecutiveFailures}' +snapshot='/absolute/path/to/health.json' # use the printed filename +jq '.instances[] | select(.pollStatus.consecutiveFailures > 0) | + {key, lastSuccess: .pollStatus.lastSuccess, failures: .pollStatus.consecutiveFailures}' "$snapshot" ``` **Check Dead Letter Queue:** ```bash -curl -s http://HOST:7655/api/monitoring/scheduler/health | \ -jq '.instances[] | select(.deadLetter.present) | {key, reason: .deadLetter.reason}' +snapshot='/absolute/path/to/health.json' # use the printed filename +jq '.instances[] | select(.deadLetter.present) | + {key, reason: .deadLetter.reason}' "$snapshot" ``` **Find Open Breakers:** ```bash -curl -s http://HOST:7655/api/monitoring/scheduler/health | \ -jq '.instances[] | select(.breaker.state != "closed") | {key, state: .breaker.state}' +snapshot='/absolute/path/to/health.json' # use the printed filename +jq '.instances[] | select(.breaker.state != "closed") | + {key, state: .breaker.state}' "$snapshot" ``` diff --git a/docs/monitoring/ADAPTIVE_POLLING.md b/docs/monitoring/ADAPTIVE_POLLING.md index 126f6b6d0f..a7036ce187 100644 --- a/docs/monitoring/ADAPTIVE_POLLING.md +++ b/docs/monitoring/ADAPTIVE_POLLING.md @@ -55,7 +55,9 @@ There is currently no dedicated UI for adaptive polling. Use environment variabl You can also set `adaptivePollingEnabled` (and related interval fields) in `system.json` and restart Pulse. ## 📊 Metrics -Exposed at `:9091/metrics`. +The separate metrics listener defaults to port 9091; `METRICS_PORT` can override +it. Use an existing authorised scrape if available, not a newly exposed port for +diagnostics. A missing scrape is not evidence of zero errors. | Metric | Type | Description | | :--- | :--- | :--- | @@ -84,3 +86,8 @@ Returns: - Dead-letter tasks. - Circuit breaker states. - Per-instance staleness. + +Use the [authenticated snapshot and local queries](../api/SCHEDULER_HEALTH.md#-common-queries-jq) +to inspect it without putting credentials in command arguments or repeatedly +requesting the same data. Follow the [rollout and rollback checks](../operations/ADAPTIVE_POLLING_ROLLOUT.md) +when changing the flag; queue depth alone is not a monitoring-health verdict. diff --git a/docs/operations/ADAPTIVE_POLLING_ROLLOUT.md b/docs/operations/ADAPTIVE_POLLING_ROLLOUT.md index 0bf8a9f16f..1637e58ffc 100644 --- a/docs/operations/ADAPTIVE_POLLING_ROLLOUT.md +++ b/docs/operations/ADAPTIVE_POLLING_ROLLOUT.md @@ -1,32 +1,70 @@ # 🚀 Adaptive Polling Rollout -Safely enable dynamic scheduling. +Adaptive polling is disabled by default. Change it during a maintenance window, +not as an unmeasured remedy for high CPU, excessive writes or a stalled backup. +A Pulse restart interrupts monitoring and notifications until it is running again. ## 📋 Pre-Flight -1. **Snapshot Health**: - ```bash - curl -s -H "X-API-Token: $TOKEN" http://localhost:7655/api/monitoring/scheduler/health | jq . - ``` -2. **Check Metrics**: Ensure `pulse_monitor_poll_queue_depth` is stable. + +1. Record the running version, fleet size, current polling settings and whether + Pulse dashboards are open. Save the existing environment overrides and + `adaptivePollingEnabled` value so rollback restores the same configuration. +2. Take one [scheduler health snapshot](../api/SCHEDULER_HEALTH.md#-common-queries-jq). + Use a signed-in browser or the private header-file recipe there. Keep the + response local: instance names, addresses and error text can be sensitive. +3. Establish each instance's last successful poll and current errors before + changing settings. A small or empty queue alone does not prove healthy polls; + scheduled tasks may be waiting for their next run, and polling may be disabled. ## đŸŸĸ Enable -Choose one method: -- **UI**: Not currently exposed in the UI (use CLI or env vars). -- **CLI**: - - systemd/LXC: `jq '.adaptivePollingEnabled=true' /etc/pulse/system.json > /tmp/system.json && sudo mv /tmp/system.json /etc/pulse/system.json` - - Docker/Kubernetes: edit `/data/system.json` in the volume and restart the container/pod -- **Env**: `ADAPTIVE_POLLING_ENABLED=true` (Docker/K8s). + +There is no dedicated UI toggle. Choose one configuration source: + +- **Managed environment**: Set `ADAPTIVE_POLLING_ENABLED=true` in the existing + service, Compose or Kubernetes deployment configuration, then restart or + redeploy Pulse using that deployment's normal procedure. +- **Persistent settings**: In the instance's actual data directory, edit only + `adaptivePollingEnabled` in `system.json`, preserving its other settings, + ownership and permissions, then restart Pulse. Back up the original privately + first; this file can contain secrets. Do not replace it through a shared + `/tmp/system.json` or replace the whole document with a single property. + +An explicit environment value overrides `system.json`. Do not change both +sources at once or assume a file edit took effect while an override remains. +See [configuration](../monitoring/ADAPTIVE_POLLING.md#-configuration) for intervals. +After restart, the health response's `enabled` field confirms the effective flag, +not successful monitoring. ## 🔍 Monitor (First 15m) -Watch for stability: -```bash -watch -n 5 'curl -s http://localhost:9091/metrics | grep pulse_monitor_poll_queue_depth' -``` -- **Success**: Queue depth < 50, no permanent errors. -- **Failure**: High queue depth, open breakers. + +For a responsive install, take another snapshot after about 1, 5 and 15 minutes; +run the local jq queries against each saved file, not a separate HTTP request per +query. Stop collection if the system becomes unresponsive. Do not add an +unbounded `watch` loop, enable Debug or expose a metrics port for this check. + +Compare per-instance `pollStatus.lastSuccess`, `consecutiveFailures`, breaker +state and dead-letter tasks with the baseline. Last-success times should advance +in line with the effective polling intervals. Check that the affected dashboard +data still refreshes. A snapshot's `updatedAt` is its collection time, not a +successful provider poll. There is no fleet-independent queue-depth success +threshold, and an open breaker is a backoff state, not proof of a new regression. + +If using an already configured Prometheus scrape, compare the queue and error +metrics too. Missing metrics or a failed read are unavailable evidence, not zero +errors. Do not prolong a failure merely to finish the 15-minute observation. ## ↩ī¸ Rollback -If instability occurs > 10m: -1. **Disable**: Remove the env var override or set `adaptivePollingEnabled=false` in `system.json`. -2. **Restart**: Required if using Env/CLI overrides. -3. **Verify**: Confirm queue drains. + +If freshness deteriorates or failures rise after the change: + +1. Restore the saved environment and persistent settings. For an originally + disabled install, an explicit `ADAPTIVE_POLLING_ENABLED=false` override disables + the flag; merely removing `true` can uncover a persisted `true` value. +2. Restart or redeploy through the same procedure. This pauses monitoring again. +3. Take a fresh snapshot to confirm the effective flag, then verify successful + polls and dashboard freshness against the baseline. An empty queue or + `enabled: false` alone is not recovery evidence. + +Keep the versions, times, changed settings and relevant redacted observations +when reporting a problem. Do not share the header file, raw health response, +deployment environment or `system.json`. diff --git a/scripts/tests/test_scheduler_polling_docs.py b/scripts/tests/test_scheduler_polling_docs.py new file mode 100644 index 0000000000..6643abf5c0 --- /dev/null +++ b/scripts/tests/test_scheduler_polling_docs.py @@ -0,0 +1,228 @@ +#!/usr/bin/env python3 +"""Exercise copied scheduler diagnostics against secret-free local HTTP fixtures. + +Run the HTTP tests through pulse-worker-source-proof for guest-local loopback. +The documentation, not a second hand-written client, supplies every command. +""" + +from contextlib import contextmanager +from http.server import BaseHTTPRequestHandler, HTTPServer +import json +import os +from pathlib import Path +import re +import shutil +import stat +import subprocess +import tempfile +import threading +import unittest + + +ROOT = Path(__file__).resolve().parents[2] +DOC = ROOT / "docs/api/SCHEDULER_HEALTH.md" +ROLLOUT = ROOT / "docs/operations/ADAPTIVE_POLLING_ROLLOUT.md" +ENDPOINT = "/api/monitoring/scheduler/health" +TOKEN = "synthetic-scheduler-doc-token" + + +def blocks(): + return re.findall(r"```bash\n(.*?)```", DOC.read_text(), re.DOTALL) + + +def request(): + # This also selects the old first query for a parent-source differential. + return next(block for block in blocks() if ENDPOINT in block) + + +def snapshot(enabled=True): + return { + "updatedAt": "2026-10-03T12:00:00Z", "enabled": enabled, + "queue": {"depth": 3, "dueWithinSeconds": 1, "perType": {"pve": 3}}, + "deadLetter": {"count": 1, "tasks": []}, "breakers": [], "staleness": [], + "instances": [ + {"key": "pve::healthy", "pollStatus": {"lastSuccess": "2026-10-03T11:59:58Z", "consecutiveFailures": 0}, + "breaker": {"state": "closed"}, "deadLetter": {"present": False}}, + {"key": "pve::retrying", "pollStatus": {"lastSuccess": "2026-10-03T11:59:00Z", "consecutiveFailures": 2}, + "breaker": {"state": "half_open"}, "deadLetter": {"present": False}}, + {"key": "pve::failed", "pollStatus": {"lastSuccess": "2026-10-03T11:55:00Z", "consecutiveFailures": 5}, + "breaker": {"state": "open"}, "deadLetter": {"present": True, "reason": "permanent_failure"}}, + ], + } + + +@contextmanager +def server(payload, status=200): + calls = [] + + class Handler(BaseHTTPRequestHandler): + def do_GET(self): + calls.append({"path": self.path, "method": self.command, "headers": dict(self.headers)}) + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(payload if isinstance(payload, bytes) else json.dumps(payload).encode()) + + def log_message(self, *_args): + pass + + httpd = HTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=httpd.serve_forever, daemon=True) + thread.start() + try: + yield httpd.server_address[1], calls + finally: + httpd.shutdown() + httpd.server_close() + thread.join(timeout=5) + + +class SchedulerPollingDocsTest(unittest.TestCase): + def exercise(self, payload=None, status=200, *, header="X-API-Token", missing_header=False, + refused_connection=False): + payload = snapshot() if payload is None else payload + with tempfile.TemporaryDirectory(prefix="pulse polling ") as temporary: + directory = Path(temporary) + home = directory / "home" + credential = home / ".config/pulse/api-header" + credential.parent.mkdir(parents=True, mode=0o700) + if not missing_header: + credential.write_text(f"{header}: {'Bearer ' if header == 'Authorization' else ''}{TOKEN}\n") + credential.chmod(0o600) + # --disable must defeat unsafe user defaults, not just omit -v itself. + (home / ".curlrc").write_text("verbose\n") + tools = directory / "tools" + tools.mkdir() + curl = tools / "curl" + curl.write_text("#!/usr/bin/env python3\nimport json, os, sys\n" + "from pathlib import Path\n" + "Path(os.environ['CURL_ARGV']).write_text(json.dumps(sys.argv[1:]))\n" + "os.execv(os.environ['REAL_CURL'], ['curl', *sys.argv[1:]])\n") + curl.chmod(0o700) + argv_file = directory / "argv.json" + env = dict(os.environ, HOME=str(home), TMPDIR=str(directory), + PATH=f"{tools}:{os.environ['PATH']}", CURL_ARGV=str(argv_file), + REAL_CURL=shutil.which("curl")) + with server(payload, status) as (port, calls): + # A closed fixture port supplies an actual connection refusal. + if refused_connection: + with server(payload) as (closed_port, _): + pass + port = closed_port + code = request().replace("http://127.0.0.1:7655", f"http://127.0.0.1:{port}") + code = code.replace("http://HOST:7655", f"http://127.0.0.1:{port}") + result = subprocess.run(["bash", "-c", code], env=env, text=True, + capture_output=True, timeout=25) + saved = list(directory.glob("pulse-scheduler.*/health.json")) + files = [(path.read_bytes(), stat.S_IMODE(path.stat().st_mode), + stat.S_IMODE(path.parent.stat().st_mode)) for path in saved] + return result, calls, json.loads(argv_file.read_text()), files + + def test_one_authenticated_request_with_both_supported_header_files(self): + for header in ("X-API-Token", "Authorization"): + with self.subTest(header=header): + result, calls, args, files = self.exercise(header=header) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(len(calls), 1) + self.assertEqual((calls[0]["path"], calls[0]["method"]), (ENDPOINT, "GET")) + self.assertEqual(calls[0]["headers"][header], + ("Bearer " if header == "Authorization" else "") + TOKEN) + self.assertNotIn("Authorization" if header == "X-API-Token" else "X-API-Token", + calls[0]["headers"]) + self.assertEqual(args[0], "--disable") + self.assertNotIn(TOKEN, " ".join(args) + result.stdout + result.stderr) + self.assertIn("Saved scheduler snapshot:", result.stdout) + self.assertEqual(json.loads(files[0][0]), snapshot()) + self.assertEqual(files[0][1:], (0o600, 0o700)) + + def test_request_has_bounded_timeouts_and_never_follows_redirects(self): + _, _, args, _ = self.exercise() + self.assertEqual(args[args.index("--connect-timeout") + 1], "5") + self.assertEqual(args[args.index("--max-time") + 1], "15") + self.assertNotIn("--location", args) + self.assertNotIn("--insecure", args) + + def test_http_errors_and_redirects_never_become_successful_snapshots(self): + for status in (301, 401, 403, 503): + with self.subTest(status=status): + # Even a body shaped like health must not hide a failed request. + result, calls, _, _ = self.exercise(snapshot(), status) + self.assertNotEqual(result.returncode, 0) + self.assertEqual(len(calls), 1) + self.assertNotIn("Saved scheduler snapshot:", result.stdout) + + def test_transport_failure_is_not_an_empty_queue(self): + result, calls, _, _ = self.exercise(refused_connection=True) + self.assertNotEqual(result.returncode, 0) + self.assertEqual(calls, []) + self.assertNotIn("Saved scheduler snapshot:", result.stdout) + + def test_missing_header_file_stops_without_an_anonymous_request(self): + result, calls, _, _ = self.exercise(missing_header=True) + self.assertNotEqual(result.returncode, 0) + self.assertEqual(calls, []) + self.assertNotIn("Saved scheduler snapshot:", result.stdout) + + def test_invalid_json_and_unexpected_shapes_are_unavailable(self): + for payload in (b"login", {}, {"enabled": True}, + dict(snapshot(), enabled="true"), dict(snapshot(), instances=None)): + with self.subTest(payload=payload): + result, _, _, _ = self.exercise(payload) + self.assertNotEqual(result.returncode, 0) + self.assertNotIn("Saved scheduler snapshot:", result.stdout) + + def test_disabled_empty_scheduler_is_preserved_not_reported_as_healthy(self): + payload = dict(snapshot(False), instances=[], queue={"depth": 0}, deadLetter={"count": 0}) + result, _, _, files = self.exercise(payload) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertFalse(json.loads(files[0][0])["enabled"]) + self.assertNotIn("healthy", result.stdout.lower()) + + def test_local_queries_reuse_one_snapshot_and_preserve_failing_instances(self): + queries = [block for block in blocks() if "jq " in block and ENDPOINT not in block] + self.assertEqual(len(queries), 4) + with tempfile.TemporaryDirectory(prefix="saved scheduler ") as temporary: + path = Path(temporary) / "health.json" + path.write_text(json.dumps(snapshot())) + results = [] + for query in queries: + self.assertNotIn("curl", query) + code = query.replace("/absolute/path/to/health.json", str(path)) + result = subprocess.run(["bash", "-eu", "-c", code], capture_output=True, + text=True, timeout=5) + self.assertEqual(result.returncode, 0, result.stderr) + decoder = json.JSONDecoder() + text = result.stdout.strip() + rows = [] + while text: + row, end = decoder.raw_decode(text) + rows.append(row) + text = text[end:].strip() + results.append(rows) + self.assertEqual(results[0], [{"updatedAt": snapshot()["updatedAt"], "enabled": True, + "queueDepth": 3, "deadLetterCount": 1, "instanceCount": 3}]) + self.assertEqual([r["key"] for r in results[1]], ["pve::retrying", "pve::failed"]) + self.assertEqual(results[2], [{"key": "pve::failed", "reason": "permanent_failure"}]) + self.assertEqual(results[3], [{"key": "pve::retrying", "state": "half_open"}, + {"key": "pve::failed", "state": "open"}]) + + def test_response_example_is_valid_json_with_the_current_field_names(self): + example = json.loads(re.findall(r"```json\n(.*?)```", DOC.read_text(), re.DOTALL)[0]) + self.assertEqual(set(example), set(snapshot())) + self.assertIsInstance(example["enabled"], bool) + self.assertIn("lastSuccess", example["instances"][0]["pollStatus"]) + + def test_rollout_restores_precedence_and_does_not_rewrite_shared_temp_settings(self): + text = ROLLOUT.read_text() + self.assertNotIn("sudo mv", text) + self.assertNotIn("watch -n", text) + self.assertNotIn('X-API-Token: $TOKEN', text) + self.assertIn("environment value overrides `system.json`", text) + self.assertIn("ADAPTIVE_POLLING_ENABLED=false", text) + self.assertIn("merely removing `true`", text) + self.assertIn("interrupts monitoring and notifications", text) + self.assertIn("not a monitoring-health verdict", (ROOT / "docs/monitoring/ADAPTIVE_POLLING.md").read_text()) + + +if __name__ == "__main__": + unittest.main()