better logging - debug to info (#306)

Author: JKL98ISR

backend/deepchecks_monitoring/bgtasks/tasks_queuer.py

@@ -84,7 +84,7 @@ def __init__(
             update(Task)
             .where(next_execution_time <= func.statement_timestamp())
             .values({Task.num_pushed: Task.num_pushed + 1})
-            .returning(Task.id)
+            .returning(Task.id, Task.bg_worker_task, Task.num_pushed)
         )
 
     async def run(self):
@@ -120,7 +120,12 @@ async def move_tasks_to_queue(self, session) -> int:
             try:
                 # Push to sorted set. if task id is already in set then do nothing.
                 pushed_count = await self.redis.zadd(GLOBAL_TASK_QUEUE, task_ids, nx=True)
-                return pushed_count
+                for task in tasks:
+                    id = task['id']
+                    worker = task['bg_worker_task']
+                    num_pushed = task['num_pushed']
+                    self.logger.info(f'pushing task {id} for {worker} that was pushed {num_pushed}')
+                return  pushed_count
             except redis_exceptions.ConnectionError:
                 # If redis failed, does not commit the update to the db
                 await session.rollback()

backend/deepchecks_monitoring/bgtasks/tasks_runner.py

@@ -80,11 +80,12 @@ async def run(self):
             raise
 
     async def wait_for_task(self, timeout=120):
+        self.logger.info('Checking for tasks')
         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.debug('Got from redis queue task_id none')
+            self.logger.info('Got from redis queue task_id none')
             return
         else:
             # Return value from redis is (redis key, value, score)
@@ -101,16 +102,16 @@ async def run_single_task(self, task_id, session, queued_timestamp):
         lock = self.redis.lock(lock_name, blocking=False, timeout=60 * 5)
         lock_acquired = await lock.acquire()
         if not lock_acquired:
-            self.logger.debug(f'Failed to acquire lock for task id: {task_id}')
+            self.logger.info(f'Failed to acquire lock for task id: {task_id}')
             return
 
         task = await session.scalar(select(Task).where(Task.id == task_id))
         # Making sure task wasn't deleted for some reason
         if task is not None:
-            self.logger.debug(f'Running task id: {task_id}')
+            self.logger.info(f'Running task id: {task_id}')
             await self._run_task(task, session, queued_timestamp, lock)
         else:
-            self.logger.debug(f'Got already removed task id: {task_id}')
+            self.logger.info(f'Got already removed task id: {task_id}')
 
         try:
             await lock.release()
@@ -130,7 +131,7 @@ async def _run_task(self, task: Task, session, queued_timestamp, lock):
                 delay = start.int_timestamp - queued_timestamp
                 self.logger.info({'duration': duration, 'task': task.bg_worker_task, 'delay': delay})
             else:
-                self.logger.error({'message': f'Unknown task type: {task.bg_worker_task}'})
+                self.logger.info({'message': f'Unknown task type: {task.bg_worker_task}'})
         except Exception:  # pylint: disable=broad-except
             await session.rollback()
             self.logger.exception({'message': 'Exception running task', 'task': task.bg_worker_task})