configd: send keepalives from streaming actions to enforce client disconnects

src/opnsense/scripts/filter/read_log.py

@@ -37,6 +37,7 @@ import argparse
 import ujson
 import subprocess
 import time
+import select
 sys.path.insert(0, "/usr/local/opnsense/site-python")
 from log_helper import reverse_log_reader
 from params import update_params
@@ -173,12 +174,18 @@ if __name__ == '__main__':
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             bufsize=0,
+            text=True
         )
 
         last_t = time.time()
         try:
             while True:
-                line = f.stdout.readline().decode('utf-8')
+                ready, _, _ = select.select([f.stdout], [], [], 1)
+                if not ready:
+                    # timeout, send keepalive
+                    print("event: keepalive\ndata:\n\n", flush=True)
+                    continue
+                line = f.stdout.readline()
                 if line and line.find('filterlog') > -1:
                     t = time.time()
                     if (t - last_t) > 30:

src/opnsense/scripts/syslog/log_matcher.py

@@ -31,6 +31,7 @@ import glob
 import re
 import sys
 import subprocess
+import select
 from logformats import FormatContainer, BaseLogFormat
 sys.path.insert(0, "/usr/local/opnsense/site-python")
 from log_helper import reverse_log_reader
@@ -60,10 +61,20 @@ class LogMatcher:
             latest = self.log_filenames[0] if len(self.log_filenames) > 0 else ''
         if os.path.exists(latest):
             format_container = FormatContainer(latest)
-            p = subprocess.Popen(['tail', '-f', '-n 0', latest], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, bufsize=0)
+            p = subprocess.Popen(
+                ['tail', '-f', '-n 0', latest],
+                stdout=subprocess.PIPE,
+                stderr=subprocess.DEVNULL,
+                bufsize=0,
+                text=True
+            )
             try:
-                for line in iter(p.stdout.readline, b''):
-                    line = line.decode()
+                while True:
+                    ready, _, _ = select.select([p.stdout], [], [], 1)
+                    if not ready:
+                        yield None
+                        continue
+                    line = p.stdout.readline()
                     if line != "" and self.filter_regexp.match(('%s' % line).lower()):
                         record = self.parse_line(line, format_container)
                         if len(self.severity) == 0 or record['severity'] is None or record['severity'] in self.severity:

src/opnsense/scripts/syslog/streamLog.py

@@ -61,4 +61,8 @@ if __name__ == '__main__':
         print(record, flush=True)
 
     for record in log_matcher.live_match_records():
-        print(f"event: message\ndata:{ujson.dumps(record)}\n\n", flush=True)
+        if record is None:
+            # keepalive
+            print("event: keepalive\ndata:\n\n", flush=True)
+        else:
+            print(f"event: log\ndata:{ujson.dumps(record)}\n\n", flush=True)