fix watcher for smb
This commit is contained in:
+140
-51
@@ -1,8 +1,10 @@
|
||||
import os
|
||||
import threading
|
||||
|
||||
import smbclient
|
||||
from watchdog.events import FileSystemEventHandler
|
||||
from watchdog.observers.polling import PollingObserver
|
||||
from pathlib import Path
|
||||
|
||||
from db_py import (
|
||||
mark_tests_completed,
|
||||
@@ -15,24 +17,23 @@ from scanner import process_result_dir, resolve_runtime_path
|
||||
from parser import parse_deleted_result_dir_name
|
||||
|
||||
|
||||
_OBSERVER = None
|
||||
_OBSERVER_LOCK = threading.RLock()
|
||||
_WATCHERS: list = []
|
||||
_WATCHERS_LOCK = threading.RLock()
|
||||
|
||||
|
||||
class ResultsDirEventHandler(FileSystemEventHandler):
|
||||
def __init__(self, results_root):
|
||||
super().__init__()
|
||||
self.results_root = str(results_root)
|
||||
self.results_root = Path(results_root)
|
||||
|
||||
def _is_direct_child(self, path_value):
|
||||
parent = os.path.dirname(path_value.rstrip("\\/"))
|
||||
return os.path.normcase(parent) == os.path.normcase(self.results_root.rstrip("\\/"))
|
||||
return Path(path_value).parent == self.results_root
|
||||
|
||||
def _mark_completed(self, dir_path):
|
||||
if not self._is_direct_child(dir_path):
|
||||
return
|
||||
|
||||
dir_name = os.path.basename(dir_path.rstrip("\\/"))
|
||||
dir_name = Path(dir_path).name
|
||||
completion = process_result_dir(self.results_root, dir_name)
|
||||
if not completion:
|
||||
return
|
||||
@@ -47,19 +48,19 @@ class ResultsDirEventHandler(FileSystemEventHandler):
|
||||
|
||||
def _mark_completed_from_file(self, file_path):
|
||||
# New measurement/log files arriving after directory creation should update that single test.
|
||||
filename = os.path.basename(file_path).lower()
|
||||
p = Path(file_path)
|
||||
filename = p.name.lower()
|
||||
is_duration_log = filename.endswith(".log") or filename.endswith(".txt") or "log" in filename
|
||||
if filename != "measurement.db" and not is_duration_log:
|
||||
return
|
||||
|
||||
parent_dir = os.path.dirname(file_path.rstrip("\\/"))
|
||||
self._mark_completed(parent_dir)
|
||||
self._mark_completed(p.parent)
|
||||
|
||||
def _clear_completed(self, dir_path):
|
||||
if not self._is_direct_child(dir_path):
|
||||
return
|
||||
|
||||
dir_name = os.path.basename(dir_path.rstrip("\\/"))
|
||||
dir_name = Path(dir_path).name
|
||||
test_id, device = parse_deleted_result_dir_name(dir_name)
|
||||
if not test_id:
|
||||
return
|
||||
@@ -92,69 +93,157 @@ class ResultsDirEventHandler(FileSystemEventHandler):
|
||||
self._mark_completed_from_file(event.dest_path)
|
||||
|
||||
|
||||
def stop_results_watchers():
|
||||
global _OBSERVER
|
||||
class _SmbPollerThread(threading.Thread):
|
||||
"""Polls a UNC results directory via smbclient and updates the DB on directory changes."""
|
||||
|
||||
with _OBSERVER_LOCK:
|
||||
if _OBSERVER is None:
|
||||
return
|
||||
_INTERVAL = 5.0
|
||||
|
||||
def __init__(self, unc_path):
|
||||
super().__init__(daemon=True, name=f"smb-poller-{unc_path}")
|
||||
self._path = unc_path.rstrip("\\")
|
||||
self._stop_evt = threading.Event()
|
||||
|
||||
def stop(self):
|
||||
self._stop_evt.set()
|
||||
|
||||
def run(self):
|
||||
prev = self._snapshot()
|
||||
while not self._stop_evt.wait(self._INTERVAL):
|
||||
try:
|
||||
curr = self._snapshot()
|
||||
for name in curr - prev:
|
||||
completion = process_result_dir(self._path, name)
|
||||
if completion:
|
||||
mark_tests_completed([completion])
|
||||
coe = update_all_p2p_coe_pairs_sql()
|
||||
p3p = update_all_p3p_pairs_sql()
|
||||
print(f"[watcher] SMB: marked completed {name} (coe={coe}, p3p={p3p})")
|
||||
for name in prev - curr:
|
||||
test_id, device = parse_deleted_result_dir_name(name)
|
||||
if test_id:
|
||||
reset_by_file_id_and_device(test_id, device)
|
||||
coe = update_all_p2p_coe_pairs_sql()
|
||||
p3p = update_all_p3p_pairs_sql()
|
||||
print(f"[watcher] SMB: cleared completed {name} (coe={coe}, p3p={p3p})")
|
||||
prev = curr
|
||||
except Exception as exc:
|
||||
print(f"[watcher] SMB poll error {self._path}: {exc}")
|
||||
|
||||
def _snapshot(self):
|
||||
try:
|
||||
_OBSERVER.stop()
|
||||
_OBSERVER.join(timeout=5)
|
||||
finally:
|
||||
_OBSERVER = None
|
||||
print("[watcher] Stopped results directory watcher")
|
||||
return {e.name for e in smbclient.scandir(self._path) if e.is_dir()}
|
||||
except Exception as exc:
|
||||
print(f"[watcher] SMB scandir error {self._path}: {exc}")
|
||||
return set()
|
||||
|
||||
|
||||
def _register_smb_session(unc_path):
|
||||
server = unc_path[2:].split("\\", 1)[0]
|
||||
if not server:
|
||||
return
|
||||
username = os.getenv("SMB_USERNAME", "").strip()
|
||||
password = os.getenv("SMB_PASSWORD", "")
|
||||
domain = os.getenv("SMB_DOMAIN", "").strip()
|
||||
if username and domain and "\\" not in username and "@" not in username:
|
||||
username = f"{domain}\\{username}"
|
||||
try:
|
||||
if username:
|
||||
smbclient.register_session(server, username=username, password=password)
|
||||
else:
|
||||
smbclient.register_session(server)
|
||||
except Exception as exc:
|
||||
print(f"[watcher] SMB session registration failed for {server}: {exc}")
|
||||
|
||||
|
||||
def stop_results_watchers():
|
||||
global _WATCHERS
|
||||
|
||||
with _WATCHERS_LOCK:
|
||||
if not _WATCHERS:
|
||||
return
|
||||
for w in _WATCHERS:
|
||||
try:
|
||||
w.stop()
|
||||
except Exception:
|
||||
pass
|
||||
for w in _WATCHERS:
|
||||
try:
|
||||
w.join(timeout=5)
|
||||
except Exception:
|
||||
pass
|
||||
_WATCHERS = []
|
||||
print("[watcher] Stopped results directory watcher(s)")
|
||||
|
||||
|
||||
def start_results_watchers(results_dir, results_dir_ref):
|
||||
global _OBSERVER
|
||||
global _WATCHERS
|
||||
|
||||
with _OBSERVER_LOCK:
|
||||
with _WATCHERS_LOCK:
|
||||
stop_results_watchers()
|
||||
|
||||
resolved = []
|
||||
local_paths = []
|
||||
smb_paths = []
|
||||
seen = set()
|
||||
|
||||
for raw_dir in (results_dir, results_dir_ref):
|
||||
if not raw_dir:
|
||||
continue
|
||||
path = resolve_runtime_path(raw_dir)
|
||||
if not path:
|
||||
continue
|
||||
normalized = os.path.normcase(str(path).rstrip("\\/"))
|
||||
if normalized and normalized not in seen:
|
||||
seen.add(normalized)
|
||||
resolved.append(str(path))
|
||||
path = str(path)
|
||||
|
||||
if not resolved:
|
||||
if path.startswith("\\\\") or path.startswith("//"):
|
||||
unc = "\\\\" + path.lstrip("/\\").replace("/", "\\")
|
||||
key = unc.lower()
|
||||
if key not in seen:
|
||||
seen.add(key)
|
||||
smb_paths.append(unc)
|
||||
else:
|
||||
p = Path(path)
|
||||
key = os.path.normcase(str(p))
|
||||
if key not in seen:
|
||||
seen.add(key)
|
||||
local_paths.append(p)
|
||||
|
||||
if not local_paths and not smb_paths:
|
||||
print("[watcher] No results directories configured, watcher not started")
|
||||
return
|
||||
|
||||
observer = PollingObserver(timeout=1.0)
|
||||
watchers = []
|
||||
|
||||
scheduled_count = 0
|
||||
for path in resolved:
|
||||
try:
|
||||
observer.schedule(ResultsDirEventHandler(path), path, recursive=False)
|
||||
scheduled_count += 1
|
||||
print(f"[watcher] Watching results directory: {path}")
|
||||
except Exception as exc:
|
||||
print(f"[watcher] Skipping watch path {path}: {exc}")
|
||||
if local_paths:
|
||||
observer = PollingObserver(timeout=1.0)
|
||||
scheduled = 0
|
||||
for p in local_paths:
|
||||
try:
|
||||
observer.schedule(ResultsDirEventHandler(p), str(p), recursive=False)
|
||||
scheduled += 1
|
||||
print(f"[watcher] Watching local directory: {p}")
|
||||
except Exception as exc:
|
||||
print(f"[watcher] Skipping local watch path {p}: {exc}")
|
||||
if scheduled:
|
||||
try:
|
||||
observer.start()
|
||||
watchers.append(observer)
|
||||
except Exception as exc:
|
||||
print(f"[watcher] Failed to start local watcher: {exc}")
|
||||
try:
|
||||
observer.stop()
|
||||
observer.join(timeout=5)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if scheduled_count == 0:
|
||||
for unc in smb_paths:
|
||||
_register_smb_session(unc)
|
||||
poller = _SmbPollerThread(unc)
|
||||
poller.start()
|
||||
watchers.append(poller)
|
||||
print(f"[watcher] Watching SMB directory: {unc}")
|
||||
|
||||
if not watchers:
|
||||
print("[watcher] No valid watch paths available, watcher not started")
|
||||
return
|
||||
|
||||
try:
|
||||
observer.start()
|
||||
except Exception as exc:
|
||||
print(f"[watcher] Failed to start results watcher: {exc}")
|
||||
try:
|
||||
observer.stop()
|
||||
observer.join(timeout=5)
|
||||
except Exception:
|
||||
pass
|
||||
return
|
||||
|
||||
_OBSERVER = observer
|
||||
print("[watcher] Results directory watcher started")
|
||||
_WATCHERS = watchers
|
||||
print(f"[watcher] Results directory watcher(s) started ({len(watchers)})")
|
||||
Reference in New Issue
Block a user