# How to filter CloudWatch logs using Python

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')
1
2

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']
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

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)
1
2
3
4
5
6
7
8
9
10
11

Newsletter

If you'd like to subscribe to my blog, please enter your details below. You can unsubscribe at any time.

Powered by Buttondown.

Last Updated: 11/20/2023, 10:04:51 AM