diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..d5165da --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,22 @@ +name: Tests + +on: + pull_request: + push: + branches: [main] + +permissions: + contents: read + +jobs: + test: + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@v6 + - uses: actions/setup-python@v6 + with: + python-version: '3.12' + - run: python -m pip install '.[dev]' build + - run: PYTHONPATH=src python -m pytest + - run: python -m build diff --git a/README.md b/README.md index 3e08f3d..d4e7897 100644 --- a/README.md +++ b/README.md @@ -252,7 +252,7 @@ baton taint clear # remove all canary data ### Service Logs ```bash -baton logs [--node N] [--level L] # show captured service logs +baton logs [-f | --follow] [--node N] [--level L] # show or follow captured service logs ``` ### Arbiter & Constrain diff --git a/docs/index.html b/docs/index.html index 2855700..d61e201 100644 --- a/docs/index.html +++ b/docs/index.html @@ -444,7 +444,7 @@

Five minutes to a running circuit

baton test --canary --tiers PII --duration 5m # Service logs and Arbiter trust -baton logs --node api --level error +baton logs --follow --node api --level error baton trust api # Build images and deploy to GCP Cloud Run @@ -764,7 +764,7 @@

baton metrics --prometheus # Prometheus text exposition format baton signals [--node N] [--path P] # recent request signals baton signals --stats # per-path statistics -baton logs [--node N] [--level L] # captured service logs +baton logs [-f | --follow] [--node N] [--level L] # show or follow captured service logs

Taint Analysis

