import boto3
import time
client = boto3.client('logs')
log_group_name = '/aws/mwaa/environment-name/task_logs'
log_stream_name = 'log-stream-name'
response = client.create_log_stream(
logGroupName=log_group_name,
logStreamName=log_stream_name
)
if response['ResponseMetadata']['HTTPStatusCode'] == 200:
print("Log stream created successfully!")
else:
print("Log stream creation failed.")
# Wait for a few seconds for the log stream to be fully created before pushing logs
time.sleep(5)
# Push logs to the newly created log stream
response = client.put_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
logEvents=[
{
'timestamp': int(round(time.time() * 1000)),
'message': 'Test log message.'
}
]
)
if response['ResponseMetadata']['HTTPStatusCode'] == 200:
print("Log event pushed successfully!")
else:
print("Log event push failed.")
上一篇:AWSMWAA定价