update-redis-settings

backend/deepchecks_monitoring/bgtasks/tasks_runner.py

@@ -14,9 +14,9 @@
 
 import anyio
 import pendulum as pdl
+import redis.exceptions as redis_exceptions
 import uvloop
 from redis.asyncio import Redis, RedisCluster
-from redis.exceptions import LockNotOwnedError
 from sqlalchemy import select
 
 from deepchecks_monitoring.bgtasks.alert_task import AlertsTask
@@ -77,17 +77,20 @@ async def run(self):
             raise
 
     async def wait_for_task(self, timeout=120):
-        task_entry = await self.redis.bzpopmin(GLOBAL_TASK_QUEUE, timeout=timeout)
-
-        # If timeout is not 0 we might get return value of None
-        if task_entry is None:
-            self.logger.info('Got from redis queue task_id none')
+        try:
+            task_entry = await self.redis.bzpopmin(GLOBAL_TASK_QUEUE, timeout=timeout)
+        except redis_exceptions.TimeoutError:
+            self.logger.info('Got timeout from redis queue polling task_id ')
             return
         else:
-            # Return value from redis is (redis key, value, score)
-            task_id = int(task_entry[1].decode())
-            queued_timestamp: int = task_entry[2]
-            return task_id, queued_timestamp
+            if task_entry is None:
+                self.logger.info('Got from redis queue task_id none')
+                return
+            else:
+                # Return value from redis is (redis key, value, score)
+                task_id = int(task_entry[1].decode())
+                queued_timestamp: int = task_entry[2]
+                return task_id, queued_timestamp
 
     async def run_single_task(self, task_id, session, queued_timestamp):
         """Run single task."""
@@ -111,7 +114,7 @@ async def run_single_task(self, task_id, session, queued_timestamp):
 
         try:
             await lock.release()
-        except LockNotOwnedError:
+        except redis_exceptions.LockNotOwnedError:
             self.logger.error(f'Failed to release lock for task id: {task_id}. probably task run for longer than '
                               f'maximum time for the lock')
 
@@ -179,7 +182,7 @@ async def main():
         from deepchecks_monitoring.bgtasks import tasks_runner  # pylint: disable=import-outside-toplevel
 
         async with ResourcesProvider(settings) as rp:
-            async_redis = await init_async_redis(rp.redis_settings)
+            async_redis = await init_async_redis()
 
             workers = [
                 ModelVersionCacheInvalidation(),