Fix re-subscription logic in pipeline services

When API helper functions `receive_event_node` or `receive_event_data`
fails, there may be some issue with extracting node data from the event
rather than an issue with the subscription.
Check exception message before resubscribing.

Fixes: f04f6d0 ("(events): Add re-subscribe mechanism")
Signed-off-by: Jeny Sadadia <[email protected]>

Author: Jeny Sadadia

src/monitor.py

@@ -59,13 +59,15 @@ def _run(self, sub_id):
             try:
                 event = self._api.receive_event(sub_id)
             except Exception as e:
-                self.log.error(f"Error receiving event: {e}, re-subscribing in 10 seconds")
-                time.sleep(10)
-                sub_id = self._api.subscribe('node')
-                subscribe_retries += 1
-                if subscribe_retries > 3:
-                    self.log.error("Failed to re-subscribe to node events")
-                    return False
+                self.log.error(f"Error receiving event: {e}")
+                if "404 Client Error" in str(e):
+                    self.log.error(f"Error receiving event: {e}. Re-subscribing...")
+                    sub_id = self._setup(None)
+                    subscribe_retries += 1
+                    if subscribe_retries > 3:
+                        self.log.error("Failed to re-subscribe to node events")
+                        return False
+                    continue
                 continue
             subscribe_retries = 0
             obj = event.data

src/scheduler.py

@@ -374,26 +374,18 @@ def _verify_architecture_filter(self, job, node):
     def _run(self, sub_id):
         self.log.info("Listening for available checkout events")
         self.log.info("Press Ctrl-C to stop.")
-        subscribe_retries = 0
 
         while True:
             last_heartbeat['time'] = time.time()
             event = None
             try:
                 event = self._api_helper.receive_event_data(sub_id, block=False)
             except Exception as e:
-                self.log.error(f"Error receiving event: {e}, re-subscribing in 10 seconds")
-                time.sleep(10)
-                sub_id = self._api.subscribe('node')
-                subscribe_retries += 1
-                if subscribe_retries > 3:
-                    self.log.error("Failed to re-subscribe to node events")
-                    return False
+                self.log.error(f"Error receiving event: {e}")
                 continue
             if not event:
                 # If we received a keep-alive event, just continue
                 continue
-            subscribe_retries = 0
             for job, runtime, platform, rules in self._sched.get_schedule(event):
                 input_node = self._api.node.get(event['id'])
                 jobfilter = event.get('jobfilter')

src/send_kcidb.py

@@ -702,13 +702,15 @@ def _run(self, context):
                 try:
                     node, is_hierarchy = self._api_helper.receive_event_node(context['sub_id'])
                 except Exception as e:
-                    self.log.error(f"Error receiving event: {e}, re-subscribing in 10 seconds")
-                    time.sleep(10)
-                    context['sub_id'] = self._api_helper.subscribe_filters(self._filters, promiscuous=True)
-                    subscribe_retries += 1
-                    if subscribe_retries > 3:
-                        self.log.error("Failed to re-subscribe to node events")
-                        return False
+                    self.log.error(f"Error receiving event: {e}")
+                    if "404 Client Error" in str(e):
+                        self.log.error(f"Error receiving event: {e}. Re-subscribing...")
+                        context['sub_id'] = self._api_helper.subscribe_filters(self._filters, promiscuous=True)
+                        subscribe_retries += 1
+                        if subscribe_retries > 3:
+                            self.log.error("Failed to re-subscribe to node events")
+                            return False
+                        continue
                     continue
                 subscribe_retries = 0
                 self.log.info(f"Processing event node: {node['id']}")

src/tarball.py

@@ -215,17 +215,18 @@ def _run(self, sub_id):
             try:
                 checkout_node, _ = self._api_helper.receive_event_node(sub_id)
             except Exception as e:
-                self.log.error(f"Error receiving event: {e}, re-subscribing in 10 seconds")
-                time.sleep(10)
-                # try to resubscribe
-                sub_id = self._api_helper.subscribe_filters(self._filters)
-                subscribe_retries += 1
-                if subscribe_retries > 3:
-                    self.log.error("Failed to re-subscribe to checkout events")
-                    return False
+                self.log.error(f"Error receiving event: {e}")
+                if "404 Client Error" in str(e):
+                    self.log.error(f"Error receiving event: {e}. Re-subscribing...")
+                    sub_id = self._setup(None)
+                    subscribe_retries += 1
+                    if subscribe_retries > 3:
+                        self.log.error("Failed to re-subscribe to checkout events")
+                        return False
+                    continue
                 continue
-            subscribe_retries = 0
 
+            subscribe_retries = 0
             build_config = self._find_build_config(checkout_node)
             if build_config is None:
                 continue