FYI: The Current date and time is Tuesday 24 September 2024 at 5:17:18 UTC.
You are an AWS Specialist with deep knowledge of AWS services and their Python SDK (boto3) operations, their options, input and outputs.
You have expertise in generating Python scripts that use boto3 to achieve a goal.
You also have the assistance of a script library which contains similar problems with working scripts for your reference.
In Addition, you follow these guidelines:
- Assume that the Python script you generate will be run against the given AWS account and with necessary permissions to access it.
- Use appropriate pagination methods provided by boto3 for handling large result sets.
- Check for all possible fields where the token can be present for fetching next page results, e.g., 'NextToken', 'Marker', etc., as applicable to the specific boto3 method.
- Avoid using hardcoded values unless absolutely necessary. Use variables for flexibility.
- Do not return a single boto3 call directly, always return a complete Python script.
- Always specify the region when creating boto3 clients or resources.
- Print only the final output to stdout.
- If no matching results are found for the provided goal, print explicitly to stdout "No matching results found".
Generate a Python script to achieve the goal provided by the user using boto3 and if necessary, libraries like json and datetime. Note that whatever code you provide will be executed as is, so ensure that the script you generate is ready to execute directly.
-
Before you generate the script, create a plan of steps that summarize your approach. You will be returning this approach as 'scriptPlan' later, but also you will be using this plan/approach for generating the script in the next steps.
Script to find EC2 instances with less than 20% CPU utilization in us-east-1
- Import necessary libraries and create boto3 clients
- List all EC2 instances
- For each instance, fetch average CPU utilization for the last 24 hours from CloudWatch
- Filter the instances where the average utilization across 24 hours is
-
Start the script with necessary imports and boto3 client creation.
-
Add the remaining script content, with proper pagination handling if necessary and logic to filter out only the relevant data from responses.
-
Remember the generated script as TEMPORARY_SCRIPT
-
Review the TEMPORARY_SCRIPT against the goal and validate that it will satisfy the goal without errors.
-
Review the TEMPORARY_SCRIPT to ensure that all boto3 calls are complete with all parameters and without placeholders. If there are errors, regenerate it with fixes.
Return the fixed TEMPORARY_SCRIPT in the response as 'script' variable.
Retrieve and filter logs from the first identified log stream
Inputs for task:
awsAcccountId: 346945241475
AwsRegion: us-east-1
query: Retrieve logs from the log stream '2024/09/24/[$LATEST]69fcc0c1dbeb42fc802c999a8217fd6a' of the Lambda function 'amplify-solvewithmeui-abh-handledatastreamlambdaA3-n27i5mHYJidl'. Filter these logs to include only those where the 'level' attribute is set to 'INFO'. Return the filtered logs.
Inputs for task:
awsAcccountId: 346945241475
AwsRegion: us-east-1
scriptPlan: |-
Script to retrieve and filter logs from a specific log stream
- Import necessary libraries (boto3, json)
- Set up boto3 client for CloudWatch Logs in us-east-1 region
- Define the log group name for the Lambda function
- Define the log stream name
- Use get_log_events to retrieve logs from the specified log stream
- Implement pagination to fetch all log events
- Parse each log event and filter for 'level': 'INFO'
- Print the filtered log events
script: |-
import boto3
import json
Set up boto3 client for CloudWatch Logs
region = 'us-east-1'
logs_client = boto3.client('logs', region_name=region)
Define log group and stream names
function_name = 'amplify-solvewithmeui-abh-handledatastreamlambdaA3-n27i5mHYJidl'
log_group_name = f'/aws/lambda/{function_name}'
log_stream_name = '2024/09/24/[$LATEST]69fcc0c1dbeb42fc802c999a8217fd6a'
Function to get and filter log events
def get_filtered_logs():
next_token = None
filtered_logs = []
while True:
if next_token:
response = logs_client.get_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
nextToken=next_token
)
else:
response = logs_client.get_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name
)
for event in response['events']:
try:
message = json.loads(event['message'])
if message.get('level') == 'INFO':
filtered_logs.append(event['message'])
except json.JSONDecodeError:
# If the message is not JSON, skip it
continue
# Check if we've reached the end of the stream
if next_token == response['nextForwardToken']:
break
next_token = response['nextForwardToken']
return filtered_logs
Get and print filtered logs
filtered_logs = get_filtered_logs()
if filtered_logs:
print("Filtered logs with 'level': 'INFO':")
for log in filtered_logs:
print(log)
else:
print("No matching results found")