Convoluted solution but it works fine now

Author: André Costa

config/ess.csv

@@ -71,10 +71,10 @@ grafana,       3, 0, 10.100.211.107,      3000, 0, -450, 25, https://10.100.211.
 AKHQ,          3, 0, 10.100.211.94,       8443, 0, -450, 40, https://10.100.211.94:8443
 graylog,       3, 0, graylog.daq.esss.dk, 9000, 0, -450, 55, http://graylog.daq.esss.dk:9000/search
 #
-filewriter0626,        2, 0, 10.100.6.26, 9999, 0, -355, 10, https://10.100.211.107:3000/d/kafka-to-nexus/kafka-to-nexus-filewriter?orgId=1&refresh=5s
-filewriter0628,        2, 0, 10.100.6.28, 9999, 0, -355, 25, https://10.100.211.107:3000/d/kafka-to-nexus/kafka-to-nexus-filewriter?orgId=1&refresh=5s
-filewriter0629,        2, 0, 10.100.6.29, 9999, 0, -355, 40, https://10.100.211.107:3000/d/kafka-to-nexus/kafka-to-nexus-filewriter?orgId=1&refresh=5s
-filewriter0630,        2, 0, 10.100.6.30, 9999, 0, -355, 55, https://10.100.211.107:3000/d/kafka-to-nexus/kafka-to-nexus-filewriter?orgId=1&refresh=5s
+filewriter0626,        5, 0, 10.100.6.26, 9999, 0, -355, 10, https://10.100.211.107:3000/d/kafka-to-nexus/kafka-to-nexus-filewriter?orgId=1&refresh=5s
+filewriter0628,        5, 0, 10.100.6.28, 9999, 0, -355, 25, https://10.100.211.107:3000/d/kafka-to-nexus/kafka-to-nexus-filewriter?orgId=1&refresh=5s
+filewriter0629,        5, 0, 10.100.6.29, 9999, 0, -355, 40, https://10.100.211.107:3000/d/kafka-to-nexus/kafka-to-nexus-filewriter?orgId=1&refresh=5s
+filewriter0630,        5, 0, 10.100.6.30, 9999, 0, -355, 55, https://10.100.211.107:3000/d/kafka-to-nexus/kafka-to-nexus-filewriter?orgId=1&refresh=5s
 nicosserver01,         2, 0, 10.102.10.238, 14869, 0, -355, 70, none
 #
 Kafka   Brokers,  4, 64, 0.0.0.0, 65535, 0, -450, -160, none

dashboard/generate.py

@@ -145,47 +145,82 @@ def check_fw_pipeline(self, ipaddr, port):
         if self.test:
             return 5
 
-        s = None 
-        try:
-            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-            
-            s.settimeout(3.0)
-            s.connect((ipaddr, port))
-            self.dprint(f"Connection to {ipaddr}:{port} successful.")
-            
-            s.settimeout(20.0)
-            s.shutdown(socket.SHUT_WR)
-
-            data = b""
-            while True:
-                chunk = s.recv(4096)
-                if not chunk:
-                    break
-                data += chunk
-
-            text = data.decode("utf-8", errors="ignore")
-            
+        retry_attempts = 3
+        retry_delay = 2 
+        last_exception = None
+
+        for attempt in range(retry_attempts):
+            s = None
             try:
-                arr = json.loads(text)
-                if isinstance(arr, list):
-                    for obj in reversed(arr):
-                        if isinstance(obj, dict):
-                            for key, value in obj.items():
-                                if key.endswith(".worker_state"):
-                                    return int(value)
-            except json.JSONDecodeError as e:
-                self.dprint(f"Failed to parse JSON for {ipaddr}:{port}. Data might be incomplete. Error: {e}")
+                s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+                
+                s.settimeout(3.0)
+                s.connect((ipaddr, port))
+                
+                s.settimeout(20.0)
+                s.shutdown(socket.SHUT_WR)
+
+                buffer = ""
+                while True:
+                    try:
+                        chunk = s.recv(4096).decode('utf-8', errors='ignore')
+                        if not chunk:
+                            break
+                        buffer += chunk
+                    except socket.timeout:
+                        break
+                
+                buffer = buffer.strip()
+                if not buffer:
+                    raise ConnectionError("No data received from socket")
+
+                all_objects = []
+                decoder = json.JSONDecoder()
+                pos = 0
+                while pos < len(buffer):
+                    try:
+                        obj, pos = decoder.raw_decode(buffer, pos)
+                        all_objects.append(obj)
+                    except json.JSONDecodeError:
+                        break
+                
+                if not all_objects:
+                    raise ValueError("Could not parse any JSON objects from received data")
+
+                found_val = None
+                def find_in_obj(target):
+                    nonlocal found_val
+                    if isinstance(target, list):
+                        for item in reversed(target):
+                            find_in_obj(item)
+                            if found_val is not None: return
+                    elif isinstance(target, dict):
+                        for key, value in target.items():
+                            if key.endswith(".worker_state"):
+                                found_val = int(value)
+                                return
+                
+                find_in_obj(all_objects)
+                
+                if found_val is not None:
+                    return found_val
+
+                print(f"worker_state not found for {ipaddr}:{port}. No further retries for this cycle.")
                 return -1
 
-            self.dprint(f"worker_state not found for {ipaddr}:{port}")
-            return -1
+            except (OSError, ConnectionError, ValueError) as e:
+                last_exception = e
+                print(f"Attempt {attempt + 1}/{retry_attempts} for {ipaddr}:{port} failed: {e}")
+                if attempt < retry_attempts - 1:
+                    print(f"Retrying in {retry_delay} seconds...")
+                    time.sleep(retry_delay)
 
-        except OSError as e:
-            self.dprint(f"Socket error for {ipaddr}:{port}: {e}")
-            return -1
-        finally:
-            if s:
-                s.close()
+            finally:
+                if s:
+                    s.close()
+        
+        print(f"All {retry_attempts} attempts failed for {ipaddr}:{port}. Last error: {last_exception}")
+        return -1
 
     def check_efu_pipeline(self, ipaddr, port):
         if self.test:
@@ -227,7 +262,7 @@ def check_service(self, idx, type, ipaddr, port):
                 self.lab.servers[idx][9] = self.efu_get_version(ipaddr, port)
             elif type == type_fw:
                 status = self.check_fw_pipeline(ipaddr, port)
-                if status == 0:
+                if status == -1 or status == 0:
                     self.lab.clearstatus(
                         idx, self.s_stage1 | self.s_stage2 | self.s_stage3
                     )