Improve tool compatibility with K8S and Slurm

README.md

@@ -120,18 +120,19 @@ Interactive docs render at [http://127.0.0.1:8000/docs](http://127.0.0.1:8000/do
 
 ### Endpoints
 
-All endpoints accept repeated `sources` params and an optional `level`.
+The `/stats/*` endpoints accept repeated `sources` params and an optional `level`.
 
 | Endpoint | Returns |
 | --- | --- |
 | `GET /stats/top` | Top-N message counts (JSON) |
 | `GET /stats/per-hour` | Per-hour entry counts (JSON) |
 | `GET /stats/regular` | Raw entries as a Server-Sent Events stream |
+| `GET /health` | `{"status": "ok"}` — liveness check, takes no params |
 
 | Query param | Applies to | Description |
 | --- | --- | --- |
-| `sources` | all | Registered source name; repeat for multiple |
-| `level` | all | `DEBUG`/`INFO`/`WARNING`/`ERROR` (case-insensitive) |
+| `sources` | all `/stats/*` | Registered source name; repeat for multiple |
+| `level` | all `/stats/*` | `DEBUG`/`INFO`/`WARNING`/`ERROR` (case-insensitive) |
 | `top` | `/stats/top` | Number of messages to return (≥ 1, default 3) |
 | `combined` | `/stats/top`, `/stats/per-hour` | Merge all sources into one result (default `false`) |
 

Tests/Performance/concurency_test.py

@@ -3,7 +3,7 @@
 
 import httpx
 
-from logstats.parser import collect
+from logstats.fetcher import collect
 from logstats.source_parser import FetchedSource
 
 

Tests/api_test.py

@@ -41,6 +41,12 @@ def api():
     app.dependency_overrides.clear()
 
 
+def test_health_reports_ok(api):
+    response = api.get("/health")
+    assert response.status_code == 200
+    assert response.json() == {"status": "ok"}
+
+
 def test_app_boots_and_serves(tmp_path, monkeypatch):
     log = tmp_path / "app.log"
     log.write_text(APP1_LOGS)

Tests/cli_test.py

@@ -3,7 +3,7 @@
 import pytest
 from click.testing import CliRunner
 
-from logstats.cli import parse_aguments
+from logstats.cli import parse_arguments
 
 
 @pytest.fixture
@@ -15,7 +15,7 @@ def sample_log(tmp_path):
 
 def test_cli_doesnt_allow_multiple_reports(sample_log):
     result = CliRunner().invoke(
-        parse_aguments, [str(sample_log), "--top", "1", "--per-hour"]
+        parse_arguments, [str(sample_log), "--top", "1", "--per-hour"]
     )
     assert result.exit_code != 0
     assert "can't be used together" in result.output
@@ -24,13 +24,13 @@ def test_cli_doesnt_allow_multiple_reports(sample_log):
 def test_cli_no_logs_available(sample_log, caplog):
     with caplog.at_level(logging.INFO):
         result = CliRunner().invoke(
-            parse_aguments, [str(sample_log), "--level", "WARNING"]
+            parse_arguments, [str(sample_log), "--level", "WARNING"]
         )
     assert result.exit_code == 0
     assert "No logs" in caplog.text
 
 
 def test_cli_end_to_end(sample_log):
-    result = CliRunner().invoke(parse_aguments, [str(sample_log), "--top", "1"])
+    result = CliRunner().invoke(parse_arguments, [str(sample_log), "--top", "1"])
     assert result.exit_code == 0
     assert "Message" in result.output

Tests/config_test.py

@@ -1,9 +1,14 @@
-import tomllib
+import re
+from pathlib import Path
 
 import pytest
-from pydantic import ValidationError
 
-from logstats.config import load_settings
+from logstats.config import (
+    DEFAULT_SOURCES,
+    ConfigError,
+    load_settings,
+    resolve_sources_path,
+)
 
 
 def write_toml(tmp_path, content: str):
@@ -12,6 +17,24 @@ def write_toml(tmp_path, content: str):
     return path
 
 
+def test_relative_path_becomes_absolute(tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    assert resolve_sources_path(Path("custom.toml")) == tmp_path / "custom.toml"
+
+
+def test_falls_back_to_default_name_without_env_var(tmp_path, monkeypatch):
+    monkeypatch.delenv("LOGSTATS_SOURCES", raising=False)
+    monkeypatch.chdir(tmp_path)
+    assert resolve_sources_path() == tmp_path / DEFAULT_SOURCES
+
+
+def test_symlinks_are_not_followed(tmp_path):
+    target = write_toml(tmp_path, '[sources]\napp1 = "sample.log"\n')
+    link = tmp_path / "link.toml"
+    link.symlink_to(target)
+    assert resolve_sources_path(link) == link
+
+
 def test_loads_named_sources(tmp_path):
     path = write_toml(tmp_path, '[sources]\napp1 = "http://logs.test/app1.log"\n')
     settings = load_settings(path)
@@ -25,16 +48,22 @@ def test_path_defaults_to_env_var(tmp_path, monkeypatch):
 
 
 @pytest.mark.parametrize(
-    "content, expected_error",
+    "content",
     [
-        ('name = "logstats"\n', ValidationError),
-        ('sources = "not-a-table"\n', ValidationError),
-        ("[sources]\napp1 = 42\n", ValidationError),
-        ("[sources\napp1 = broken", tomllib.TOMLDecodeError),
+        'name = "logstats"\n',
+        'sources = "not-a-table"\n',
+        "[sources]\napp1 = 42\n",
+        "[sources\napp1 = broken",
     ],
     ids=["missing-table", "sources-not-a-table", "url-not-a-string", "malformed-toml"],
 )
-def test_invalid_config_is_rejected(tmp_path, content, expected_error):
+def test_invalid_config_is_rejected(tmp_path, content):
     path = write_toml(tmp_path, content)
-    with pytest.raises(expected_error):
+    with pytest.raises(ConfigError, match=re.escape(str(path))):
         load_settings(path)
+
+
+def test_missing_config_reports_the_path(tmp_path):
+    missing = tmp_path / "nope.toml"
+    with pytest.raises(ConfigError, match=re.escape(str(missing))):
+        load_settings(missing)

Tests/data_test.py

@@ -4,16 +4,16 @@
 from hypothesis import given
 from hypothesis import strategies as st
 
-from logstats.data import InvalidLogFormat, LogEntry, LogType, get_name_capitalize
+from logstats.data import InvalidLogFormat, LogEntry, LogType, level_label
 
 
 @given(level=st.sampled_from(LogType))
-def test_get_name_capitalize_when_enum_values_used(level):
-    assert get_name_capitalize(level) == level.name.capitalize()
+def test_level_label_when_enum_values_used(level):
+    assert level_label(level) == level.name.capitalize()
 
 
-def test_get_name_capitalize_when_none_used():
-    assert get_name_capitalize(None) == "All"
+def test_level_label_when_none_used():
+    assert level_label(None) == "All"
 
 
 def test_produce_log_entry():

Tests/parser_test.py → Tests/fetcher_test.py

@@ -2,7 +2,7 @@
 
 import httpx
 
-from logstats.parser import collect
+from logstats.fetcher import collect
 from logstats.source_parser import FetchedSource
 
 
@@ -21,7 +21,7 @@ def test_parse_fetches():
     }
     result = asyncio.run(collect_logs(logs))
     assert {r.source for r in result} == set(logs)
-    assert (sum(len(r.log_lines) for r in result)) == 4
+    assert (sum(len(r.entries) for r in result)) == 4
 
 
 def test_unknown_source_types():
@@ -30,7 +30,29 @@ def test_unknown_source_types():
     }
     result = asyncio.run(collect_logs(logs))
     assert {r.source for r in result} == set(logs)
