Merge pull request #34 from clowder-framework/add-max-retry-flag

Add --max-retry CLI flag

Author: robkooper

.github/workflows/ci.yaml

@@ -1,13 +1,15 @@
 name: Python package
 
-on: [push]
+on:
+  push:
+  pull_request:
 
 jobs:
   build:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: [3.5, 3.6, 3.7, 3.8, 3.9]
+        python-version: [3.6, 3.7, 3.8, 3.9]
 
     steps:
     - uses: actions/checkout@v2

.github/workflows/pypi.yaml

@@ -5,6 +5,7 @@ on:
     branches:
       - main
       - master
+  pull_request:
 
 jobs:
   publish:
@@ -49,7 +50,7 @@ jobs:
           python setup.py sdist bdist_wheel
 
       - name: Publish distribution to PyPI
-        if: steps.release_info.outputs.version != 'unreleased'
+        if: github.event_name == 'push' && env.BRANCH == 'master' && steps.release_info.outputs.version != 'unreleased'
         uses: pypa/gh-action-pypi-publish@master
         with:
           password: ${{ secrets.pypi_password }}

.github/workflows/release.yaml

@@ -3,7 +3,6 @@ name: Create Release
 on:
   push:
     branches:
-      - main
       - master
 
 jobs:
@@ -27,7 +26,7 @@ jobs:
           echo "::set-output name=changelog::$changelog"
 
       - name: create release
-        if: steps.release_info.outputs.version != 'unreleased'
+        if: github.event_name == 'push' && env.BRANCH == 'master' && steps.release_info.outputs.version != 'unreleased'
         uses: actions/create-release@v1
         id: create_release
         env:

CHANGELOG.md

@@ -4,7 +4,10 @@ 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
+## UNRELEASED
+
+### Added
+- Add `--max_retry` CLI flag and `CLOWDER_MAX_RETRY` environment variable.
 
 ### Changed
 - updated all of the requirements to latest versions

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)