diff --git a/pyproject.toml b/pyproject.toml index 8d6f010..44b2b1f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,7 +25,7 @@ dependencies = [ dev = [ "pytest>=8.0", "pytest-asyncio>=0.23", - "mcp[cli]>=1.26.0", + "mcp[cli]>=1.26.0,<2", ] gcp = [ "google-cloud-run>=0.10", @@ -36,7 +36,7 @@ aws = [ ] otel = [] mcp = [ - "mcp[cli]>=1.26.0", + "mcp[cli]>=1.26.0,<2", ] certs = [ "cryptography>=41.0", diff --git a/src/baton/cli.py b/src/baton/cli.py index cf9288b..3d0493a 100644 --- a/src/baton/cli.py +++ b/src/baton/cli.py @@ -378,6 +378,7 @@ def main(argv: list[str] | None = None) -> int: p_logs.add_argument("--node", default="", help="Filter by node") p_logs.add_argument("--level", default="", help="Minimum severity level (debug/info/warning/error/critical)") p_logs.add_argument("--last", type=int, default=50, help="Number of entries (default: 50)") + p_logs.add_argument("-f", "--follow", action="store_true", help="Follow new log entries") p_logs.add_argument("--dir", default=".", help="Project directory") p_logs.set_defaults(func=_cmd_logs) @@ -1514,6 +1515,17 @@ def _cmd_dora(args: argparse.Namespace) -> int: def _cmd_logs(args: argparse.Namespace) -> int: from baton.service_log import ServiceLogCollector + if getattr(args, "follow", False): + records = ServiceLogCollector.follow_history( + args.dir, + node=args.node or None, + severity=args.level or None, + last_n=args.last, + ) + for record in records: + _print_service_log(record) + return 0 + records = ServiceLogCollector.load_history( args.dir, node=args.node or None, @@ -1524,17 +1536,22 @@ def _cmd_logs(args: argparse.Namespace) -> int: print("No service logs found") return 0 - for r in records: - sev = r.get("severity", "info").upper() - node = r.get("node_name", "?") - ts = r.get("timestamp", "")[:19] # Trim to seconds - stream = r.get("stream", "") - msg = r.get("message", "") - print(f"[{ts}] [{sev:<8}] [{node}:{stream}] {msg}") + for record in records: + _print_service_log(record) return 0 +def _print_service_log(record: dict) -> None: + """Print one structured service log record.""" + sev = record.get("severity", "info").upper() + node = record.get("node_name", "?") + ts = record.get("timestamp", "")[:19] # Trim to seconds + stream = record.get("stream", "") + msg = record.get("message", "") + print(f"[{ts}] [{sev:<8}] [{node}:{stream}] {msg}", flush=True) + + async def _cmd_dashboard(args: argparse.Namespace) -> int: import json as json_mod from baton.dashboard import collect, format_table diff --git a/src/baton/service_log.py b/src/baton/service_log.py index 021d68c..0e47425 100644 --- a/src/baton/service_log.py +++ b/src/baton/service_log.py @@ -7,11 +7,14 @@ from __future__ import annotations +import json import re +import time +from collections.abc import Iterator from datetime import datetime, timezone from pathlib import Path -from baton.state import append_jsonl, read_jsonl +from baton.state import BATON_DIR, append_jsonl, read_jsonl LOGS_FILE = "service_logs.jsonl" @@ -40,6 +43,20 @@ def parse_severity(line: str) -> str: return "" # caller decides default based on stream +def _matches_filters( + record: dict, + node: str | None = None, + severity: str | None = None, +) -> bool: + """Return whether a structured log record matches CLI filters.""" + if node and record.get("node_name") != node: + return False + if severity and severity in SEVERITIES: + minimum = SEVERITIES.index(severity) + return SEVERITIES.index(record.get("severity", "info")) >= minimum + return True + + class ServiceLogCollector: """Captures and structures service log output. @@ -113,12 +130,55 @@ def load_history( ) -> list[dict]: """Read from .baton/service_logs.jsonl.""" records = read_jsonl(project_dir, LOGS_FILE, last_n=last_n) - if node: - records = [r for r in records if r.get("node_name") == node] - if severity and severity in SEVERITIES: - sev_idx = SEVERITIES.index(severity) - records = [ - r for r in records - if SEVERITIES.index(r.get("severity", "info")) >= sev_idx - ] - return records + return [ + record for record in records + if _matches_filters(record, node=node, severity=severity) + ] + + @staticmethod + def follow_history( + project_dir: str | Path, + node: str | None = None, + severity: str | None = None, + last_n: int | None = 50, + poll_interval: float = 0.2, + ) -> Iterator[dict]: + """Yield recent records, then wait for and yield newly appended records.""" + path = Path(project_dir) / BATON_DIR / LOGS_FILE + while not path.exists(): + time.sleep(poll_interval) + + with open(path) as log_file: + history: list[str] = [] + while True: + offset = log_file.tell() + line = log_file.readline() + if not line: + break + if not line.endswith("\n"): + log_file.seek(offset) + break + history.append(line) + + if last_n is not None: + history = history[-last_n:] + for line in history: + if not line.strip(): + continue + record = json.loads(line) + if _matches_filters(record, node=node, severity=severity): + yield record + + while True: + offset = log_file.tell() + line = log_file.readline() + if not line or not line.endswith("\n"): + log_file.seek(offset) + time.sleep(poll_interval) + continue + + if not line.strip(): + continue + record = json.loads(line) + if _matches_filters(record, node=node, severity=severity): + yield record diff --git a/tests/test_cli.py b/tests/test_cli.py index 448f2fc..5e578a6 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -633,6 +633,57 @@ def test_signals_path_filter(self, project_dir: Path, capsys): assert "/users" not in out +class TestLogsCLI: + @pytest.mark.parametrize("follow_flag", ["-f", "--follow"]) + def test_follow_streams_records_with_filters( + self, + project_dir: Path, + monkeypatch, + capsys, + follow_flag: str, + ): + from baton.service_log import ServiceLogCollector + + captured = {} + + def follow_history(project_dir, *, node, severity, last_n): + captured.update( + project_dir=project_dir, + node=node, + severity=severity, + last_n=last_n, + ) + return iter([ + { + "node_name": "api", + "stream": "stderr", + "severity": "error", + "message": "request failed", + "timestamp": "2026-01-01T00:00:00+00:00", + } + ]) + + monkeypatch.setattr(ServiceLogCollector, "follow_history", follow_history) + + rc = main([ + "logs", + follow_flag, + "--node", "api", + "--level", "warning", + "--last", "7", + "--dir", str(project_dir), + ]) + + assert rc == 0 + assert captured == { + "project_dir": str(project_dir), + "node": "api", + "severity": "warning", + "last_n": 7, + } + assert "[ERROR ] [api:stderr] request failed" in capsys.readouterr().out + + class TestMetricsCLI: def _init_with_metrics(self, d: Path) -> Path: """Initialize a project and write telemetry data.""" diff --git a/tests/test_service_log.py b/tests/test_service_log.py new file mode 100644 index 0000000..3cee006 --- /dev/null +++ b/tests/test_service_log.py @@ -0,0 +1,119 @@ +"""Tests for service log capture and following.""" + +from __future__ import annotations + +import json + +from baton import service_log as service_log_module +from baton.service_log import ServiceLogCollector + + +def test_follow_history_yields_existing_and_appended_records(tmp_path): + collector = ServiceLogCollector(tmp_path) + collector.handler("api", "stdout", "service ready") + + records = ServiceLogCollector.follow_history(tmp_path, poll_interval=0) + first = next(records) + + collector.handler("api", "stderr", "request failed") + second = next(records) + records.close() + + assert first["message"] == "service ready" + assert second["message"] == "request failed" + + +def test_follow_history_applies_node_and_severity_filters(tmp_path): + collector = ServiceLogCollector(tmp_path) + collector.handler("api", "stdout", "service ready") + collector.handler("api", "stderr", "request failed") + collector.handler("worker", "stderr", "worker failed") + + records = ServiceLogCollector.follow_history( + tmp_path, + node="api", + severity="warning", + poll_interval=0, + ) + record = next(records) + records.close() + + assert record["node_name"] == "api" + assert record["severity"] == "error" + assert record["message"] == "request failed" + + +def test_follow_history_waits_for_log_file(tmp_path, monkeypatch): + collector = ServiceLogCollector(tmp_path) + + def create_log_file(_interval): + collector.handler("api", "stdout", "first record") + + monkeypatch.setattr(service_log_module.time, "sleep", create_log_file) + + records = ServiceLogCollector.follow_history(tmp_path) + record = next(records) + records.close() + + assert record["message"] == "first record" + + +def test_follow_history_limits_history_and_keeps_appends(tmp_path): + collector = ServiceLogCollector(tmp_path) + for message in ("old", "recent", "newest"): + collector.handler("api", "stdout", message) + + records = ServiceLogCollector.follow_history(tmp_path, last_n=2) + assert next(records)["message"] == "recent" + # Append while the caller is still consuming the initial snapshot. + collector.handler("api", "stdout", "appended") + assert next(records)["message"] == "newest" + assert next(records)["message"] == "appended" + records.close() + + +def test_follow_history_waits_for_complete_record(tmp_path, monkeypatch): + collector = ServiceLogCollector(tmp_path) + collector.handler("api", "stdout", "initial") + path = tmp_path / ".baton" / "service_logs.jsonl" + payload = json.dumps({"message": "completed"}) + with path.open("a") as stream: + stream.write(payload[:8]) + + def finish_record(_interval): + with path.open("a") as stream: + stream.write(payload[8:] + "\n") + + monkeypatch.setattr(service_log_module.time, "sleep", finish_record) + records = ServiceLogCollector.follow_history(tmp_path) + assert next(records)["message"] == "initial" + assert next(records)["message"] == "completed" + records.close() + + +def test_follow_history_skips_blank_history_and_appended_lines(tmp_path): + collector = ServiceLogCollector(tmp_path) + collector.handler("api", "stdout", "initial") + path = tmp_path / ".baton" / "service_logs.jsonl" + with path.open("a") as stream: + stream.write("\n \n") + + records = ServiceLogCollector.follow_history(tmp_path) + assert next(records)["message"] == "initial" + with path.open("a") as stream: + stream.write("\n") + collector.handler("api", "stdout", "appended") + assert next(records)["message"] == "appended" + records.close() + + +def test_follow_history_filters_new_records(tmp_path): + collector = ServiceLogCollector(tmp_path) + collector.handler("api", "stderr", "initial") + records = ServiceLogCollector.follow_history(tmp_path, node="api", severity="error") + assert next(records)["message"] == "initial" + collector.handler("worker", "stderr", "wrong node") + collector.handler("api", "stdout", "wrong level") + collector.handler("api", "stderr", "matching") + assert next(records)["message"] == "matching" + records.close()