Merge pull request #2 from skiedude/filewatchsensor

nzlosh · web-flow · commit a0526d392cff · 2026-03-27T19:06:27.000+01:00
Convert FileWatchSensor to use threading over eventlent and stackstorm/logshipper
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
@@ -1,8 +1,8 @@
 Changelog
 =========
 
-in development - unreleased
----------------------------
+in development
+--------------
 
 This release is tested with and supports the following Python versions and 3rd party dependencies.
 
@@ -24,11 +24,11 @@ Changed
  * Removed mongodb 7.0, rabbitmq 3.13 and redis 8.0
  * Replaced deprecated `pkg_resources` module with `importlib-metadata` and `importlib-resources`.
  * Replaced abandoned `flex` module by `openapi-spec-validator`
+ * Replaced Stackstorm/logshipper (stops working with Python 3.12) and eventlet in the `linux.file_watch_sensor` with threading. (by @skiedude)
 
 Added
 ~~~~~
 
-
 3.9.0 - October 10, 2025
 ------------------------
 
diff --git a/contrib/linux/README.md b/contrib/linux/README.md
@@ -2,28 +2,37 @@
 
 This pack contains actions for commonly used Linux commands and tools.
 
-## Configuration
+## Sensors
+
+### FileWatchSensor
 
-* ``file_watch_sensor.file_paths`` - A list of paths to the files to monitor.
-  Note: Those need to be full paths to the files (e.g. ``/var/log/auth.log``)
-  and not directories (files don't need to exist yet when the sensor is ran
-  though).
+This sensor monitors files that are declared by a rule (one file per rule). Once a new line is
+detected, a trigger is emitted.
 
-Example:
+### Example Rule:
 
-```yaml
----
-file_watch_sensor:
-  file_paths:
-    - /opt/data/absolute_path_to_file.log
+This example rule will register with the FileWatchSensor to watch the file `/tmp/st2_test`. When a new line is
+detected, the trigger will be emitted, and the action `core.local` will echo the trigger data.
 ```
+---
+name: sample_rule_file_watch
+pack: "examples"
+description: Sample rule custom trigger type - add a file to be watched by file_watch_sensor in linux pack.
+enabled: true
 
-## Sensors
+trigger:
+  parameters:
+    file_path: /tmp/st2_test
+  type: linux.file_watch.line
 
-### FileWatchSensor
+criteria: {}
 