-    assert (sum(len(r.log_lines) for r in result)) == 0
+    assert (sum(len(r.entries) for r in result)) == 0
+
+
+def test_collect_never_exceeds_max_concurrent():
+    in_flight = 0
+    peak = 0
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        nonlocal in_flight, peak
+        in_flight += 1
+        peak = max(peak, in_flight)
+        await asyncio.sleep(0.01)
+        in_flight -= 1
+        return httpx.Response(200, text="2026-07-13T09:00:00 INFO ok")
+
+    async def run() -> list[FetchedSource]:
+        sources = [f"http://x/app{i}.log" for i in range(8)]
+        async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+            return await collect(sources, None, client, max_concurrent=3)
+
+    result = asyncio.run(run())
+    assert len(result) == 8
+    assert peak == 3
 
 
 def test_connectivity_issues():
@@ -46,5 +68,5 @@ async def run() -> list[FetchedSource]:
             )
 
     by_source = {r.source: r for r in asyncio.run(run())}
-    assert len(by_source["http://x/good.log"].log_lines) == 1  # good one still parsed
+    assert len(by_source["http://x/good.log"].entries) == 1  # good one still parsed
     assert by_source["http://x/bad.log"].error is not None  # bad one flagged, not fatal

pyproject.toml

@@ -18,7 +18,7 @@ requires = ["uv_build >= 0.11.23, <0.12.0"]
 build-backend = "uv_build"
 
 [project.scripts]
-logstats="logstats.cli:parse_aguments"
+logstats="logstats.cli:parse_arguments"
 
 [tool.pytest.ini_options]
 testpaths = ["Tests"]

src/logstats/api.py

@@ -8,6 +8,7 @@
 
 from logstats.config import Settings, load_settings
 from logstats.data import LogType
