我可以确定哪个 AWS 服务触发了我的 Lambda 函数吗？

Posted 2023-03-08

【中文标题】我可以确定哪个 AWS 服务触发了我的 Lambda 函数吗？

【英文标题】：Can I determine which AWS service triggered my Lambda function?

【发布时间】：2017-07-22 13:58:08

【问题描述】：

我有一个 Python Lambda 函数，它可以响应 IoT Button 和 Alexa 技能。

有没有办法读取传递给处理函数的event 或context 以确定哪个服务触发了该函数（Alexa 或 IoT）？

【问题讨论】：

我没有使用过 Alexa 或 IoT，但我相信它们的事件对象的方案应该如此不同，以至于很难将一个与另一个混淆（除非事件结构其中一个或两个完全在用户的控制之下，并且您故意使它们无法区分，但在这种情况下，您只需通过添加例如eventSource 字段来使它们可区分）

@Leon：我不明白。这有什么关系？

请参阅this 问题，该问题的答案取决于event template 的格式。

@KhalidT.：哇，有趣。所以我真的需要基本上看看那里有哪些字段并据此猜测：没有简单标识服务的字段？

@KhalidT.: Unforutnaley，列出的事件模板都不是用于 IoT 或 ASK。

【参考方案1】：

您可以在 Cloudwatch 中读取和记录事件，如下所示

import logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def my_logging_handler(event, context):
logger.info('got event'.format(event))

这将在您看到的云手表中记录事件数据并确定触发 lambda 的事件

【讨论】：

问题是我可以访问event 或context 的哪一部分，我可以使用它来确定是什么派生（物联网或ASK）调用了该函数。或者，如果（且仅当）那是不可能的，我如何检查其他来源（比如日志，也许）这样做。

这并没有真正提供问题的答案，但它确实向您展示了如何调试它以自己发现答案。

【参考方案2】：

没有办法可靠地做到这一点。您可以获得的最接近的方法是熟悉不同服务生成的各种事件的内容，并（希望）识别出您感兴趣的每个系列中存在的可靠唯一键，然后您可以在代码中检查，例如与

if 'distinctKey' in event.keys():
    # ...

然而这不是一个可靠的方法，因为它需要你

检查每个可能的每个潜在服务生成的事件结构，并 成功而自信地为每个感兴趣的服务识别一个或一组密钥，该密钥或一组密钥始终可靠地存在在服务的事件中并且对它们唯一。

【参考方案3】：

希望 AWS 现在已经简化了这一点，但遗憾的是事实并非如此。没有一个参数可供您检查以确定所有 AWS 服务的事件类型。

但是，在网上找到了这个很好的发音here

function getLambdaEventSource(event) 
if (event.Records && event.Records[0].cf) return 'isCloudfront';

if (event.configRuleId && event.configRuleName && event.configRuleArn) return 'isAwsConfig';

if (event.Records && (event.Records[0].eventSource === 'aws:codecommit')) return 'isCodeCommit';

if (event.authorizationToken === "incoming-client-token") return 'isApiGatewayAuthorizer';

if (event.StackId && event.RequestType && event.ResourceType) return 'isCloudFormation';

if (event.Records && (event.Records[0].eventSource === 'aws:ses')) return 'isSes';

if (event.pathParameters && event.pathParameters.proxy) return 'isApiGatewayAwsProxy';

if (event.source === 'aws.events') return 'isScheduledEvent';

if (event.awslogs && event.awslogs.data) return 'isCloudWatchLogs';

if (event.Records && (event.Records[0].EventSource === 'aws:sns')) return 'isSns';

if (event.Records && (event.Records[0].eventSource === 'aws:dynamodb')) return 'isDynamoDb';

if (event.records && event.records[0].approximateArrivalTimestamp) return 'isKinesisFirehose';

if (event.records && event.deliveryStreamArn && event.deliveryStreamArn.startsWith('arn:aws:kinesis:')) return 'isKinesisFirehose';

if (event.eventType === 'SyncTrigger' && event.identityId && event.identityPoolId) return 'isCognitoSyncTrigger';

if (event.Records && event.Records[0].eventSource === 'aws:kinesis') return 'isKinesis';

if (event.Records && event.Records[0].eventSource === 'aws:s3') return 'isS3';

if (event.operation && event.message) return 'isMobileBackend';

【参考方案4】：

对于 Python 开发人员：您可以查看 https://gist.github.com/Necromancerx/abed07138690d37d170a6cf15b40d749。

def get_lambda_event_source(self, event: dict):
    if 'pathParameters' in event and 'proxy' in event['pathParameters']:
        return 'api_gateway_aws_proxy'
    elif 'Records' in event and len(event['Records']) > 0 and 'eventSource' in event['Records'][0] and event['Records'][0]['eventSource'] == 'aws:s3':
        return 's3'
    elif 'Records' in event and len(event['Records']) > 0 and 'EventSource' in event['Records'][0] and event['Records'][0]['EventSource'] == 'aws:sns':
        return 'sns'
    elif 'Records' in event and len(event['Records']) > 0 and 'eventSource' in event['Records'][0] and event['Records'][0]['eventSource'] == 'aws:dynamodb':
        return 'dynamo_db'
    elif 'Records' in event and len(event['Records']) > 0 and 'cf' in event['Records'][0]:
        return 'cloudfront'
    elif 'source' in event and event['source'] == 'aws.events':
        return 'scheduled_event'
    elif 'awslogs' in event and 'data' in event['awslogs']:
        return 'cloud_watch_logs'
    elif 'authorizationToken' in event and event['authorizationToken'] == "incoming-client-token":
        return 'api_gateway_authorizer'
    elif 'configRuleId' in event and 'configRuleName' in event and 'configRuleArn' in event:
        return 'aws_config'
    elif 'StackId' in event and 'RequestType' in event and 'ResourceType' in event:
        return 'cloud_formation'
    elif 'Records' in event and len(event['Records']) > 0 and 'eventSource' in event['Records'][0] and event['Records'][0]['eventSource'] == 'aws:codecommit':
        return 'code_commit'
    elif 'Records' in event and len(event['Records']) > 0 and 'eventSource' in event['Records'][0] and event['Records'][0]['eventSource'] == 'aws:ses':
        return 'ses'
    elif 'Records' in event and len(event['Records']) > 0 and 'eventSource' in event['Records'][0] and event['Records'][0]['eventSource'] == 'aws:kinesis':
        return 'kinesis'
    elif 'records' in event and len(event['Records']) > 0 and 'approximateArrivalTimestamp' in event['records'][0]:
        return 'kinesis_firehose'
    elif 'records' in event and len(event['Records']) > 0 and 'deliveryStreamArn' in event and event['deliveryStreamArn'] is str and event['deliveryStreamArn'].startswith('arn:aws:kinesis:'):
        return 'kinesis_firehose'
    elif 'eventType' in event and event['eventType'] == 'SyncTrigger' and 'identityId' in event and 'identityPoolId' in event:
        return 'cognito_sync_trigger'
    elif 'operation' in event and 'message' in event:
        return 'is_mobile_backend'

基于这个 javascript gist https://gist.github.com/jeshan/52cb021fd20d871c56ad5ce6d2654d7b