如何通过API密钥从AWS API网关获取使用度量,例如使用计数器、使用日志等?
目前我正在将x-api-key
发送到我的集成端点以进行手动记录。我希望不这样做,并让AWS API Gateway为我测量和报告此指标。
如何通过API密钥从AWS API网关获取使用度量,例如使用计数器、使用日志等?
目前我正在将x-api-key
发送到我的集成端点以进行手动记录。我希望不这样做,并让AWS API Gateway为我测量和报告此指标。
我发现所有的日志记录都不能满足我所需的要求,特别是因为它还没有按API Key进行记录(希望这个问题早日解决)。
所以我创建了自己的自定义日志 -
通过这种方式,我可以搜索我的CloudWatch日志,并获取我需要的精确数据,如果需要,甚至可以根据API Key进行搜索;
在我的阶段下,我启用了 "Custom Access Logging" 并使用了以下格式:
注意:这些自定义日志目前只支持context
变量。
input
变量的支持:文档可以在这里找到:
有了这个自定义日志,我可以打开CloudWatch
,按日期过滤并使用搜索字符串搜索任何我想要的东西(符合我的自定义记录);
/fees
上,所有以 BcxvY1
结尾的API Key所进行的GET请求。[RequestId,APIKeyText,APIKeyValue="*BcxvY1*",HTTPText,MethodText,HTTPMethodType="*GET*",PathText,PathValue="*/fees,",StatusText,StatusCode,ErrorsText,ErrorsValue,DescriptionText,DescriptionValue=custom_log]
希望这有所帮助。
ApiGateway
API的GetUsage
端点,该端点接受一个UsagePlanId
参数。
https://docs.aws.amazon.com/apigateway/latest/api/API_GetUsage.html
这是一个Python(boto3
)的示例:response = client.get_usage(
usagePlanId='string',
keyId='string',
startDate='string',
endDate='string',
position='string',
limit=123
)
from boto3 import session
from argparse import ArgumentParser
def parse_args():
parser = ArgumentParser(
prog="CreateNewUsagePlanWithKey",
description="Creates a new usage plan with key for an AWS API gateway api"
)
parser.add_argument(
"-p","--aws-profile",
type=str,
help="The AWS profile to use for the client session"
)
parser.add_argument(
"-r","--aws-region",
type=str,
default="us-east-1",
help="The AWS profile to use for the client session"
)
parser.add_argument(
"-g","--gateway-id",
type=str,
help="The api gateway id to create the key and usage plan for"
)
parser.add_argument(
"-s","--stage",
type=str,
help="The api gateway id to create the key and usage plan for"
)
parser.add_argument(
"-n","--usage-plan-name",
type=str,
help="The name to give the usage plan (will be propagated to the key with a \"_key\" suffix)"
)
return parser.parse_args().__dict__
def create_or_update_usage_plan(client,usage_plan_name,api_id,stage) -> str:
res = client.create_usage_plan(
name=usage_plan_name,
description=f"Usage plan {usage_plan_name} for API {api_id}",
apiStages=[
{
"apiId": api_id,
"stage": stage
}
]
)
return res["id"]
def create_api_key(client,usage_plan_id) -> (str,str):
res = client.create_api_key(
name=f"{usage_plan_id}_key",
description=f"API key for usage plan {usage_plan_id}",
enabled=True,
)
return res["id"],res["value"]
def create_usage_plan_key(client,usage_plan_id,key_id) -> None:
client.create_usage_plan_key(
usagePlanId=usage_plan_id,
keyId=key_id,
keyType="API_KEY"
)
def main(aws_profile,aws_region,gateway_id,stage,usage_plan_name) -> None:
sess = session.Session(profile_name=aws_profile,region_name=aws_region)
client = sess.client('apigateway')
usage_plan_id = create_or_update_usage_plan(client,usage_plan_name,gateway_id)
key_id,key_value = create_api_key(client,usage_plan_id)
create_usage_plan_key(client,usage_plan_id,key_id)
print(f"Successfully created usage plan, new key: {key_value}")
if __name__ == "__main__":
args = parse_args()
main(**args)