Add --max-retry CLI flag

Author: ka7eh

CHANGELOG.md

@@ -4,6 +4,11 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/) 
 and this project adheres to [Semantic Versioning](http://semver.org/).
 
+## UNRELEASED
+
+### Added
+- Add `--max_retry` CLI flag and `CLOWDER_MAX_RETRY` environment variable.
+
 ## 2.4.0 - 2021-02-22
 
 ### Changed

pyclowder/connectors.py

@@ -66,7 +66,7 @@ class Connector(object):
     registered_clowder = list()
 
     def __init__(self, extractor_name, extractor_info, check_message=None, process_message=None, ssl_verify=True,
-                 mounted_paths=None, clowder_url=None):
+                 mounted_paths=None, clowder_url=None, max_retry=10):
         self.extractor_name = extractor_name
         self.extractor_info = extractor_info
         self.check_message = check_message
@@ -77,6 +77,7 @@ def __init__(self, extractor_name, extractor_info, check_message=None, process_m
         else:
             self.mounted_paths = mounted_paths
         self.clowder_url = clowder_url
+        self.max_retry = max_retry
 
         filename = 'notifications.json'
         self.smtp_server = None
@@ -490,7 +491,7 @@ def _process_message(self, body):
         except Exception as exc:  # pylint: disable=broad-except
             message = str(exc)
             logger.exception("[%s] %s", resource['id'], message)
-            if retry_count < 10:
+            if retry_count < self.max_retry:
                 message = "(#%s) %s" % (retry_count+1, message)
                 self.message_resubmit(resource, retry_count+1, message)
             else:
@@ -629,9 +630,9 @@ class RabbitMQConnector(Connector):
     def __init__(self, extractor_name, extractor_info,
                  rabbitmq_uri, rabbitmq_exchange=None, rabbitmq_key=None, rabbitmq_queue=None,
                  check_message=None, process_message=None, ssl_verify=True, mounted_paths=None,
-                 heartbeat=5*60, clowder_url=None):
+                 heartbeat=5*60, clowder_url=None, max_retry=10):
         super(RabbitMQConnector, self).__init__(extractor_name, extractor_info, check_message, process_message,
-                                                ssl_verify, mounted_paths, clowder_url)
+                                                ssl_verify, mounted_paths, clowder_url, max_retry)
         self.rabbitmq_uri = rabbitmq_uri
         self.rabbitmq_exchange = rabbitmq_exchange
         self.rabbitmq_key = rabbitmq_key
@@ -852,10 +853,10 @@ class RabbitMQHandler(Connector):
     """
 
     def __init__(self, extractor_name, extractor_info, job_id, check_message=None, process_message=None, ssl_verify=True,
-                 mounted_paths=None, clowder_url=None, method=None, header=None, body=None):
+                 mounted_paths=None, clowder_url=None, method=None, header=None, body=None, max_retry=10):
 
         super(RabbitMQHandler, self).__init__(extractor_name, extractor_info, check_message, process_message,
-                                              ssl_verify, mounted_paths, clowder_url)
+                                              ssl_verify, mounted_paths, clowder_url, max_retry)
         self.method = method
         self.header = header
         self.body = body
@@ -977,9 +978,9 @@ class HPCConnector(Connector):
 
     # pylint: disable=too-many-arguments
     def __init__(self, extractor_name, extractor_info, picklefile, job_id=None,
-                 check_message=None, process_message=None, ssl_verify=True, mounted_paths=None):
+                 check_message=None, process_message=None, ssl_verify=True, mounted_paths=None, max_retry=10):
         super(HPCConnector, self).__init__(extractor_name, extractor_info, check_message, process_message,
-                                           ssl_verify, mounted_paths)
+                                           ssl_verify, mounted_paths, max_retry=max_retry)
         self.job_id = job_id
         self.picklefile = picklefile
         self.logfile = None
@@ -1036,8 +1037,8 @@ class LocalConnector(Connector):
 
     """
 
-    def __init__(self, extractor_name, extractor_info, input_file_path, process_message=None, output_file_path=None):
-        super(LocalConnector, self).__init__(extractor_name, extractor_info, process_message=process_message)
+    def __init__(self, extractor_name, extractor_info, input_file_path, process_message=None, output_file_path=None, max_retry=10):
+        super(LocalConnector, self).__init__(extractor_name, extractor_info, process_message=process_message, max_retry=max_retry)
         self.input_file_path = input_file_path
         self.output_file_path = output_file_path
         self.completed_processing = False

pyclowder/extractors.py

@@ -73,6 +73,7 @@ def __init__(self):
         connector_default = "RabbitMQ"
         if os.getenv('LOCAL_PROCESSING', "False").lower() == "true":
             connector_default = "Local"
+        max_retry = os.getenv('CLOWDER_MAX_RETRY', 10)
 
         # create the actual extractor
         self.parser = argparse.ArgumentParser(description=self.extractor_info['description'])
@@ -108,6 +109,8 @@ def __init__(self):
         self.parser.add_argument('--version', action='version', version='%(prog)s 1.0')
         self.parser.add_argument('--no-bind', dest="nobind", action='store_true',
                                  help='instance will bind itself to RabbitMQ by name but NOT file type')
+        self.parser.add_argument('--max-retry', dest='max_retry', default=max_retry,
+                                 help='Maximum number of retries if an error happens in the extractor')
 
     def setup(self):
         """Parse command line arguments and so some setup
@@ -161,7 +164,8 @@ def start(self):
                                               rabbitmq_key=rabbitmq_key,
                                               rabbitmq_queue=self.args.rabbitmq_queuename,
                                               mounted_paths=json.loads(self.args.mounted_paths),
-                                              clowder_url=self.args.clowder_url)
+                                              clowder_url=self.args.clowder_url,
+                                              max_retry=self.args.max_retry)
                 connector.connect()
                 connector.register_extractor(self.args.registration_endpoints)
                 threading.Thread(target=connector.listen, name="RabbitMQConnector").start()
@@ -175,7 +179,8 @@ def start(self):
                                          check_message=self.check_message,
                                          process_message=self.process_message,
                                          picklefile=self.args.hpc_picklefile,
-                                         mounted_paths=json.loads(self.args.mounted_paths))
+                                         mounted_paths=json.loads(self.args.mounted_paths),
+                                         max_retry=self.args.max_retry)
                 connector.register_extractor(self.args.registration_endpoints)
                 threading.Thread(target=connector.listen, name="HPCConnector").start()
 
@@ -191,7 +196,8 @@ def start(self):
                                            self.extractor_info,
                                            self.args.input_file_path,
                                            process_message=self.process_message,
-                                           output_file_path=self.args.output_file_path)
+                                           output_file_path=self.args.output_file_path,
+                                           max_retry=self.args.max_retry)
                 threading.Thread(target=connector.listen, name="LocalConnector").start()
         else:
             logger.error("Could not create instance of %s connector.", self.args.connector)