Merge branch 'deployment-est-gpu-Bimsara' of https://github.com/rootcodelabs/Global-Classifier into deployment-est-gpu

Author: erangi-ar

DSL/CronManager/DSL/callback_formatter.yml

@@ -2,4 +2,4 @@ callback_format:
   trigger: off
   type: exec
   command: "../app/scripts/callback_format.sh"
-  allowedEnvs: ['filePath', 'results', 'taskId']
+  allowedEnvs: ['filePath', 'results', 'taskId', 'metricsFile']

DSL/CronManager/script/callback_format.sh

@@ -3,8 +3,8 @@
 echo "Started Shell Script for Dataset Generation Callback Processing"
 
 # Check if environment variables are set
-if [ -z "$filePath" ] || [ -z "$results" ] || [ -z "$taskId" ]; then
-  echo "Please set the filePath, results, and taskId environment variables."
+if [ -z "$filePath" ] || [ -z "$results" ] || [ -z "$taskId" ] || [ -z "$metricsFile" ]; then
+  echo "Please set the filePath, results, taskId, and metricsFile environment variables."
   exit 1
 fi
 
@@ -62,6 +62,7 @@ python3 "$CALLBACK_SCRIPT" \
   --encoded-results "$results" \
   --output-json "$temp_response" \
   --session-id "$taskId" \
+  --metrics-file "$metricsFile" \
   > /tmp/callback_stdout.log 2> /tmp/callback_stderr.log
 exit_code=$?
 

DSL/Ruuter.public/global-classifier/POST/data/callback.yml

@@ -23,6 +23,9 @@ declare:
       - field: results
         type: array
         description: "List of agency IDs for which the dataset generation was completed"
+      - field: metrics_file
+        type: string
+        description: "Path to the metrics file for the dataset generation"
 
 log_callback_received:
   log: "Dataset generation callback received - Task ID: ${incoming.body.task_id}, Status: ${incoming.body.status}, File Path: ${incoming.body.filePath}, Result: ${incoming.body.results}"
@@ -35,10 +38,11 @@ extract_callback_data:
     message: ${incoming.body.message}
     file_path: ${incoming.body.filePath}
     results: ${encodeURIComponent(JSON.stringify(incoming.body.results))}
+    metrics_file: ${incoming.body.metrics_file}
   next: log_detailed_info
 
 log_detailed_info:
-  log: "Callback Details - Task: ${task_id}, Status: ${status}, Message: ${message}, filePath: ${file_path}, results: ${results}"
+  log: "Callback Details - Task: ${task_id}, Status: ${status}, Message: ${message}, filePath: ${file_path}, results: ${results}, metrics_file: ${metrics_file}"
   next: check_for_request_data
 
 check_for_request_data:
@@ -76,6 +80,7 @@ execute_cron_manager:
       taskId: ${task_id}
       filePath: ${file_path}
       results: ${results}
+      metricsFile: ${metrics_file}
   result: cron_res
   next: assign_success_cron_response
 

src/s3_dataset_processor/dataset_generation_callback_processor.py

@@ -101,6 +101,23 @@ def upload_csv_to_s3(local_csv_path: str, dataset_id: int) -> None:
         raise RuntimeError(f"Failed to upload CSV to S3: {response.text}")
 
 
+def upload_metrics_to_s3(metrics_file_path: str, dataset_id: int) -> None:
+    """Upload the metrics file to S3 using S3Ferry."""
+    destination_file_path = f"/datasets/{dataset_id}/metrics.json"
+    source_file_path = metrics_file_path.replace("/app/", "")
+    logger.info(f"Uploading {metrics_file_path} to S3 as {destination_file_path}")
+    response = s3_ferry_service.transfer_file(
+        destination_file_path=destination_file_path,
+        destination_storage_type="S3",
+        source_file_path=source_file_path,
+        source_storage_type="FS",
+    )
+    logger.info(f"Metrics S3 upload status: {response.status_code}")
+    logger.info(f"Metrics S3 upload response: {response.text}")
+    if response.status_code not in [200, 201]:
+        raise RuntimeError(f"Failed to upload metrics to S3: {response.text}")
+
+
 def notify_progress_uploading_to_s3(session_id: int) -> None:
     """Notify progress update: New Dataset Uploading to S3."""
     payload = {
@@ -239,7 +256,7 @@ def _log_cleanup_results(cleanup_summary: list) -> None:
 
 
 def process_callback_background(
-    file_path: str, encoded_results: str, session_id: int
+    file_path: str, encoded_results: str, session_id: int, metrics_file: str
 ) -> None:
     """Process the dataset generation callback: upload CSV to S3 and send status update."""
     try:
@@ -302,6 +319,13 @@ def process_callback_background(
 
         notify_dataset_update(output_csv_path)
         upload_csv_to_s3(output_csv_path, dataset_id)
+        
+        try:
+            upload_metrics_to_s3(metrics_file, dataset_id)
+            logger.info(f"Metrics file uploaded successfully for dataset {dataset_id}")
+        except Exception as e:
+            logger.warning(f"S3 upload failure for metrics file: {e}")
+            
         send_status_update(dataset_id, encoded_results)
 
         logger.info("Processing completed successfully")
@@ -330,6 +354,9 @@ def parse_args():
     parser.add_argument(
         "--session-id", required=True, help="Session ID for the callback"
     )
+    parser.add_argument(
+        "--metrics-file", required=True, help="Metrics file path for the callback"
+    )
     return parser.parse_args()
 
 
@@ -339,10 +366,11 @@ def main():
     try:
         logger.info("Starting callback processing...")
         logger.info(f"File path: {args.file_path}")
+        logger.info(f"Metrics file: {args.metrics_file}")
         logger.info(f"Encoded results length: {len(args.encoded_results)} characters")
 
         process_callback_background(
-            args.file_path, args.encoded_results, args.session_id
+            args.file_path, args.encoded_results, args.session_id, args.metrics_file
         )
 
         response = {