+from logstats.fetcher import TIMEOUT
 from logstats.schemas import (
     PerHourResponse,
     TopResponse,
@@ -20,14 +21,13 @@
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
     app.state.settings = load_settings()
-    app.state.client = httpx.AsyncClient()
+    app.state.client = httpx.AsyncClient(timeout=TIMEOUT)
     yield
     await app.state.client.aclose()
 
 
 app = FastAPI(lifespan=lifespan)
 router = APIRouter(prefix="/stats", tags=["stats"])
-app.include_router(router)
 
 
 def get_settings(request: Request) -> Settings:
@@ -87,3 +87,11 @@ async def get_regular(
     return StreamingResponse(
         stream_all(urls, level, client), media_type="text/event-stream"
     )
+
+
+@app.get("/health")
+def health() -> dict[str, str]:
+    return {"status": "ok"}
+
+
+app.include_router(router)

src/logstats/cli.py

@@ -20,7 +20,7 @@
 @click.option(
     "-c", "combined", is_flag=True, help="combine reports from different sources"
 )
-def parse_aguments(
+def parse_arguments(
     sources: Sequence[str],
     level: LogType | None,
     top: int | None,

src/logstats/config.py

@@ -2,7 +2,11 @@
 import tomllib
 from pathlib import Path
 
-from pydantic import BaseModel
+from pydantic import BaseModel, ValidationError
+
+
+class ConfigError(Exception):
+    """Raised when the sources config cannot be read or is not valid."""
 
 
 class Settings(BaseModel):
@@ -12,8 +16,25 @@ def get_source_url(self, name: str) -> str:
         return self.sources[name]
 
 
-def load_settings(path: Path | None = None) -> Settings:
+DEFAULT_SOURCES = "sources.toml"
+
+
+def resolve_sources_path(path: Path | None = None) -> Path:
+    """Return the sources config as an absolute path, without following symlinks.
+
+    Falls back to $LOGSTATS_SOURCES, then to sources.toml in the working directory.
+    """
     if path is None:
-        path = Path(os.environ.get("LOGSTATS_SOURCES", "sources.toml"))
-    with path.open("rb") as f:
-        return Settings(**tomllib.load(f))
+        path = Path(os.environ.get("LOGSTATS_SOURCES", DEFAULT_SOURCES))
+    return path.absolute()
+
+
+def load_settings(path: Path | None = None) -> Settings:
+    resolved = resolve_sources_path(path)
+    try:
+        with resolved.open("rb") as f:
+            return Settings(**tomllib.load(f))
+    except OSError as exc:
+        raise ConfigError(f"Cannot read sources config at {resolved}: {exc}") from exc
+    except (tomllib.TOMLDecodeError, ValidationError) as exc:
+        raise ConfigError(f"Invalid sources config at {resolved}: {exc}") from exc

src/logstats/core.py

@@ -4,7 +4,7 @@
 from functools import partial
 
 from logstats.data import LogType, ReportRequest
-from logstats.parser import fetch
+from logstats.fetcher import fetch
 from logstats.per_hour_stats import compute_per_hour
 from logstats.report import Report, format_hourly, format_regular, format_top
 from logstats.source_parser import FetchedSource
@@ -47,7 +47,7 @@ def main(
     if len(fetched_sources) == 0:
         logger.warning("No available sources.")
         return
-    if all(not fs.log_lines for fs in fetched_sources):
+    if all(not fs.entries for fs in fetched_sources):
         logger.info("No logs available with the selected filters.")
         return
     for report in create_reports(fetched_sources, request):

src/logstats/data.py

@@ -14,7 +14,7 @@ class LogType(Enum):
     ERROR = 3
 
 
-def get_name_capitalize(level: LogType | None) -> str:
+def level_label(level: LogType | None) -> str:
     return level.name.capitalize() if level is not None else "All"
 
 

src/logstats/fetcher.py

@@ -0,0 +1,35 @@
+import asyncio
+import logging
+from collections.abc import Sequence
+
+import httpx
+
+from logstats.data import LogType
+from logstats.source_parser import FetchedSource, fetch_source
+
+logger = logging.getLogger(__name__)
+
+MAX_CONCURRENT_FETCHES = 10
+TIMEOUT = httpx.Timeout(connect=5.0, read=30.0, write=10.0, pool=5.0)
+
+
+async def fetch(sources: Sequence[str], level: LogType | None) -> list[FetchedSource]:
+    async with httpx.AsyncClient(timeout=TIMEOUT) as client:
+        return await collect(sources, level, client)
+
+
+async def collect(
+    sources: Sequence[str],
+    level: LogType | None,
+    client: httpx.AsyncClient,
+    max_concurrent: int = MAX_CONCURRENT_FETCHES,
+) -> list[FetchedSource]:
+    limit = asyncio.Semaphore(max_concurrent)
+
+    async def fetch_one(source: str) -> FetchedSource:
+        async with limit:
+            return await fetch_source(source, level, client)
+
+    async with asyncio.TaskGroup() as group:
+        results = [group.create_task(fetch_one(s)) for s in sources]
+    return [t.result() for t in results]