How to filter CloudWatch logs using Python

Fri May 28 2021

You can use boto3's CloudWatchLogs client (opens new window) to filter and fetch logs from CloudWatch.

First create the client:

import boto3
client = boto3.client('logs')

Then start (opens new window) a query using the start_query method:

import datetime

def get_seconds_since_epoch(d: datetime) -> int:
    epoch = datetime(1970, 1, 1)
    return int((d - epoch).total_seconds())

# query, see https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/CWL_QuerySyntax.html#CWL_Insights_Aggregation_Functions
query = f"""
    fields event
    | filter @message like 'User created'
    | limit 1
"""
# log groups you want to search:
log_groups = ["ecs/your-cluster/your-service"]
# start time in seconds since Unix Epch
start = get_seconds_since_epoch(datetime(2020, 6, 7, 11, 00))
# end time in seconds since Unix Epch
end = get_seconds_since_epoch(datetime(2020, 6, 7, 11, 00))
# number of logs to fetch
limit = 10


response = client.start_query(
    logGroupNames=log_groups,
    startTime=start,
    endTime=end,
    queryString=query,
    limit=limit,
)
query_id = response['queryId']

Then you can fetch the result of the query using the get_query_results (opens new window) method. You'll need to wait for the query to finish before you can see the results:

SLEEP_TIME = 3 # seconds

results = client.get_query_results(queryId=query_id)

time.sleep(SLEEP_TIME)

while results['status'] == 'Running':
    results = client.get_query_results(queryId=query_id)
    time.sleep(SLEEP_TIME)

print(results)