[DL Streamer Pipeline Server] : fix: enhance error message for InfluxDB initialization failures (open-edge-platform#600)

Co-authored-by: Ashish Jagadish <[email protected]>

Author: sugnanprabhu

microservices/dlstreamer-pipeline-server/src/publisher/influx/influx_writer.py

@@ -39,9 +39,11 @@ def __init__(self, config, qsize=DEFAULT_APPDEST_INFLUX_QUEUE_SIZE):
         self.influx_bucket_name = config.get("bucket")
         self.influx_measurement = config.get("measurement", "dlsps")
         self.influxwrite_complete = th.Event()
-
         self.th = None
         self.log = get_logger(f'{__name__} ({self.influx_bucket_name})')
+        if not self.influx_bucket_name:
+            self.log.error(f'Empty value given for bucket name. It cannot be blank')
+            self.initialized=False
         if not self.host:
             self.log.error(f'Empty value given for INFLUXDB_HOST. It cannot be blank')
             self.initialized=False
@@ -59,8 +61,12 @@ def __init__(self, config, qsize=DEFAULT_APPDEST_INFLUX_QUEUE_SIZE):
 
         self.log.info(f'Initializing Influx Writer for bucket - {self.influx_bucket_name}')
         self.influx_client = InfluxClient(self.host, self.port, self.org, self.username, self.password)
-        self.initialized=True
-        self.log.info("InfluxDB Writer initialized")
+        if not self.influx_client.bucket_exists(self.influx_bucket_name):
+            self.log.error(f"Given bucket name - {self.influx_bucket_name} does NOT exist or server is inaccessible")
+            self.initialized=False
+        else:
+            self.initialized=True
+            self.log.info("InfluxDB Writer initialized")
 
     def start(self):
         """Start publisher.

microservices/dlstreamer-pipeline-server/utils/influx_client.py

@@ -10,6 +10,7 @@
 from influxdb_client.client.write_api import SYNCHRONOUS
 from src.publisher.influx.influx_schema import DataSchema
 from src.common.log import get_logger
+import urllib3
 
 class InfluxClient():
     """Influx Client.
@@ -43,6 +44,8 @@ def upload_metadata(self, influx_bucket_name, point, img_handle):
         try:
             self.write_api.write(bucket=influx_bucket_name, org=self.influx_org, record=point)
             self.log.debug(f"Successfully wrote data in influx for image handle: {img_handle}")
+        except urllib3.exceptions.NameResolutionError as e:
+            self.log.exception("Unable to resolve InfluxDB hostname. Please verify that InfluxDB is running.")
         except Exception as e:
             self.log.exception(f"Error writing data to InfluxDB for image handle: {img_handle}", e)
 
@@ -87,4 +90,12 @@ def stop(self):
         """
         # for compatibility with other publisher.
         # TODO: Implement safe disconnect. 
-        pass
+        pass
+
+    def bucket_exists(self, bucket_name):
+        try:
+            buckets_api = self.client.buckets_api()
+            buckets = buckets_api.find_buckets().buckets
+            return any(bucket.name == bucket_name for bucket in buckets)
+        except:
+            return False