我有一个 AWS Lambda (python3.7
运行时)运行某个任务,及其MaximumRetryAttempts
被设定为1
(这意味着它可能会失败一次,然后再尝试一次)。
由于某种原因,我不会进入,我想知道,在运行时,这是第一次尝试还是第二次尝试。有什么办法可以实现这一点吗?
import json
import boto3
from datetime import datetime, timedelta
import time
aws_cloud_watch_logs_client = boto3.client('logs')
lambda_invocation_count_query = "fields @retry_count | filter @message like 'REPORT' AND @requestId = '{lambda_request_id}' | stats count(@message) as retry_count"
aws_cloud_watch_log_group = '/aws/lambda/mylambdaname' # Set aws cloudwatch log group name as per your setup
def lambda_handler(event, context):
lambda_retry_counts=get_lambda_retry_count(context)
print("lambda_retry_counts=",lambda_retry_counts)
#TODO : your logic here
# refactor to reduce complexity, might extract method to cover single responsibility principle
def get_lambda_retry_count(context):
print("context.aws_request_id=", context.aws_request_id)
qry=lambda_invocation_count_query.format(lambda_request_id=context.aws_request_id)
print(qry)
start_query_response = aws_cloud_watch_logs_client.start_query(
logGroupName=aws_cloud_watch_log_group,
startTime=int((datetime.today() - timedelta(hours=1)).timestamp()), # Refactor as per your requirement
endTime=int(datetime.now().timestamp()),
queryString=qry,
)
query_id = start_query_response['queryId']
response = None
while response == None or response['status'] == 'Running':
print('Waiting for cloud watch query to complete ...')
time.sleep(10) # Refactor as per your requirement
response = aws_cloud_watch_logs_client.get_query_results(
queryId=query_id
)
if response['results']:
return response['results'][0][0]['value'] # refactor for readability, maintenance
else:
return 0
或者,要查询 aws cloud watch,您还可以使用 AWS System Parameter Store 并将 aws_request_id 作为参数名称的一部分来唯一标识它。此外,保留 lambda 重试计数作为此参数的值。每次执行 lambda 时根据 aws_request_id 递增它。如果达到阈值,则根据您的逻辑进行处理,并删除名称具有达到阈值的 aws_request_id 的 AWS System Parameter Store。摘要 - 您只需要一些存储来保存 aws_request_id 和重试计数的映射。
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)