commit da19f6f

Author: Kyle Knoepfel <knoepfel@fnal.gov>
Date:   Thu Sep 30 21:45:50 2021 +0000

    Support retries for contacting Graphite.

    (cherry picked from commit a9c1f3c)

Author: vitodb

src/decisionengine_modules/AWS/publishers/AWS_generic_publisher.py

@@ -18,14 +18,13 @@ class AWSGenericPublisher(Publisher.Publisher, metaclass=abc.ABCMeta):
     def __init__(self, config):
         super().__init__(config)
         self.logger = self.logger.bind(class_module=__name__.split(".")[-1], )
-        self.graphite_host = config.get(
-            'graphite_host', DEFAULT_GRAPHITE_HOST)
-        self.graphite_port = config.get(
-            'graphite_port', DEFAULT_GRAPHITE_PORT)
-        self.graphite_context_header = config.get(
-            'graphite_context', DEFAULT_GRAPHITE_CONTEXT)
-        self.publish_to_graphite = config.get('publish_to_graphite')
-        self.output_file = config.get('output_file')
+        self.graphite_host = config.get("graphite_host", DEFAULT_GRAPHITE_HOST)
+        self.graphite_port = config.get("graphite_port", DEFAULT_GRAPHITE_PORT)
+        self.graphite_context_header = config.get("graphite_context", DEFAULT_GRAPHITE_CONTEXT)
+        self.max_retries = config.get("max_retries", 2)
+        self.retry_interval = config.get("retry_interval", 60)
+        self.publish_to_graphite = config.get("publish_to_graphite")
+        self.output_file = config.get("output_file")
 
     @classmethod
     def consumes_dataframe(cls, product_name):
@@ -52,12 +51,15 @@ def publish(self, data_block):
             return
         data = data_block[list(self._consumes.keys())[0]]
         if self.graphite_host and self.publish_to_graphite:
-            end_point = graphite.Graphite(host=self.graphite_host,
-                                          pickle_port=self.graphite_port)
-            end_point.send_dict(self.graphite_context(data)[0],
-                                self.graphite_context(data)[1],
-                                debug_print=False,
-                                send_data=True)
+            end_point = graphite.Graphite(host=self.graphite_host, pickle_port=self.graphite_port)
+            end_point.send_dict(
+                self.graphite_context(data)[0],
+                self.graphite_context(data)[1],
+                debug_print=False,
+                send_data=True,
+                max_retries=self.max_retries,
+                retry_interval=self.retry_interval,
+            )
         csv_data = data.to_csv(self.output_file, index=False)
         if not self.output_file:
             print(csv_data)

src/decisionengine_modules/graphite/publishers/generic_publisher.py

@@ -14,20 +14,26 @@
 DEFAULT_GRAPHITE_CONTEXT = ""
 
 
-@Publisher.supports_config(Parameter('graphite_host', default=DEFAULT_GRAPHITE_HOST),
-                           Parameter('graphite_port', default=DEFAULT_GRAPHITE_PORT),
-                           Parameter('graphite_context', default=DEFAULT_GRAPHITE_CONTEXT),
-                           Parameter('publish_to_graphite', type=bool),
-                           Parameter('output_file', type=str))
+@Publisher.supports_config(
+    Parameter("graphite_host", default=DEFAULT_GRAPHITE_HOST),
+    Parameter("graphite_port", default=DEFAULT_GRAPHITE_PORT),
+    Parameter("max_retries", default=2, comment="Number of retries allowed to send data to Graphite."),
+    Parameter("retry_interval", default=60, comment="Number of seconds to wait between retries."),
+    Parameter("graphite_context", default=DEFAULT_GRAPHITE_CONTEXT),
+    Parameter("publish_to_graphite", type=bool),
+    Parameter("output_file", type=str),
+)
 class GenericPublisher(Publisher.Publisher, metaclass=abc.ABCMeta):
 
     def __init__(self, config):
         super().__init__(config)
-        self.graphite_host = config.get('graphite_host', DEFAULT_GRAPHITE_HOST)
-        self.graphite_port = config.get('graphite_port', DEFAULT_GRAPHITE_PORT)
-        self.graphite_context_header = config.get('graphite_context', DEFAULT_GRAPHITE_CONTEXT)
-        self.publish_to_graphite = config.get('publish_to_graphite')
-        self.output_file = config.get('output_file')
+        self.graphite_host = config.get("graphite_host", DEFAULT_GRAPHITE_HOST)
+        self.graphite_port = config.get("graphite_port", DEFAULT_GRAPHITE_PORT)
+        self.graphite_context_header = config.get("graphite_context", DEFAULT_GRAPHITE_CONTEXT)
+        self.max_retries = config.get("max_retries", 2)
+        self.retry_interval = config.get("retry_interval", 60)
+        self.publish_to_graphite = config.get("publish_to_graphite")
+        self.output_file = config.get("output_file")
         self.logger = self.logger.bind(class_module=__name__.split(".")[-1], )
 
     @classmethod
@@ -60,11 +66,14 @@ def publish(self, data_block):
             self.logger.exception(f"Failed to extract {product} from data block.")
             return
         if self.graphite_host and self.publish_to_graphite:
-            end_point = graphite.Graphite(host=self.graphite_host,
-                                          pickle_port=self.graphite_port)
-            end_point.send_dict(self.graphite_context(data)[0],
-                                self.graphite_context(data)[1],
-                                debug_print=False,
-                                send_data=True)
+            end_point = graphite.Graphite(host=self.graphite_host, pickle_port=self.graphite_port)
+            end_point.send_dict(
+                self.graphite_context(data)[0],
+                self.graphite_context(data)[1],
+                debug_print=False,
+                send_data=True,
+                max_retries=self.max_retries,
+                retry_interval=self.retry_interval,
+            )
         if not self.output_file:
             self.logger.debug(data.to_csv(self.output_file, index=False))

src/decisionengine_modules/graphite_client.py

@@ -5,6 +5,10 @@
 import structlog
 from decisionengine.framework.modules.logging_configDict import CHANNELLOGGERNAME
 
+from functools import partial
+
+from decisionengine_modules.util.retry_function import retry_wrapper
+
 
 def sanitize_key(key):
     if key is None:
@@ -25,12 +29,13 @@ def __init__(self, host="fifemondata.fnal.gov", pickle_port=2004):
         self.logger = structlog.getLogger(CHANNELLOGGERNAME)
         self.logger = self.logger.bind(module_class=__name__.split(".")[-1], channel="")
 
-    def send_dict(self, namespace, data, debug_print=True, send_data=True):
+    def send_dict(self, namespace, data, debug_print=True, send_data=True, max_retries=2, retry_interval=60):
         """send data contained in dictionary as {k: v} to graphite dataset
         $namespace.k with current timestamp"""
         if data is None:
             self.logger.warning("Warning: send_dict called with no data")
             return
+
         now = int(time.time())
         post_data = []
         # turning data dict into [('$path.$key',($timestamp,$value)),...]]
@@ -48,14 +53,12 @@ def send_dict(self, namespace, data, debug_print=True, send_data=True):
             return
         # throw data at graphite
 
-        s = socket.socket()
-        try:
+        retry_wrapper(partial(self._send_to_graphite, message), max_retries, retry_interval)
+
+    def _send_to_graphite(self, message):
+        with socket.socket() as s:
             s.connect((self.graphite_host, self.graphite_pickle_port))
             s.sendall(message)
-        except socket.error:
-            self.logger.exception(f"Error sending data to graphite at {self.graphite_host}:{self.graphite_pickle_port}")
-        finally:
-            s.close()
 
 
 if __name__ == "__main__":