From e4b35871379ffa9d04cd45e5c1874a2da3be6b60 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 13 Aug 2026 15:51:29 +0000 Subject: [PATCH] Make worker startup logging visible, bound dataset downloads, persist blocked list Three fixes prompted by an arax_pathfinder deployment that picked up no tasks, emitted no logs at all, and was terminated after ~8 minutes. Startup logging was silently discarded. setup_logging attached its handlers to the `shepherd` logger and left root unconfigured, but every worker logged its startup phase through `logging.getLogger(STREAM)` ("arax.pathfinder") or a bare `logging.info`. Those names sit outside the `shepherd` namespace, so records reached a handler-less root and were dropped by logging's lastResort fallback, which only emits WARNING+. Dataset downloads, pool sizing and poll-loop errors were invisible, and a worker wedged before `get_tasks` looked exactly like a healthy idle one. Handlers now live on root, and `shepherd` sets only its level and propagates to them -- so our logging still emits at DEBUG, third-party libraries stay quiet below WARNING (httpx logs a line per request at INFO), nothing double-logs, and pytest's caplog keeps working. Workers get their logger from a new `get_worker_logger` helper that namespaces the stream name, applied across all 21 workers plus the two module-named loggers in data_download and db. Dataset downloads had no timeout. urllib defaults to none, so a connection that opens and then stalls -- an egress proxy that swallows the request, a server that accepts and never responds -- hung worker startup indefinitely, before the poll loop and before any heartbeat, without ever crashing. They are now bounded by DATASET_DOWNLOAD_TIMEOUT_SEC (default 60), applied per socket operation so a multi-GB file still downloads for as long as it keeps making progress. A stall part-way through the body raises TimeoutError straight out of resp.read() rather than wrapped in URLError, so it gets its own handler and a message naming the likely cause. The ARAX blocked-concept list was written to the worker's working directory, which is the container's writable layer -- discarded on every restart, so each new pod re-fetched it from GitHub during startup on that same invisible code path. It now lives alongside the pathfinder sqlite databases on the mounted volume, reached through an `arax_blocked_list_path` helper that keeps the download and read sites in agreement. It is only fetched when absent, so delete it from the volume to pick up an updated upstream list, and the volume must be writable for the first run (or the file preloaded). Also documents the deployment trap behind the original incident: the "already present" check is an exact match on ARAX_PATHFINDER_DBS_DIR (a relative path by default, resolved against /app) plus the tier-versioned filenames, so a mount path or ARAX_PATHFINDER_TIER_VERSION that disagrees re-downloads the databases onto ephemeral storage instead of using the mounted copies -- which is what got the pod evicted. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01VqmpAvRW688q7KqX9sQjGX --- README.md | 45 ++++++++++--- compose.yml | 7 +- shepherd_utils/config.py | 13 ++++ shepherd_utils/data_download.py | 77 +++++++++++++++++++-- shepherd_utils/db.py | 2 +- shepherd_utils/logger.py | 43 +++++++++++- tests/unit/test_data_download.py | 90 +++++++++++++++++++++++++ tests/unit/test_logger.py | 67 +++++++++++++++++- workers/aragorn/worker.py | 6 +- workers/aragorn_lookup/worker.py | 6 +- workers/aragorn_omnicorp/worker.py | 12 ++-- workers/aragorn_pathfinder/worker.py | 6 +- workers/aragorn_score/worker.py | 10 +-- workers/arax/worker.py | 6 +- workers/arax_pathfinder/worker.py | 62 +++++++---------- workers/arax_rank/worker.py | 10 +-- workers/bte/worker.py | 6 +- workers/bte_lookup/worker.py | 6 +- workers/example_ara/worker.py | 6 +- workers/example_lookup/worker.py | 6 +- workers/example_score/worker.py | 6 +- workers/filter_analyses_top_n/worker.py | 6 +- workers/filter_kgraph_orphans/worker.py | 6 +- workers/filter_results_top_n/worker.py | 6 +- workers/finish_query/worker.py | 6 +- workers/merge_message/worker.py | 13 ++-- workers/score_paths/worker.py | 11 +-- workers/sipr/worker.py | 6 +- workers/sort_results_score/worker.py | 6 +- 29 files changed, 440 insertions(+), 112 deletions(-) diff --git a/README.md b/README.md index 4921dfc..aa2f94e 100644 --- a/README.md +++ b/README.md @@ -21,7 +21,7 @@ commit to git (they're gitignored and volume-mounted from the host): - **`aragorn_omnicorp`** → `./omnicorp_lmdb/` (`curies.lmdb`, `shared_counts.lmdb`) - **`score_paths`** → `./pathfinder_embeddings/` (a directory-style LMDB) -- **`arax_pathfinder`** → `./arax_pathfinder_dbs/` (`curie_ngd_v1.0_.sqlite`, `tier0-info-for-overlay_v1.0_.sqlite`) +- **`arax_pathfinder`** → `./arax_pathfinder_dbs/` (`curie_ngd_v1.0_.sqlite`, `tier0-info-for-overlay_v1.0_.sqlite`, `general_concepts.json`) So a new developer doesn't have to source these by hand, each worker can fetch its dataset on first startup. Two download mechanisms are supported, depending on where the dataset lives: @@ -37,7 +37,7 @@ PATHFINDER_EMBEDDINGS_URL=https://example.org/path/pathfinder_embeddings.tar.gz The archive for each dataset should contain the expected files at its top level: `curies.lmdb` and `shared_counts.lmdb` for omnicorp, `data.mdb` (and `lock.mdb`) for the embeddings. -**arax_pathfinder's sqlite databases** are served as plain files over HTTPS. The +**arax_pathfinder's sqlite databases** are served as plain files over HTTPS, no credentials needed. The filenames embed a Knowledge Graph version that changes periodically, so only one variable needs updating when a new Knowledge Graph ships: @@ -45,14 +45,43 @@ when a new Knowledge Graph ships: ARAX_PATHFINDER_TIER_VERSION=tier0-20260621 ``` -This requires an SSH key with access to that host, mounted read-only into the container -(`~/.ssh:/home/nru/.ssh:ro` in compose.yml). +The ARAX blocked-concept list (`general_concepts.json`, fetched from GitHub) lands in this same +directory, so it is downloaded once and then persists with the databases rather than being re-fetched +by every new container. It is only fetched when absent — delete it from the volume to pick up an +updated upstream list. On startup, each worker checks whether its files already exist in the volume-mounted directory. If -they're missing and a source is configured (URL or scp path), it fetches them into that directory — -which lives on the host, so the data persists across restarts and is only downloaded once. If the -files are already present, or no source is configured, the download is skipped (production mounts -this data out of band, so it's unaffected). +they're missing and a URL is configured, it fetches them into that directory — which lives on the +host, so the data persists across restarts and is only downloaded once. If the files are already +present, or no source is configured, the download is skipped (production mounts this data out of +band, so it's unaffected). + +Downloads are bounded by `DATASET_DOWNLOAD_TIMEOUT_SEC` (default 60), which applies per socket +operation rather than to the whole transfer — a large file downloads for as long as it needs, but a +connection that opens and then stalls fails loudly instead of hanging worker startup. + +#### Deploying these workers + +The presence check is an exact match on the configured directory **and** the tier-versioned filenames. +A deployment that mounts the data somewhere else, or whose `ARAX_PATHFINDER_TIER_VERSION` doesn't match +the filenames on the volume, will not use the mounted copies — it will decide the dataset is missing and +download it again, into whatever path the settings do point at. If that path isn't the mount, the files +land on the container's writable layer and the pod is eventually evicted for exceeding its ephemeral +storage. So when deploying, set the directory explicitly and confirm it resolves to your mount: + +```dotenv +ARAX_PATHFINDER_DBS_DIR=/data/arax_pathfinder_dbs # default is relative: resolved against /app +``` + +```console +$ kubectl exec deploy/arax-pathfinder -- python -c \ + "from shepherd_utils.data_download import arax_pathfinder_sqlite_paths as p; print(*p(), sep='\n')" +``` + +Because `general_concepts.json` now shares that directory, the volume needs to be writable for the +first startup that fetches it — or the file can be preloaded alongside the sqlite databases, after +which the worker only ever reads it. A read-only mount with no preloaded copy fails at startup with a +permission error rather than silently continuing. ### Worker diff --git a/compose.yml b/compose.yml index 0e1a39a..bd6d77d 100644 --- a/compose.yml +++ b/compose.yml @@ -359,9 +359,14 @@ services: - ./logs:/app/logs - ./.env:/app/.env # First run? The worker downloads its two sqlite dbs from - # kg2webhost.rtx.ai on startup, no credentials needed. Set + # kg2webhost.rtx.ai, plus the ARAX blocked-concept list from GitHub, on + # startup -- no credentials needed. All three live here, so they persist + # across restarts instead of being re-fetched by each new container. Set # ARAX_PATHFINDER_TIER_VERSION in your .env if you need a tier other than # the default (see README "Worker data (LMDB / sqlite) downloads"). + # Deploying elsewhere? ARAX_PATHFINDER_DBS_DIR must point at the mount -- + # it defaults to a path relative to /app, and a mismatch silently + # re-downloads everything instead of using the mounted copies. - ./arax_pathfinder_dbs:/app/arax_pathfinder_dbs arax_rank: diff --git a/shepherd_utils/config.py b/shepherd_utils/config.py index 7f7e65c..78e9907 100644 --- a/shepherd_utils/config.py +++ b/shepherd_utils/config.py @@ -122,6 +122,19 @@ class Settings(BaseSettings): omnicorp_lmdb_url: str = "" pathfinder_embeddings_url: str = "" + # Socket timeout (seconds) for those startup dataset downloads. This bounds + # each individual blocking socket operation -- the connect and every + # subsequent read -- not the transfer as a whole, so a genuinely large file + # still downloads for as long as it needs provided it keeps making progress. + # Without it, urllib inherits Python's default of no timeout: a connection + # that opens and then stalls (an egress proxy that swallows the request, a + # server that accepts and never responds) hangs the worker's startup + # forever, before it ever reaches the poll loop. That is silent -- the + # worker never registers a heartbeat and never picks up a task, but it also + # never crashes. With a timeout the download fails loudly instead. Set to 0 + # to restore the unbounded behavior. + dataset_download_timeout_sec: float = 60.0 + otel_enabled: bool = True jaeger_host: str = "http://jaeger" jaeger_port: int = 4317 diff --git a/shepherd_utils/data_download.py b/shepherd_utils/data_download.py index 610a948..e10998f 100644 --- a/shepherd_utils/data_download.py +++ b/shepherd_utils/data_download.py @@ -15,7 +15,8 @@ ``aragorn_omnicorp`` and ``score_paths`` below). * **Per-file** -- individual files fetched directly, no archive/extract step (``arax_pathfinder`` below, whose two sqlite databases are served as plain - files rather than bundled into one archive). + files rather than bundled into one archive, plus the ARAX blocked-concept + list that shares their directory). When a download source is configured (read via :mod:`shepherd_utils.config`), each worker calls its matching ``ensure_*`` helper at startup: @@ -50,14 +51,25 @@ def _missing_files(target_dir: str, required_files: List[str]) -> List[str]: def _download(url: str, dest_path: str, logger: logging.Logger) -> None: - """Stream ``url`` to ``dest_path``, logging progress periodically.""" + """Stream ``url`` to ``dest_path``, logging progress periodically. + + Bounded by ``dataset_download_timeout_sec`` (see config). The timeout is + per socket operation rather than for the whole transfer, so a multi-GB file + downloads for as long as it needs while a stalled connection raises instead + of hanging worker startup indefinitely. + """ logger.info(f"Downloading dataset from {url} ...") + timeout = float(settings.dataset_download_timeout_sec) + # urlopen treats timeout=None as "no timeout"; 0 is the documented opt-out. + kwargs = {"timeout": timeout} if timeout > 0 else {} + # Hoisted so the timeout handler below can report how far the transfer got + # even when it never made it past opening the connection. + read = 0 try: # nosec B310: the URL is operator-configured (an env var), not user input. - with urllib.request.urlopen(url) as resp: # noqa: S310 + with urllib.request.urlopen(url, **kwargs) as resp: # noqa: S310 header = resp.headers.get("Content-Length") total = int(header) if header and header.isdigit() else None - read = 0 step = 50 * 1024 * 1024 # log roughly every 50 MB next_log = step with open(dest_path, "wb") as out: @@ -82,10 +94,22 @@ def _download(url: str, dest_path: str, logger: logging.Logger) -> None: f"URL is correct and reachable." ) from e except urllib.error.URLError as e: + # A connect timeout arrives here, wrapped, with e.reason set to the + # underlying socket.timeout. raise RuntimeError( f"download failed for {url}: {e.reason}. Confirm the URL is correct " f"and reachable from inside the container." ) from e + except TimeoutError as e: + # A stall part-way through the body raises straight out of resp.read() + # rather than being wrapped in URLError, so it needs its own handler -- + # otherwise this surfaces as a bare TimeoutError with no context. + raise RuntimeError( + f"download failed for {url}: no data for {timeout:.0f}s " + f"({read / 1e6:.0f} MB transferred). The host is reachable but the " + f"transfer stalled -- check for an egress proxy or a rate limit, or " + f"raise DATASET_DOWNLOAD_TIMEOUT_SEC." + ) from e logger.info(f"Download complete: {read / 1e6:.0f} MB") @@ -123,7 +147,7 @@ def ensure_lmdb_dataset( Idempotent: once the files are present this returns immediately, so it's safe to call unconditionally on every worker startup. """ - logger = logger or logging.getLogger(__name__) + logger = logger or logging.getLogger("shepherd.data_download") missing = _missing_files(target_dir, required_files) if not missing: @@ -191,7 +215,7 @@ def ensure_http_files_dataset( Idempotent: once a file is present it's left alone, so it's safe to call unconditionally on every worker startup. """ - logger = logger or logging.getLogger(__name__) + logger = logger or logging.getLogger("shepherd.data_download") required_files = list(file_sources.keys()) missing = _missing_files(target_dir, required_files) @@ -344,3 +368,44 @@ def ensure_arax_pathfinder_dbs(logger: Optional[logging.Logger] = None) -> None: }, logger=logger, ) + + +# The ARAX blocked-concept list lives alongside the pathfinder sqlite databases +# so it lands on the same mounted volume. It previously went to the worker's +# working directory (``/app``), which is the container's writable layer and +# therefore thrown away on every restart -- meaning each new pod re-fetched it +# from GitHub during startup, before the poll loop, on a code path whose logs +# were being discarded. On the volume it is fetched once and persists. +ARAX_BLOCKED_LIST_FILENAME = "general_concepts.json" + + +def arax_blocked_list_path() -> str: + """Return the on-disk path of the ARAX blocked-concept list. + + Single source of truth, in the same spirit as + ``arax_pathfinder_sqlite_paths``: ``ensure_arax_blocked_list`` uses it to + know where to download, and worker.py uses it to know what to open. + """ + return os.path.join(settings.arax_pathfinder_dbs_dir, ARAX_BLOCKED_LIST_FILENAME) + + +def ensure_arax_blocked_list(logger: Optional[logging.Logger] = None) -> None: + """Ensure the ARAX blocked-concept list is present next to the sqlite dbs. + + Fetched via the shared downloader so it lands through a temp file + atomic + rename (a direct write let concurrent tasks race on a half-written file) + and inherits the download timeout. Idempotent, so it is safe to call at + startup and again lazily from a pool child. + + Note the flip side of persisting this on the volume: it is now only fetched + when absent, so a refreshed upstream list is not picked up until the file is + deleted. Delete it from the volume to force a re-fetch on the next restart. + """ + ensure_http_files_dataset( + name="arax_blocked_list", + # ``or "."`` so an unset/blank dbs dir degrades to the working directory + # rather than handing makedirs an empty path. + target_dir=os.path.dirname(arax_blocked_list_path()) or ".", + file_sources={ARAX_BLOCKED_LIST_FILENAME: settings.arax_blocked_list_url}, + logger=logger, + ) diff --git a/shepherd_utils/db.py b/shepherd_utils/db.py index 0380903..dfdff43 100644 --- a/shepherd_utils/db.py +++ b/shepherd_utils/db.py @@ -193,7 +193,7 @@ async def initialize_db() -> None: except Exception: # A failed upgrade must never keep a worker from starting: the schema # additions are performance aids, and the janitor/next boot retries. - logging.getLogger(__name__).warning( + logging.getLogger("shepherd.db").warning( "Failed to apply startup schema upgrades", exc_info=True ) diff --git a/shepherd_utils/logger.py b/shepherd_utils/logger.py index 791de4c..e43957b 100644 --- a/shepherd_utils/logger.py +++ b/shepherd_utils/logger.py @@ -128,10 +128,32 @@ def get_logging_config(): "default": {"format": "[%(asctime)s: %(levelname)s/%(name)s]: %(message)s"} }, "handlers": handlers, + # The output handlers live on root, and root alone. Previously they were + # attached to ``shepherd`` and root was left unconfigured, so any record + # logged outside that namespace -- a stray ``logging.info``, a + # third-party library, a logger named for its module or its Redis stream + # -- reached a handler-less root and was dropped by logging's + # ``lastResort`` fallback, which only emits WARNING+ and ignores our + # formatter. That silently swallowed every worker's entire startup + # phase; see ``get_worker_logger``. + # + # Root's level applies only to records logged directly on root, not to + # ones propagated up from a child (those are level-checked at the + # originating logger). So WARNING here means third-party libraries are + # quiet below WARNING -- httpx logs a line per request at INFO, which + # would bury our own output -- while ``shepherd.*`` still emits at DEBUG + # via the entry below. Everything then reaches these handlers by + # propagation, which also keeps pytest's ``caplog`` working. + "root": { + "level": "WARNING", + "handlers": logger_handlers, + }, "loggers": { + # No handlers: records propagate to root's. Only the level is set + # here, which is what lets our own logging through at DEBUG while + # leaving third-party loggers at root's WARNING. "shepherd": { "level": "DEBUG", - "handlers": logger_handlers, }, # psycopg's pool retries to keep min_size connections warm and logs # a WARNING on every failed attempt. When the DB is down that floods @@ -150,6 +172,25 @@ def get_logging_config(): return logging_config +def get_worker_logger(name: str) -> logging.Logger: + """Return a logger under the configured ``shepherd`` namespace. + + ``setup_logging`` only attaches handlers to ``shepherd`` (and, as a + WARNING-level backstop, root), so a logger named for the stream alone + (``logging.getLogger("arax.pathfinder")``) inherits no handler at INFO and + its records vanish. Every worker's startup phase -- dataset downloads, pool + sizing, poll-loop errors -- logged through exactly such a logger, so a + worker that hung before reaching ``get_tasks`` produced no output at all + and looked identical to a healthy idle one. + + Passing a name that is already namespaced is a no-op, so this is safe to + apply to existing ``shepherd.``-prefixed names. + """ + if name == "shepherd" or name.startswith("shepherd."): + return logging.getLogger(name) + return logging.getLogger(f"shepherd.{name}") + + def setup_logging(): """Set up logging.""" config = get_logging_config() diff --git a/tests/unit/test_data_download.py b/tests/unit/test_data_download.py index 851cdae..1b32170 100644 --- a/tests/unit/test_data_download.py +++ b/tests/unit/test_data_download.py @@ -5,7 +5,9 @@ """ import logging +import os import tarfile +import urllib.request # noqa: F401 (patched by name in the timeout tests) import pytest @@ -155,3 +157,91 @@ def test_ensure_pathfinder_embeddings_wires_settings(tmp_path, mocker): assert (target / "data.mdb").exists() assert (target / "lock.mdb").exists() + + +# --- download timeout -------------------------------------------------------- +# +# urllib defaults to no timeout, so a connection that opens and then stalls hung +# worker startup forever -- before the poll loop, before any heartbeat, and with +# no crash to point at. + + +def test_download_passes_configured_timeout(tmp_path, mocker): + mocker.patch.object(data_download.settings, "dataset_download_timeout_sec", 12.5) + urlopen = mocker.patch.object(data_download.urllib.request, "urlopen") + urlopen.return_value.__enter__.return_value.headers.get.return_value = None + urlopen.return_value.__enter__.return_value.read.return_value = b"" + + data_download._download("http://example.invalid/x", str(tmp_path / "out"), logger) + + assert urlopen.call_args.kwargs["timeout"] == 12.5 + + +def test_download_timeout_zero_restores_unbounded_behavior(tmp_path, mocker): + mocker.patch.object(data_download.settings, "dataset_download_timeout_sec", 0) + urlopen = mocker.patch.object(data_download.urllib.request, "urlopen") + urlopen.return_value.__enter__.return_value.headers.get.return_value = None + urlopen.return_value.__enter__.return_value.read.return_value = b"" + + data_download._download("http://example.invalid/x", str(tmp_path / "out"), logger) + + assert "timeout" not in urlopen.call_args.kwargs + + +def test_download_reports_a_mid_transfer_stall(tmp_path, mocker): + """A stall inside resp.read() raises TimeoutError directly rather than being + wrapped in URLError, so it needs its own handler to get a useful message.""" + mocker.patch.object(data_download.settings, "dataset_download_timeout_sec", 30) + urlopen = mocker.patch.object(data_download.urllib.request, "urlopen") + resp = urlopen.return_value.__enter__.return_value + resp.headers.get.return_value = None + resp.read.side_effect = [b"partial", TimeoutError("timed out")] + + with pytest.raises(RuntimeError, match="transfer stalled"): + data_download._download( + "http://example.invalid/x", str(tmp_path / "out"), logger + ) + + +# --- ARAX blocked list ------------------------------------------------------- + + +def test_blocked_list_lives_with_the_pathfinder_sqlite_dbs(tmp_path, mocker): + """It used to be written to the working directory -- the container's + writable layer -- so every new pod re-fetched it from GitHub at startup.""" + mocker.patch.object( + data_download.settings, "arax_pathfinder_dbs_dir", str(tmp_path / "dbs") + ) + + path = data_download.arax_blocked_list_path() + + assert path == str(tmp_path / "dbs" / "general_concepts.json") + curie_ngd, _ = data_download.arax_pathfinder_sqlite_paths() + # Same volume as the sqlite databases, which is the whole point. + assert os.path.dirname(path) == os.path.dirname(curie_ngd) + + +def test_ensure_arax_blocked_list_downloads_into_the_volume(tmp_path, mocker): + target = tmp_path / "dbs" + source = tmp_path / "general_concepts.json" + source.write_text('{"curies": ["CHEBI:1"], "synonyms": ["Water"]}') + mocker.patch.object(data_download.settings, "arax_pathfinder_dbs_dir", str(target)) + mocker.patch.object( + data_download.settings, "arax_blocked_list_url", source.as_uri() + ) + + data_download.ensure_arax_blocked_list(logger) + + assert (target / "general_concepts.json").exists() + + +def test_ensure_arax_blocked_list_is_a_noop_once_present(tmp_path, mocker): + target = tmp_path / "dbs" + target.mkdir() + (target / "general_concepts.json").write_text('{"curies": [], "synonyms": []}') + mocker.patch.object(data_download.settings, "arax_pathfinder_dbs_dir", str(target)) + spy = mocker.patch.object(data_download, "_download") + + data_download.ensure_arax_blocked_list(logger) + + spy.assert_not_called() diff --git a/tests/unit/test_logger.py b/tests/unit/test_logger.py index c0d37c2..df97bfd 100644 --- a/tests/unit/test_logger.py +++ b/tests/unit/test_logger.py @@ -7,6 +7,7 @@ QueryLogger, ReasonerLogEntryFormatter, get_logging_config, + get_worker_logger, ) @@ -91,7 +92,7 @@ def test_get_logging_config_local_includes_file_handler(monkeypatch, tmp_path): config = get_logging_config() assert "file" in config["handlers"] assert "console" in config["handlers"] - assert set(config["loggers"]["shepherd"]["handlers"]) == {"console", "file"} + assert set(config["root"]["handlers"]) == {"console", "file"} # The function eagerly creates the logs/ dir for file output. assert os.path.isdir(tmp_path / "logs") @@ -100,7 +101,7 @@ def test_get_logging_config_kubernetes_skips_file_handler(monkeypatch): monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.0.0.1") config = get_logging_config() assert "file" not in config["handlers"] - assert config["loggers"]["shepherd"]["handlers"] == ["console"] + assert config["root"]["handlers"] == ["console"] def test_get_logging_config_pool_child_skips_file_handler(monkeypatch, tmp_path): @@ -119,4 +120,64 @@ def test_get_logging_config_pool_child_skips_file_handler(monkeypatch, tmp_path) ) config = get_logging_config() assert "file" not in config["handlers"] - assert config["loggers"]["shepherd"]["handlers"] == ["console"] + assert config["root"]["handlers"] == ["console"] + + +# --- worker logger namespacing ---------------------------------------------- +# +# Regression cover for a silent-startup bug: workers logged through +# ``logging.getLogger(STREAM)`` (e.g. "arax.pathfinder"), which sits outside the +# only namespace ``setup_logging`` attaches handlers to. Those records reached a +# handler-less root and were dropped by logging's lastResort fallback, so a +# worker that hung during startup produced no output whatsoever. + + +def test_get_worker_logger_namespaces_stream_names(): + assert get_worker_logger("arax.pathfinder").name == "shepherd.arax.pathfinder" + + +def test_get_worker_logger_does_not_double_prefix(): + """Safe to apply to names that are already namespaced.""" + assert get_worker_logger("shepherd.monitor").name == "shepherd.monitor" + assert get_worker_logger("shepherd").name == "shepherd" + + +def test_worker_logger_inherits_configured_handlers(monkeypatch): + """The whole point: a worker logger must resolve to a real handler at INFO.""" + monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.0.0.1") + logging.config.dictConfig(get_logging_config()) + + worker_logger = get_worker_logger("arax.pathfinder") + # Walk the ancestry the way logging does when emitting a record. + effective = [] + node = worker_logger + while node: + effective.extend(node.handlers) + node = node.parent if node.propagate else None + + assert effective, "worker logger resolved to no handler" + assert worker_logger.getEffectiveLevel() <= logging.INFO + + +def test_root_logger_configured_as_warning_backstop(monkeypatch): + """Stray records outside ``shepherd.*`` still land somewhere, formatted. + + WARNING rather than INFO on purpose -- at INFO, libraries such as httpx emit + a line per request and drown out our own logs. + """ + monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.0.0.1") + config = get_logging_config() + assert config["root"]["level"] == "WARNING" + assert config["root"]["handlers"] == ["console"] + + +def test_handlers_are_attached_only_once(monkeypatch): + """Handlers live on root only; shepherd sets a level and propagates to them. + + Attaching them in both places would emit every shepherd record twice. + """ + monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.0.0.1") + config = get_logging_config() + assert config["root"]["handlers"] == ["console"] + assert "handlers" not in config["loggers"]["shepherd"] + assert config["loggers"]["shepherd"]["level"] == "DEBUG" diff --git a/workers/aragorn/worker.py b/workers/aragorn/worker.py index 4bdde0d..e58d8c7 100644 --- a/workers/aragorn/worker.py +++ b/workers/aragorn/worker.py @@ -6,6 +6,7 @@ import uuid from shepherd_utils.db import get_message +from shepherd_utils.logger import get_worker_logger from shepherd_utils.otel import setup_tracer from shepherd_utils.shared import ( examine_query, @@ -20,6 +21,7 @@ CONSUMER = str(uuid.uuid4())[:8] TASK_LIMIT = 100 tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) async def aragorn(task, logger: logging.Logger): @@ -76,9 +78,9 @@ async def poll_for_tasks(): ): asyncio.create_task(process_task(task, parent_ctx, logger, limiter)) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) # back off before retrying diff --git a/workers/aragorn_lookup/worker.py b/workers/aragorn_lookup/worker.py index e529717..50beea9 100644 --- a/workers/aragorn_lookup/worker.py +++ b/workers/aragorn_lookup/worker.py @@ -23,6 +23,7 @@ remove_callback_id, save_message, ) +from shepherd_utils.logger import get_worker_logger from shepherd_utils.otel import setup_tracer from shepherd_utils.shared import get_tasks, run_task_lifecycle @@ -33,6 +34,7 @@ CONSUMER = str(uuid.uuid4())[:8] TASK_LIMIT = 100 tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) def examine_query(message): @@ -367,9 +369,9 @@ async def poll_for_tasks(): ): asyncio.create_task(process_task(task, parent_ctx, logger, limiter)) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) # back off before retrying diff --git a/workers/aragorn_omnicorp/worker.py b/workers/aragorn_omnicorp/worker.py index f72dd23..cff432b 100644 --- a/workers/aragorn_omnicorp/worker.py +++ b/workers/aragorn_omnicorp/worker.py @@ -29,6 +29,7 @@ from shepherd_utils.cpu import resolve_pool_workers from shepherd_utils.data_download import ensure_omnicorp_lmdb from shepherd_utils.db import get_message_sync, save_message_sync +from shepherd_utils.logger import get_worker_logger from shepherd_utils.otel import setup_tracer from shepherd_utils.process_pool import ProcessPoolManager from shepherd_utils.shared import get_tasks, run_task_lifecycle @@ -43,6 +44,7 @@ # many run in parallel; keep it modest since each message can be large. TASK_LIMIT = 10 tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) # Matches the upstream `redis_batch_size`. LMDB_BATCH_SIZE = 1000 @@ -560,7 +562,7 @@ async def poll_for_tasks(): # to open them lazily (a first-run local `docker compose up` starts with the # volume-mounted directory empty). No-op once present or when no download URL # is configured (e.g. production, where the data is mounted out of band). - ensure_omnicorp_lmdb(logging.getLogger(STREAM)) + ensure_omnicorp_lmdb(LOGGER) loop = asyncio.get_running_loop() # The overlay is CPU-bound, so cap real parallelism at the number of cores. @@ -571,8 +573,8 @@ async def poll_for_tasks(): # messages at once and OOM-killing the pod). resolve_pool_workers reads the # cgroup CPU limit and honours a POOL_MAX_WORKERS override for memory-tight # deployments. Extra in-flight tasks queue against the pool without blocking. - max_workers = resolve_pool_workers(TASK_LIMIT, logging.getLogger(STREAM)) - logging.info(f"{STREAM}: process pool sized to {max_workers} worker(s).") + max_workers = resolve_pool_workers(TASK_LIMIT, LOGGER) + LOGGER.info(f"{STREAM}: process pool sized to {max_workers} worker(s).") pool = ProcessPoolManager( max_workers, max_tasks_per_child=settings.pool_max_tasks_per_child, @@ -588,9 +590,9 @@ async def poll_for_tasks(): process_task(task, parent_ctx, logger, limiter, loop, pool) ) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) # back off before retrying diff --git a/workers/aragorn_pathfinder/worker.py b/workers/aragorn_pathfinder/worker.py index 9cc0f3b..dc003d5 100644 --- a/workers/aragorn_pathfinder/worker.py +++ b/workers/aragorn_pathfinder/worker.py @@ -16,6 +16,7 @@ get_running_callbacks, save_message, ) +from shepherd_utils.logger import get_worker_logger from shepherd_utils.otel import setup_tracer from shepherd_utils.shared import ( get_tasks, @@ -29,6 +30,7 @@ CONSUMER = str(uuid.uuid4())[:8] TASK_LIMIT = 10 tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) async def shadowfax(task, logger: logging.Logger) -> str: @@ -278,9 +280,9 @@ async def poll_for_tasks(): ): asyncio.create_task(process_task(task, parent_ctx, logger, limiter)) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) # back off before retrying diff --git a/workers/aragorn_score/worker.py b/workers/aragorn_score/worker.py index c4f7c77..d5db81b 100644 --- a/workers/aragorn_score/worker.py +++ b/workers/aragorn_score/worker.py @@ -14,6 +14,7 @@ from shepherd_utils.config import settings from shepherd_utils.cpu import resolve_pool_workers from shepherd_utils.db import get_message_sync, save_message_sync +from shepherd_utils.logger import get_worker_logger from shepherd_utils.otel import setup_tracer from shepherd_utils.process_pool import ProcessPoolManager from shepherd_utils.shared import get_tasks, run_task_lifecycle @@ -25,6 +26,7 @@ CONSUMER = str(uuid.uuid4())[:8] TASK_LIMIT = 10 tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) DEFAULT_WEIGHT = 1e-2 @@ -1255,8 +1257,8 @@ async def poll_for_tasks(): # Size the pool by the pod's actual CPU allocation (cgroup limit), not # os.cpu_count() -- see aragorn_omnicorp.poll_for_tasks. Each child loads a # full message, so this also bounds peak memory. POOL_MAX_WORKERS overrides. - max_workers = resolve_pool_workers(TASK_LIMIT, logging.getLogger(STREAM)) - logging.info(f"{STREAM}: process pool sized to {max_workers} worker(s).") + max_workers = resolve_pool_workers(TASK_LIMIT, LOGGER) + LOGGER.info(f"{STREAM}: process pool sized to {max_workers} worker(s).") pool = ProcessPoolManager( max_workers, max_tasks_per_child=settings.pool_max_tasks_per_child, @@ -1272,9 +1274,9 @@ async def poll_for_tasks(): process_task(task, parent_ctx, logger, limiter, loop, pool) ) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) # back off before retrying diff --git a/workers/arax/worker.py b/workers/arax/worker.py index 84ab9e4..34a9b46 100644 --- a/workers/arax/worker.py +++ b/workers/arax/worker.py @@ -11,6 +11,7 @@ from shepherd_utils.config import settings from shepherd_utils.db import get_message, save_message +from shepherd_utils.logger import get_worker_logger from shepherd_utils.otel import setup_tracer from shepherd_utils.shared import get_tasks, run_task_lifecycle @@ -20,6 +21,7 @@ CONSUMER = str(uuid.uuid4())[:8] TASK_LIMIT = 10 tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) def is_pathfinder_query(message): @@ -82,9 +84,9 @@ async def poll_for_tasks(): ): asyncio.create_task(process_task(task, parent_ctx, logger, limiter)) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) # back off before retrying diff --git a/workers/arax_pathfinder/worker.py b/workers/arax_pathfinder/worker.py index e1e8e9f..bbe2d0b 100644 --- a/workers/arax_pathfinder/worker.py +++ b/workers/arax_pathfinder/worker.py @@ -14,9 +14,10 @@ from shepherd_utils.config import settings from shepherd_utils.cpu import resolve_pool_workers from shepherd_utils.data_download import ( + arax_blocked_list_path, arax_pathfinder_sqlite_paths, + ensure_arax_blocked_list, ensure_arax_pathfinder_dbs, - ensure_http_files_dataset, ) from shepherd_utils.db import ( get_message_sync, @@ -25,6 +26,7 @@ from shepherd_utils.inject_shepherd_arax_provenance import ( add_shepherd_arax_to_edge_sources, ) +from shepherd_utils.logger import get_worker_logger from shepherd_utils.otel import setup_tracer from shepherd_utils.process_pool import ProcessPoolManager from shepherd_utils.shared import get_tasks, run_task_lifecycle @@ -36,6 +38,7 @@ CONSUMER = str(uuid.uuid4())[:8] TASK_LIMIT = 10 tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) NUM_TOTAL_HOPS = 4 MAX_HOPS_TO_EXPLORE = 4 @@ -43,13 +46,6 @@ PRUNE_TOP_K = 75 NODE_DEGREE_THRESHOLD = 10000 -# The ARAX blocked-concept list, fetched once at worker startup (see -# poll_for_tasks) and read by each pool child. Kept in the working directory -# (/app in the image) so it needs no extra volume mount. -BLOCKED_LIST_DIR = "." -BLOCKED_LIST_FILENAME = "general_concepts.json" -BLOCKED_LIST_PATH = Path(BLOCKED_LIST_DIR) / BLOCKED_LIST_FILENAME - BIOLINK_CACHE_DIR = "/tmp/biolink" REHYDRATE_TIMEOUT_SEC = 30.0 @@ -65,30 +61,16 @@ _descendants_cache: dict = {} -def ensure_blocked_list(logger: logging.Logger) -> None: - """Fetch the ARAX blocked-concept list if it isn't on disk yet. - - Uses the shared downloader so the file lands via a temp file + atomic - rename; the previous per-task ``requests.get`` wrote the destination - directly, so concurrent tasks could race on a half-written file. - Idempotent, so it is safe to call at startup and again lazily in a child. - """ - ensure_http_files_dataset( - name="arax_blocked_list", - target_dir=BLOCKED_LIST_DIR, - file_sources={BLOCKED_LIST_FILENAME: settings.arax_blocked_list_url}, - logger=logger, - ) - - def get_blocked_list(logger: logging.Logger): """``(blocked_curies, blocked_synonyms)``, parsed once per pool child.""" global _blocked_list_cache if _blocked_list_cache is None: - if not BLOCKED_LIST_PATH.exists(): - # Startup fetch failed or this child outlived a wiped working dir. - ensure_blocked_list(logger) - with open(BLOCKED_LIST_PATH, "r") as file: + blocked_list_path = Path(arax_blocked_list_path()) + if not blocked_list_path.exists(): + # Startup fetch failed, or the volume was replaced under a + # long-lived child. + ensure_arax_blocked_list(logger) + with open(blocked_list_path, "r") as file: json_block_list = json.load(file) synonyms = set(s.lower() for s in json_block_list["synonyms"]) _blocked_list_cache = (set(json_block_list["curies"]), synonyms) @@ -326,21 +308,25 @@ async def _run(task, logger): async def poll_for_tasks(): """On initialization, poll indefinitely for available tasks.""" - startup_logger = logging.getLogger(STREAM) # Ensure the two sqlite databases exist before any task tries to open them # (a first-run local `docker compose up` starts with the volume-mounted - # directory empty). No-op once present or when no scp source is configured - # (e.g. production, where the data is mounted out of band). - ensure_arax_pathfinder_dbs(startup_logger) + # directory empty). No-op once present -- which is the case in production, + # where the data is mounted out of band. Note the "already present" check is + # an exact match on ARAX_PATHFINDER_DBS_DIR plus the tier-versioned + # filenames, so a deployment whose mount path or ARAX_PATHFINDER_TIER_VERSION + # disagrees with those settings downloads the databases again rather than + # using the mounted copies. + ensure_arax_pathfinder_dbs(LOGGER) # Fetch the blocked-concept list once here rather than per task, so the pool - # children only ever read it. - ensure_blocked_list(startup_logger) + # children only ever read it. It lives on the same volume as the sqlite dbs, + # so this is a no-op after the first run. + ensure_arax_blocked_list(LOGGER) loop = asyncio.get_running_loop() # Size the pool by the pod's actual CPU allocation (cgroup limit), not # os.cpu_count() -- see aragorn_score.poll_for_tasks. Each child loads a full # message, so this also bounds peak memory. POOL_MAX_WORKERS overrides. - max_workers = resolve_pool_workers(TASK_LIMIT, startup_logger) - logging.info(f"{STREAM}: process pool sized to {max_workers} worker(s).") + max_workers = resolve_pool_workers(TASK_LIMIT, LOGGER) + LOGGER.info(f"{STREAM}: process pool sized to {max_workers} worker(s).") pool = ProcessPoolManager( max_workers, max_tasks_per_child=settings.pool_max_tasks_per_child, @@ -356,9 +342,9 @@ async def poll_for_tasks(): process_task(task, parent_ctx, logger, limiter, loop, pool) ) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) # back off before retrying diff --git a/workers/arax_rank/worker.py b/workers/arax_rank/worker.py index af9cb06..197dc12 100644 --- a/workers/arax_rank/worker.py +++ b/workers/arax_rank/worker.py @@ -18,6 +18,7 @@ from shepherd_utils.config import settings from shepherd_utils.cpu import resolve_pool_workers from shepherd_utils.db import get_message_sync, save_message_sync +from shepherd_utils.logger import get_worker_logger from shepherd_utils.otel import setup_tracer from shepherd_utils.process_pool import ProcessPoolManager from shepherd_utils.shared import get_tasks, run_task_lifecycle @@ -31,6 +32,7 @@ CONSUMER = str(uuid.uuid4())[:8] TASK_LIMIT = 10 tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) def rank_message(in_message: dict, logger: logging.Logger) -> dict: @@ -124,8 +126,8 @@ async def poll_for_tasks() -> None: # Size the pool by the pod's actual CPU allocation (cgroup limit), not # os.cpu_count() -- see aragorn_omnicorp.poll_for_tasks. Each child loads a # full message, so this also bounds peak memory. POOL_MAX_WORKERS overrides. - max_workers = resolve_pool_workers(TASK_LIMIT, logging.getLogger(STREAM)) - logging.info(f"{STREAM}: process pool sized to {max_workers} worker(s).") + max_workers = resolve_pool_workers(TASK_LIMIT, LOGGER) + LOGGER.info(f"{STREAM}: process pool sized to {max_workers} worker(s).") pool = ProcessPoolManager( max_workers, max_tasks_per_child=settings.pool_max_tasks_per_child, @@ -142,9 +144,9 @@ async def poll_for_tasks() -> None: process_task(task, parent_ctx, logger, limiter, loop, pool) ) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) # back off before retrying diff --git a/workers/bte/worker.py b/workers/bte/worker.py index c882263..f1aee0d 100644 --- a/workers/bte/worker.py +++ b/workers/bte/worker.py @@ -6,6 +6,7 @@ import uuid from shepherd_utils.db import get_message +from shepherd_utils.logger import get_worker_logger from shepherd_utils.otel import setup_tracer from shepherd_utils.shared import ( examine_query, @@ -20,6 +21,7 @@ CONSUMER = str(uuid.uuid4())[:8] TASK_LIMIT = 100 tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) async def bte(task, logger: logging.Logger): @@ -69,9 +71,9 @@ async def poll_for_tasks(): ): asyncio.create_task(process_task(task, parent_ctx, logger, limiter)) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) # back off before retrying diff --git a/workers/bte_lookup/worker.py b/workers/bte_lookup/worker.py index 21f7391..5e3c1e4 100644 --- a/workers/bte_lookup/worker.py +++ b/workers/bte_lookup/worker.py @@ -23,6 +23,7 @@ remove_callback_id, save_message, ) +from shepherd_utils.logger import get_worker_logger from shepherd_utils.otel import setup_tracer from shepherd_utils.shared import get_tasks, run_task_lifecycle @@ -33,6 +34,7 @@ CONSUMER = str(uuid.uuid4())[:8] TASK_LIMIT = 100 tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) def examine_query(message): @@ -443,9 +445,9 @@ async def poll_for_tasks(): ): asyncio.create_task(process_task(task, parent_ctx, logger, limiter)) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) # back off before retrying diff --git a/workers/example_ara/worker.py b/workers/example_ara/worker.py index 12f0449..db8069d 100644 --- a/workers/example_ara/worker.py +++ b/workers/example_ara/worker.py @@ -7,6 +7,7 @@ import uuid from shepherd_utils.db import get_message from shepherd_utils.shared import get_tasks, run_task_lifecycle +from shepherd_utils.logger import get_worker_logger from shepherd_utils.otel import setup_tracer # Queue name @@ -15,6 +16,7 @@ CONSUMER = str(uuid.uuid4())[:8] TASK_LIMIT = 100 tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) async def example_ara(task, logger: logging.Logger): @@ -50,9 +52,9 @@ async def poll_for_tasks(): ): asyncio.create_task(process_task(task, parent_ctx, logger, limiter)) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) # back off before retrying diff --git a/workers/example_lookup/worker.py b/workers/example_lookup/worker.py index d95b458..0865d03 100644 --- a/workers/example_lookup/worker.py +++ b/workers/example_lookup/worker.py @@ -17,6 +17,7 @@ get_running_callbacks, save_message, ) +from shepherd_utils.logger import get_worker_logger from shepherd_utils.otel import setup_tracer from shepherd_utils.shared import get_tasks, run_task_lifecycle @@ -26,6 +27,7 @@ CONSUMER = str(uuid.uuid4())[:8] TASK_LIMIT = 100 tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) async def example_lookup(task, logger: logging.Logger): @@ -135,9 +137,9 @@ async def poll_for_tasks(): ): asyncio.create_task(process_task(task, parent_ctx, logger, limiter)) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) # back off before retrying diff --git a/workers/example_score/worker.py b/workers/example_score/worker.py index 8236f53..0b935ce 100644 --- a/workers/example_score/worker.py +++ b/workers/example_score/worker.py @@ -7,6 +7,7 @@ import uuid from shepherd_utils.db import get_message, save_message from shepherd_utils.shared import get_tasks, run_task_lifecycle +from shepherd_utils.logger import get_worker_logger from shepherd_utils.otel import setup_tracer # Queue name @@ -15,6 +16,7 @@ CONSUMER = str(uuid.uuid4())[:8] TASK_LIMIT = 3 tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) async def example_score(task, logger: logging.Logger): @@ -47,9 +49,9 @@ async def poll_for_tasks(): ): asyncio.create_task(process_task(task, parent_ctx, logger, limiter)) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) # back off before retrying diff --git a/workers/filter_analyses_top_n/worker.py b/workers/filter_analyses_top_n/worker.py index 5db85de..d8402c1 100644 --- a/workers/filter_analyses_top_n/worker.py +++ b/workers/filter_analyses_top_n/worker.py @@ -6,6 +6,7 @@ import uuid from shepherd_utils.db import get_message, save_message, get_query_state from shepherd_utils.shared import get_tasks, run_task_lifecycle +from shepherd_utils.logger import get_worker_logger from shepherd_utils.otel import setup_tracer # Queue name @@ -14,6 +15,7 @@ CONSUMER = str(uuid.uuid4())[:8] TASK_LIMIT = 5 tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) async def filter_analyses_top_n(task, logger: logging.Logger): @@ -56,9 +58,9 @@ async def poll_for_tasks(): ): asyncio.create_task(process_task(task, parent_ctx, logger, limiter)) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) # back off before retrying diff --git a/workers/filter_kgraph_orphans/worker.py b/workers/filter_kgraph_orphans/worker.py index 7cffeda..5627589 100644 --- a/workers/filter_kgraph_orphans/worker.py +++ b/workers/filter_kgraph_orphans/worker.py @@ -6,6 +6,7 @@ import uuid from shepherd_utils.db import get_message, save_message +from shepherd_utils.logger import get_worker_logger from shepherd_utils.otel import setup_tracer from shepherd_utils.shared import ( filter_kgraph_orphans, @@ -19,6 +20,7 @@ CONSUMER = str(uuid.uuid4())[:8] TASK_LIMIT = 10 tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) async def do_filter_kgraph_orphans(task, logger: logging.Logger): @@ -51,9 +53,9 @@ async def poll_for_tasks(): ): asyncio.create_task(process_task(task, parent_ctx, logger, limiter)) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) # back off before retrying diff --git a/workers/filter_results_top_n/worker.py b/workers/filter_results_top_n/worker.py index 4652669..d535e2a 100644 --- a/workers/filter_results_top_n/worker.py +++ b/workers/filter_results_top_n/worker.py @@ -6,6 +6,7 @@ import uuid from shepherd_utils.db import get_message, save_message, get_query_state from shepherd_utils.shared import get_tasks, run_task_lifecycle +from shepherd_utils.logger import get_worker_logger from shepherd_utils.otel import setup_tracer # Queue name @@ -14,6 +15,7 @@ CONSUMER = str(uuid.uuid4())[:8] TASK_LIMIT = 5 tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) async def filter_results_top_n(task, logger: logging.Logger): @@ -59,9 +61,9 @@ async def poll_for_tasks(): ): asyncio.create_task(process_task(task, parent_ctx, logger, limiter)) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) # back off before retrying diff --git a/workers/finish_query/worker.py b/workers/finish_query/worker.py index 3546f2e..2cbed87 100644 --- a/workers/finish_query/worker.py +++ b/workers/finish_query/worker.py @@ -19,6 +19,7 @@ set_query_completed, ) from shepherd_utils.shared import get_tasks +from shepherd_utils.logger import get_worker_logger from shepherd_utils.otel import setup_tracer # Queue name @@ -27,6 +28,7 @@ CONSUMER = str(uuid.uuid4())[:8] TASK_LIMIT = 10 tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) CALLBACK_RETRIES = 3 @@ -130,9 +132,9 @@ async def poll_for_tasks(): ): asyncio.create_task(process_task(task, parent_ctx, logger, limiter)) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) # back off before retrying diff --git a/workers/merge_message/worker.py b/workers/merge_message/worker.py index 6b52901..bdfa0ba 100644 --- a/workers/merge_message/worker.py +++ b/workers/merge_message/worker.py @@ -30,7 +30,7 @@ save_logs, save_message_sync, ) -from shepherd_utils.logger import QueryLogger +from shepherd_utils.logger import QueryLogger, get_worker_logger from shepherd_utils.otel import setup_tracer from shepherd_utils.process_pool import ProcessPoolManager from shepherd_utils.shared import filter_kgraph_orphans, get_tasks, merge_kgraph @@ -41,6 +41,7 @@ CONSUMER = str(uuid.uuid4())[:8] TASK_LIMIT = 10 tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) def get_edgeset(result): @@ -690,7 +691,7 @@ def merge_messages_by_ids( # otherwise handlers would accumulate across the child's successive tasks # and leak one query's logs into the next. query_log_handler = QueryLogger().log_handler - worker_logger = logging.getLogger(f"merge_message.worker.{os.getpid()}") + worker_logger = get_worker_logger(f"merge_message.worker.{os.getpid()}") worker_logger.setLevel(log_level) worker_logger.addHandler(query_log_handler) try: @@ -766,8 +767,8 @@ async def poll_for_tasks(): # and the in-flight task limit below: each merge runs a child that loads the # growing response blob, so pool size == concurrency bounds peak memory. # POOL_MAX_WORKERS overrides. - max_workers = resolve_pool_workers(TASK_LIMIT, logging.getLogger(STREAM)) - logging.info(f"{STREAM}: process pool sized to {max_workers} worker(s).") + max_workers = resolve_pool_workers(TASK_LIMIT, LOGGER) + LOGGER.info(f"{STREAM}: process pool sized to {max_workers} worker(s).") # Shared self-healing pool: spawn-context executor that replaces itself in # place on a BrokenProcessPool (same implementation the aragorn.omnicorp / # aragorn.score / arax.rank workers use). run() swaps the dead pool before @@ -957,13 +958,13 @@ async def process_query(task, parent_ctx, logger, limiter): inflight.add(t) t.add_done_callback(inflight.discard) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") for t in inflight: t.cancel() pool.shutdown() return except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) # back off before retrying diff --git a/workers/score_paths/worker.py b/workers/score_paths/worker.py index 635264e..1c84e63 100644 --- a/workers/score_paths/worker.py +++ b/workers/score_paths/worker.py @@ -1,7 +1,6 @@ """Path scoring module""" import asyncio -import logging import time import uuid from concurrent.futures import ThreadPoolExecutor @@ -17,6 +16,7 @@ from shepherd_utils.config import settings from shepherd_utils.data_download import ensure_pathfinder_embeddings from shepherd_utils.db import get_message_sync, save_message_sync +from shepherd_utils.logger import get_worker_logger from shepherd_utils.otel import setup_tracer from shepherd_utils.shared import get_tasks, run_task_lifecycle @@ -26,6 +26,7 @@ TASK_LIMIT = 4 EMBEDDING_DIR = settings.pathfinder_embeddings_dir tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) def convert_path_to_components(source, target, path, knowledge_graph, logger): @@ -275,7 +276,7 @@ async def poll_for_tasks(): # local `docker compose up` starts with the volume-mounted directory empty). # No-op once present or when no download URL is configured (e.g. production, # where the data is mounted out of band). - ensure_pathfinder_embeddings(logging.getLogger(STREAM)) + ensure_pathfinder_embeddings(LOGGER) clf = XGBClassifier() clf.load_model("model_weights/squashbert_classifier_weights.json") bmt = Toolkit() @@ -283,7 +284,7 @@ async def poll_for_tasks(): EMBEDDING_DIR, readonly=True, lock=False, readahead=False, subdir=True ) count, sample = _probe_cache(embedding_env) - logging.info(f"embeddings cache: {count} entries (sample key: {sample!r})") + LOGGER.info(f"embeddings cache: {count} entries (sample key: {sample!r})") mlp = nn.Sequential( nn.Linear(11 * 768, 1536), nn.GELU(), @@ -304,9 +305,9 @@ async def poll_for_tasks(): ): asyncio.create_task(process_task(task, parent_ctx, logger, limiter)) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) diff --git a/workers/sipr/worker.py b/workers/sipr/worker.py index 83d2610..6e472a7 100644 --- a/workers/sipr/worker.py +++ b/workers/sipr/worker.py @@ -14,6 +14,7 @@ get_message, save_message, ) +from shepherd_utils.logger import get_worker_logger from shepherd_utils.otel import setup_tracer from shepherd_utils.shared import get_tasks, run_task_lifecycle @@ -24,6 +25,7 @@ TASK_LIMIT = 10 MAX_QUERY_TIME = 2400 tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) async def get_neighborhood(id_list: list[str], depth: int, logger): @@ -349,9 +351,9 @@ async def poll_for_tasks(): ): asyncio.create_task(process_task(task, parent_ctx, logger, limiter)) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) # back off before retrying diff --git a/workers/sort_results_score/worker.py b/workers/sort_results_score/worker.py index e3ac1d9..e0790fc 100644 --- a/workers/sort_results_score/worker.py +++ b/workers/sort_results_score/worker.py @@ -11,6 +11,7 @@ get_query_state, ) from shepherd_utils.shared import get_tasks, run_task_lifecycle +from shepherd_utils.logger import get_worker_logger from shepherd_utils.otel import setup_tracer # Queue name @@ -19,6 +20,7 @@ CONSUMER = str(uuid.uuid4())[:8] TASK_LIMIT = 3 tracer = setup_tracer(STREAM) +LOGGER = get_worker_logger(STREAM) async def sort_results_score(task, logger: logging.Logger): @@ -82,9 +84,9 @@ async def poll_for_tasks(): ): asyncio.create_task(process_task(task, parent_ctx, logger, limiter)) except asyncio.CancelledError: - logging.info("Poll loop cancelled, shutting down.") + LOGGER.info("Poll loop cancelled, shutting down.") except Exception as e: - logging.error(f"Error in task polling loop: {e}", exc_info=True) + LOGGER.error(f"Error in task polling loop: {e}", exc_info=True) await asyncio.sleep(5) # back off before retrying