[celerytask] Add support for fetching and deleting logs in CloudWatch and S3

cubicweb_celerytask/entities.py

@@ -27,7 +27,16 @@ from cubicweb.view import EntityAdapter
 from cubicweb.predicates import is_instance
 from cubicweb.server.hook import DataOperationMixIn, Operation
 
-from cw_celerytask_helpers.filelogger import get_task_logs
+from cw_celerytask_helpers import filelogger
+# CloudWatch and S3 dependencies are optional
+try:
+    from cw_celerytask_helpers import cloudwatchlogger
+except ImportError:
+    cloudwatchlogger = None
+try:
+    from cw_celerytask_helpers import s3logger
+except ImportError:
+    s3logger = None
 
 from cubicweb_celerytask import STATES, FINAL_STATES
 
@@ -193,7 +202,12 @@ class ICeleryTask(EntityAdapter):
 
     @property
     def logs(self):
-        return get_task_logs(self.task_id) or b''
+        for logger in [cloudwatchlogger, s3logger, filelogger]:
+            if logger:
+                logs = logger.get_task_logs(self.task_id)
+                if logs:
+                    return logs
+        return b''
 
     @property
     def result(self):

cubicweb_celerytask/hooks.py

@@ -28,7 +28,16 @@ from cubicweb import ConfigurationError
 from cubicweb.predicates import is_instance
 from cubicweb.server.hook import Hook, DataOperationMixIn, Operation
 
-from cw_celerytask_helpers.filelogger import flush_task_logs
+from cw_celerytask_helpers import filelogger
+# CloudWatch and S3 dependencies are optional
+try:
+    from cw_celerytask_helpers import cloudwatchlogger
+except ImportError:
+    cloudwatchlogger = None
+try:
+    from cw_celerytask_helpers import s3logger
+except ImportError:
+    s3logger = None
 
 
 class DeleteCeleryTaskOp(DataOperationMixIn, Operation):
@@ -39,7 +48,12 @@ class DeleteCeleryTaskOp(DataOperationMixIn, Operation):
             celery.task.control.revoke(list(tasks), terminate=True,
                                        signal='SIGKILL')
         for task_id in tasks:
-            flush_task_logs(task_id)
+            for logger in [filelogger, cloudwatchlogger, s3logger]:
+                if logger:
+                    try:
+                        logger.flush_task_logs(task_id)
+                    except Exception:
+                        pass
 
 
 class CeleryTaskDeletedHook(Hook):