Remove re-subscription logic from pipeline services

When API helper functions `receive_event_node` or `receive_event_data`
fails, there must be some issue with extracting node data from the event
rather than an issue with the subscription.
Hence, remove the re-subscription logic.

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

Author: Jeny Sadadia

src/monitor.py

@@ -53,21 +53,13 @@ def _run(self, sub_id):
         self.log.info("Monitor: Listening for events... ")
         self.log.info("Press Ctrl-C to stop.")
         print(self._log_titles, flush=True)
-        subscribe_retries = 0
         while True:
             event = None
             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}")
                 continue
-            subscribe_retries = 0
             obj = event.data
             dt = datetime.datetime.fromisoformat(event['time'])
             try:

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

@@ -688,7 +688,6 @@ def _run(self, context):
         self.log.info("Listening for events... Press Ctrl-C to stop.")
 
         chunksize = 500
-        subscribe_retries = 0
         while True:
             is_hierarchy = False
 
@@ -702,15 +701,8 @@ 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}")
                     continue
-                subscribe_retries = 0
                 self.log.info(f"Processing event node: {node['id']}")
                 nodes = [node]
             else:

src/tarball.py

@@ -208,23 +208,14 @@ def _stop(self, sub_id):
     def _run(self, sub_id):
         self.log.info("Listening for new trigger events")
         self.log.info("Press Ctrl-C to stop.")
-        subscribe_retries = 0
 
         while True:
             checkout_node = None
             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}")
                 continue
-            subscribe_retries = 0
 
             build_config = self._find_build_config(checkout_node)
             if build_config is None: