feat: Complete Pattern 3 Lambda metering implementation

- Add Lambda metering to all 5 Pattern 3 core processing functions
- OCR: Both normal and skip execution paths with metering
- Classification: Both normal and skip execution paths (SageMaker UDOP backend)
- Extraction: Both normal and skip execution paths with metering
- Assessment: Both normal and skip execution paths with metering
- Summarization: Successful execution path with metering

Pattern 3 now has complete Lambda execution cost tracking matching
Pattern 2 implementation. All functions track invocation counts and
GB-seconds duration for comprehensive cost visibility.

Author: Bob Strahan

patterns/pattern-3/src/assessment_function/index.py

@@ -9,6 +9,7 @@
 from idp_common import get_config, assessment
 from idp_common.models import Document, Status
 from idp_common.docs_service import create_document_service
+from idp_common.utils import calculate_lambda_metering, merge_metering_data
 
 # Configuration will be loaded in handler function
 
@@ -22,6 +23,7 @@ def handler(event, context):
     This function assesses the confidence of extraction results for a document section
     using the Assessment service from the idp_common library.
     """
+    start_time = time.time()  # Capture start time for Lambda metering
     logger.info(f"Starting assessment processing for event: {json.dumps(event, default=str)}")
 
     # Load configuration
@@ -94,6 +96,13 @@ def handler(event, context):
                 # Add only the section being processed (preserve existing data)
                 section_document.sections = [section]
 
+                # Add Lambda metering for assessment skip execution
+                try:
+                    lambda_metering = calculate_lambda_metering("Assessment", context, start_time)
+                    section_document.metering = merge_metering_data(section_document.metering, lambda_metering)
+                except Exception as e:
+                    logger.warning(f"Failed to add Lambda metering for assessment skip: {str(e)}")
+                
                 # Return consistent format for Map state collation
                 response = {
                     "section_id": section_id, 
@@ -138,6 +147,13 @@ def handler(event, context):
         logger.error(error_message)
         raise Exception(error_message)
 
+    # Add Lambda metering for successful assessment execution
+    try:
+        lambda_metering = calculate_lambda_metering("Assessment", context, start_time)
+        updated_document.metering = merge_metering_data(updated_document.metering, lambda_metering)
+    except Exception as e:
+        logger.warning(f"Failed to add Lambda metering for assessment: {str(e)}")
+    
     # Prepare output with automatic compression if needed
     result = {
         'document': updated_document.serialize_document(working_bucket, f"assessment_{section_id}", logger),

patterns/pattern-3/src/classification_function/index.py

@@ -14,6 +14,7 @@
 from idp_common import classification, metrics, get_config
 from idp_common.models import Document, Status
 from idp_common.docs_service import create_document_service
+from idp_common.utils import calculate_lambda_metering, merge_metering_data
 
 # Configuration will be loaded in handler function
 region = os.environ['AWS_REGION']
@@ -28,6 +29,7 @@ def handler(event, context):
     """
     Lambda handler for document classification using SageMaker UDOP model.
     """
+    start_time = time.time()  # Capture start time for Lambda metering
     logger.info(f"Event: {json.dumps(event)}")
 
     # Extract document from the OCR result - handle both compressed and uncompressed
@@ -58,6 +60,13 @@ def handler(event, context):
         logger.info(f"Updating document execution ARN for classification skip")
         document_service.update_document(document)
 
+        # Add Lambda metering for classification skip execution
+        try:
+            lambda_metering = calculate_lambda_metering("Classification", context, start_time)
+            document.metering = merge_metering_data(document.metering, lambda_metering)
+        except Exception as e:
+            logger.warning(f"Failed to add Lambda metering for classification skip: {str(e)}")
+        
         # Prepare output with existing document data
         response = {
             "document": document.serialize_document(working_bucket, "classification_skip", logger)
@@ -140,6 +149,13 @@ def handler(event, context):
     t1 = time.time()
     logger.info(f"Time taken for classification: {t1-t0:.2f} seconds")
 
+    # Add Lambda metering for successful classification execution
+    try:
+        lambda_metering = calculate_lambda_metering("Classification", context, start_time)
+        document.metering = merge_metering_data(document.metering, lambda_metering)
+    except Exception as e:
+        logger.warning(f"Failed to add Lambda metering for classification: {str(e)}")
+    
     # Prepare output with automatic compression if needed
     response = {
         "document": document.serialize_document(working_bucket, "classification", logger)

patterns/pattern-3/src/extraction_function/index.py

@@ -10,6 +10,7 @@
 from idp_common import metrics, get_config, extraction
 from idp_common.models import Document, Section, Status
 from idp_common.docs_service import create_document_service
+from idp_common.utils import calculate_lambda_metering, merge_metering_data
 
 # Configuration will be loaded in handler function
 
@@ -24,6 +25,7 @@ def handler(event, context):
     """
     Process a single section of a document for information extraction
     """
+    start_time = time.time()  # Capture start time for Lambda metering
     logger.info(f"Event: {json.dumps(event)}")
 
     # Load configuration
@@ -65,6 +67,13 @@ def handler(event, context):
     if section.extraction_result_uri and section.extraction_result_uri.strip():
         logger.info(f"Skipping extraction for section {section_id} - already has extraction data: {section.extraction_result_uri}")
 
+        # Add Lambda metering for extraction skip execution
+        try:
+            lambda_metering = calculate_lambda_metering("Extraction", context, start_time)
+            full_document.metering = merge_metering_data(full_document.metering, lambda_metering)
+        except Exception as e:
+            logger.warning(f"Failed to add Lambda metering for extraction skip: {str(e)}")
+        
         # Return the section without processing
         response = {
             "section_id": section_id,
@@ -117,6 +126,13 @@ def handler(event, context):
         logger.error(error_message)
         raise Exception(error_message)
 
+    # Add Lambda metering for successful extraction execution
+    try:
+        lambda_metering = calculate_lambda_metering("Extraction", context, start_time)
+        section_document.metering = merge_metering_data(section_document.metering, lambda_metering)
+    except Exception as e:
+        logger.warning(f"Failed to add Lambda metering for extraction: {str(e)}")
+    
     # Prepare output with automatic compression if needed
     response = {
         "section_id": section_id,

patterns/pattern-3/src/ocr_function/index.py

@@ -14,6 +14,7 @@
 from idp_common import get_config, ocr
 from idp_common.models import Document, Status
 from idp_common.docs_service import create_document_service
+from idp_common.utils import calculate_lambda_metering, merge_metering_data
 
 # Configuration will be loaded in handler function
 
@@ -29,7 +30,8 @@
 def handler(event, context): 
     """
     Lambda handler for OCR processing.
