#!/usr/bin/env python3 """ AWS Step Functions Failed Executions Lister This program lists failed executions for a specified AWS Step Functions state machine from the past N days (default: 5 days). Usage: python program_list_failed_executions.py [--days N] [--max-results N] Examples: python program_list_failed_executions.py MyStateMachine python program_list_failed_executions.py MyStateMachine --days 7 python program_list_failed_executions.py MyStateMachine --days 1 --max-results 50 """ import boto3 import json import sys import os import argparse from datetime import datetime, timedelta, timezone from botocore.exceptions import ClientError, NoCredentialsError from dotenv import load_dotenv # Load environment variables from .env file load_dotenv() def setup_aws_session(): """ Setup AWS session with proper error handling for credentials. Returns boto3 Step Functions client. """ try: # Check if .env file exists and report what we found env_file_path = os.path.join(os.getcwd(), '.env') if os.path.exists(env_file_path): print("✓ Found .env file - loading AWS credentials") # Create session - this will use environment variables or AWS CLI config session = boto3.Session() # Test credentials by getting caller identity sts = session.client('sts') identity = sts.get_caller_identity() print(f"✓ Connected to AWS as: {identity.get('Arn', 'Unknown')}") print(f"✓ Account ID: {identity.get('Account', 'Unknown')}") # Show current region region = session.region_name or os.environ.get('AWS_DEFAULT_REGION', 'Unknown') print(f"✓ Region: {region}") # Create Step Functions client sfn_client = session.client('stepfunctions') return sfn_client except NoCredentialsError: print("❌ Error: No AWS credentials found!") print("\nPlease set up your AWS credentials using one of these methods:") print("1. Create a .env file in this directory with:") print(" AWS_ACCESS_KEY_ID=your_access_key") print(" AWS_SECRET_ACCESS_KEY=your_secret_key") print(" AWS_DEFAULT_REGION=your_region") print(" AWS_SESSION_TOKEN=your_session_token # (if using temporary credentials)") print("\n2. Environment variables:") print(" export AWS_ACCESS_KEY_ID=your_access_key") print(" export AWS_SECRET_ACCESS_KEY=your_secret_key") print(" export AWS_DEFAULT_REGION=your_region") print("\n3. AWS CLI configuration:") print(" aws configure") print("\n4. IAM roles (if running on EC2)") sys.exit(1) except ClientError as e: print(f"❌ Error connecting to AWS: {e}") sys.exit(1) def find_state_machine_arn(sfn_client, state_machine_name): """ Find the ARN of a state machine by name. """ try: # List all state machines and find the one with matching name response = sfn_client.list_state_machines() for sm in response.get('stateMachines', []): if sm['name'] == state_machine_name: return sm['stateMachineArn'] # If not found, print available state machines print(f"❌ State machine '{state_machine_name}' not found!") print("\nAvailable state machines:") for sm in response.get('stateMachines', []): print(f" - {sm['name']}") return None except ClientError as e: print(f"❌ Error listing state machines: {e}") return None def list_failed_executions(sfn_client, state_machine_arn, max_results=100, days_back=5): """ List failed executions for a given state machine within the past specified days. """ try: # Calculate the cutoff date (5 days ago) cutoff_date = datetime.now(timezone.utc) - timedelta(days=days_back) print(f"\n🔍 Searching for failed executions from the past {days_back} days (since {cutoff_date.strftime('%Y-%m-%d %H:%M:%S UTC')})...") print(f" Maximum results: {max_results}") response = sfn_client.list_executions( stateMachineArn=state_machine_arn, statusFilter='FAILED', maxResults=max_results ) all_failed_executions = response.get('executions', []) # Filter executions to only include those from the past 5 days recent_failed_executions = [] for execution in all_failed_executions: start_date = execution.get('startDate') if start_date and start_date >= cutoff_date: recent_failed_executions.append(execution) if not recent_failed_executions: if all_failed_executions: print(f"✅ No failed executions found in the past {days_back} days!") print(f" (Found {len(all_failed_executions)} older failed executions)") else: print("✅ No failed executions found!") return [] print(f"\n📋 Found {len(recent_failed_executions)} failed execution(s) in the past {days_back} days:") print("-" * 100) for i, execution in enumerate(recent_failed_executions, 1): execution_name = execution.get('name', 'Unknown') execution_arn = execution.get('executionArn', 'Unknown') start_date = execution.get('startDate', 'Unknown') stop_date = execution.get('stopDate', 'Unknown') # Format dates if they exist if isinstance(start_date, datetime): start_date_str = start_date.strftime('%Y-%m-%d %H:%M:%S UTC') # Calculate how many days ago days_ago = (datetime.now(timezone.utc) - start_date).days if days_ago == 0: days_ago_str = "today" elif days_ago == 1: days_ago_str = "1 day ago" else: days_ago_str = f"{days_ago} days ago" start_date_str += f" ({days_ago_str})" else: start_date_str = str(start_date) if isinstance(stop_date, datetime): stop_date_str = stop_date.strftime('%Y-%m-%d %H:%M:%S UTC') else: stop_date_str = str(stop_date) print(f"{i}. Execution Name: {execution_name}") print(f" ARN: {execution_arn}") print(f" Started: {start_date_str}") print(f" Stopped: {stop_date_str}") # Get execution details for error information try: details = sfn_client.describe_execution(executionArn=execution_arn) if 'error' in details: print(f" Error: {details.get('error', 'Unknown error')}") if 'cause' in details: cause = details.get('cause', '') # Truncate long causes if len(cause) > 200: cause = cause[:200] + "..." print(f" Cause: {cause}") except ClientError as e: print(f" Could not fetch details: {e}") print("-" * 100) # Show summary of older executions if any older_executions_count = len(all_failed_executions) - len(recent_failed_executions) if older_executions_count > 0: print(f"\n📊 Additional info: {older_executions_count} older failed execution(s) found (older than {days_back} days)") return recent_failed_executions except ClientError as e: print(f"❌ Error listing executions: {e}") return [] def main(): """ Main function to handle command line arguments and orchestrate the process. """ print("🚀 AWS Step Functions Failed Executions Lister") print("=" * 50) # Set up command line argument parsing parser = argparse.ArgumentParser( description='List failed executions for AWS Step Functions state machine from the past N days', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: python program_list_failed_executions.py MyStateMachine python program_list_failed_executions.py MyStateMachine --days 7 python program_list_failed_executions.py MyStateMachine --days 1 --max-results 50 """ ) parser.add_argument('state_machine_name', help='Name of the Step Functions state machine') parser.add_argument('--days', '-d', type=int, default=5, help='Number of days back to search for failed executions (default: 5)') parser.add_argument('--max-results', '-m', type=int, default=100, help='Maximum number of results to return (default: 100)') args = parser.parse_args() print(f"🎯 Target state machine: {args.state_machine_name}") print(f"📅 Looking back: {args.days} days") print(f"📊 Max results: {args.max_results}") # Setup AWS session sfn_client = setup_aws_session() # Find state machine ARN state_machine_arn = find_state_machine_arn(sfn_client, args.state_machine_name) if not state_machine_arn: sys.exit(1) print(f"✓ Found state machine ARN: {state_machine_arn}") # List failed executions failed_executions = list_failed_executions(sfn_client, state_machine_arn, max_results=args.max_results, days_back=args.days) print(f"\n📊 Summary: {len(failed_executions)} failed execution(s) found in the past {args.days} days") if failed_executions: print("\n💡 Next steps:") print("- Review the error messages and causes above") print("- Check CloudWatch logs for detailed error information") print("- Consider implementing retry logic or fixing the underlying issues") if __name__ == "__main__": main()