-This sensor monitors specified files for new new lines. Once a new line is
-detected, a trigger is emitted.
+action:
+  parameters:
+    cmd: echo "{{trigger}}"
+  ref: core.local
+
+```
 
 ### linux.file_watch.line trigger
 
diff --git a/contrib/linux/requirements.txt b/contrib/linux/requirements.txt
@@ -1,3 +1 @@
-# used by file watcher sensor
-pyinotify>=0.9.5,<=0.10 ; platform_system=="Linux"
-logshipper@ git+https://github.com/StackStorm/logshipper.git@stackstorm_patched ; platform_system=="Linux"
+
diff --git a/contrib/linux/sensors/README.md b/contrib/linux/sensors/README.md
diff --git a/contrib/linux/sensors/file_watch_sensor.py b/contrib/linux/sensors/file_watch_sensor.py
@@ -1,4 +1,4 @@
-# Copyright 2020 The StackStorm Authors.
+# Copyright 2020-2026 The StackStorm Authors.
 # Copyright 2019 Extreme Networks, Inc.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -14,9 +14,8 @@
 # limitations under the License.
 
 import os
-import eventlet
-
-from logshipper.tail import Tail
+import time
+import threading
 
 from st2reactor.sensor.base import Sensor
 
@@ -27,25 +26,19 @@ def __init__(self, sensor_service, config=None):
             sensor_service=sensor_service, config=config
         )
         self.log = self._sensor_service.get_logger(__name__)
-        self.tail = None
+        self._watchers = {}  # file_path -> (thread, stop_event)
         self.file_ref = {}
 
     def setup(self):
-        self.tail = Tail(filenames=[])
-        self.tail.handler = self._handle_line
-        self.tail.should_run = True
+        pass
 
     def run(self):
-        self.tail.run()
+        while True:
+            time.sleep(1)
 
     def cleanup(self):
-        if self.tail:
-            self.tail.should_run = False
-
-            try:
-                self.tail.notifier.stop()
-            except Exception:
-                self.log.exception("Unable to stop the tail notifier")
+        for file_path in list(self._watchers):
+            self._stop_watcher(file_path)
 
     def add_trigger(self, trigger):
         file_path = trigger["parameters"].get("file_path", None)
@@ -54,18 +47,19 @@ def add_trigger(self, trigger):
             self.log.error('Received trigger type without "file_path" field.')
             return
 
-        trigger = trigger.get("ref", None)
+        ref = trigger.get("ref", None)
 
-        if not trigger:
+        if not ref:
             raise Exception(f"Trigger {trigger} did not contain a ref.")
 
-        # Wait a bit to avoid initialization race in logshipper library
-        eventlet.sleep(1.0)
+        self.file_ref[file_path] = ref
 
-        self.tail.add_file(filename=file_path)
-        self.file_ref[file_path] = trigger
+        stop_event = threading.Event()
+        t = threading.Thread(target=self._tail, args=(file_path, stop_event), daemon=True)
+        self._watchers[file_path] = (t, stop_event)
+        t.start()
 
-        self.log.info(f"Added file '{file_path}' ({trigger}) to watch list.")
+        self.log.info(f"Added file '{file_path}' ({ref}) to watch list.")
 
     def update_trigger(self, trigger):
         pass
@@ -77,10 +71,25 @@ def remove_trigger(self, trigger):
             self.log.error("Received trigger type without 'file_path' field.")
             return
 
-        self.tail.remove_file(filename=file_path)
-        self.file_ref.pop(file_path)
-
-        self.log.info(f"Removed file '{file_path}' ({trigger}) from watch list.")
+        self._stop_watcher(file_path)
+        self.file_ref.pop(file_path, None)
+
+        self.log.info(f"Removed file '{file_path}' from watch list.")
+
+    def _stop_watcher(self, file_path):
+        if file_path in self._watchers:
+            _, stop_event = self._watchers.pop(file_path)
+            stop_event.set()
+
+    def _tail(self, file_path, stop_event):
+        with open(file_path, "r") as f:
+            f.seek(0, 2)  # seek to EOF
+            while not stop_event.is_set():
+                line = f.readline()
+                if line:
+                    self._handle_line(file_path, line.strip())
+                else:
+                    time.sleep(0.1)
 
     def _handle_line(self, file_path, line):
         if file_path not in self.file_ref:
diff --git a/fixed-requirements.txt b/fixed-requirements.txt
@@ -39,7 +39,6 @@ oslo.utils==7.3.0
 paramiko==3.5.1
 # 202403: bump to 3.0.43 for py3.10 support
 prompt-toolkit==3.0.52
-pyinotify==0.9.6 ; platform_system=="Linux"
 pymongo==4.6.3
 pyparsing==3.1.4
 zstandard==0.23.0
@@ -76,8 +75,6 @@ tooz==6.3.0
 # virtualenv==20.30.0 (<21) has pip==25.0.1 wheel==0.45.1 setuptools==75.3.2
 # lockfiles/st2.lock has pip==25.0.1 wheel==0.45.1 setuptools==75.3.2
 pip==25.3
-# This setuptools version number is in the Makefile, but CircleCI builds are pulling a version
-# that is incompatible with our logshipper fork.
 setuptools==80.10.2
 webob==1.8.9
 webtest==3.0.1
diff --git a/st2actions/in-requirements.txt b/st2actions/in-requirements.txt
@@ -16,11 +16,6 @@ python-json-logger
 # needed by core "packs" pack
 gitpython
 lockfile
-# needed by core "linux" pack - TODO: create virtualenv for linux pack on postinst
-pyinotify
-logshipper@ git+https://github.com/StackStorm/logshipper.git ; platform_system=="Linux"
-# logshipper has metadata in setup.cfg that is not supported by setuptools 78, so we need
-# an explicit dep (from fixed-requirements.txt) to prevent CircleCI from pulling that in.
 setuptools
 # required by pack_mgmt/setup_virtualenv.py#L135
 virtualenv
diff --git a/st2common/tests/fixtures/requirements-used-for-tests.txt b/st2common/tests/fixtures/requirements-used-for-tests.txt
@@ -7,10 +7,8 @@ flex==6.14.0
 # some
 # more
 # commments....
-git+https://github.com/Kami/logshipper.git@stackstorm_patched#egg=logshipper
 git+https://github.com/StackStorm/orquesta.git@224c1a589a6007eb0598a62ee99d674e7836d369#egg=orquesta
 git+https://github.com/StackStorm/st2-auth-backend-flat-file.git@master#egg=st2-auth-backend-flat-file
--e git+https://github.com/Kami/logshipper.git@stackstorm_patched#egg=logshipper-editable
 git+https://github.com/StackStorm/st2.git#egg=python_runner&subdirectory=contrib/runners/python_runner
 hg+https://hg.repo/some_pkg.git#egg=SomePackageHq
 svn+svn://svn.repo/some_pkg/trunk/@ma-branch#egg=SomePackageSvn
diff --git a/st2common/tests/unit/test_dist_utils.py b/st2common/tests/unit/test_dist_utils.py
@@ -74,10 +74,8 @@ def test_fetch_requirements(self):
             "amqp==2.5.1",
             "argcomplete",
             "bcrypt==3.1.6",
-            "logshipper",
             "orquesta",
             "st2-auth-backend-flat-file",
-            "logshipper-editable",
             "python_runner",
             "SomePackageHq",
             "SomePackageSvn",
@@ -89,10 +87,8 @@ def test_fetch_requirements(self):
             "zake==0.2.2",
         ]
         expected_links = [
-            "git+https://github.com/Kami/logshipper.git@stackstorm_patched#egg=logshipper",
             "git+https://github.com/StackStorm/orquesta.git@224c1a589a6007eb0598a62ee99d674e7836d369#egg=orquesta",  # NOQA
             "git+https://github.com/StackStorm/st2-auth-backend-flat-file.git@master#egg=st2-auth-backend-flat-file",  # NOQA
-            "git+https://github.com/Kami/logshipper.git@stackstorm_patched#egg=logshipper-editable",
             "git+https://github.com/StackStorm/st2.git#egg=python_runner&subdirectory=contrib/runners/python_runner",  # NOQA
             "hg+https://hg.repo/some_pkg.git#egg=SomePackageHq",
             "svn+svn://svn.repo/some_pkg/trunk/@ma-branch#egg=SomePackageSvn",
