Improve harvester offline handling

.gitignore

@@ -2,6 +2,9 @@
 config.yaml
 config.*.yaml
 
+# pygtail offset
+debug.log.offset
+
 # dev files
 .idea
 venv

requirements.txt

@@ -1,3 +1,5 @@
 paramiko==2.7.2
 python-dateutil~=2.8.1
-PyYAML==5.4
+PyYAML==5.4
+retry==0.9.2
+pygtail==0.11.1

src/chia_log/log_consumer.py

@@ -8,18 +8,22 @@
 
 # std
 import logging
-import subprocess
 from abc import ABC, abstractmethod
 from pathlib import Path, PurePosixPath, PureWindowsPath, PurePath
 from threading import Thread
+from time import sleep
 from typing import List, Optional, Tuple
 
 # project
+
 from src.config import check_keys, is_win_platform
 from src.util import OS
 
 # lib
 import paramiko
+from paramiko.channel import ChannelStdinFile, ChannelStderrFile, ChannelFile
+from pygtail import Pygtail  # type: ignore
+from retry import retry
 
 
 class LogConsumerSubscriber(ABC):
@@ -50,49 +54,56 @@ def _notify_subscribers(self, logs: str):
 
 
 class FileLogConsumer(LogConsumer):
-    """Specific implementation for a simple file consumer"""
-
     def __init__(self, log_path: Path):
-        logging.info("Enabled file log consumer.")
         super().__init__()
-        self._log_path = log_path
+        self._expanded_log_path = str(log_path.expanduser())
+        self._offset_path = PurePath("debug.log.offset")
         self._is_running = True
         self._thread = Thread(target=self._consume_loop)
         self._thread.start()
+        self._log_size = 0
 
     def stop(self):
         logging.info("Stopping")
         self._is_running = False
 
+    @retry((FileNotFoundError, PermissionError), delay=2)
     def _consume_loop(self):
-        expanded_user_log_path = str(self._log_path.expanduser())
-        logging.info(f"Consuming log file from {expanded_user_log_path}")
+        while self._is_running:
+            for log_line in Pygtail(self._expanded_log_path, read_from_end=True, offset_file=self._offset_path):
+                self._notify_subscribers(log_line)
 
-        if is_win_platform():
-            consume_command_args = ["powershell.exe", "get-content", expanded_user_log_path, "-tail", "1", "-wait"]
-        else:
-            consume_command_args = ["tail", "-F", expanded_user_log_path]
 
-        f = subprocess.Popen(consume_command_args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        while self._is_running:
-            log_line = f.stdout.readline().decode(encoding="utf-8")
-            self._notify_subscribers(log_line)
+class PosixFileLogConsumer(FileLogConsumer):
+    """Specific implementation for a simple file consumer for Linux/MacOS"""
+
+    def __init__(self, log_path: Path):
+        logging.info("Enabled Posix file log consumer.")
+        super(PosixFileLogConsumer, self).__init__(log_path)
+
+
+class WindowsFileLogConsumer(FileLogConsumer):
+    """Specific implementation for a simple file consumer for Windows"""
+
+    def __init__(self, log_path: Path):
+        logging.info("Enabled Windows file log consumer.")
+        super(WindowsFileLogConsumer, self).__init__(log_path)
 
 
 class NetworkLogConsumer(LogConsumer):
-    """Consume logs over the network"""
+    """Consume logs over SSH from a remote harvester"""
 
     def __init__(
         self, remote_log_path: PurePath, remote_user: str, remote_host: str, remote_port: int, remote_platform: OS
     ):
-        logging.info("Enabled network log consumer.")
         super().__init__()
 
         self._remote_user = remote_user
         self._remote_host = remote_host
         self._remote_port = remote_port
         self._remote_log_path = remote_log_path
         self._remote_platform = remote_platform
+        self._log_size = 0
 
         self._ssh_client = paramiko.client.SSHClient()
         self._ssh_client.load_system_host_keys()
@@ -113,18 +124,70 @@ def _consume_loop(self):
             + f" from {self._remote_host}:{self._remote_port} ({self._remote_platform})"
         )
 
