create-alarms.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import os
import json
import boto3

api_gw = boto3.client('apigateway')
cw = boto3.client('cloudwatch')

alarmActions = os.environ['ALARM_SNS_TOPIC']


def get_rest_api_name(api_id):
    response = api_gw.get_rest_api(restApiId=api_id)
    return response['name']


def get_rest_endpoints(api_id):
    endpoints = []
    response = api_gw.get_resources(restApiId=api_id)
    for item in response['items']:
        methods = item['resourceMethods']
        for method in methods:
            endpoint = {"method": method, "resource": item['path']}
            endpoints.append(endpoint)
    return endpoints


def enable_detailed_metrics(api_id, stage):
    response = api_gw.get_stages(restApiId=api_id)
    response.pop("ResponseMetadata")
    for item in response['item']:
        method_settings = item['methodSettings']
        if not method_settings['*/*']['metricsEnabled']:
            resp = api_gw.update_stage(
                restApiId=api_id,
                stageName=stage,
                patchOperations=[
                    {
                        "op": "replace",
                        "path": "/*/*/metrics/enabled",
                        "value": "True"
                    },
                ],
            )
            print(resp)


def create_alarms_for_endpoints(api_id, stage):
    api_name = get_rest_api_name(api_id)
    print("API Name is ", api_name)
    rest_endpoints = get_rest_endpoints(api_id)
    print("got REST endpoints", rest_endpoints)
    for endpoint in rest_endpoints:
        resp = cw.put_metric_alarm(
            AlarmName="API [{}] stage [{}] {} {} :p99 > 1s".format(
                api_name, stage, endpoint["method"], endpoint["resource"]),
            MetricName="Latency",
            Dimensions=[{
                'Name': 'ApiName',
                'Value': api_name
            }, {
                'Name': 'Resource',
                'Value': endpoint['resource']
            }, {
                'Name': 'Method',
                'Value': endpoint['method']
            }, {
                'Name': 'Stage',
                'Value': stage
            }],
            Namespace='AWS/ApiGateway',
            Threshold=1000,
            ComparisonOperator='GreaterThanThreshold',
            Period=60,
            EvaluationPeriods=5,
            DatapointsToAlarm=5,
            ExtendedStatistic='p99',
            ActionsEnabled=True,
            AlarmActions=[alarmActions],
            AlarmDescription=
            "auto-generated by lambda [{process.env.AWS_LAMBDA_FUNCTION_NAME}]",
            Unit="Milliseconds")
        print(resp)


def handler(event, context):
    event_msg = event['details']['requestParameters']
    rest_api_id = event_msg['restApiId']
    stage_name = event_msg['createDeploymentInput']['stageName']
    enable_detailed_metrics(rest_api_id, stage_name)
    create_alarms_for_endpoints(rest_api_id, stage_name)