(events): Add re-subscribe mechanism

If we got HTTP error, likely we lost subscription due remote server
restart. Instead of crashing, try to resubscribe several times,
with delay, to allow remote server restart.

Signed-off-by: Denys Fedoryshchenko <[email protected]>

Author: nuclearcat

src/monitor.py

@@ -8,7 +8,7 @@
 
 import datetime
 import sys
-
+import time
 import kernelci
 import kernelci.config
 from kernelci.legacy.cli import Args, Command, parse_opts
@@ -53,9 +53,21 @@ 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 = self._api.receive_event(sub_id)
+            event = None
+            try:
+                event = self._api.receive_event(sub_id)
+            except Exception as e:
+                self.log.error(f"Error receiving event: {e}")
+                time.sleep(10)
+                sub_id = self._api.subscribe('node')
+                subscribe_retries += 1
+                if subscribe_retries > 3:
+                    self.log.error("Failed to subscribe to node events")
+                    return False
+                continue
+            subscribe_retries = 0
             obj = event.data
             dt = datetime.datetime.fromisoformat(event['time'])
             try:

src/scheduler.py

@@ -14,6 +14,7 @@
 import requests
 import re
 import datetime
+import time
 
 import kernelci
 import kernelci.config
@@ -304,9 +305,22 @@ 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:
-            event = self._api_helper.receive_event_data(sub_id)
+            event = None
+            try:
+                event = self._api_helper.receive_event_data(sub_id)
+            except Exception as e:
+                self.log.error(f"Error receiving event: {e}")
+                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
+                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

@@ -656,7 +656,7 @@ def _run(self, context):
         self.log.info("Listening for events... Press Ctrl-C to stop.")
 
         chunksize = 20
-
+        subscribe_retries = 0
         while True:
             is_hierarchy = False
 
@@ -666,7 +666,23 @@ def _run(self, context):
             if not nodes:
                 # Switch to event mode if no unprocessed nodes
                 # Listen and wait for a node instead of processing the queue
-                node, is_hierarchy = self._api_helper.receive_event_node(context['sub_id'])
+                node = None
+                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}")
+                    time.sleep(10)
+                    context['sub_id'] = self._api_helper.subscribe_filters({
+                        'op': 'created',
+                        'kind': 'node',
+                        'state': 'done',
+                    })
+                    subscribe_retries += 1
+                    if subscribe_retries > 3:
+                        self.log.error("Failed to re-subscribe to node events")
+                        return False
+                    continue
+                subscribe_retries = 0
                 self.log.info(f"Processing event node: {node['id']}")
                 nodes = [node]
             else:

src/tarball.py

@@ -14,7 +14,7 @@
 import sys
 import json
 import requests
-
+import time
 import kernelci
 import kernelci.build
 import kernelci.config
@@ -206,9 +206,27 @@ 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, _ = self._api_helper.receive_event_node(sub_id)
+            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}")
+                time.sleep(10)
+                # try to resubscribe
+                sub_id = self._api_helper.subscribe_filters({
+                    'op': 'created',
+                    'kind': 'checkout',
+                    'state': 'running',
+                })
+                subscribe_retries += 1
+                if subscribe_retries > 3:
+                    self.log.error("Failed to re-subscribe to checkout events")
+                    return False
+                continue
+            subscribe_retries = 0
 
             build_config = self._find_build_config(checkout_node)
             if build_config is None: