Mark 404'd Repos to be Ignored and Re-collect Errored Repos

augur/application/cli/backend.py

@@ -19,7 +19,8 @@
 from datetime import datetime
 
 from augur import instance_id
-from augur.tasks.start_tasks import augur_collection_monitor, CollectionState, create_collection_status_records
+from augur.tasks.util.collection_state import CollectionState
+from augur.tasks.start_tasks import augur_collection_monitor, create_collection_status_records
 from augur.tasks.git.facade_tasks import clone_repos
 from augur.tasks.data_analysis.contributor_breadth_worker.contributor_breadth_worker import contributor_breadth_model
 from augur.tasks.init.redis_connection import redis_connection 
@@ -38,7 +39,7 @@
 
 
 @click.group('server', short_help='Commands for controlling the backend API server & data collection workers')
 def cli():
     pass
 
 @cli.command("start")
@@ -47,7 +48,7 @@
 @click.option('--port')
 @test_connection
 @test_db_connection
 def start(disable_collection, development, port):
     """Start Augur's backend server."""
 
     try:
@@ -79,13 +80,13 @@
         worker_vmem_cap = config.get_value("Celery", 'worker_process_vmem_cap')
 
     gunicorn_command = f"gunicorn -c {gunicorn_location} -b {host}:{port} augur.api.server:app --log-file gunicorn.log"
     server = subprocess.Popen(gunicorn_command.split(" "))
 
     time.sleep(3)
     logger.info('Gunicorn webserver started...')
     logger.info(f'Augur is running at: {"http" if development else "https"}://{host}:{port}')
 
     processes = start_celery_worker_processes(float(worker_vmem_cap), disable_collection)
 
     if os.path.exists("celerybeat-schedule.db"):
             logger.info("Deleting old task schedule")
@@ -96,7 +97,7 @@
         log_level = config.get_value("Logging", "log_level")
         celery_beat_process = None
         celery_command = f"celery -A augur.tasks.init.celery_app.celery_app beat -l {log_level.lower()}"
         celery_beat_process = subprocess.Popen(celery_command.split(" "))    
 
     if not disable_collection:
 
@@ -142,7 +143,7 @@
             except RedisConnectionError:
                 pass
 
 def start_celery_worker_processes(vmem_cap_ratio, disable_collection=False):
 
     #Calculate process scaling based on how much memory is available on the system in bytes.
     #Each celery process takes ~500MB or 500 * 1024^2 bytes
@@ -161,7 +162,7 @@
 
     frontend_worker = f"celery -A augur.tasks.init.celery_app.celery_app worker -l info --concurrency=1 -n frontend:{uuid.uuid4().hex}@%h -Q frontend"
     max_process_estimate -= 1
     process_list.append(subprocess.Popen(frontend_worker.split(" ")))
     sleep_time += 6
 
     if not disable_collection:
@@ -169,7 +170,7 @@
         #2 processes are always reserved as a baseline.
         scheduling_worker = f"celery -A augur.tasks.init.celery_app.celery_app worker -l info --concurrency=2 -n scheduling:{uuid.uuid4().hex}@%h -Q scheduling"
         max_process_estimate -= 2
         process_list.append(subprocess.Popen(scheduling_worker.split(" ")))
         sleep_time += 6
 
         #60% of estimate, Maximum value of 45
@@ -307,7 +308,7 @@
         SET facade_status='Pending', facade_task_id=NULL
         WHERE facade_status='Failed Clone' OR facade_status='Initializing';
     """))
     #TODO: write timestamp for currently running repos.
 
 def assign_orphan_repos_to_default_user(session):
     query = s.sql.text("""

augur/tasks/git/facade_tasks.py

@@ -31,7 +31,8 @@
 from augur.tasks.git.util.facade_worker.facade_worker.utilitymethods import get_facade_weight_time_factor, get_repo_commit_count, update_facade_scheduling_fields, get_facade_weight_with_commit_count, facade_bulk_insert_commits
 
 from augur.tasks.github.facade_github.tasks import *
