Just needs test coverage

local/configs/package.yaml

@@ -40,5 +40,9 @@ init_local: |
   METRICS_NAME = "metrics"
   DEFAULT_EXT = "yaml"
 
+mypy_local: |
+  [mypy-aiofiles.*]
+  ignore_missing_imports = True
+
 ci_local:
   - "- run: mk python-editable"

mypy.ini

@@ -13,3 +13,7 @@ warn_unused_ignores = False
 strict = False
 disallow_any_generics = False
 strict_equality = False
+
+# runtimepy-specific configurations.
+[mypy-aiofiles.*]
+ignore_missing_imports = True

runtimepy/data/factories.yaml

@@ -35,6 +35,7 @@ factories:
   - {name: runtimepy.task.sample.Sample}
   - {name: runtimepy.task.sample.SampleApp}
   - {name: runtimepy.control.step.StepperToggler}
+  - {name: runtimepy.task.file.LogCapture}
 
   # Useful structs.
   - {name: runtimepy.net.arbiter.info.TrigStruct}

runtimepy/mixins/logging.py

@@ -3,10 +3,17 @@
 """
 
 # built-in
+from contextlib import AsyncExitStack
+import io
 import logging
+import os
+from pathlib import Path
+from typing import Any
 
 # third-party
-from vcorelib.logging import LoggerMixin
+import aiofiles
+from vcorelib.logging import LoggerMixin, LoggerType
+from vcorelib.paths import Pathlike, normalize
 
 # internal
 from runtimepy.channel.environment import ChannelEnvironment
@@ -55,3 +62,53 @@ def setup_level_channel(
         env.set(name, initial)
 
         del chan
+
+
+SYSLOG = Path(os.sep, "var", "log", "syslog")
+
+
+class LogCaptureMixin:
+    """A simple async file-reading interface."""
+
+    logger: LoggerType
+
+    # Open aiofiles handles.
+    streams: list[Any]
+
+    async def init_log_capture(
+        self, stack: AsyncExitStack, log_paths: list[Pathlike]
+    ) -> None:
+        """Initialize this task with application information."""
+
+        self.streams = [
+            await stack.enter_async_context(aiofiles.open(x, mode="r"))
+            for x in log_paths
+            if normalize(x).is_file()
+        ]
+
+        # Don't handle any kind of backhaul.
+        for stream in self.streams:
+            await stream.seek(0, io.SEEK_END)
+
+    def log_line(self, data: str) -> None:
+        """Log a line for output."""
+        self.logger.info(data, extra={"external": True})
+
+    async def next_lines(self) -> list[str]:
+        """Get the next line from this log stream."""
+
+        result = []
+        for stream in self.streams:
+            line = (await stream.readline()).rstrip()
+            if line:
+                result.append(line)
+        return result
+
+    async def dispatch_log_capture(self) -> None:
+        """Check for new log data."""
+
+        data = await self.next_lines()
+        while data:
+            for item in data:
+                self.log_line(item)
+            data = await self.next_lines()

runtimepy/task/file/__init__.py

@@ -0,0 +1,42 @@
+"""
+A module implementing file-related task interfaces.
+"""
+
+# third-party
+from vcorelib.paths import Pathlike
+
+# internal
+from runtimepy.mixins.logging import SYSLOG, LogCaptureMixin
+from runtimepy.net.arbiter.info import AppInfo
+from runtimepy.net.arbiter.task import ArbiterTask as _ArbiterTask
+from runtimepy.net.arbiter.task import TaskFactory as _TaskFactory
+
+
+class LogCaptureTask(_ArbiterTask, LogCaptureMixin):
+    """
+    A task that captures all log messages emitted by this program instance.
+    """
+
+    auto_finalize = True
+
+    log_paths: list[Pathlike] = [SYSLOG]
+
+    async def init(self, app: AppInfo) -> None:
+        """Initialize this task with application information."""
+
+        await super().init(app)
+
+        # Allow additional paths via configuration?
+        await self.init_log_capture(app.stack, self.log_paths)
+
+    async def dispatch(self) -> bool:
+        """Dispatch an iteration of this task."""
+
+        await self.dispatch_log_capture()
+        return True
+
+
+class LogCapture(_TaskFactory[LogCaptureTask]):
+    """A factory for the syslog capture task."""
+
+    kind = LogCaptureTask

runtimepy/util.py

@@ -37,7 +37,15 @@ def drain(self) -> list[logging.LogRecord]:
     def drain_str(self) -> list[str]:
         """Drain formatted messages."""
 
-        return [self.format(x) for x in self.drain()]
+        result = []
+        for record in self.drain():
+            result.append(
+                self.format(record)
+                # Respect 'external' logs that don't warrant full formatting.
+                if not getattr(record, "external", False)
+                else record.getMessage()
+            )
+        return result
 
     def __bool__(self) -> bool:
         """Evaluate this instance as boolean."""

tasks/dev.yaml

@@ -4,6 +4,9 @@ includes:
   - dev_no_wait.yaml
   - ../tests/data/valid/connection_arbiter/test_ssl.yaml
 
+tasks:
+  - {name: root_log, factory: log_capture, period_s: 0.1}
+
 port_overrides:
   runtimepy_https_server: 8443
 

tests/data/valid/connection_arbiter/test_ssl.yaml

@@ -4,6 +4,10 @@ ports:
   - {name: runtimepy_secure_websocket_json_server, type: tcp}
   - {name: runtimepy_secure_websocket_data_server, type: tcp}
 
+# This should probably be tested somewhere else.
+tasks:
+  - {name: root_log, factory: log_capture, period_s: 0.1}
+
 servers:
   - factory: runtimepy_http
     kwargs: