Swag ondemand fix history

root/app/ondemand/container_thread.py

@@ -1,12 +1,17 @@
-from data_classes import DockerHost, OnDemandContainer
-from shared_state import last_accessed_urls, last_accessed_urls_lock, websocket_terminated_urls, websocket_terminated_urls_lock
-
-from datetime import datetime
 import logging
 import os
 import threading
 import time
+from datetime import datetime
+
 import wakeonlan
+from data_classes import DockerHost, OnDemandContainer
+from shared_state import (
+    last_accessed_urls,
+    last_accessed_urls_lock,
+    websocket_terminated_urls,
+    websocket_terminated_urls_lock,
+)
 
 CONTAINER_QUERY_SLEEP = float(os.environ.get("SWAG_ONDEMAND_CONTAINER_QUERY_SLEEP", "5.0"))
 STOP_THRESHOLD = int(os.environ.get("SWAG_ONDEMAND_STOP_THRESHOLD", "600"))
@@ -22,21 +27,24 @@ def __init__(self):
 
     def init_docker_hosts(self):
         docker_host_url = os.environ.get("DOCKER_HOST", None)
-        if docker_host_url and not docker_host_url.startswith("tcp://"):
-            docker_host_url = f"tcp://{docker_host_url}:2375"
-        self.docker_hosts.append(DockerHost(url=docker_host_url))
-
+        if docker_host_url:
+            if not docker_host_url.startswith("tcp://"):
+                docker_host_url = f"tcp://{docker_host_url}:2375"
+            self.docker_hosts.append(DockerHost(url=docker_host_url))
+
         remote_hosts_env_vars = {key: value for key, value in os.environ.items() if key.startswith(REMOTE_HOSTS_PREFIX)}
         for i in range(1, 21):
             if f"{REMOTE_HOSTS_PREFIX}{i}" not in remote_hosts_env_vars:
                 break
-            
+
             docker_host_url = remote_hosts_env_vars[f"{REMOTE_HOSTS_PREFIX}{i}"]
             if docker_host_url and not docker_host_url.startswith("tcp://"):
                 docker_host_url = f"tcp://{docker_host_url}:2375"
             remote_host = DockerHost(url=docker_host_url)
             remote_host.wol_mac = remote_hosts_env_vars.get(f"{REMOTE_HOSTS_PREFIX}{i}_WOL_MAC", None)
-            remote_host.wol_broadcast = remote_hosts_env_vars.get(f"{REMOTE_HOSTS_PREFIX}{i}_WOL_BROADCAST", "255.255.255.255")
+            remote_host.wol_broadcast = remote_hosts_env_vars.get(
+                f"{REMOTE_HOSTS_PREFIX}{i}_WOL_BROADCAST", "255.255.255.255"
+            )
             remote_host.wol_urls = remote_hosts_env_vars.get(f"{REMOTE_HOSTS_PREFIX}{i}_WOL_URLS", None)
             remote_host.wol_port = int(remote_hosts_env_vars.get(f"{REMOTE_HOSTS_PREFIX}{i}_WOL_PORT", "9"))
             remote_host.wol_interface = remote_hosts_env_vars.get(f"{REMOTE_HOSTS_PREFIX}{i}_WOL_INTERFACE", None)
@@ -60,9 +68,11 @@ def process_containers(self):
 
             for container in containers:
                 default_url = container.labels.get("swag_url", f"{container.name}.").rstrip("*")
-                container_urls = container.labels.get("swag_ondemand_urls", f"https://{default_url},http://{default_url}")
+                container_urls = container.labels.get(
+                    "swag_ondemand_urls", f"https://{default_url},http://{default_url}"
+                )
                 websocket = container.labels.get("swag_ondemand_websocket", "0").lower() in ("true", "1")
-                
+
                 if container.name in docker_host.ondemand_containers:
                     docker_host.ondemand_containers[container.name].status = container.status
                     docker_host.ondemand_containers[container.name].websocket = websocket
@@ -71,13 +81,9 @@ def process_containers(self):
                         logging.info(f"Updated urls for {container.name} on {docker_host.url} to: {container_urls}")
                 else:
                     docker_host.ondemand_containers[container.name] = OnDemandContainer(
-                        status=container.status,
-                        urls=container_urls,
-                        last_accessed=datetime.now(),
-                        websocket=websocket
+                        status=container.status, urls=container_urls, last_accessed=datetime.now(), websocket=websocket
                     )
                     logging.info(f"Started monitoring {container.name} on {docker_host.url} for urls: {container_urls}")
-
 
     def stop_containers(self, websocket_terminated_urls_combined: str):
         for docker_host in self.docker_hosts:
@@ -86,7 +92,7 @@ def stop_containers(self, websocket_terminated_urls_combined: str):
             for container_name, ondemand_container in docker_host.ondemand_containers.items():
                 if ondemand_container.status != "running":
                     continue
-                
+
                 if ondemand_container.websocket and not ondemand_container.terminated:
                     for ondemand_url in ondemand_container.urls.split(","):
                         if ondemand_url in websocket_terminated_urls_combined:
@@ -99,19 +105,19 @@ def stop_containers(self, websocket_terminated_urls_combined: str):
                 inactive_seconds = (datetime.now() - ondemand_container.last_accessed).total_seconds()
                 if inactive_seconds < STOP_THRESHOLD:
                     continue
-                
+
                 container = docker_host.get_container(container_name)
                 if not container:
                     continue
-                
+
                 container.stop()
                 ondemand_container.status = "exited"
                 logging.info(f"Stopped {container_name} on {docker_host.url} after {STOP_THRESHOLD}s of inactivity")
 
     def start_containers(self, last_accessed_urls_combined: str):
         if not last_accessed_urls_combined:
             return
-        
+
         for docker_host in self.docker_hosts:
             if not docker_host.is_connected:
                 continue
@@ -123,10 +129,10 @@ def start_containers(self, last_accessed_urls_combined: str):
                         ondemand_container.terminated = False
                         accessed = True
                         break
-                
+
                 if not accessed or ondemand_container.status == "running":
                     continue
-                
+
                 container = docker_host.get_container(container_name)
                 if not container:
                     continue
@@ -138,19 +144,23 @@ def start_containers(self, last_accessed_urls_combined: str):
     def send_wol(self, last_accessed_urls_combined: str):
         if not last_accessed_urls_combined:
             return
-        
+
         for docker_host in self.docker_hosts:
             if not docker_host.wol_mac or not docker_host.wol_urls or docker_host.is_connected:
                 continue
             for wol_url in docker_host.wol_urls.split(","):
                 if wol_url in last_accessed_urls_combined:
                     wakeonlan.send_magic_packet(
-                        docker_host.wol_mac, 
-                        ip_address=docker_host.wol_broadcast, 
-                        port=docker_host.wol_port, 
-                        interface=docker_host.wol_interface
+                        docker_host.wol_mac,
+                        ip_address=docker_host.wol_broadcast,
+                        port=docker_host.wol_port,
+                        interface=docker_host.wol_interface,
+                    )
+                    logging.info(
+                        "Sent a WoL packet to mac {docker_host.wol_mac} via broadcast "
+                        "{docker_host.wol_broadcast} on port {docker_host.wol_port} on "
+                        "interface {docker_host.wol_interface or 'default'} activated by {wol_url}"
                     )
-                    logging.info(f"Sent a WoL packet to mac {docker_host.wol_mac} via broadcast {docker_host.wol_broadcast} on port {docker_host.wol_port} on interface {docker_host.wol_interface or 'default'} activated by {wol_url}")
                     break
 
     def run(self):
@@ -159,11 +169,11 @@ def run(self):
                 with last_accessed_urls_lock:
                     last_accessed_urls_combined = ",".join(last_accessed_urls)
                     last_accessed_urls.clear()
-                
+
                 with websocket_terminated_urls_lock:
                     websocket_terminated_urls_combined = ",".join(websocket_terminated_urls)
                     websocket_terminated_urls.clear()