-        if self._remote_platform == OS.WINDOWS:
-            stdin, stdout, stderr = self._ssh_client.exec_command(
-                f"powershell.exe Get-Content {self._remote_log_path} -Wait -Tail 1"
-            )
-        else:
-            stdin, stdout, stderr = self._ssh_client.exec_command(f"tail -F {self._remote_log_path}")
+
+class PosixNetworkLogConsumer(NetworkLogConsumer):
+    """Consume logs over SSH from a remote Linux/MacOS harvester"""
+
+    def __init__(
+        self, remote_log_path: PurePath, remote_user: str, remote_host: str, remote_port: int, remote_platform: OS
+    ):
+        logging.info("Enabled Posix network log consumer.")
+        super(PosixNetworkLogConsumer, self).__init__(
+            remote_log_path, remote_user, remote_host, remote_port, remote_platform
+        )
+
+    def _consume_loop(self):
+        super(PosixNetworkLogConsumer, self)._consume_loop()
+
+        stdin, stdout, stderr = self._ssh_client.exec_command(f"tail -F {self._remote_log_path}")
 
         while self._is_running:
             log_line = stdout.readline()
             self._notify_subscribers(log_line)
 
 
+class WindowsNetworkLogConsumer(NetworkLogConsumer):
+    """Consume logs over SSH from a remote Windows harvester"""
+
+    def __init__(
+        self, remote_log_path: PurePath, remote_user: str, remote_host: str, remote_port: int, remote_platform: OS
+    ):
+        logging.info("Enabled Windows network log consumer.")
+        super(WindowsNetworkLogConsumer, self).__init__(
+            remote_log_path, remote_user, remote_host, remote_port, remote_platform
+        )
+
+    def _consume_loop(self):
+        super(WindowsNetworkLogConsumer, self)._consume_loop()
+
+        stdin, stdout, stderr = self._read_log()
+
+        while self._is_running:
+            if self._has_rotated(self._remote_log_path):
+                sleep(1)
+                stdin, stdout, stderr = self._read_log()
+
+            log_line = stdout.readline()
+            self._notify_subscribers(log_line)
+
+    def _read_log(self) -> Tuple[ChannelStdinFile, ChannelFile, ChannelStderrFile]:
+        stdin, stdout, stderr = self._ssh_client.exec_command(
+            f"powershell.exe Get-Content {self._remote_log_path} -Wait -Tail 1"
+        )
+
+        return stdin, stdout, stderr
+
+    def _has_rotated(self, path: PurePath) -> bool:
+        stdin, stdout, stderr = self._ssh_client.exec_command(
+            f"powershell.exe Write-Host((Get-Item {str(path)}).length"
+        )
+
+        old_size = self._log_size
+        self._log_size = int(stdout.readline())
+
+        return old_size > self._log_size
+
+
 def get_host_info(host: str, user: str, path: str, port: int) -> Tuple[OS, PurePath]:
 
     client = paramiko.client.SSHClient()
@@ -164,12 +227,15 @@ def create_log_consumer_from_config(config: dict) -> Optional[LogConsumer]:
     if enabled_consumer == "file_log_consumer":
         if not check_keys(required_keys=["file_path"], config=enabled_consumer_config):
             return None
-        return FileLogConsumer(log_path=Path(enabled_consumer_config["file_path"]))
+
+        if is_win_platform():
+            return WindowsFileLogConsumer(log_path=Path(enabled_consumer_config["file_path"]))
+        else:
+            return PosixFileLogConsumer(log_path=Path(enabled_consumer_config["file_path"]))
 
     if enabled_consumer == "network_log_consumer":
         if not check_keys(
-            required_keys=["remote_file_path", "remote_host", "remote_user"],
-            config=enabled_consumer_config,
+            required_keys=["remote_file_path", "remote_host", "remote_user"], config=enabled_consumer_config
         ):
             return None
 
@@ -183,13 +249,22 @@ def create_log_consumer_from_config(config: dict) -> Optional[LogConsumer]:
             remote_port,
         )
 
-        return NetworkLogConsumer(
-            remote_log_path=path,
-            remote_host=enabled_consumer_config["remote_host"],
-            remote_user=enabled_consumer_config["remote_user"],
-            remote_port=remote_port,
-            remote_platform=platform,
-        )
+        if platform == OS.WINDOWS:
+            return WindowsNetworkLogConsumer(
+                remote_log_path=path,
+                remote_host=enabled_consumer_config["remote_host"],
+                remote_user=enabled_consumer_config["remote_user"],
+                remote_port=remote_port,
+                remote_platform=platform,
+            )
+        else:
+            return PosixNetworkLogConsumer(
+                remote_log_path=path,
+                remote_host=enabled_consumer_config["remote_host"],
+                remote_user=enabled_consumer_config["remote_user"],
+                remote_port=remote_port,
+                remote_platform=platform,
+            )
 
     logging.error("Unhandled consumer type")
     return None