如何在Boto3 ECS ServicesStable等待器执行期间定期打印状态信息？

Absolutely! The built-in Boto3 waiters are convenient but don’t include default progress output. Here are two clean approaches to add periodic status messages while waiting for your ECS service to stabilize:

Option 1: Custom Polling Loop (Full Control) #

Instead of relying on the built-in waiter, implement a manual polling loop. This gives you complete control over what to print and when, and ensures your status checks match exactly what the built-in services_stable waiter looks for.

import boto3
import time

ecs_client = boto3.client('ecs', config=my_config)
cluster_name = "myCluster"
service_name = "myService"
delay = 15  # Seconds between checks
max_attempts = 60  # Total wait time: 15*60=900s (15 mins)

attempt = 0
service_stable = False

while attempt < max_attempts:
    attempt += 1
    # Fetch current service details
    response = ecs_client.describe_services(cluster=cluster_name, services=[service_name])
    service = response['services'][0]

    # Match Boto3's built-in `services_stable` waiter conditions exactly
    is_stable = (
        service['status'] == 'ACTIVE'
        and service['desiredCount'] == service['runningCount']
        and service['pendingCount'] == 0
        and len(service['deployments']) > 0
        and service['deployments'][0]['status'] == 'PRIMARY'
        and service['deployments'][0]['desiredCount'] == service['deployments'][0]['runningCount']
    )

    if is_stable:
        service_stable = True
        break

    # Print progress (overwrites the same line each time to avoid clutter)
    print(f"Waiting for service to stabilize... Attempt {attempt}/{max_attempts} (next check in {delay}s)", end='\r')
    time.sleep(delay)

if service_stable:
    print("\nService is in steady state.")
else:
    print("\nTimeout: Service did not stabilize within the maximum allowed attempts.")

Why this works: #

• You replicate the exact stability checks used by Boto3's native waiter
• The end='\r' flag keeps progress updates on a single line, preventing terminal clutter
• You can easily customize the message to include additional details like pending tasks or deployment status

Option 2: Threaded Progress Updates (Keep Using Built-in Waiter) #

If you want to keep using the built-in waiter but add progress messages, use a daemon thread to print updates in parallel. This requires minimal changes to your original code.

import boto3
import threading
import time

def print_progress(delay, max_attempts):
    """Thread function to print periodic progress updates"""
    for attempt in range(1, max_attempts + 1):
        print(f"Waiting for service to stabilize... Attempt {attempt}/{max_attempts}", end='\r')
        time.sleep(delay)

ecs_client = boto3.client('ecs', config=my_config)
cluster_name = "myCluster"
service_name = "myService"
waiter_config = {'Delay': 15, 'MaxAttempts': 60}

# Start progress thread (daemon thread exits when main thread finishes)
progress_thread = threading.Thread(
    target=print_progress,
    args=(waiter_config['Delay'], waiter_config['MaxAttempts'])
)
progress_thread.daemon = True
progress_thread.start()

try:
    # Run the original waiter
    waiter = ecs_client.get_waiter('services_stable')
    waiter.wait(
        cluster=cluster_name,
        services=[service_name],
        WaiterConfig=waiter_config
    )
    print("\nService is in steady state.")
except Exception as e:
    print(f"\nWaiter failed: {str(e)}")

Why this works: #

• You retain all the benefits of Boto3's tested waiter logic
• The daemon thread runs in the background, printing updates without interfering with the waiter
• No need to replicate the stability check logic yourself

Both approaches will eliminate the "blank terminal" issue and keep you informed about the wait progress. Pick the one that fits your workflow best!

内容的提问来源于stack exchange，提问作者Dipendra Dangal