[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

releases/unreleased/struclog-added-for-logging-platform-messages.yml

@@ -0,0 +1,14 @@
+---
+title: struclog added for logging platform messages
+category: other
+author: Santiago Dueñas <[email protected]>
+issue: null
+notes: >
+  Organizing log data in a structured format makes easier
+  to read and analyze it. After adding 'structlog',
+  log messages will always have a structured format.
+
+  The default mode prints the messages to the console
+  in plain format, but it can also be configured to print
+  them in JSON format by setting the environment variable
+  'GRIMOIRELAB_LOGS_JSON' to 1 or 'true'.

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,9 @@ 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 +202,10 @@ def recover_stream_entries(self, recover_idle_time: int = RECOVER_IDLE_TIME) ->
             if self._stop_event.is_set():
                 break
 
+        self.logger.debug(
+            "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 +229,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
 
@@ -265,5 +269,5 @@ def _run_consumer(
     try:
         consumer = consumer_class(connection=connection, *args, **kwargs)
         consumer.start(burst=burst)
-    except Exception as e:
-        logging.error(f"Consumer {consumer_class.__name__} failed: {e}")
+    except Exception as exc:
+        logger.error(f"Consumer {consumer_class.__name__} failed", err=exc)

src/grimoirelab/core/runner/commands/run.py

@@ -31,6 +31,7 @@
 import django_rq
 import opensearchpy
 import redis
+import structlog
 
 from django.conf import settings
 from django.db import connections, OperationalError
@@ -44,6 +45,9 @@
 DEFAULT_MAX_RETRIES = 10
 
 
+logger = structlog.get_logger(__name__)
+
+
 @click.group()
 @click.pass_context
 def run(ctx: Context):
@@ -122,17 +126,17 @@ def periodic_maintain_tasks(interval):
     while True:
         try:
             maintain_tasks()
-            logging.info("Maintenance tasks executed successfully.")
-        except redis.exceptions.ConnectionError as e:
-            logging.error(f"Redis connection error during maintenance tasks: {e}")
-        except django.db.utils.OperationalError as e:
-            logging.error(f"Database connection error during maintenance tasks: {e}")
+            logger.info("Maintenance tasks executed successfully")
+        except redis.exceptions.ConnectionError as exc:
+            logger.error("Redis connection error during maintenance tasks", err=exc)
+        except django.db.utils.OperationalError as exc:
+            logger.error("Database connection error during maintenance tasks", err=exc)
             connections.close_all()
-        except Exception as e:
-            logging.error("Error during maintenance tasks: %s", e)
+        except Exception as exc:
+            logger.error("Unexpected error during maintenance tasks", err=exc)
             raise
         except KeyboardInterrupt:
-            logging.info("Maintenance task interrupted. Exiting...")
+            logger.info("Maintenance task interrupted. Exiting...")
             return
 
         time.sleep(interval)
@@ -146,7 +150,7 @@ def _maintenance_process(maintenance_interval):
     )
     process.start()
 
-    logging.info("Started maintenance process with PID %s", process.pid)
+    logger.info("Started maintenance process", pid=process.pid)
 
     return process
 
@@ -217,6 +221,10 @@ def _wait_opensearch_ready(
 ) -> None:
     """Wait for OpenSearch to be available before starting"""
 
+    # The 'opensearch' library writes logs with the exceptions while
+    # connecting to the database. Disable them temporarily until
+    # the service is up. We have to use logging library because structlog
+    # doesn't allow to disable a logger dynamically.
     os_logger = logging.getLogger("opensearch")
     os_logger.disabled = True
 
@@ -245,24 +253,26 @@ def _wait_opensearch_ready(
             # Index still not created, but OpenSearch is up
             break
         except opensearchpy.exceptions.AuthorizationException:
-            logging.error("OpenSearch Authorization failed. Check your credentials.")
+            logger.error("OpenSearch Authorization failed. Check your credentials.")
             exit(1)
         except (
             opensearchpy.exceptions.ConnectionError,
             opensearchpy.exceptions.TransportError,
-        ) as e:
-            logging.warning(
-                f"[{attempt + 1}/{DEFAULT_MAX_RETRIES}] OpenSearch connection not ready: {e}"
+        ) as exc:
+            logger.warning(
+                f"[{attempt + 1}/{DEFAULT_MAX_RETRIES}] OpenSearch connection not ready",
+                err=exc,
             )
             _sleep_backoff(attempt)
 
     else:
-        logging.error("Failed to connect to OpenSearch.")
+        logger.error("Failed to connect to OpenSearch.")
         exit(1)
 
+    # Enable back 'opensearch' library logs
     os_logger.disabled = False
 
-    logging.info("OpenSearch is ready.")
+    logger.info("OpenSearch is ready.")
 
 
 def _wait_redis_ready():
@@ -273,16 +283,17 @@ def _wait_redis_ready():
             redis_conn = django_rq.get_connection()
             redis_conn.ping()
             break
-        except redis.exceptions.ConnectionError as e:
-            logging.warning(
-                f"[{attempt + 1}/{DEFAULT_MAX_RETRIES}] Redis connection not ready: {e}"
+        except redis.exceptions.ConnectionError as exc:
+            logger.warning(
+                f"[{attempt + 1}/{DEFAULT_MAX_RETRIES}] Redis connection not ready",
+                err=exc,
             )
             _sleep_backoff(attempt)
     else:
-        logging.error("Failed to connect to Redis server")
+        logger.error("Failed to connect to Redis server")
         exit(1)
 
-    logging.info("Redis is ready")
+    logger.info("Redis is ready")
 
 
 def _wait_database_ready():
@@ -296,17 +307,18 @@ def _wait_database_ready():
                     pass  # Just test the connection
                 break
 
-        except OperationalError as e:
-            logging.warning(
-                f"[{attempt + 1}/{DEFAULT_MAX_RETRIES}] Database connection not ready: {e.__cause__}"
+        except OperationalError as exc:
+            logger.warning(
+                f"[{attempt + 1}/{DEFAULT_MAX_RETRIES}] Database connection not ready",
+                err=exc.__cause__,
             )
             _sleep_backoff(attempt)
     else:
         error_msg = "Failed to connect to the database after all retries"
-        logging.error(error_msg)
+        logger.error(error_msg)
         raise ConnectionError(error_msg)
 
-    logging.info("Database is ready.")
+    logger.info("Database is ready.")
 
     # Close all database connections to avoid timed out connections
     connections.close_all()

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,9 @@ 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()