-                
+
                 self.send_wol(last_accessed_urls_combined)
                 self.process_containers()
                 self.start_containers(last_accessed_urls_combined)

root/app/ondemand/data_classes.py

@@ -1,9 +1,10 @@
+import logging
 from dataclasses import dataclass, field
 from datetime import datetime
+
 import docker
-import logging
 import requests
-from typing import Optional
+
 
 @dataclass
 class OnDemandContainer:
@@ -13,15 +14,16 @@ class OnDemandContainer:
     websocket: bool
     terminated: bool = False
 
+
 @dataclass
 class DockerHost:
     url: str
-    client: Optional[docker.DockerClient] = None
-    wol_mac: Optional[str] = None
+    client: docker.DockerClient | None = None
+    wol_mac: str | None = None
     wol_broadcast: str = "255.255.255.255"
     wol_port: int = 9
-    wol_interface: Optional[str] = None
-    wol_urls: Optional[str] = None
+    wol_interface: str | None = None
+    wol_urls: str | None = None
     is_connected: bool = False
     was_connected: bool = False
     ondemand_containers: dict[str, OnDemandContainer] = field(default_factory=dict)
@@ -32,17 +34,17 @@ def check_connection(self, timeout: int):
             if self.client and self.client.ping():
                 self.is_connected = True
                 return
-            
+
             if self.url:
                 self.client = docker.DockerClient(base_url=self.url, timeout=timeout)
             else:
                 self.client = docker.from_env(timeout=timeout)
                 self.url = "unix:///var/run/docker.sock"
-            
+
             self.is_connected = True
             if not self.was_connected:
                 logging.info(f"Connection to {self.url} has been restored")
-        except (docker.errors.DockerException, requests.exceptions.ConnectionError, requests.exceptions.ReadTimeout):
+        except (docker.errors.DockerException, requests.exceptions.ConnectionError, requests.exceptions.ReadTimeout):  # ty:ignore[possibly-missing-submodule]
             self.client = None
             self.is_connected = False
             if self.was_connected:
@@ -59,7 +61,7 @@ def get_container(self, container_name: str):
             if not client or not self.is_connected:
                 return None
             return client.containers.get(container_name)
-        except (docker.errors.DockerException, requests.exceptions.ConnectionError, requests.exceptions.ReadTimeout):
+        except (docker.errors.DockerException, requests.exceptions.ConnectionError, requests.exceptions.ReadTimeout):  # ty:ignore[possibly-missing-submodule]
             self.handle_disconnect()
             return None
 
@@ -69,6 +71,6 @@ def get_containers(self):
             if not client or not self.is_connected:
                 return None
             return client.containers.list(all=True, filters={"label": ["swag_ondemand=enable"]})
-        except (docker.errors.DockerException, requests.exceptions.ConnectionError, requests.exceptions.ReadTimeout):
+        except (docker.errors.DockerException, requests.exceptions.ConnectionError, requests.exceptions.ReadTimeout):  # ty:ignore[possibly-missing-submodule]
             self.handle_disconnect()
             return None

root/app/ondemand/healthcheck_thread.py

@@ -1,10 +1,10 @@
-from data_classes import DockerHost
-
-from concurrent.futures import ThreadPoolExecutor
 import logging
 import os
 import threading
 import time
+from concurrent.futures import ThreadPoolExecutor
+
+from data_classes import DockerHost
 
 DOCKER_API_TIMEOUT = int(os.environ.get("SWAG_ONDEMAND_DOCKER_API_TIMEOUT", "5"))
 
@@ -25,11 +25,11 @@ def run(self):
                     executor.submit(docker_host.check_connection, DOCKER_API_TIMEOUT)
                     for docker_host in self.docker_hosts
                 ]
-                
+
                 for future in futures:
                     try:
                         future.result()
                     except Exception as e:
                         logging.exception(e)
-                
+
                 time.sleep(1)

root/app/ondemand/log_reader_thread.py

@@ -1,55 +1,96 @@
-from shared_state import last_accessed_urls, last_accessed_urls_lock, websocket_terminated_urls, websocket_terminated_urls_lock
-
 import logging
 import os
