Merge pull request #9 from nasa/GESDISCUMU-5133-Refactor-and-Clean-Up-Logging-Across-Codebase

refactored all logging across codebase

Author: dhsing23

src/gapConfig/gapConfig.py

@@ -84,7 +84,8 @@ def get_cmr_time(collection_id: str) -> Tuple[str, str]:
         url = f"https://cmr.earthdata.nasa.gov/search/collections.umm_json_v1_4?short_name={short_name}&version={version}"
     else:
         url = f"https://cmr.{cmr_env}.earthdata.nasa.gov/search/collections.umm_json_v1_4?short_name={short_name}&version={version}"
-    logger.info(f"Requesting granule time from: {url}")
+    
+    logger.debug(f"Requesting granule time from: {url}")
     res = requests.get(url)
     data = res.json()
     if not data["items"]:
@@ -116,7 +117,7 @@ def init_collection(collection_name, collection_version, conn) -> str:
     collection_id = f"{collection_name}___{collection_version}"
     try:
         start, end = get_cmr_time(collection_id)
-        logger.info(f"Initializing {collection_id} with {start, end}")
+        logger.debug(f"Retrieved temporal extent for {collection_id}: {start} to {end}")
 
         # For new collection, partition `gaps` and `reasons` tables  and insert into `collections` table
         with conn.cursor() as cur:
@@ -148,6 +149,7 @@ def init_collection(collection_name, collection_version, conn) -> str:
                 logger.info(
                     f"Created gaps partition {partition_name} for collection {collection_id}"
                 )
+            
             # Create partition on `reasons` table
             reasons_partition_name = f"reasons_{safe_collection_id}"
             cur.execute(
@@ -189,11 +191,13 @@ def init_collection(collection_name, collection_version, conn) -> str:
             )
 
             conn.commit()
+        logger.info(f"Successfully initialized collection {collection_id}")
         return f"Initialized collection {collection_id} in table"
 
     except Exception as e:
         conn.rollback()
-        logger.warning(traceback.format_exc())
+        logger.error(f"Collection {collection_id} initialization failed: {str(e)}")
+        logger.debug(traceback.format_exc())
         return f"Collection {collection_id} initialization failed: {str(e)}"
 
 
@@ -236,8 +240,10 @@ def init_migration_stream(collection_name, collection_version):
     )
     payload_response = json.loads(response["Payload"].read().decode())
     if response["StatusCode"] != 200 or payload_response.get("statusCode") != 200:
+        logger.error(f"Migration stream invocation failed for {collection_name} v{collection_version}")
         raise Exception(f"Collection backfill failed: {payload_response.get('body')}")
 
+    logger.info(f"Migration stream completed for {collection_name} v{collection_version}")
     return {
         "status": "success",
         "statusCode": response["StatusCode"],
@@ -260,11 +266,11 @@ def save_tolerance_to_dynamodb(shortname: str, versionid: str, tolerance: int):
                 "granulegap": tolerance,
             }
         )
-        logger.info(
+        logger.debug(
             f"Saved tolerance for {shortname}___{versionid}: {tolerance} seconds. PutItem Response: {response['ResponseMetadata']['HTTPStatusCode']}"
         )
     except Exception as e:
-        logger.error(f"Failed to save tolerance to DynamoDB: {str(e)}")
+        logger.error(f"Failed to save tolerance to DynamoDB for {shortname}___{versionid}: {str(e)}")
         raise
 
 
@@ -291,13 +297,13 @@ def lambda_handler(event: events.SQSEvent, context: Context) -> Dict[str, Any]:
     try:
         http_method = event.get("httpMethod", "")
         resource_path = event.get("path", "")
-        logger.info(f"Got HTTP {http_method} for {resource_path}")
+        logger.debug(f"Got HTTP {http_method} for {resource_path}")
 
         try:
             collections, backfill_behavior = parse_event(event)
         except Exception as e:
-            message = f"Error processing request: {str(e)}"
-            logger.error(traceback.format_exc())
+            message = f"Invalid request format: {str(e)}"
+            logger.warning(message)
             return build_response(400, {"message": message})
 
         if http_method != "POST":
@@ -308,6 +314,7 @@ def lambda_handler(event: events.SQSEvent, context: Context) -> Dict[str, Any]:
             for collection in collections:
                 collection_id = f"{collection['name']}___{collection['version']}"
                 tolerance = collection.get("tolerance")
+                
                 # Update tolerance table even if the collection already exists
                 if tolerance is not None:
                     try:
@@ -316,45 +323,42 @@ def lambda_handler(event: events.SQSEvent, context: Context) -> Dict[str, Any]:
                             collection["raw_version"],
                             int(tolerance),
                         )
