Merge pull request #7 from DACCS-Climate/support-pipes

support reading log files from pipe-like files

Author: mishaschwartz

README.md

@@ -73,7 +73,7 @@ asyncio.run(my_coroutine())
 
 `track_file` has the following parameters:
 
-- `log_file`: path to a log file to track 
+- `log_file`: path to a log file to track. This can be a regular file, stream (e.g. stdout/stderr), or named FIFO pipe.
 - `line_parsers` a list of [line parsers](#what-is-a-line-parser)
 - `poll_delay`: (see [parameters](#track-parameters) for `track` and `track_async`)
 - `tail`: (see [parameters](#track-parameters) for `track` and `track_async`)
@@ -237,7 +237,8 @@ these scenarios in the following ways:
 
 #### How do I track logs that haven't been written to a file?
 
-Log Parser currently only supports reading logs from files. We'd like to support other log sources in the future though.
+Log Parser currently only supports reading logs from files and named pipes (such as the stdout and stderr streams). 
+We'd like to support other log sources in the future though.
 Feel free to [contribute](#development) to this project if you want to speed up support for this feature.
 
 ## Extensions

log_parser/log_parser.py

@@ -40,6 +40,8 @@ async def _check_file_state(log_io: AsyncFile[str]) -> _FileState:
         return _FileState.DELETED
     file_stat = os.stat(log_io.fileno())
     if same_name_file_stat == file_stat:
+        if not log_io.seekable():
+            return _FileState.NOCHANGE
         if await log_io.tell() > file_stat.st_size:
             return _FileState.TRUNCATED
         else:
@@ -79,7 +81,7 @@ async def track_file(
         log_io = await open_file(log_file)
         logger.debug(f"file '{log_file}' opened for reading")
         try:
-            if tail:
+            if tail and log_io.seekable():
                 await log_io.seek(0, os.SEEK_END)
             while True:
                 file_state = await _check_file_state(log_io)

test/test_log_parser.py

@@ -2,6 +2,7 @@
 import asyncio
 import inspect
 import itertools
+import os
 from contextlib import asynccontextmanager
 
 import anyio
@@ -28,6 +29,12 @@ async def tmp_log(self, tmp_path):
         await log_file.touch()
         return log_file
 
+    @pytest.fixture
+    async def tmp_log_pipe(self, tmp_path):
+        log_file = anyio.Path(tmp_path / "test.log")
+        os.mkfifo(log_file)
+        return log_file
+
     @staticmethod
     async def delayed_write(file, text, delay, mode="w"):
         await asyncio.sleep(delay)
@@ -162,6 +169,14 @@ async def test_file_contains_line_mixed_line_parsers(self, tmp_log):
             )
         assert output == text * 2
 
+    @pytest.mark.parametrize("text", [["test 123"], ["test 123", "test 456"]])
+    async def test_pipe_contains_lines(self, tmp_log_pipe, text):
+        output = []
+        async with silent_timeout(1.1), asyncio.TaskGroup() as tg:
+            tg.create_task(log_parser.track_file(str(tmp_log_pipe), [self.basic_line_parser(output)]))
+            tg.create_task(self.delayed_write(tmp_log_pipe, "\n".join(text), 0.1))
+        assert output == text
+
 
 class _TrackTests(abc.ABC):
     @abc.abstractmethod