Find AWS Lambda Functions Executed in a Specific Time Range Using Boto3 and CloudWatch Logs

Motivation

At times, we need to identify which Lambda functions were executed within a specific time range, primarily for troubleshooting or monitoring purposes.

In this blog, I'll share a Python Boto3 script that allows you to search and discover which Lambda functions have execution results in their CloudWatch logs.

(Note: You can also use CloudWatch Logs Insights to query the logs. If the number of target log groups is smaller than the maximum limit of 50, using Logs Insights on the Console might be more convenient.)

Code

	"""
	This script finds and prints AWS Lambda functions executed in a specific time range using boto3 and CloudWatch Logs.
	Parameters:
	prefix (str): Log group name prefix to filter the desired log groups.
	Example: '/aws/lambda' to fetch log groups for Lambda functions.
	filter_pattern (str): The filter pattern to apply on the logs.
	Example: 'START RequestId' to find logs that contain the string 'START RequestId'.
	end_datetime (datetime): The end datetime of the time range in which you want to search for executed Lambda functions.
	Example: datetime(2023, 4, 27, 19, 10, 0, 0) for April 27, 2023 at 19:10:00.
	range_minutes (int): The duration in minutes to search for executed Lambda functions before the end_datetime.
	Example: 10 to search for Lambda functions executed in the 10 minutes before end_datetime.
	timezone (str, optional): The timezone to be used when calculating the time range. Defaults to "UTC".
	Example: "Asia/Tokyo" for Tokyo timezone.
	The script will:
	1. Get the log groups with the given prefix.
	2. Retrieve the events from these log groups within the specified time range.
	3. Filter the events based on the provided filter pattern.
	4. Print the log group name and the number of events that match the filter pattern.
	Usage:
	Ensure you have the AWS SDK for Python (Boto3) installed and properly configured with your AWS credentials before running the script.
	Run the script using `python <script_name>.py`.
	"""
	from datetime import datetime, timedelta
	from zoneinfo import ZoneInfo
	import boto3
	def get_time_range(end_datetime, minutes, timezone="UTC"):
	tz = ZoneInfo(timezone)
	time_end = end_datetime.astimezone(tz)
	time_start = time_end - timedelta(minutes=minutes)
	return time_start, time_end
	def print_header(time_start, time_end):
	print(f'Start. {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}\n===')
	print("From", time_start)
	print("To", time_end)
	def print_events(group, events):
	print("---")
	print(group["logGroupName"])
	print(len(events), "events")
	def retrieve_events(log_group_name, start_time, end_time, filter_pattern):
	client = boto3.client("logs")
	next_token = ""
	response = {}
	kwargs = {
	"logGroupName": log_group_name,
	"startTime": start_time,
	"endTime": end_time,
	"filterPattern": filter_pattern,
	}
	while True:
	if next_token:
	kwargs["nextToken"] = next_token
	response = client.filter_log_events(**kwargs)
	if "nextToken" in response:
	next_token = response["nextToken"]
	if len(response["events"]) > 0:
	yield response["events"]
	else:
	break
	def get_log_groups(prefix):
	client = boto3.client("logs")
	next_token = ""
	response = {}
	kwargs = {"logGroupNamePrefix": prefix}
	while True:
	if next_token:
	kwargs["nextToken"] = next_token
	response = client.describe_log_groups(**kwargs)
	if "nextToken" in response:
	next_token = response["nextToken"]
	yield response["logGroups"]
	else:
	yield response["logGroups"]
	break
	def main():
	prefix = "/aws/lambda"
	filter_pattern = "START RequestId"
	end_datetime = datetime(2023, 4, 30, 21, 10, 0, 0)
	range_minutes = 60
	timezone = "Asia/Tokyo"
	time_start, time_end = get_time_range(end_datetime, range_minutes, timezone)
	time_start_unix = int(time_start.timestamp() * 1000)
	time_end_unix = int(time_end.timestamp() * 1000)
	print_header(time_start, time_end)
	groups = [group for group_list in get_log_groups(prefix) for group in group_list]
	print(f"Check {len(groups)} log groups... wait...")
	for group in groups:
	for events in retrieve_events(
	group["logGroupName"], time_start_unix, time_end_unix, filter_pattern
	):
	if events:
	print_events(group, events)
	print("===\nFinished. ", datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
	if __name__ == "__main__":
	main()

view raw find_lambda_executed_in_specific_time_range.py hosted with ❤ by GitHub

Example results

This is the example of the results. I've found that the Lambda function test-my2 was executed twice (2 events) between 2023-04-30 20:10:00+09:00 and 2023-04-30 21:10:00+09:00.

Start. 2023-04-30 21:20:37
===
From 2023-04-30 20:10:00+09:00
To 2023-04-30 21:10:00+09:00
Check 65 log groups... wait...
---
/aws/lambda/test-my2
2 events
===
Finished.  2023-04-30 21:21:17

Appendix

You can also use logGroupNamePattern instead of logGroupNamePrefix. (These are mutually exclusive. ) Note that boto3 version after 1.26.17 is required.

Refer to the boto3 document.