-from augur.tasks.util.collection_util import CollectionState, get_collection_status_repo_git_from_filter
+from augur.tasks.util.collection_state import CollectionState
+from augur.tasks.util.collection_util import get_collection_status_repo_git_from_filter
 from augur.tasks.git.util.facade_worker.facade_worker.repofetch import GitCloneError, git_repo_initialize
 
 

augur/tasks/github/detect_move/core.py

@@ -6,14 +6,9 @@
 from augur.tasks.github.util.util import parse_json_response
 import logging
 from datetime import datetime
-from enum import Enum
+from augur.tasks.utl.collection_state import CollectionState
 from augur.application.db.util import execute_session_query
 
-class CollectionState(Enum):
-    SUCCESS = "Success"
-    PENDING = "Pending"
-    ERROR = "Error"
-    COLLECTING = "Collecting"
 
 
 def update_repo_with_dict(current_dict,new_dict,logger,db):
@@ -104,16 +99,16 @@ def ping_github_for_repo_move(augur_db, key_auth, repo, logger,collection_hook='
 
     collectionRecord = execute_session_query(statusQuery,'one')
     if collection_hook == 'core':
-        collectionRecord.core_status = CollectionState.PENDING.value
+        collectionRecord.core_status = CollectionState.STANDBY.value
         collectionRecord.core_task_id = None
         collectionRecord.core_data_last_collected = datetime.today().strftime('%Y-%m-%dT%H:%M:%SZ')
     elif collection_hook == 'secondary':
-        collectionRecord.secondary_status = CollectionState.PENDING.value
+        collectionRecord.secondary_status = CollectionState.STANDBY.value
         collectionRecord.secondary_task_id = None
         collectionRecord.secondary_data_last_collected = datetime.today().strftime('%Y-%m-%dT%H:%M:%SZ')
 
     augur_db.session.commit()
 
-    raise Exception("ERROR: Repo has moved! Marked repo as pending and stopped collection")
+    raise Exception("ERROR: Repo has moved! Marked repo as standby and stopped collection")
 
 

augur/tasks/init/celery_app.py

@@ -20,16 +20,7 @@
 from augur.application.db.engine import get_database_string
 from augur.tasks.init import get_redis_conn_values, get_rabbitmq_conn_string
 from augur.application.db.models import CollectionStatus, Repo
-
-class CollectionState(Enum):
-    SUCCESS = "Success"
-    PENDING = "Pending"
-    ERROR = "Error"
-    COLLECTING = "Collecting"
-    INITIALIZING = "Initializing"
-    UPDATE = "Update"
-    FAILED_CLONE = "Failed Clone"
-
+from augur.tasks.util.collection_state import CollectionState
 
 logger = logging.getLogger(__name__)
 
@@ -85,7 +76,7 @@ class CollectionState(Enum):
 #Classes for tasks that take a repo_git as an argument.
 class AugurCoreRepoCollectionTask(celery.Task):
 
-    def augur_handle_task_failure(self,exc,task_id,repo_git,logger_name,collection_hook='core'):
+    def augur_handle_task_failure(self,exc,task_id,repo_git,logger_name,collection_hook='core',after_fail=CollectionState.ERROR.value):
         from augur.tasks.init.celery_app import engine
 
         logger = AugurLogger(logger_name).get_logger()
@@ -104,7 +95,7 @@ def augur_handle_task_failure(self,exc,task_id,repo_git,logger_name,collection_h
             prevStatus = getattr(repoStatus, f"{collection_hook}_status")
 
             if prevStatus == CollectionState.COLLECTING.value or prevStatus == CollectionState.INITIALIZING.value:
-                setattr(repoStatus, f"{collection_hook}_status", CollectionState.ERROR.value)
+                setattr(repoStatus, f"{collection_hook}_status", after_fail)
                 setattr(repoStatus, f"{collection_hook}_task_id", None)
                 session.commit()
 
@@ -129,6 +120,7 @@ def on_failure(self,exc,task_id,args,kwargs,einfo):
         repo_git = args[0]
         self.augur_handle_task_failure(exc,task_id,repo_git, "ml_task_failure", collection_hook='ml')
 
+
 #task_cls='augur.tasks.init.celery_app:AugurCoreRepoCollectionTask'
 celery_app = Celery('tasks', broker=BROKER_URL, backend=BACKEND_URL, include=tasks)
 
@@ -209,7 +201,7 @@ def setup_periodic_tasks(sender, **kwargs):
     """
     from celery.schedules import crontab
     from augur.tasks.start_tasks import augur_collection_monitor, augur_collection_update_weights
-    from augur.tasks.start_tasks import non_repo_domain_tasks
+    from augur.tasks.start_tasks import non_repo_domain_tasks, retry_404_repos
     from augur.tasks.git.facade_tasks import clone_repos
     from augur.tasks.db.refresh_materialized_views import refresh_materialized_views
     from augur.tasks.data_analysis.contributor_breadth_worker.contributor_breadth_worker import contributor_breadth_model
@@ -234,6 +226,9 @@ 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_404_repos.s())
+
         logger.info(f"Scheduling contributor breadth every 30 days")
         thirty_days_in_seconds = 30*24*60*60
         sender.add_periodic_task(thirty_days_in_seconds, contributor_breadth_model.s())

augur/tasks/start_tasks.py

@@ -36,6 +36,7 @@
 from enum import Enum
 from augur.tasks.util.redis_list import RedisList
 from augur.application.db.models import CollectionStatus, Repo
+from augur.tasks.util.collection_state import CollectionState
 from augur.tasks.util.collection_util import *
 from augur.tasks.git.util.facade_worker.facade_worker.utilitymethods import get_facade_weight_time_factor
 
@@ -328,6 +329,22 @@ def augur_collection_update_weights():
             session.commit()
             #git_update_commit_count_weight(repo_git)
 
+@celery.task
+def retry_404_repos():
+    from augur.tasks.init.celery_app import engine
+    logger = logging.getLogger(create_collection_status_records.__name__)
+
+    with DatabaseSession(logger,engine) as session:
+        query = s.sql.text(f"""UPDATE repo SET secondary_staus = {CollectionState.STANDBY.value}"""
+        f""" WHERE secondary_status = '{CollectionState.PENDING.value}' ;"""
+        f"""UPDATE repo SET core_status = {CollectionState.STANDBY.value}"""
+        f""" WHERE core_status = '{CollectionState.PENDING.value}' ;"""
+        )
+
+        session.execute_sql(query)
+
+
+
 #Retry this task for every issue so that repos that were added manually get the chance to be added to the collection_status table.
 @celery.task(autoretry_for=(Exception,), retry_backoff=True, retry_backoff_max=300, retry_jitter=True, max_retries=None)
 def create_collection_status_records():

augur/tasks/util/collection_state.py

@@ -0,0 +1,11 @@
+from enum import Enum
+
+class CollectionState(Enum):
+    SUCCESS = "Success"
+    PENDING = "Pending"
+    ERROR = "Error"
+    COLLECTING = "Collecting"
+    INITIALIZING = "Initializing"
+    UPDATE = "Update"
+    FAILED_CLONE = "Failed Clone"
+    STANDBY = "Standby"

augur/tasks/util/collection_util.py

@@ -24,18 +24,9 @@
 from augur.tasks.github.util.github_task_session import GithubTaskManifest
 from augur.application.db.session import DatabaseSession
 from augur.tasks.util.worker_util import calculate_date_weight_from_timestamps
+from augur.tasks.util.collection_state import CollectionState
 
 
-# class syntax
-class CollectionState(Enum):
-    SUCCESS = "Success"
-    PENDING = "Pending"
-    ERROR = "Error"
-    COLLECTING = "Collecting"
-    INITIALIZING = "Initializing"
-    UPDATE = "Update"
-    FAILED_CLONE = "Failed Clone"
-
 def get_list_of_all_users(session):
     #Get a list of all users.
     query = s.sql.text("""