+import re
 import threading
 import time
+from datetime import datetime
+
+from shared_state import (
+    last_accessed_urls,
+    last_accessed_urls_lock,
+    websocket_terminated_urls,
+    websocket_terminated_urls_lock,
+)
 
 ACCESS_LOG_FILE = "/config/log/nginx/access.log"
 LOG_READER_SLEEP = float(os.environ.get("SWAG_ONDEMAND_LOG_READER_SLEEP", "1.0"))
+STOP_THRESHOLD = int(os.environ.get("SWAG_ONDEMAND_STOP_THRESHOLD", "600"))
+TIMESTAMP_REGEX = re.compile(r"\[(\d{2}/\w{3}/\d{4}:\d{2}:\d{2}:\d{2} [+-]\d{4})\]")
 
 
 class LogReaderThread(threading.Thread):
     def __init__(self):
         super().__init__(name="LogReaderThread")
         self.daemon = True
 
-    def tail(self, f):
-        f.seek(0,2)
-        inode = os.fstat(f.fileno()).st_ino
-
-        while True:
-            line = f.readline()
-            if not line:
-                time.sleep(LOG_READER_SLEEP)
-                if os.stat(ACCESS_LOG_FILE).st_ino != inode:
-                    f.close()
-                    f = open(ACCESS_LOG_FILE, 'r')
-                    inode = os.fstat(f.fileno()).st_ino
-                continue
-            yield line
-
     def run(self):
+        self._process_historical_logs()
+
         while True:
             try:
                 if not os.path.exists(ACCESS_LOG_FILE):
                     time.sleep(1)
                     continue
 
                 logfile = open(ACCESS_LOG_FILE, "r")
-                for line in self.tail(logfile):
-                    if '" 302 ' in line:
-                        continue
-                    for part in line.split():
-                        if not part.startswith("http"):
-                            continue
-                        if '" 101 ' in line:
-                            with websocket_terminated_urls_lock:
-                                websocket_terminated_urls.add(part)
-                        else:
-                            with last_accessed_urls_lock:
-                                last_accessed_urls.add(part)
-                        break
+                for line in self._tail(logfile):
+                    self._process_line(line, startup_mode=False)
             except Exception as e:
                 logging.exception(e)
                 time.sleep(1)
+
+    def _process_historical_logs(self):
+        if not os.path.exists(ACCESS_LOG_FILE):
+            return
+
+        try:
+            with open(ACCESS_LOG_FILE, "r") as f:
+                for line in f:
+                    log_time = self._parse_nginx_time(line)
+                    if not log_time:
+                        continue
+                    seconds_delta = (datetime.now() - log_time).total_seconds()
+                    if seconds_delta > STOP_THRESHOLD:
+                        continue
+                    self._process_line(line, startup_mode=True)
+        except Exception as e:
+            logging.error(f"Error processing historical logs: {e}")
+
+    def _parse_nginx_time(self, line):
+        match = TIMESTAMP_REGEX.search(line)
+        if match:
+            try:
+                return datetime.strptime(match.group(1), "%d/%b/%Y:%H:%M:%S %z")
+            except ValueError:
+                return None
+        return None
+
+    def _process_line(self, line, startup_mode=False):
+        if '" 302 ' in line:
+            return
+        for part in line.split():
+            if not part.startswith("http"):
+                continue
+
+            if '" 101 ' in line:
+                with websocket_terminated_urls_lock:
+                    websocket_terminated_urls.add(part)
+            elif not startup_mode:
+                with last_accessed_urls_lock:
+                    last_accessed_urls.add(part)
+            break
+
+    def _tail(self, f):
+        f.seek(0, 2)
+        inode = os.fstat(f.fileno()).st_ino
+
+        while True:
+            line = f.readline()
+            if not line:
+                time.sleep(LOG_READER_SLEEP)
+                if os.stat(ACCESS_LOG_FILE).st_ino != inode:
+                    f.close()
+                    f = open(ACCESS_LOG_FILE, "r")
+                    inode = os.fstat(f.fileno()).st_ino
+                continue
+            yield line