Skip to content

Commit

Permalink
Kubernetes watcher tweaks (PR #41)
Browse files Browse the repository at this point in the history
* change timeout_seconds parameter for k8s api connection from 0 to 300 to reset the connection periodically; add lock for events like subscibe, unsubscribe and event dispatching

* change timeout_seconds watcher param back to 0 to try and hold connection forever
  • Loading branch information
vlerkin authored Nov 28, 2024
1 parent d5a72a7 commit a857ae5
Showing 1 changed file with 15 additions and 11 deletions.
26 changes: 15 additions & 11 deletions scrapyd_k8s/k8s_resource_watcher.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ def __init__(self, namespace, config):
self._stop_event = threading.Event()
self.watcher_thread = threading.Thread(target=self.watch_pods, daemon=True)
self.watcher_thread.start()
self._lock = threading.Lock()
logger.info(f"ResourceWatcher thread started for namespace '{self.namespace}'.")

def subscribe(self, callback: Callable):
Expand All @@ -46,9 +47,10 @@ def subscribe(self, callback: Callable):
callback : Callable
A function to call when an event is received.
"""
if callback not in self.subscribers:
self.subscribers.append(callback)
logger.debug(f"Subscriber {callback.__name__} added.")
with self._lock:
if callback not in self.subscribers:
self.subscribers.append(callback)
logger.debug(f"Subscriber {callback.__name__} added.")

def unsubscribe(self, callback: Callable):
"""
Expand All @@ -59,9 +61,10 @@ def unsubscribe(self, callback: Callable):
callback : Callable
The subscriber function to remove.
"""
if callback in self.subscribers:
self.subscribers.remove(callback)
logger.debug(f"Subscriber {callback.__name__} removed.")
with self._lock:
if callback in self.subscribers:
self.subscribers.remove(callback)
logger.debug(f"Subscriber {callback.__name__} removed.")

def notify_subscribers(self, event: dict):
"""
Expand All @@ -72,11 +75,12 @@ def notify_subscribers(self, event: dict):
event : dict
The Kubernetes event data.
"""
for subscriber in self.subscribers:
try:
subscriber(event)
except Exception as e:
logger.exception(f"Error notifying subscriber {subscriber.__name__}: {e}")
with self._lock:
for subscriber in self.subscribers:
try:
subscriber(event)
except Exception as e:
logger.exception(f"Error notifying subscriber {subscriber.__name__}: {e}")

def watch_pods(self):
"""
Expand Down

0 comments on commit a857ae5

Please sign in to comment.