Merge pull request #722 from mskcc/develop

Hotfixes for 1.38.0
mskcc · Feb 17, 2021 · 59d3280 · 59d3280
2 parents f8d4581 + 8f8c7c4
commit 59d3280
Show file tree

Hide file tree

Showing 4 changed files with 25 additions and 23 deletions.
diff --git a/container/Readme.md b/container/Readme.md
@@ -58,6 +58,7 @@ SINGULARITYENV_BEAGLE_RABIX_URL
 SINGULARITYENV_BEAGLE_RABIX_PATH
 SINGULARITYENV_BEAGLE_RABBITMQ_USERNAME
 SINGULARITYENV_BEAGLE_RABBITMQ_PASSWORD
+SINGULARITYENV_CELERY_BROKER_URL
 SINGULARITYENV_BEAGLE_LIMS_USERNAME
 SINGULARITYENV_BEAGLE_LIMS_PASSWORD
 SINGULARITYENV_BEAGLE_RUNNER_QUEUE

diff --git a/container/celery_services.def b/container/celery_services.def
@@ -1,5 +1,5 @@
 Bootstrap: docker
-From: ubuntu:19.10
+From: ubuntu:18.04
 Includecmd: no
 
 %startscript
@@ -36,7 +36,7 @@ Includecmd: no
 
     nohup celery -A beagle_etl worker --workdir ${BEAGLE_PATH} -l info -Q ${BEAGLE_RUNNER_QUEUE} -f ${CELERY_LOG_PATH}/beagle_runner.log --pidfile ${CELERY_PID_PATH}/${CELERY_EVENT_QUEUE_PREFIX}.beagle_runner.pid -n ${CELERY_EVENT_QUEUE_PREFIX}.beagle_runner &
 
-    nohup celery flower -A beagle_etl --port=${FLOWER_PORT} --broker=$CELERY_BROKER_URL// &
+    nohup celery -A beagle_etl flower --workdir ${BEAGLE_PATH} -f ${CELERY_LOG_PATH}/beagle_flower.log --pidfile ${CELERY_PID_PATH}/${CELERY_EVENT_QUEUE_PREFIX}.beagle_flower.pid --port=${FLOWER_PORT} --broker=$CELERY_BROKER_URL// &
 
 %post
     export DEBIAN_FRONTEND=noninteractive \

diff --git a/docs/ENVIRONMENT_VARIABLES.md b/docs/ENVIRONMENT_VARIABLES.md
@@ -38,6 +38,7 @@ BEAGLE_DEFAULT_QUEUE | Rabbitmq default queue | example.runner.queue
 BEAGLE_JOB_SCHEDULER_QUEUE | Rabbitmq scheduler queue | example.runner.queue
 CELERY_EVENT_QUEUE_PREFIX | Prefix for Celery event | beagle.celery
 CELERY_LOG_PATH | Log path for Celery | /path/to/celey.log
+CELERY_BROKER_URL | URL to Celery broker | amqp://[RABBITMQ_USERNAME]:[RABBITMQ_PASSWORD]@localhost:5672
 JIRA_USERNAME | JIRA username | example_username
 JIRA_PASSWORD | JIRA password | example_password
 JIRA_URL | JIRA URL | http://jira.example.org:5000

diff --git a/runner/tasks.py b/runner/tasks.py
@@ -529,27 +529,27 @@ def check_jobs_status():
             status = remote_statuses[str(run.execution_id)]
             if status['started'] and not run.started:
                 run.started = status['started']
-                if status['submitted'] and not run.submitted:
-                    run.submitted = status['submitted']
-                    if status['commandlinetooljob_set']:
-                        update_commandline_job_status(run, status['commandlinetooljob_set'])
-                    if status['status'] == 'FAILED':
-                        logger.info("Job %s [%s] FAILED" % (run.id, run.execution_id))
-                        message = dict(details=status.get('message'))
-                        fail_job(str(run.id),
-                                 message)
-                        continue
-                    if status['status'] == 'COMPLETED':
-                        logger.info("Job %s [%s] COMPLETED" % (run.id, run.execution_id))
-                        complete_job(str(run.id), status['outputs'])
-                        continue
-                    if status['status'] == 'CREATED' or status['status'] == 'PENDING' or status['status'] == 'RUNNING':
-                        logger.info("Job %s [%s] RUNNING" % (run.id, run.execution_id))
-                        running_job(run)
-                        continue
-                    if status['status'] == 'ABORTED':
-                        logger.info("Job %s [%s] ABORTED" % (run.id, run.execution_id))
-                        abort_job(run)
+            if status['submitted'] and not run.submitted:
+                run.submitted = status['submitted']
+            if status['commandlinetooljob_set']:
+                update_commandline_job_status(run, status['commandlinetooljob_set'])
+            if status['status'] == 'FAILED':
+                logger.info("Job %s [%s] FAILED" % (run.id, run.execution_id))
+                message = dict(details=status.get('message'))
+                fail_job(str(run.id),
+                         message)
+                continue
+            if status['status'] == 'COMPLETED':
+                logger.info("Job %s [%s] COMPLETED" % (run.id, run.execution_id))
+                complete_job(str(run.id), status['outputs'])
+                continue
+            if status['status'] == 'CREATED' or status['status'] == 'PENDING' or status['status'] == 'RUNNING':
+                logger.info("Job %s [%s] RUNNING" % (run.id, run.execution_id))
+                running_job(run)
+                continue
+            if status['status'] == 'ABORTED':
+                logger.info("Job %s [%s] ABORTED" % (run.id, run.execution_id))
+                abort_job(run)
 
 def run_routine_operator_job(operator, job_group_id=None):
     """