schedule metrics reporters in their own tasks

osf/management/commands/daily_reporters_go.py

@@ -2,11 +2,11 @@
 import logging
 
 from django.core.management.base import BaseCommand
+from django.db.utils import OperationalError
 from django.utils import timezone
 
-from framework import sentry
 from framework.celery_tasks import app as celery_app
-from osf.metrics.reporters import DAILY_REPORTERS
+from osf.metrics.reporters import AllDailyReporters
 from website.app import init_app
 
 
@@ -20,25 +20,27 @@ def daily_reporters_go(also_send_to_keen=False, report_date=None, reporter_filte
     if report_date is None:  # default to yesterday
         report_date = (timezone.now() - datetime.timedelta(days=1)).date()
 
-    errors = {}
-    for reporter_class in DAILY_REPORTERS:
-        if reporter_filter and (reporter_filter.lower() not in reporter_class.__name__.lower()):
+    for _reporter_key, _reporter_class in AllDailyReporters.__members__.items():
+        if reporter_filter and (reporter_filter.lower() not in _reporter_class.__name__.lower()):
             continue
-        try:
-            reporter_class().run_and_record_for_date(
-                report_date=report_date,
-                also_send_to_keen=also_send_to_keen,
-            )
-        except Exception as e:
-            errors[reporter_class.__name__] = repr(e)
-            logger.exception(e)
-            sentry.log_exception(e)
-            # continue with the next reporter
-    return errors
+        daily_reporter_go.apply_async(kwargs={
+            'reporter_key': _reporter_key,
+            'report_date': report_date.isoformat(),
+        })
 
 
-def date_fromisoformat(date_str):
-    return datetime.datetime.strptime(date_str, '%Y-%m-%d').date()
+@celery_app.task(
+    name='management.commands.daily_reporter_go',
+    max_retries=5,
+    bind=True,
+)
+def daily_reporter_go(task, reporter_key: str, report_date: str):
+    _reporter_class = AllDailyReporters[reporter_key].value
+    _parsed_date = datetime.date.fromisoformat(report_date)
+    try:
+        _reporter_class().run_and_record_for_date(report_date=_parsed_date)
+    except OperationalError as e:
+        raise task.retry(exc=e)
 
 
 class Command(BaseCommand):
@@ -51,7 +53,7 @@ def add_arguments(self, parser):
         )
         parser.add_argument(
             '--date',
-            type=date_fromisoformat,  # in python 3.7+, could pass datetime.date.fromisoformat
+            type=datetime.date.fromisoformat,
             help='run for a specific date (default: yesterday)',
         )
         parser.add_argument(

osf/management/commands/monthly_reporters_go.py

@@ -1,11 +1,11 @@
 import logging
 
 from django.core.management.base import BaseCommand
+from django.db.utils import OperationalError
 from django.utils import timezone
 
-from framework import sentry
 from framework.celery_tasks import app as celery_app
-from osf.metrics.reporters import MONTHLY_REPORTERS
+from osf.metrics.reporters import AllMonthlyReporters
 from osf.metrics.utils import YearMonth
 from website.app import init_app
 
@@ -28,17 +28,25 @@ def monthly_reporters_go(report_year=None, report_month=None):
             year=today.year if today.month > 1 else today.year - 1,
             month=today.month - 1 or MAXMONTH,
         )
+    for _reporter_key in AllMonthlyReporters.__members__.keys():
+        monthly_reporter_go.apply_async(kwargs={
+            'reporter_key': _reporter_key,
+            'yearmonth': str(report_yearmonth),
+        })
 
-    errors = {}
-    for reporter_class in MONTHLY_REPORTERS:
-        try:
-            reporter_class().run_and_record_for_month(report_yearmonth)
-        except Exception as e:
-            errors[reporter_class.__name__] = str(e)
-            logger.exception(e)
-            sentry.log_exception(e)
-            # continue with the next reporter
-    return errors
+
+@celery_app.task(
+    name='management.commands.monthly_reporter_go',
+    max_retries=5,
+    bind=True,
+)
+def monthly_reporter_go(task, reporter_key: str, yearmonth: str):
+    _reporter_class = AllMonthlyReporters[reporter_key].value
+    _parsed_yearmonth = YearMonth.from_str(yearmonth)
+    try:
+        _reporter_class().run_and_record_for_month(_parsed_yearmonth)
+    except OperationalError as e:
+        raise task.retry(exc=e)
 
 
 class Command(BaseCommand):

osf/metrics/reporters/__init__.py

@@ -1,3 +1,5 @@
+import enum
+
 # from .active_users import ActiveUserReporter
 from .storage_addon_usage import StorageAddonUsageReporter
 from .download_count import DownloadCountReporter
@@ -10,18 +12,17 @@
 from .spam_count import SpamCountReporter
 
 
-DAILY_REPORTERS = (
-    # ActiveUserReporter,
-    DownloadCountReporter,
-    InstitutionSummaryReporter,
-    NewUserDomainReporter,
-    NodeCountReporter,
-    OsfstorageFileCountReporter,
-    PreprintCountReporter,
-    StorageAddonUsageReporter,
-    UserCountReporter,
-)
+class AllDailyReporters(enum.Enum):
+    # ACTIVE_USER = ActiveUserReporter
+    DOWNLOAD_COUNT = DownloadCountReporter
+    INSTITUTION_SUMMARY = InstitutionSummaryReporter
+    NEW_USER_DOMAIN = NewUserDomainReporter
+    NODE_COUNT = NodeCountReporter
+    OSFSTORAGE_FILE_COUNT = OsfstorageFileCountReporter
+    PREPRINT_COUNT = PreprintCountReporter
+    STORAGE_ADDON_USAGE = StorageAddonUsageReporter
+    USER_COUNT = UserCountReporter
+
 
-MONTHLY_REPORTERS = (
-    SpamCountReporter,
-)
+class AllMonthlyReporters(enum.Enum):
+    SPAM_COUNT = SpamCountReporter