[core] Modify logs messages to be structured

Some messages were updated to follow the structured format.

Signed-off-by: Santiago Dueñas <[email protected]>

Author: sduenas

src/grimoirelab/core/consumers/consumer.py

@@ -27,6 +27,7 @@
 from multiprocessing import Event as ProcessEvent
 
 import redis
+import structlog
 
 if typing.TYPE_CHECKING:
     from typing import Iterable
@@ -170,7 +171,7 @@ def fetch_new_entries(self) -> Iterable[Entry]:
     def recover_stream_entries(self, recover_idle_time: int = RECOVER_IDLE_TIME) -> Iterable[Entry]:
         """Transfers ownership of pending entries idle for 'recover_idle_time'."""
 
-        self.logger.debug(f"Recovering events from '{self.stream_name}:{self.consumer_group}'")
+        self.logger.debug("Recovering events", stream=self.stream_name, consumer_group=self.consumer_group)
 
         while True:
             response = self.connection.xautoclaim(
@@ -199,6 +200,8 @@ def recover_stream_entries(self, recover_idle_time: int = RECOVER_IDLE_TIME) ->
             if self._stop_event.is_set():
                 break
 
+        self.logger.info("events recovered", stream=self.stream_name, consumer_group=self.consumer_group)
+
     def process_entries(self, entries: Iterable[Entry], recovery: bool = False):
         """Process entries (implement this method in subclasses).
 
@@ -222,12 +225,11 @@ def ack_entries(self, message_ids: list):
     def stop(self):
         """Stop the consumer gracefully."""
 
-        self.logger.info(f"Stopping consumer '{self.consumer_name}'.")
-
         self._stop_event.set()
+        self.logger.info("consumer stopped", consumer=self.consumer_name)
 
     def _create_logger(self):
-        logger = logging.getLogger(name=f"{self.__class__.__name__}")
+        logger = structlog.get_logger(self.__class__.__name__)
         logger.setLevel(self.logging_level)
         return logger
 

src/grimoirelab/core/consumers/consumer_pool.py

@@ -27,13 +27,17 @@
 
 import redis
 from rq.connections import parse_connection
+import structlog
 
 from .consumer import Consumer
 
 
 ConsumerData = namedtuple("ConsumerData", ["name", "pid", "process"])
 
 
+logger = structlog.get_logger(__name__)
+
+
 class ConsumerPool:
     """Base class to create a pool of consumers.
 
@@ -198,7 +202,7 @@ def cleanup_consumers(self):
                 self._consumers.pop(name)
 
     def _create_logger(self):
-        logger = logging.getLogger(self.__class__.__name__)
+        logger = structlog.get_logger(self.__class__.__name__)
         logger.setLevel(self.log_level)
         return logger
 
@@ -266,4 +270,4 @@ def _run_consumer(
         consumer = consumer_class(connection=connection, *args, **kwargs)
         consumer.start(burst=burst)
     except Exception as e:
-        logging.error(f"Consumer {consumer_class.__name__} failed: {e}")
+        logger.error(f"Consumer {consumer_class.__name__} failed: {e}")

src/grimoirelab/core/scheduler/jobs.py

@@ -22,14 +22,15 @@
 import typing
 
 import rq.job
+import structlog
 
 if typing.TYPE_CHECKING:
     from typing import Any
     from logging import LogRecord
     from rq.types import FunctionReferenceType
 
 
-logger = logging.getLogger(__name__)
+logger = structlog.get_logger(__name__)
 
 
 class GrimoireLabJob(rq.job.Job):
@@ -129,9 +130,13 @@ def _execute(self) -> Any:
         try:
             self._add_log_handler()
             return super()._execute()
-        except Exception:
-            logger.exception(f"Error running job {self.id}.")
-            raise
+        except Exception as ex:
+            logger.error(
+                "job exception",
+                job_id=self.id,
+                exc_info=ex
+            )
+            raise ex
         finally:
             self._remove_log_handler()
 

src/grimoirelab/core/scheduler/scheduler.py

@@ -19,14 +19,14 @@
 from __future__ import annotations
 
 import datetime
-import logging
 import typing
 import uuid
 
 import django.db
 import django_rq
 import rq.exceptions
 import rq.job
+import structlog
 
 from django.conf import settings
 from rq.registry import StartedJobRegistry
@@ -59,7 +59,8 @@
     rq.job.JobStatus.CANCELED,
 ]
 
-logger = logging.getLogger(__name__)
+
+logger = structlog.get_logger(__name__)
 
 
 def schedule_task(
@@ -88,6 +89,15 @@ def schedule_task(
     )
     _enqueue_task(task, scheduled_at=datetime_utcnow())
 
+    logger.info(
+        "task scheduled",
+        task_uuid=task.uuid,
+        task_args=task_args,
+        job_interval=job_interval,
+        job_max_retries=job_max_retries,
+        burst=burst
+    )
+
     return task
 
 
@@ -122,6 +132,10 @@ def cancel_task(task_uuid: str) -> None:
         task.status = SchedulerStatus.CANCELED
         task.save()
 
+    logger.info(
+        "task canceled", task_uuid=task.uuid
+    )
+
 
 def reschedule_task(task_uuid: str) -> None:
     """Reschedule a task
@@ -147,7 +161,6 @@ def reschedule_task(task_uuid: str) -> None:
         except (rq.exceptions.NoSuchJobError, rq.exceptions.InvalidJobOperation):
             pass
         _schedule_job(task, job, datetime_utcnow(), job.job_args)
-
     elif task.status == SchedulerStatus.RUNNING:
         # Make sure it is running
         job = task.jobs.order_by("-scheduled_at").first()
@@ -157,6 +170,10 @@ def reschedule_task(task_uuid: str) -> None:
     else:
         _enqueue_task(task)
 
+    logger.info(
+        "task rescheduled", task_uuid=task.uuid
+    )
+
 
 def maintain_tasks() -> None:
     """Maintain the tasks that are scheduled to be executed.
@@ -178,14 +195,21 @@ def maintain_tasks() -> None:
         if not _is_job_removed_or_stopped(job_db, task.default_job_queue):
             continue
 
-        logger.info(f"Job #{job_db.job_id} in queue (task: {task.task_id}) stopped. Rescheduling.")
+        logger.info(
+            "Job stopped but task wasn't updated; rescheduling",
+            task_uuid=task.uuid,
+            job_uuid=job_db.uuid,
+            queue=job_db.queue
+        )
 
         current_time = datetime_utcnow()
         scheduled_at = max(task.scheduled_at, current_time)
 
         job_db.save_run(SchedulerStatus.CANCELED)
         _enqueue_task(task, scheduled_at=scheduled_at)
 
+    logger.debug("Maintenance of tasks completed")
+
 
 def _is_job_removed_or_stopped(job: Job, queue: str) -> bool:
     """
@@ -246,10 +270,6 @@ def _enqueue_task(task: Task, scheduled_at: datetime.datetime | None = None) ->
 
     _schedule_job(task, job, scheduled_at, job_args)
 
-    logger.info(
-        f"Job #{job.job_id} (task: {task.task_id}) enqueued in '{job.queue}' at {scheduled_at}"
-    )
-
     return job
 
 
@@ -277,15 +297,29 @@ def _schedule_job(
         task.status = SchedulerStatus.ENQUEUED
         job.scheduled_at = scheduled_at
         task.scheduled_at = scheduled_at
-    except Exception as e:
-        logger.error(f"Error enqueuing job of task {task.task_id}. Not scheduled. Error: {e}")
+    except Exception as ex:
         job.status = SchedulerStatus.FAILED
         task.status = SchedulerStatus.FAILED
-        raise e
+        logger.error(
+            "job scheduling",
+            job_uuid=job.uuid,
+            task_uuid=task.uuid,
+            exc_info=ex
+        )
+        raise ex
     finally:
         job.save()
         task.save()
 
+    logger.info(
+        "job scheduled",
+        job_uuid=job.uuid,
+        job_args=job_args,
+        task_uuid=task.uuid,
+        queue=job.queue,
+        scheduled_at=scheduled_at
+    )
+
     return rq_job
 
 
@@ -304,17 +338,25 @@ def _on_success_callback(
     try:
         job_db = find_job(job.id)
     except NotFoundError:
-        logger.error("Job not found. Not rescheduling.")
+        logger.error("job not found", job_uuid=job.uuid)
         return
 
-    job_db.save_run(SchedulerStatus.COMPLETED, progress=result, logs=job.meta.get("log", None))
+    job_db.save_run(
+        SchedulerStatus.COMPLETED,
+        progress=result,
+        logs=job.meta.get('log', None)
+    )
     task = job_db.task
 
-    logger.info(f"Job #{job_db.job_id} (task: {task.task_id}) completed.")
+    logger.info(
+        "job completed",
+        job_uuid=job_db.uuid,
+        task_uuid=task.uuid
+    )
 
     # Reschedule task
     if task.burst:
-        logger.info(f"Task: {task.task_id} finished. It was a burst task. It won't be rescheduled.")
+        logger.info("task completed", task_uuid=task.uuid, burst=True)
         return
     else:
         scheduled_at = datetime_utcnow() + datetime.timedelta(seconds=task.job_interval)
@@ -342,27 +384,39 @@ def _on_failure_callback(
     try:
         job_db = find_job(job.id)
     except NotFoundError:
-        logger.error("Job not found. Not rescheduling.")
+        logger.error("job not found", job_uuid=job.uuid)
         return
 
     job_db.save_run(
-        SchedulerStatus.FAILED, progress=job.meta["progress"], logs=job.meta.get("log", None)
+        SchedulerStatus.FAILED,
+        progress=job.meta['progress'],
+        logs=job.meta.get('log', None)
     )
     task = job_db.task
 
-    logger.error(f"Job #{job_db.job_id} (task: {task.task_id}) failed; error: {value}")
+    logger.info(
+        "job failed",
+        job_uuid=job_db.uuid,
+        task_uuid=task.uuid,
+        error=value
+    )
+
+    # Define new log to reuse parameters
+    log = logger.bind(task_uuid=task.uuid, job_uuid=job_db.uuid)
 
-    # Try to retry the task
+    # Retry the task, if possible
     if task.failures >= task.job_max_retries:
-        logger.error(f"Task: {task.task_id} max retries reached; cancelled")
+        log.info(
+            "task failed; max retries reached",
+            max_failures=task.failures
+        )
         return
     elif not task.can_be_retried():
-        logger.error(f"Task: {task.task_id} can't be retried")
+        log.info("task failed; can't be retried")
         return
     else:
-        logger.error(f"Task: {task.task_id} failed but task will be retried")
         task.status = SchedulerStatus.RECOVERY
         task.save()
-
-    scheduled_at = datetime_utcnow() + datetime.timedelta(seconds=task.job_interval)
-    _enqueue_task(task, scheduled_at=scheduled_at)
+        scheduled_at = datetime_utcnow() + datetime.timedelta(seconds=task.job_interval)
+        _enqueue_task(task, scheduled_at=scheduled_at)
+        log.info("task failed; recovered")

src/grimoirelab/core/scheduler/tasks/chronicler.py

@@ -18,11 +18,11 @@
 
 from __future__ import annotations
 
-import logging
 import typing
 
 import cloudevents.conversion
-import rq.job
+import rq
+import structlog
 
 import perceval.backend
 import perceval.backends
@@ -37,7 +37,7 @@
     from datetime import datetime
 
 
-logger = logging.getLogger("chronicler")
+logger = structlog.get_logger('__name__')
 
 
 def chronicler_job(

src/grimoirelab/core/scheduler/worker.py

@@ -18,11 +18,11 @@
 
 from __future__ import annotations
 
-import logging
 import typing
 
 import django.db.transaction
 import rq.worker
+import structlog
 
 from grimoirelab_toolkit.datetime import datetime_utcnow
 
@@ -33,7 +33,7 @@
     from .jobs import GrimoireLabJob
 
 
-logger = logging.getLogger(__name__)
+logger = structlog.get_logger(__name__)
 
 
 class GrimoireLabWorker(rq.worker.Worker):