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
This commit is contained in:
pulse-triage[bot]
2026-10-03 13:13:21 +01:00
parent 43b81bbb2b
commit ec5c96a290
4 changed files with 358 additions and 31 deletions
+62 -8
View File
@@ -1,7 +1,8 @@
# 🩺 Scheduler Health API
**Endpoint**: `GET /api/monitoring/scheduler/health`
**Auth**: Required (`Authorization: Bearer <token>`, `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"
```
+8 -1
View File
@@ -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.
+60 -22
View File
@@ -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`.
@@ -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"<html>login</html>", {}, {"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()