#issue3406

Signed-off-by: Aayush Patidar <patidaraayush053@gmail.com>

Author: aayush-patidar

augur/application/config.py

@@ -90,7 +90,12 @@ def redact_setting_value(section_name, setting_name, value):
                 "core_collection_interval_days": 15,
                 "secondary_collection_interval_days": 10,
                 "facade_collection_interval_days": 10,
-                "ml_collection_interval_days": 40
+                "ml_collection_interval_days": 40,
+                "non_repo_domain_tasks_interval_in_days": 30,
+                "retry_errored_repos_cron_hour": 0,
+                "retry_errored_repos_cron_minute": 0,
+                "process_contributors_interval_in_seconds": 3600,
+                "create_collection_status_records_interval_in_seconds": 86400
             },
             "Message_Insights": {
                     "insight_days": 30,

augur/tasks/init/celery_app.py

@@ -235,10 +235,15 @@ def setup_periodic_tasks(sender, **kwargs):
         logger.info(f"Scheduling collection every {collection_interval/60} minutes")
         sender.add_periodic_task(collection_interval, augur_collection_monitor.s())
 
-        #Do longer tasks less often
-        logger.info(f"Scheduling data analysis every 30 days")
-        thirty_days_in_seconds = 30*24*60*60
-        sender.add_periodic_task(thirty_days_in_seconds, non_repo_domain_tasks.s())
+        # non_repo_domain_tasks
+        non_repo_domain_tasks_val = config.get_value('Tasks', 'non_repo_domain_tasks_interval_in_days')
+        non_repo_domain_tasks_interval = int(non_repo_domain_tasks_val) if non_repo_domain_tasks_val is not None else 30
+        if non_repo_domain_tasks_interval > 0:
+            logger.info(f"Scheduling non_repo_domain_tasks every {non_repo_domain_tasks_interval} days")
+            sender.add_periodic_task(datetime.timedelta(days=non_repo_domain_tasks_interval), non_repo_domain_tasks.s())
+        else:
+            logger.info("non_repo_domain_tasks is disabled (interval <= 0)")
+
 
         mat_views_interval = int(config.get_value('Celery', 'refresh_materialized_views_interval_in_days'))
         if mat_views_interval > 0: 
@@ -250,14 +255,32 @@ def setup_periodic_tasks(sender, **kwargs):
         # logger.info(f"Scheduling update of collection weights on midnight each day")
         # sender.add_periodic_task(crontab(hour=0, minute=0),augur_collection_update_weights.s())
 
-        logger.info(f"Setting 404 repos to be marked for retry on midnight each day")
-        sender.add_periodic_task(crontab(hour=0, minute=0),retry_errored_repos.s())
-
-        one_hour_in_seconds = 60*60
-        sender.add_periodic_task(one_hour_in_seconds, process_contributors.s())
-
-        one_day_in_seconds = 24*60*60
-        sender.add_periodic_task(one_day_in_seconds, create_collection_status_records.s())
+        # retry_errored_repos
+        retry_repos_hour_val = config.get_value('Tasks', 'retry_errored_repos_cron_hour')
+        retry_repos_minute_val = config.get_value('Tasks', 'retry_errored_repos_cron_minute')
+        retry_repos_hour = int(retry_repos_hour_val) if retry_repos_hour_val is not None else 0
+        retry_repos_minute = int(retry_repos_minute_val) if retry_repos_minute_val is not None else 0
+        
+        logger.info(f"Scheduling retry_errored_repos at {retry_repos_hour}:{retry_repos_minute}")
+        sender.add_periodic_task(crontab(hour=retry_repos_hour, minute=retry_repos_minute), retry_errored_repos.s())
+
+        # process_contributors
+        process_contributors_val = config.get_value('Tasks', 'process_contributors_interval_in_seconds')
+        process_contributors_interval = int(process_contributors_val) if process_contributors_val is not None else 3600
+        if process_contributors_interval > 0:
+            logger.info(f"Scheduling process_contributors every {process_contributors_interval} seconds")
+            sender.add_periodic_task(process_contributors_interval, process_contributors.s())
+        else:
+            logger.info("process_contributors is disabled")
+
+        # create_collection_status_records
+        status_records_val = config.get_value('Tasks', 'create_collection_status_records_interval_in_seconds')
+        create_collection_status_records_interval = int(status_records_val) if status_records_val is not None else 86400
+        if create_collection_status_records_interval > 0:
+            logger.info(f"Scheduling create_collection_status_records every {create_collection_status_records_interval} seconds")
+            sender.add_periodic_task(create_collection_status_records_interval, create_collection_status_records.s())
+        else:
+            logger.info("create_collection_status_records is disabled")
 
 @after_setup_logger.connect
 def setup_loggers(*args,**kwargs):

docs/source/development-guide/configuration-file-reference.rst

@@ -3,6 +3,47 @@ Configuration file reference
 
 Augur's configuration template file, which generates your locally deployed ``augur.config.json`` file, is found at ``augur/config.py``. You will notice a small collection of workers are turned on to start with, by examining the ``switch`` variable within the ``Workers`` block of the config file. You can also specify the number of processes to spawn for each worker using the ``workers`` command. The default is one, and we recommend you start here. If you are going to spawn multiple workers, be sure you have enough credentials cached in the ``augur_operations.worker_oath`` table for the platforms you use.
 
+Task Scheduling Configuration
+-----------------------------
+
+The following keys in the ``[Tasks]`` section control the scheduling of periodic Celery tasks. You can adjust these intervals or disable tasks by setting the value to <= 0 (for interval-based tasks).
+
+- ``collection_interval``: Interval in seconds for the main collection monitor. Default: 30.
+- ``core_collection_interval_days``: Interval in days for core collection. Default: 15.
+- ``secondary_collection_interval_days``: Interval in days for secondary collection. Default: 10.
+- ``facade_collection_interval_days``: Interval in days for facade collection. Default: 10.
+- ``ml_collection_interval_days``: Interval in days for ML collection. Default: 40.
+
+**New Periodic Task Intervals:**
+
+- ``non_repo_domain_tasks_interval_in_days``: 
+  - Description: Schedule interval for non-repository domain tasks (e.g., data analysis).
+  - Default: 30 days.
+  - Unit: Days.
+  - Disable: Set to <= 0.
+
+- ``retry_errored_repos_cron_hour``: 
+  - Description: Hour (0-23) to run the retry errored repos task.
+  - Default: 0 (Midnight).
+  - Unit: Hour of day.
+
+- ``retry_errored_repos_cron_minute``: 
+  - Description: Minute (0-59) to run the retry errored repos task.
+  - Default: 0.
+  - Unit: Minute of hour.
+
+- ``process_contributors_interval_in_seconds``: 
+  - Description: Interval to process contributors.
+  - Default: 3600 (1 hour).
+  - Unit: Seconds.
+  - Disable: Set to <= 0.
+
+- ``create_collection_status_records_interval_in_seconds``: 
+  - Description: Interval to create collection status records.
+  - Default: 86400 (1 day).
+  - Unit: Seconds.
+  - Disable: Set to <= 0.
+
 If you have questions or would like to help please open an issue on GitHub_.
 
 .. _GitHub: https://github.com/chaoss/augur/issues

tests/tasks/test_celery_scheduler.py

@@ -0,0 +1,153 @@
+
+import pytest
+from unittest.mock import MagicMock, patch, ANY
+import datetime
+import sys
+
+# Mock modules that might cause side effects on import or are hard to set up
+sys.modules['augur.tasks.init'] = MagicMock()
+sys.modules['augur.tasks.init'].get_redis_conn_values = MagicMock(return_value=(0, 'redis://localhost:6379/'))
+sys.modules['augur.tasks.init'].get_rabbitmq_conn_string = MagicMock(return_value='amqp://guest:guest@localhost:5672//')
+
+# We need to ensure we can import the module now
+# The module augur.tasks.init.celery_app imports get_redis_conn_values etc from augur.tasks.init
+# which we just mocked.
+
+from augur.tasks.init.celery_app import setup_periodic_tasks, split_tasks_into_groups
+
+@pytest.fixture
+def mock_sender():
+    sender = MagicMock()
+    return sender
+
+@pytest.fixture
+def mock_config_factory():
+    def _create_config(settings):
+        config_mock = MagicMock()
+        
+        def get_value(section, key):
+            val = settings.get(section, {}).get(key)
+            return val
+            
+        config_mock.get_value.side_effect = get_value
+        return config_mock
+    return _create_config
+
+class TestCeleryScheduler:
+
+    @patch('augur.tasks.init.celery_app.AugurConfig')
+    @patch('augur.tasks.init.celery_app.DatabaseSession')
+    @patch('augur.tasks.init.celery_app.temporary_database_engine')
+    @patch('augur.tasks.init.celery_app.non_repo_domain_tasks')
+    @patch('augur.tasks.init.celery_app.retry_errored_repos')
+    @patch('augur.tasks.init.celery_app.process_contributors')
+    @patch('augur.tasks.init.celery_app.create_collection_status_records')
+    @patch('augur.tasks.init.celery_app.refresh_materialized_views')
+    @patch('augur.tasks.init.celery_app.augur_collection_monitor')
+    def test_setup_periodic_tasks_scheduling(
+        self,
+        mock_monitor,
+        mock_refresh_views,
+        mock_create_status,
+        mock_process_contributors,
+        mock_retry_repos,
+        mock_non_repo_tasks,
+        mock_temp_engine,
+        mock_db_session,
+        mock_AugurConfig,
+        mock_sender,
+        mock_config_factory
+    ):
+        # Setup config
+        settings = {
+            'Tasks': {
+                'collection_interval': 30,
+                'non_repo_domain_tasks_interval_in_days': 15,
+                'retry_errored_repos_cron_hour': 2,
+                'retry_errored_repos_cron_minute': 30,
+                'process_contributors_interval_in_seconds': 1200,
+                'create_collection_status_records_interval_in_seconds': 4000
+            },
+            'Celery': {
+                'refresh_materialized_views_interval_in_days': 1
+            }
+        }
+        mock_AugurConfig.return_value = mock_config_factory(settings)
+        
+        # Run setup
+        setup_periodic_tasks(mock_sender)
+        
+        # Verify calls
+        
+        # 1. non_repo_domain_tasks: 15 days
+        # The code usually converts days to seconds or timedelta
+        # 15 days = 15 * 24 * 60 * 60 = 1296000 seconds
+        
+        # We need to inspect what arguments add_periodic_task was called with.
+        # It's called multiple times.
+        
+        # Look for non_repo_domain_tasks
+        # We can check if it was called with approximately the right value or check if the task signature was passed
+        
+        # Let's verify that the task was scheduled with the correct interval
+        # Since we haven't implemented the change yet, we expect this to fail if we ran it, or pass if we check for the NEW behavior we want.
+        # This test defines the expected behavior.
+        
+        # Expected:
+        # non_repo_domain_tasks.s() passed as second arg
+        # 15 * 86400 or timedelta(days=15) passed as first arg
+        
+        # Find call for non_repo_domain_tasks
+        found_non_repo = False
+        for call in mock_sender.add_periodic_task.call_args_list:
+            args, _ = call
+            interval = args[0]
+            task_sig = args[1]
+            
+            if task_sig == mock_non_repo_tasks.s():
+                found_non_repo = True
+                # Check interval. 
+                # Our implementation will likely use timedelta or seconds.
+                # 15 days in seconds is 1296000
+                if isinstance(interval, (int, float)):
+                    assert interval == 1296000
+                elif isinstance(interval, datetime.timedelta):
+                    assert interval.days == 15
+        
+        assert found_non_repo, "non_repo_domain_tasks was not scheduled"
+
+    @patch('augur.tasks.init.celery_app.AugurConfig')
+    @patch('augur.tasks.init.celery_app.DatabaseSession')
+    @patch('augur.tasks.init.celery_app.temporary_database_engine')
+    @patch('augur.tasks.init.celery_app.process_contributors')
+    def test_disabled_tasks(
+        self,
+        mock_process_contributors,
+        mock_temp_engine,
+        mock_db_session,
+        mock_AugurConfig,
+        mock_sender,
+        mock_config_factory
+    ):
+        settings = {
+            'Tasks': {
+                'collection_interval': 30,
+                'process_contributors_interval_in_seconds': 0, # Disabled
+            },
+            'Celery': {
+                 'refresh_materialized_views_interval_in_days': 1
+            }
+        }
+        mock_AugurConfig.return_value = mock_config_factory(settings)
+        
+        setup_periodic_tasks(mock_sender)
+        
+        # Verify process_contributors was NOT scheduled
+        found_process = False
+        for call in mock_sender.add_periodic_task.call_args_list:
+            args, _ = call
+            task_sig = args[1]
+            if task_sig == mock_process_contributors.s():
+                found_process = True
+        
+        assert not found_process, "process_contributors should be disabled"