Merge pull request #1061 from dod-ccpo/csp-retry-failure

Record job failures with application context.

atst/app.py

@@ -169,6 +169,12 @@ def map_config(config):
         "LIMIT_CONCURRENT_SESSIONS": config.getboolean(
             "default", "LIMIT_CONCURRENT_SESSIONS"
         ),
+        # Store the celery task results in a database table (celery_taskmeta)
+        "CELERY_RESULT_BACKEND": "db+{}".format(config.get("default", "DATABASE_URI")),
+        # Do not automatically delete results (by default, Celery will do this
+        # with a Beat job once a day)
+        "CELERY_RESULT_EXPIRES": 0,
+        "CELERY_RESULT_EXTENDED": True,
     }
 
 

atst/jobs.py

@@ -1,14 +1,36 @@
 from flask import current_app as app
 
 from atst.queue import celery
+from atst.database import db
+from atst.models import EnvironmentJobFailure, EnvironmentRoleJobFailure
 
 
-@celery.task()
+class RecordEnvironmentFailure(celery.Task):
+    def on_failure(self, exc, task_id, args, kwargs, einfo):
+        if "environment_id" in kwargs:
+            failure = EnvironmentJobFailure(
+                environment_id=kwargs["environment_id"], task_id=task_id
+            )
+            db.session.add(failure)
+            db.session.commit()
+
+
+class RecordEnvironmentRoleFailure(celery.Task):
+    def on_failure(self, exc, task_id, args, kwargs, einfo):
+        if "environment_role_id" in kwargs:
+            failure = EnvironmentRoleJobFailure(
+                environment_role_id=kwargs["environment_role_id"], task_id=task_id
+            )
+            db.session.add(failure)
+            db.session.commit()
+
+
+@celery.task(ignore_result=True)
 def send_mail(recipients, subject, body):
     app.mailer.send(recipients, subject, body)
 
 
-@celery.task()
+@celery.task(ignore_result=True)
 def send_notification_mail(recipients, subject, body):
     app.logger.info(
         "Sending a notification to these recipients: {}\n\nSubject: {}\n\n{}".format(

atst/models/__init__.py

@@ -11,6 +11,7 @@ from .audit_event import AuditEvent
 from .clin import CLIN, JEDICLINType
 from .environment import Environment
 from .environment_role import EnvironmentRole, CSPRole
+from .job_failure import EnvironmentJobFailure, EnvironmentRoleJobFailure
 from .notification_recipient import NotificationRecipient
 from .permissions import Permissions
 from .permission_set import PermissionSet

atst/models/environment.py

@@ -19,6 +19,8 @@ class Environment(
 
     cloud_id = Column(String)
 
+    job_failures = relationship("EnvironmentJobFailure")
+
     @property
     def users(self):
         return {r.application_role.user for r in self.roles}

atst/models/environment_role.py

@@ -31,6 +31,8 @@ class EnvironmentRole(
     )
     application_role = relationship("ApplicationRole")
 
+    job_failures = relationship("EnvironmentRoleJobFailure")
+
     def __repr__(self):
         return "<EnvironmentRole(role='{}', user='{}', environment='{}', id='{}')>".format(
             self.role, self.application_role.user_name, self.environment.name, self.id

atst/models/job_failure.py

@@ -0,0 +1,16 @@
+from sqlalchemy import Column, ForeignKey
+
+from atst.models import Base
+from atst.models import mixins
+
+
+class EnvironmentJobFailure(Base, mixins.JobFailureMixin):
+    __tablename__ = "environment_job_failures"
+
+    environment_id = Column(ForeignKey("environments.id"), nullable=False)
+
+
+class EnvironmentRoleJobFailure(Base, mixins.JobFailureMixin):
+    __tablename__ = "environment_role_job_failures"
+
+    environment_role_id = Column(ForeignKey("environment_roles.id"), nullable=False)

atst/models/mixins/__init__.py

@@ -3,3 +3,4 @@ from .auditable import AuditableMixin
 from .permissions import PermissionsMixin
 from .deletable import DeletableMixin
 from .invites import InvitesMixin
+from .job_failure import JobFailureMixin

atst/models/mixins/job_failure.py

@@ -0,0 +1,14 @@
+from celery.result import AsyncResult
+from sqlalchemy import Column, String, Integer
+
+
+class JobFailureMixin(object):
+    id = Column(Integer(), primary_key=True)
+    task_id = Column(String(), nullable=False)
+
+    @property
+    def task(self):
+        if not hasattr(self, "_task"):
+            self._task = AsyncResult(self.task_id)
+
+        return self._task