queueworker: prevent stop event on WorkerSleepException (PROJQUAY-1857)…

… (#737)

Prevents the queueworker from setting the event to stop the poll_queue
job when a WorkerSleepException is raised. On WorkerSleepException,
the worker should instead skip this iteration (go to sleep). e.g when
the NamespaceGCWorker can't acquire a lock because it is already taken
by some other worker.

Reverts the gcworkers job timeout from 24h to 3h. In case of a
deadlock between processes (for example, redeploying the app will not
clear the existing Redis keys), 24h is too long waiting for the locks to
expires so that the workers can resume work.

Add missing Counter increment for on row deletion on the Manifest table.

data/model/gc.py

@@ -586,6 +586,8 @@ def _garbage_collect_manifest(manifest_id, context):
     if deleted_manifest_legacy_image:
         gc_table_rows_deleted.labels(table="ManifestLegacyImage").inc(deleted_manifest_legacy_image)
 
+    gc_table_rows_deleted.labels(table="Manifest").inc()
+
     return True
 
 

workers/gc/gcworker.py

@@ -17,7 +17,7 @@
 
 logger = logging.getLogger(__name__)
 
-REPOSITORY_GC_TIMEOUT = 24 * 60 * 60  # 24h
+REPOSITORY_GC_TIMEOUT = 3 * 60 * 60  # 3h
 LOCK_TIMEOUT_PADDING = 60  # 60 seconds
 
 
@@ -60,14 +60,14 @@ def _garbage_collection_repos(self, skip_lock_for_testing=False):
                     except Repository.DoesNotExist:
                         return
 
-                    gc_iterations.inc()
                     logger.debug(
                         "Starting GC of repository #%s (%s)", repository.id, repository.name
                     )
                     garbage_collect_repo(repository)
                     logger.debug(
                         "Finished GC of repository #%s (%s)", repository.id, repository.name
                     )
+                    gc_iterations.inc()
             except LockNotAcquiredException:
                 logger.debug("Could not acquire repo lock for garbage collection")
 

workers/namespacegcworker.py

@@ -15,7 +15,7 @@
 
 
 POLL_PERIOD_SECONDS = 60
-NAMESPACE_GC_TIMEOUT = 24 * 60 * 60  # 24h
+NAMESPACE_GC_TIMEOUT = 3 * 60 * 60  # 3h
 LOCK_TIMEOUT_PADDING = 60  # 60 seconds
 
 

workers/queueworker.py

@@ -1,5 +1,6 @@
 import logging
 import json
+import time
 
 from threading import Event, Lock
 
@@ -10,6 +11,8 @@
 
 logger = logging.getLogger(__name__)
 
+QUEUE_WORKER_SLEEP_DURATION = 1
+
 
 class JobException(Exception):
     """
@@ -142,7 +145,7 @@ def poll_queue(self):
             except WorkerSleepException as exc:
                 logger.debug("Worker has been requested to go to sleep")
                 self.mark_current_incomplete(restore_retry=True)
-                self._stop.set()
+                time.sleep(QUEUE_WORKER_SLEEP_DURATION)
 
             except WorkerUnhealthyException as exc:
                 logger.error(

workers/repositorygcworker.py

@@ -15,7 +15,7 @@
 
 
 POLL_PERIOD_SECONDS = 60
-REPOSITORY_GC_TIMEOUT = 24 * 60 * 60  # 24h
+REPOSITORY_GC_TIMEOUT = 3 * 60 * 60  # 3h
 LOCK_TIMEOUT_PADDING = 60  # 60 seconds