diff --git a/backend/tests/flow/test_workers.py b/backend/tests/flow/test_workers.py index 4d8c9a0..a2be684 100644 --- a/backend/tests/flow/test_workers.py +++ b/backend/tests/flow/test_workers.py @@ -69,6 +69,24 @@ def test_a_node_returns_its_value_and_what_it_printed(pool, capsys): assert "seen 21" in capsys.readouterr().out +def test_a_node_can_log_through_the_module_it_imports(pool, capsys): + """The SDK exports `logger`; inside a worker `fluksio` is the reporter. + + Without it, `fluksio.logger.info(...)` died with AttributeError — after + the training it was reporting on had already succeeded. + """ + result = run( + pool, + "import fluksio\n\n\n" + "def process(value):\n" + " fluksio.logger.info('tuned %s', value)\n" + " return {'out': value}\n", + value=7, + ) + assert result == {"out": 7} + assert "tuned 7" in capsys.readouterr().out + + def test_a_failure_keeps_its_class_and_points_at_the_node(pool): with pytest.raises(Exception) as caught: run(pool, "def process():\n raise ValueError('bad input')\n") diff --git a/worker/fluksio_worker/worker_main.py b/worker/fluksio_worker/worker_main.py index 9d7c604..5ba0880 100644 --- a/worker/fluksio_worker/worker_main.py +++ b/worker/fluksio_worker/worker_main.py @@ -45,6 +45,7 @@ import hashlib import inspect import io import json +import logging import tempfile import time import traceback @@ -150,6 +151,18 @@ class _Reporter(ModuleType): return _Declared() +class _LogHandler(logging.Handler): + """Writes to whatever ``sys.stderr`` is at the time of the record. + + A call runs under ``redirect_stderr`` into the buffer that becomes the + node's logs, so a handler holding the stream it was built with would write + past every one of them. + """ + + def emit(self, record: logging.LogRecord) -> None: + sys.stderr.write(self.format(record) + "\n") + + class _Declared: """Stands in for a declaration whose work was done before the run. @@ -294,6 +307,17 @@ def _install_reporter() -> None: """Put ``fluksio`` on the import path of every node this worker runs.""" module = _Reporter("fluksio") module.__doc__ = "Report metrics and progress from inside a node." + # The SDK exports a logger at top level, so a node written against it says + # `fluksio.logger.info(...)` — which used to die with AttributeError, after + # whatever it was reporting on had already succeeded. Its records land in + # the same capture as a print. + logger = logging.getLogger("fluksio") + logger.setLevel(logging.INFO) + logger.addHandler(_LogHandler()) + # Nothing above it should see these twice: a node calling `basicConfig` for + # its own logging installs a root handler on the same stream. + logger.propagate = False + module.logger = logger sys.modules["fluksio"] = module