chore: improve logs on GBFS processing functions (#1213)

Author: davidgamez

functions-python/gbfs_validator/src/gbfs_data_processor.py

@@ -1,6 +1,5 @@
 import json
 import language_tags
-import logging
 from datetime import datetime
 from http import HTTPMethod
 from typing import Dict, Any, Optional, List, Tuple
@@ -28,6 +27,7 @@
 )
 from sqlalchemy.orm import Session
 from shared.database.database import with_db_session
+from shared.helpers.logger import get_logger
 from shared.helpers.utils import create_http_task
 
 FEATURE_ENDPOINTS = [
@@ -51,6 +51,7 @@ def __init__(self, stable_id: str, feed_id: str):
         self.gbfs_versions: List[GBFSVersion] = []
         self.gbfs_endpoints: Dict[str, List[GBFSEndpoint]] = {}
         self.validation_reports: Dict[str, Dict[str, Any]] = {}
+        self.logger = get_logger(GBFSDataProcessor.__name__, stable_id)
 
     def process_gbfs_data(self, autodiscovery_url: str) -> None:
         """Process the GBFS data from the autodiscovery URL."""
@@ -76,8 +77,10 @@ def record_autodiscovery_request(
         self, autodiscovery_url: str, db_session: Session
     ) -> None:
         """Record the request to the autodiscovery URL."""
-        logging.info(f"Accessing auto-discovery URL: {autodiscovery_url}")
-        request_metadata = GBFSEndpoint.get_request_metadata(autodiscovery_url)
+        self.logger.info("Accessing auto-discovery URL: %s", autodiscovery_url)
+        request_metadata = GBFSEndpoint.get_request_metadata(
+            autodiscovery_url, self.logger
+        )
         gbfs_feed = (
             db_session.query(Gbfsfeed).filter(Gbfsfeed.id == self.feed_id).one_or_none()
         )
@@ -96,28 +99,28 @@ def record_autodiscovery_request(
         if request_metadata.get("response_size_bytes") is None:
             raise ValueError(f"Error fetching {autodiscovery_url}")
 
-    @staticmethod
     def extract_gbfs_endpoints(
+        self,
         gbfs_json_url: str,
     ) -> Tuple[Optional[List[GBFSEndpoint]], GBFSVersion]:
         """
         Extract GBFS endpoints from the GBFS JSON URL.
         @:returns: GBFS endpoints and the version of the GBFS feed.
         """
-        logging.info(f"Fetching GBFS data from {gbfs_json_url}")
+        self.logger.info("Fetching GBFS data from %s", gbfs_json_url)
         gbfs_json = fetch_gbfs_data(gbfs_json_url)
         feeds_matches = parse("$..feeds").find(gbfs_json)
         version_match = parse("$..version").find(gbfs_json)
         if not version_match:
-            logging.warning(
+            self.logger.warning(
                 "No version found in the GBFS data. Defaulting to version 1.0."
             )
             gbfs_version = GBFSVersion("1.0", gbfs_json_url)
         else:
             gbfs_version = GBFSVersion(version_match[0].value, gbfs_json_url)
         if not feeds_matches:
-            logging.error(
-                f"No feeds found in the GBFS data for version {gbfs_version.version}."
+            self.logger.error(
+                "No feeds found in the GBFS data for version %s.", gbfs_version.version
             )
             return None, gbfs_version
         endpoints = []
@@ -144,15 +147,15 @@ def extract_gbfs_endpoints(
                 for endpoint in endpoints
             }.values()
         )
-        logging.info(f"Found version {gbfs_version.version}.")
-        logging.info(
-            f"Found endpoints {', '.join([endpoint.name for endpoint in endpoints])}."
+        self.logger.info("Found version %s.", gbfs_version.version)
+        self.logger.info(
+            "Found endpoints %s.", ", ".join([endpoint.name for endpoint in endpoints])
         )
         return unique_endpoints, gbfs_version
 
     def extract_gbfs_versions(self, gbfs_json_url: str) -> Optional[List[GBFSVersion]]:
         """Extract GBFS versions from the autodiscovery URL"""
-        all_endpoints, version = GBFSDataProcessor.extract_gbfs_endpoints(gbfs_json_url)
+        all_endpoints, version = self.extract_gbfs_endpoints(gbfs_json_url)
         if not all_endpoints or not version:
             return None
         self.gbfs_endpoints[version.version] = all_endpoints
@@ -163,17 +166,20 @@ def extract_gbfs_versions(self, gbfs_json_url: str) -> Optional[List[GBFSVersion
         )
 
         if gbfs_versions_endpoint:
-            logging.info(f"Fetching GBFS versions from {gbfs_versions_endpoint.url}")
+            self.logger.info(
+                "Fetching GBFS versions from %s", gbfs_versions_endpoint.url
+            )
             gbfs_versions_json = fetch_gbfs_data(gbfs_versions_endpoint.url)
             versions_matches = parse("$..versions").find(gbfs_versions_json)
             if versions_matches:
                 gbfs_versions = GBFSVersion.from_dict(versions_matches[0].value)
-                logging.info(
-                    f"Found versions {', '.join([version.version for version in gbfs_versions])}"
+                self.logger.info(
+                    "Found versions %s",
+                    ", ".join([version.version for version in gbfs_versions]),
                 )
                 return gbfs_versions
             else:
-                logging.warning(
+                self.logger.warning(
                     "No versions found in the GBFS versions data. Defaulting to the autodiscovery URL version."
                 )
         return [
@@ -192,14 +198,14 @@ def get_latest_version(self) -> Optional[str]:
             default=None,
         )
         if not max_version:
-            logging.error(
+            self.logger.error(
                 "No non-RC versions found. Trying to set the latest to a RC version."
             )
             max_version = max(
                 self.gbfs_versions, key=lambda version: version.version, default=None
             )
         if not max_version:
-            logging.error("No versions found.")
+            self.logger.error("No versions found.")
             return None
         return max_version.version
 
@@ -213,7 +219,7 @@ def update_database_entities(self, db_session: Session) -> None:
             .one_or_none()
         )
         if not gbfs_feed:
-            logging.error(f"GBFS feed with ID {self.feed_id} not found.")
+            self.logger.error("GBFS feed with ID %s not found.", self.feed_id)
             return
         gbfs_versions_orm = []
         latest_version = self.get_latest_version()
@@ -329,7 +335,8 @@ def validate_gbfs_feed_versions(self) -> None:
                 response.raise_for_status()
                 json_report_summary = response.json()
             except RequestException as e:
-                logging.error(f"Validation request failed for {version.url}: {e}")
+                self.logger.error("Validation request failed for %s", version.url)
+                self.logger.error(e)
                 continue
 
             report_summary_blob = bucket.blob(
@@ -362,7 +369,7 @@ def create_validation_report_entities(
             "validatorVersion", None
         )
         if validator_version is None or validation_time is None:
-            logging.error("Validation version or time not found.")
+            self.logger.error("Validation version or time not found.")
             return None
 
         validation_report_id = (
@@ -400,13 +407,13 @@ def extract_endpoints_for_all_versions(self):
             if endpoints:
                 self.gbfs_endpoints[version.version] = endpoints
             else:
-                logging.error(f"No endpoints found for version {version.version}.")
+                self.logger.error("No endpoints found for version %s.", version.version)
 
     def trigger_location_extraction(self):
         """Trigger the location extraction process."""
         latest_version = self.get_latest_version()
         if not latest_version:
-            logging.error("No latest version found.")
+            self.logger.error("No latest version found.")
             return
         endpoints = self.gbfs_endpoints.get(latest_version, [])
         # Find the station_information_url endpoint
@@ -428,7 +435,9 @@ def trigger_location_extraction(self):
             None,
         )
         if not station_information_url and not vehicle_status_url:
-            logging.warning("No station_information_url or vehicle_status_url found.")
+            self.logger.warning(
+                "No station_information_url or vehicle_status_url found."
+            )
             return
         client = tasks_v2.CloudTasksClient()
         body = json.dumps(

functions-python/gbfs_validator/src/gbfs_utils.py

@@ -24,9 +24,12 @@ class GBFSEndpoint:
     language: Optional[str] = None
 
     @staticmethod
-    def get_request_metadata(url: str) -> Optional[Dict[str, Any]]:
+    def get_request_metadata(
+        url: str, logger: Optional[logging.Logger] = None
+    ) -> Optional[Dict[str, Any]]:
         """Fetch the endpoint and return latency, status code, and response size."""
         try:
+            logger = logger or logging.getLogger(__name__)
             response = requests.get(url)
             response.raise_for_status()
             return {
@@ -35,7 +38,7 @@ def get_request_metadata(url: str) -> Optional[Dict[str, Any]]:
                 "response_size_bytes": len(response.content),
             }
         except requests.exceptions.RequestException as error:
-            logging.error(f"Error fetching {url}: {error}")
+            logger.error("Error fetching %s. Error %s", url, error)
             return {
                 "latency": None,
                 "status_code": error.response.status_code if error.response else 400,

functions-python/gbfs_validator/src/main.py

@@ -24,10 +24,10 @@
     MaxExecutionsReachedError,
 )
 from shared.database.database import with_db_session
-from shared.helpers.logger import Logger, StableIdFilter
+from shared.helpers.logger import init_logger, get_logger
 from shared.helpers.parser import jsonify_pubsub
 
-logging.basicConfig(level=logging.INFO)
+init_logger()
 
 
 def fetch_all_gbfs_feeds(db_session: Session) -> List[Gbfsfeed]:
@@ -38,15 +38,14 @@ def fetch_all_gbfs_feeds(db_session: Session) -> List[Gbfsfeed]:
         db_session.expunge_all()
         return gbfs_feeds
     except Exception as e:
-        logging.error(f"Error fetching all GBFS feeds: {e}")
+        logging.error("Error fetching all GBFS feeds: %s", e)
         raise e
 
 
 @functions_framework.cloud_event
 def gbfs_validator_pubsub(cloud_event: CloudEvent):
-    Logger.init_logger()
     data = cloud_event.data
-    logging.info(f"Function triggered with Pub/Sub event data: {data}")
+    logging.info("Function triggered with Pub/Sub event data: %s", data)
 
     # Get the pubsub message data
     message_json = jsonify_pubsub(data)
@@ -59,50 +58,45 @@ def gbfs_validator_pubsub(cloud_event: CloudEvent):
         url = message_json["url"]
         feed_id = message_json["feed_id"]
     except KeyError as e:
-        logging.error(f"Missing required field: {e}")
+        logging.error("Missing required field: %s", e)
         return f"Invalid Pub/Sub message data. Missing {e}."
 
-    # Add stable_id to logs
-    stable_id_filter = StableIdFilter(stable_id)
-    logging.getLogger().addFilter(stable_id_filter)
+    # get logger with stable_id
+    logger = get_logger(__name__, stable_id)
+    # Save trace and validate if the execution is allowed
+    trace_service = DatasetTraceService()
+    trace_id = str(uuid.uuid4())
+    trace = DatasetTrace(
+        trace_id=trace_id,
+        stable_id=stable_id,
+        execution_id=execution_id,
+        status=Status.PROCESSING,
+        timestamp=datetime.now(),
+        pipeline_stage=PipelineStage.GBFS_VALIDATION,
+    )
+
     try:
-        # Save trace and validate if the execution is allowed
-        trace_service = DatasetTraceService()
-        trace_id = str(uuid.uuid4())
-        trace = DatasetTrace(
-            trace_id=trace_id,
-            stable_id=stable_id,
-            execution_id=execution_id,
-            status=Status.PROCESSING,
-            timestamp=datetime.now(),
-            pipeline_stage=PipelineStage.GBFS_VALIDATION,
-        )
+        trace_service.validate_and_save(trace, int(os.getenv("MAXIMUM_EXECUTIONS", 1)))
+    except (ValueError, MaxExecutionsReachedError) as e:
+        error_message = str(e)
+        logger.error(error_message)
+        save_trace_with_error(trace, error_message, trace_service)
+        return error_message
+
+    # Process GBFS data
+    try:
+        processor = GBFSDataProcessor(stable_id, feed_id)
+        processor.process_gbfs_data(url)
+    except Exception as e:
+        error_message = f"Error processing GBFS data: {e}"
+        logger.error(error_message)
+        logger.error("Traceback: %s", traceback.format_exc())
+        save_trace_with_error(trace, error_message, trace_service)
+        return error_message
 
-        try:
-            trace_service.validate_and_save(
-                trace, int(os.getenv("MAXIMUM_EXECUTIONS", 1))
-            )
-        except (ValueError, MaxExecutionsReachedError) as e:
-            error_message = str(e)
-            logging.error(error_message)
-            save_trace_with_error(trace, error_message, trace_service)
-            return error_message
-
-        # Process GBFS data
-        try:
-            processor = GBFSDataProcessor(stable_id, feed_id)
-            processor.process_gbfs_data(url)
-        except Exception as e:
-            error_message = f"Error processing GBFS data: {e}"
-            logging.error(f"{error_message}\nTraceback:\n{traceback.format_exc()}")
-            save_trace_with_error(trace, error_message, trace_service)
-            return error_message
-
-        trace.status = Status.SUCCESS
-        trace_service.save(trace)
-        return "GBFS data processed and stored successfully."
-    finally:
-        logging.getLogger().removeFilter(stable_id_filter)
+    trace.status = Status.SUCCESS
+    trace_service.save(trace)
+    return "GBFS data processed and stored successfully."
 
 
 @with_db_session
@@ -113,7 +107,6 @@ def gbfs_validator_batch(_, db_session: Session):
     @param _: The request object.
     @return: The response of the function.
     """
-    Logger.init_logger()
     logging.info("Batch function triggered.")
     pubsub_topic_name = os.getenv("PUBSUB_TOPIC_NAME", None)
     if pubsub_topic_name is None:
@@ -137,7 +130,7 @@ def gbfs_validator_batch(_, db_session: Session):
             "url": gbfs_feed.auto_discovery_url,
         }
         feeds_data.append(feed_data)
-        logging.info(f"Feed {gbfs_feed.stable_id} added to the batch.")
+        logging.info("Feed %s added to the batch.", gbfs_feed.stable_id)
 
     # Publish to Pub/Sub topic
     try:
@@ -148,9 +141,9 @@ def gbfs_validator_batch(_, db_session: Session):
             message_data = json.dumps(feed_data).encode("utf-8")
             future = publisher.publish(topic_path, message_data)
             future.result()  # Ensure message was published
-            logging.info(f"Published feed {feed_data['stable_id']} to Pub/Sub.")
+            logging.info("Published feed %s to Pub/Sub.", feed_data["stable_id"])
     except Exception as e:
-        logging.error(f"Error publishing feeds to Pub/Sub: {e}")
+        logging.error("Error publishing feeds to Pub/Sub: %s", e)
         return "Error publishing feeds to Pub/Sub.", 500
 
     return (

functions-python/gbfs_validator/tests/test_gbfs_data_processor.py

@@ -49,7 +49,7 @@ def mock_fetch_gbfs_data(url):
         return {}
 
 
-def mock_get_request_metadata(_):
+def mock_get_request_metadata(*args, **kwargs):
     return {
         "latency": 100,
         "status_code": 200,