fix: implement robust pipeline tracking using version ID to eliminate race conditions

Author: Taniya Mathur

scripts/integration_test_deployment.py

@@ -92,53 +92,92 @@ def upload_to_s3(bucket_name):
         sys.exit(1)
 
 
-def monitor_pipeline(pipeline_name, max_wait=7200):
-    """Monitor CodePipeline execution until completion"""
-    print(f"Monitoring pipeline: {pipeline_name}")
-
+def find_pipeline_execution_by_version(pipeline_name, version_id, max_wait=300):
+    """Find pipeline execution that corresponds to specific S3 version ID"""
+    print(f"Finding pipeline execution for version: {version_id}")
+    
+    codepipeline = boto3.client("codepipeline")
+    start_time = time.time()
+    
+    while time.time() - start_time < max_wait:
+        try:
+            response = codepipeline.list_pipeline_executions(
+                pipelineName=pipeline_name, maxResults=10
+            )
+            
+            for execution in response["pipelineExecutionSummaries"]:
+                execution_id = execution["pipelineExecutionId"]
+                
+                # Get execution details to check source version
+                details = codepipeline.get_pipeline_execution(
+                    pipelineName=pipeline_name,
+                    pipelineExecutionId=execution_id
+                )
+                
+                # Check if this execution matches our version ID
+                for artifact in details["pipelineExecution"].get("artifactRevisions", []):
+                    if artifact.get("revisionId") == version_id:
+                        print(f"✅ Found matching execution: {execution_id}")
+                        return execution_id
+                        
+        except Exception as e:
+            print(f"Error finding execution: {e}")
+            
+        time.sleep(10)
+    
+    print(f"❌ Could not find pipeline execution for version {version_id}")
+    return None
+
+
+def monitor_pipeline_execution(pipeline_name, execution_id, max_wait=7200):
+    """Monitor specific pipeline execution until completion"""
+    print(f"Monitoring execution: {execution_id}")
+    
     codepipeline = boto3.client("codepipeline")
     wait_time = 0
     poll_interval = 30
-
-    # Initial wait for pipeline to start
-    print("Waiting for pipeline to start...")
-    time.sleep(30)
-
+    
     while wait_time < max_wait:
         try:
-            # Get latest pipeline execution
-            response = codepipeline.list_pipeline_executions(
-                pipelineName=pipeline_name, maxResults=1
+            response = codepipeline.get_pipeline_execution(
+                pipelineName=pipeline_name,
+                pipelineExecutionId=execution_id
             )
-
-            if not response["pipelineExecutionSummaries"]:
-                print("⏳ No pipeline executions found, waiting...")
-            else:
-                execution = response["pipelineExecutionSummaries"][0]
-                execution_id = execution["pipelineExecutionId"]
-                status = execution["status"]
-
-                print(f"Pipeline execution {execution_id}: {status}")
-
-                if status == "Succeeded":
-                    print("✅ Pipeline completed successfully!")
-                    return True
-                elif status in ["Failed", "Cancelled", "Superseded"]:
-                    print(f"❌ Pipeline failed with status: {status}")
-                    return False
-                elif status == "InProgress":
-                    print(f"⏳ Pipeline still running... ({wait_time}s elapsed)")
-
+            
+            status = response["pipelineExecution"]["status"]
+            print(f"Pipeline execution {execution_id}: {status}")
+            
+            if status == "Succeeded":
+                print("✅ Pipeline completed successfully!")
+                return True
+            elif status in ["Failed", "Cancelled", "Superseded"]:
+                print(f"❌ Pipeline failed with status: {status}")
+                return False
+            elif status == "InProgress":
+                print(f"⏳ Pipeline still running... ({wait_time}s elapsed)")
+                
         except Exception as e:
             print(f"Error checking pipeline status: {e}")
-
+            
         time.sleep(poll_interval)
         wait_time += poll_interval
-
+    
     print(f"❌ Pipeline monitoring timed out after {max_wait} seconds")
     return False
 
 
+def monitor_pipeline(pipeline_name, version_id, max_wait=7200):
+    """Monitor pipeline using version-based tracking"""
+    # First find the execution that matches our version
+    execution_id = find_pipeline_execution_by_version(pipeline_name, version_id)
+    
+    if not execution_id:
+        return False
+        
+    # Then monitor that specific execution
+    return monitor_pipeline_execution(pipeline_name, execution_id, max_wait)
+
+
 def main():
     """Main execution function"""
     print("Starting integration test deployment...")
@@ -156,9 +195,9 @@ def main():
 
     # Execute deployment steps
     create_deployment_package()
-    upload_to_s3(bucket_name)
+    version_id = upload_to_s3(bucket_name)
 
-    success = monitor_pipeline(pipeline_name)
+    success = monitor_pipeline(pipeline_name, version_id)
 
     if success:
         print("🎉 Integration test deployment completed successfully!")