+                        logger.info(f"Updated tolerance for {collection['name']} v{collection['raw_version']}: {tolerance}s")
                     except Exception as e:
                         logger.error(
                             f"Error saving tolerance for {collection['name']}___{collection['raw_version']}: {str(e)}"
                         )
+                        
                 # Add collection to collections table, create partition for gaps table, set initial full gap
                 if collection_id not in current_collections:
-                    message = init_collection(
-                        collection["name"], collection["version"], conn
-                    )
-                    logger.info(message)
+                    init_collection(collection["name"], collection["version"], conn)
 
                     # Kick off the migration stream
                     try:
-                        logger.info(f"Starting collection backfill")
                         migration_result = init_migration_stream(
                             collection["name"], collection["version"].replace("_", ".")
                         )
-                        logger.info(f"Backfill result: {migration_result}")
+                        logger.debug(f"Backfill result: {migration_result}")
                     except Exception as e:
                         message = (
                             f"Collection backfill failed for {collection_id}: {str(e)}"
                         )
                         logger.error(message)
-                        logger.warn(
+                        logger.warning(
                             f"Collection {collection_id} left in incomplete state, use force=True to rectify"
                         )
                         return build_response(500, {"message": message})
+                        
                 # Skip DB init but still backfill granules from CMR
                 elif backfill_behavior.lower() == "force":
                     logger.info(
                         f"Force flag detected, proceeding with backfill for existing collection: {collection_id}"
                     )
-                    # Kick off the migration stream
                     try:
-                        logger.info(f"Starting collection backfill")
                         migration_result = init_migration_stream(
                             collection["name"], collection["version"].replace("_", ".")
                         )
-                        logger.info(f"Backfill result: {migration_result}")
+                        logger.debug(f"Backfill result: {migration_result}")
                     except Exception as e:
                         message = (
                             f"Collection backfill failed for {collection_id}: {str(e)}"
@@ -366,11 +370,12 @@ def lambda_handler(event: events.SQSEvent, context: Context) -> Dict[str, Any]:
                         f"Skipping initialization of {collection_id}: already exists in collection table"
                     )
 
+        logger.info(f"Collection initialization completed for {len(collections)} collection(s)")
         return build_response(
             200, {"message": f"Collection initialization complete for {collections}"}
         )
 
     except Exception as e:
-        logger.error(f"Error processing request: {str(e)}")
-        logger.error(traceback.format_exc())
+        logger.error(f"Unexpected error in lambda handler: {str(e)}")
+        logger.debug(traceback.format_exc())
         return build_response(500, {"message": "Unexpected error occurred"})

src/gapMigrationStreamMessageCompiler/gapMigrationStreamMessageCompiler.py

@@ -19,7 +19,6 @@
     formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
     handler.setFormatter(formatter)
     logger.addHandler(handler)
-logger.info("Logger initialized successfully.")
 
 loop = asyncio.get_event_loop()
 
@@ -106,11 +105,12 @@ def get_params(short_name, version, max_producers=8, consumer_ratio=1.5):
         queue_size = n_producers * 2 * 2000
         date_ranges = split_date_ranges(beginning_date, ending_date, n_producers)
 
+        logger.info(f"Collection {short_name} v{version}: {num_granules} granules, {n_producers} producers, {n_consumers} consumers")
         return date_ranges, n_consumers, queue_size, num_granules
 
     except Exception as e:
         logger.error(f"Error occurred: {e}")
-        logger.error(traceback.format_exc())
+        logger.debug(traceback.format_exc())
         return None, {
             "statusCode": 400,
             "body": json.dumps(
@@ -141,6 +141,7 @@ async def fetch_cmr_range(session, url, params, result_queue, fetch_stats):
     """
     search_after = None
     max_retries = 3
+    
     while True:
         headers = {"CMR-Search-After": search_after} if search_after else {}
 
@@ -152,7 +153,7 @@ async def fetch_cmr_range(session, url, params, result_queue, fetch_stats):
                     if response.status != 200:
                         error_body = await response.text()
                         if retry < max_retries:
-                            logger.warning(
+                            logger.debug(
                                 f"CMR API error: HTTP {response.status} on {params}: {error_body} "
                                 f"Retrying in {retry ** 2}s ({retry+1}/{max_retries})"
                             )
@@ -178,14 +179,12 @@ async def fetch_cmr_range(session, url, params, result_queue, fetch_stats):
                     fetch_stats["total"] += len(granules)
                     if not search_after:
                         return
-                    if fetch_stats["total"] % 10000 == 0:
-                        logger.info(f"{fetch_stats["total"]} granules fetched")
                     break
 
             except Exception as e:
                 if retry < max_retries:
                     retry_delay = retry**2
-                    logger.warning(
+                    logger.debug(
                         f"Error fetching CMR page for {params}: {str(e)}. "
                         f"Retrying in {retry ** 2}s ({retry+1}/{max_retries})"
                     )
@@ -279,7 +278,7 @@ async def process_collection(
 
     fetch_stats = {"total": 0}
     send_stats = {"total": 0}
-    logger.info(f"Processing {short_name} v{version} with {len(partitions)} producers")
+    logger.info(f"Starting collection processing: {short_name} v{version} ({len(partitions)} producers, {n_consumers} consumers)")
 
     async with aiohttp.ClientSession() as http_session:
         async with aioboto3.Session().client(
@@ -322,16 +321,15 @@ async def process_collection(
 
                     # Wait for producers to finish
                     await asyncio.gather(*producers)
-                    logger.info("All producers completed")
+                    logger.debug("All producers completed")
 
                     # Signal consumers to complete
                     for _ in range(n_consumers):
                         await result_queue.put(None)
 
                     # Wait for consumers
                     await asyncio.gather(*consumers)
-                    logger.info("All tasks complete")
-
+                    logger.debug("All consumers completed")
 
             except Exception as e:
                 logger.error(f"Failed to process collection {short_name} v{version}: {e}")
@@ -342,7 +340,8 @@ async def process_collection(
                     send_stats["total"] / total_duration if total_duration > 0 else 0
                 )
                 logger.info(
-                    f"{fetch_stats['total']} fetched, {send_stats['total']} sent in {total_duration:.1f}s ({throughput:.1f}/s)"
+                    f"Collection processing complete: {short_name} v{version} - "
+                    f"{fetch_stats['total']} fetched, {send_stats['total']} sent in {total_duration:.1f}s ({throughput:.1f} msg/s)"
                 )
 
 
@@ -365,7 +364,7 @@ def lambda_handler(event, context):
         short_name = sns_message.get("short_name")
         version = sns_message.get("version")
         if not short_name or not version:
-            logger.error("Missing short_name or version in the event")
+            logger.warning("Missing short_name or version in the event")
             return {
                 "statusCode": 400,
                 "body": json.dumps(
@@ -375,7 +374,7 @@ def lambda_handler(event, context):
 
     except Exception as e:
         logger.error(f"Input Error: {e}")
-        logger.error(traceback.format_exc())
+        logger.debug(traceback.format_exc())
         return None, {
             "statusCode": 400,
             "body": json.dumps(
@@ -407,11 +406,12 @@ def lambda_handler(event, context):
                 total_granules,
             )
         )
+        logger.info(f"Lambda execution completed successfully for {short_name} v{version}")
         return {
             "statusCode": 200,
             "body": json.dumps({"message": "Processing complete"}),
         }
     except Exception as e:
-        logger.error(f"Error occurred: {e}")
-        logger.error(traceback.format_exc())
-        return {"statusCode": 500, "body": json.dumps({"error": str(e)})}
+        logger.error(f"Lambda execution failed for {short_name} v{version}: {str(e)}")
+        logger.debug(traceback.format_exc())
+        return {"statusCode": 500, "body": json.dumps({"error": str(e)})}

src/gapReporter/gapReporter.py

@@ -28,43 +28,45 @@ def parse_collection_id(collection_id):
     shortname, versionid = collection_id.rsplit('___', 1)
     return shortname, versionid.replace('_', '.')  # Reverse sanitize_versionid
 
-
 def lambda_handler(event, context):
     """
     AWS Lambda handler that processes all collections from the DB.
     For each collection:
-      - gets granule gap from DynamoDB
-      - fetches time gaps exceeding granule gap
-      - creates and uploads a CSV to S3 if gaps exist
-
+    - gets granule gap from DynamoDB
+    - fetches time gaps exceeding granule gap
+    - creates and uploads a CSV to S3 if gaps exist
     Returns summary of uploads.
     """
     validate_environment_variables(['GAP_REPORT_BUCKET'])
-
     results = []
+    
     with get_db_connection() as conn:
         try:
             collections = check_collections(conn)
-            logger.info(f"Found {len(collections)} collections in collections table.")
+            logger.info(f"Processing gap reports for {len(collections)} collections")
         except Exception as e:
             logger.error(f"Failed to fetch collections: {e}")
             return {'statusCode': 500, 'body': 'Failed to fetch collections'}
 
+        upload_count = 0
+        skip_count = 0
+        error_count = 0
+
         for collection_id in collections:
             try:
                 shortname, versionid = parse_collection_id(collection_id)
-                logger.info(f"Processing collection: {shortname} version {versionid}")
-
+                logger.debug(f"Processing collection: {shortname} version {versionid}")
+                
                 granule_gap = get_granule_gap(shortname, versionid)
-                logger.info(f"Granule gap: {granule_gap} seconds")
-
+                logger.debug(f"Granule gap threshold for {shortname} version {versionid}: {granule_gap}s")
+                
                 with conn.cursor() as cursor:
                     time_gaps = fetch_time_gaps(shortname, versionid, granule_gap, cursor)
-                    logger.info(f"Found {len(time_gaps)} time gaps exceeding threshold.")
-
+                
                 if not time_gaps:
-                    logger.info(f"No qualifying time gaps for {collection_id}. Skipping upload.")
+                    logger.debug(f"No qualifying time gaps for {collection_id}. Skipping upload.")
                     results.append({'collection_id': collection_id, 'status': 'no gaps'})
+                    skip_count += 1
                     continue
 
                 # Create CSV
@@ -73,24 +75,38 @@ def lambda_handler(event, context):
                     csvwriter = csv.writer(csvfile)
                     csvwriter.writerow(['gap_begin', 'gap_end'])
                     csvwriter.writerows(time_gaps)
-                logger.info(f"Created CSV file {output_csv}")
 
                 # Upload to S3
                 s3 = boto3.client('s3')
                 bucket_name = os.environ['GAP_REPORT_BUCKET']
                 s3_output_key = os.path.basename(output_csv)
-
-                s3.upload_file(output_csv, bucket_name, s3_output_key)
-                logger.info(f"Uploaded CSV to s3://{bucket_name}/{s3_output_key}")
-
-                os.remove(output_csv)
-                logger.info(f"Deleted temporary CSV file: {output_csv}")
-
-                results.append({'collection_id': collection_id, 'status': 'uploaded', 's3_key': s3_output_key})
-
+                
+                try:
+                    s3.upload_file(output_csv, bucket_name, s3_output_key)
+                    logger.info(f"Gap report uploaded: {collection_id} ({len(time_gaps)} gaps) -> s3://{bucket_name}/{s3_output_key}")
+                    upload_count += 1
+                    results.append({'collection_id': collection_id, 'status': 'uploaded', 's3_key': s3_output_key})
+                except ClientError as e:
+                    logger.error(f"S3 upload failed for {collection_id}: {str(e)}")
+                    results.append({'collection_id': collection_id, 'status': 'upload_failed', 'error': str(e)})
+                    error_count += 1
+                finally:
+                    # Clean up temp file
+                    if os.path.exists(output_csv):
+                        os.remove(output_csv)
+                        logger.debug(f"Cleaned up temp file: {output_csv}")
+
+            except ValueError as e:
+                logger.warning(f"Invalid collection ID format: {collection_id}")
+                results.append({'collection_id': collection_id, 'status': 'invalid_format', 'error': str(e)})
+                error_count += 1
             except Exception as e:
-                logger.error(f"Error processing collection {collection_id}: {e}")
+                logger.error(f"Failed to process collection {collection_id}: {str(e)}")
                 results.append({'collection_id': collection_id, 'status': 'error', 'error': str(e)})
+                error_count += 1
+
+        # Summary logging
+        logger.info(f"Gap report generation complete: {upload_count} uploaded, {skip_count} skipped, {error_count} errors")
 
     return {
         'statusCode': 200,