我能否确定是哪个AWS服务触发了我的Lambda函数？

Question

我有一个Python Lambda函数，可以响应物联网按钮和Alexa技能。

是否有一种方法可以读取传递给处理程序函数的 event 或 context 来识别触发函数的服务（Alexa或IoT）？

- orome

---

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

请查看此问题，其中有一个被接受的答案依赖于event template的格式。 - Khalid T.

@KhalidT: 哇，有趣。所以我基本上需要查看存在哪些字段，并根据那些字段进行猜测：没有一个仅仅用于识别服务的字段吗？ - orome

@KhalidT.：不幸的是，列出的所有事件模板都不是针对物联网或ASK的。 - orome

遗憾的是，AWS仍然没有将事件源作为事件对象本身的一部分提供。这就是为什么我引用了那个答案，基本上是依靠猜测（是的，我知道这不太好看）。此外，AWS文档到目前为止还没有列出样例IoT或ASK事件，但您可以使用lambda函数将其打印到控制台或CloudWatch。 - Khalid T.

Answer 1

希望AWS现在已经改进了这一点，但遗憾的是还没有。在所有AWS服务中，没有一个单一的参数可以用来确定事件类型。

不过，在网上找到了这篇不错的文章（链接）

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';

}

- Kdroid

Answer 2

无法可靠地完成此操作。你能做的最接近的方法是熟悉不同服务生成的各种事件内容，并（希望）在你感兴趣的每个系列中识别出一个可靠的唯一键，然后在代码中检查它，例如使用

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

然而，这几乎不是一种可靠的方法，因为它要求您：

1. 检查每个潜在服务生成的所有可能的事件结构，并且
2. 成功并自信地为每个感兴趣的服务识别一个关键或一组始终可靠出现在服务的事件中且唯一的密钥。

- orome

Answer 3

针对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代码片段https://gist.github.com/jeshan/52cb021fd20d871c56ad5ce6d2654d7b。

- John Paulo Rodriguez

Answer 4

您可以像下面这样在Cloudwatch中读取并记录事件。

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

这将在云监视器中记录事件数据，以便您可以确定触发 Lambda 的事件

- Dhanesh H S

---

问题是我可以访问的event或context的哪个部分，可以用来确定是什么衍生（IoT或ASK）调用了该函数。或者，如果（并且只有在）这不可能，那么我如何检查其他某些来源（例如日志）来确定呢？ - orome

这并没有真正回答问题，但它确实向你展示了如何调试以自己发现答案。 - Mark B