-    """       
+    """
+    start_time = time.time()  # Capture start time for Lambda metering
     logger.info(f"Event: {json.dumps(event)}")
 
     # Get document from event - handle both compressed and uncompressed
@@ -61,6 +63,13 @@ def handler(event, context):
             logger.info(f"Updating document execution ARN for OCR skip")
             document_service.update_document(document)
 
+        # Add Lambda metering for OCR skip execution
+        try:
+            lambda_metering = calculate_lambda_metering("OCR", context, start_time)
+            document.metering = merge_metering_data(document.metering, lambda_metering)
+        except Exception as e:
+            logger.warning(f"Failed to add Lambda metering for OCR skip: {str(e)}")
+        
         # Prepare output with existing document data
         working_bucket = os.environ.get('WORKING_BUCKET')
         response = {
@@ -105,6 +114,13 @@ def handler(event, context):
     t1 = time.time()
     logger.info(f"Total OCR processing time: {t1-t0:.2f} seconds")
 
+    # Add Lambda metering for successful OCR execution
+    try:
+        lambda_metering = calculate_lambda_metering("OCR", context, start_time)
+        document.metering = merge_metering_data(document.metering, lambda_metering)
+    except Exception as e:
+        logger.warning(f"Failed to add Lambda metering for OCR: {str(e)}")
+    
     # Prepare output with automatic compression if needed
     working_bucket = os.environ.get('WORKING_BUCKET')
     response = {

patterns/pattern-3/src/summarization_function/index.py

@@ -13,6 +13,7 @@
 from idp_common import get_config, summarization
 from idp_common.models import Document, Status
 from idp_common.docs_service import create_document_service
+from idp_common.utils import calculate_lambda_metering, merge_metering_data
 
 # Configuration will be loaded in handler function
 
@@ -72,6 +73,13 @@ def handler(event, context):
         else:
             logger.warning("Document summarization completed but no summary report URI was set")
 
+        # Add Lambda metering for successful summarization execution
+        try:
+            lambda_metering = calculate_lambda_metering("Summarization", context, start_time)
+            processed_document.metering = merge_metering_data(processed_document.metering, lambda_metering)
+        except Exception as e:
+            logger.warning(f"Failed to add Lambda metering for summarization: {str(e)}")
+        
         # Prepare output with automatic compression if needed
         return {
             'document': processed_document.serialize_document(working_bucket, "summarization", logger),