Introduction to AWS Log Insights as CloudWatch Metrics

16:39:11,156 INFO  [celery.worker.strategy] Received task: lib.tasks.lists.trigger_list_log_notification[9b33b464-d4f9-4909-8d4e-1a3134fead97]  

parse @message /\[celery\.(?<source>[a-z.]+)\].*Received task: (?<task>[a-z._]+)\[/
 | filter not isblank(source)
 | stats count(*) as number by task
 | sort number desc
 | limit 9

1 lib.tasks.lists.trigger_list_log_notification 4559
2 lib.tasks.notify.notify_recipient 397
3 lib.message._send_mobile_push_notification 353
4 lib.tasks.jobs.check_job_cutoffs 178
5 lib.tasks.notify.check_message_cutoffs 177
6 lib.tasks.notify.check_notification_retry 177
7 lib.tasks.notify.async_list_response 81
8 lib.tasks.hmrc_poll.govtalk_periodic_poll 59
9 lib.tasks.lists.recalculate_list_entry 56

parse @message /\[celery\.(?<source>[a-z.]+)\].*Received task: (?<task>[a-z._]+)\[/
 | filter not isblank(source)
 | stats count(*) by bin(30s)

+ common
  + lambda-loginsights2metrics
    + cloudwatch.tf
    + dynamodb.tf
    + iam.tf
    + lambda.tf
    + variables.tf
+ account1
  + lambda-loginsights2metrics
    + main.tf
    + variables.tf

variable "items" {
  type    = "list"
  default = []
}

variable "prefix" {
  type    = "string"
  default = "loginsights2metrics"
}

variable "security_groups" {
  type    = "list"
  default = []
}

variable "vpc_subnets" {
  type    = "list"
  default = []
}

resource "aws_dynamodb_table" "loginsights2metrics" {
  name = "${var.prefix}-items"

  billing_mode = "PAY_PER_REQUEST"

  hash_key = "SlotName"

  attribute {
    name = "SlotName"
    type = "S"
  }
}

resource "aws_dynamodb_table_item" "item" {
  count = length(var.items)

  table_name = aws_dynamodb_table.loginsights2metrics.name
  hash_key   = aws_dynamodb_table.loginsights2metrics.hash_key

  item = jsonencode(element(var.items, count.index))
}

parse @message /\[celery\.(?<source>[a-z.]+)\].*Received task: (?<task>[a-z._]+)\[/
 | filter not isblank(source)
 | stats count(*) as Value by bin(1m)

"parse @message /\\[celery\\.(?<source>[a-z.]+)\\].*Received task: (?<task>[a-z._]+)\\[/ | filter not isblank(source) | stats count(*) as Value by bin(1m)"

data "aws_iam_policy_document" "trust" {
  statement {

    actions = ["sts:AssumeRole"]

    principals {
      type = "Service"
      identifiers = [
        "lambda.amazonaws.com"
      ]
    }
  }
}

resource "aws_iam_role" "loginsights2metrics" {
  name               = var.prefix
  assume_role_policy = data.aws_iam_policy_document.trust.json
}

$ aws iam get-policy --policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole
{
    "Policy": {
        "PolicyName": "AWSLambdaVPCAccessExecutionRole",
        "PolicyId": "ANPAJVTME3YLVNL72YR2K",
        "Arn": "arn:aws:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole",
        "Path": "/service-role/",
        "DefaultVersionId": "v2",
        "AttachmentCount": 0,
        "PermissionsBoundaryUsageCount": 0,
        "IsAttachable": true,
        "Description": "Provides minimum permissions for a Lambda function to execute while accessing a resource within a VPC - create, describe, delete network interfaces and write permissions to CloudWatch Logs. ",
        "CreateDate": "2016-02-11T23:15:26Z",
        "UpdateDate": "2020-10-15T22:53:03Z"
    }
}
$ aws iam get-policy-version --policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole --version-id v2
{
    "PolicyVersion": {
        "Document": {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": [
                        "logs:CreateLogGroup",
                        "logs:CreateLogStream",
                        "logs:PutLogEvents",
                        "ec2:CreateNetworkInterface",
                        "ec2:DescribeNetworkInterfaces",
                        "ec2:DeleteNetworkInterface",
                        "ec2:AssignPrivateIpAddresses",
                        "ec2:UnassignPrivateIpAddresses"
                    ],
                    "Resource": "*"
                }
            ]
        },
        "VersionId": "v2",
        "IsDefaultVersion": true,
        "CreateDate": "2020-10-15T22:53:03Z"
    }
}

resource "aws_iam_role_policy_attachment" "loginsights2metrics-" {
  role       = aws_iam_role.loginsights2metrics.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole"
}

data "aws_iam_policy_document" "loginsights2metrics" {
  statement {
    actions = [
      "cloudwatch:PutMetricData",
      "cloudwatch:PutMetricAlarm",
      "logs:StartQuery",
      "logs:GetQueryResults",
      "logs:GetLogEvents",
    ]

    resources = ["*"]
  }

  statement {
    actions = [
      "dynamodb:Scan"
    ]

    resources = [aws_dynamodb_table.loginsights2metrics.arn]
  }
}

resource "aws_iam_role_policy" "loginsights2metrics" {
  name   = var.prefix
  role   = aws_iam_role.loginsights2metrics.name
  policy = data.aws_iam_policy_document.loginsights2metrics.json
}

data "archive_file" "dummy" {
  type        = "zip"
  output_path = "${path.module}/lambda.zip"

  source {
    content  = "dummy"
    filename = "dummy.txt"
  }
}

resource "aws_lambda_function" "loginsights2metrics" {
  function_name = "loginsights2metrics"
  handler       = "zappa.handler.lambda_handler"
  runtime       = "python3.8"
  filename      = data.archive_file.dummy.output_path

  role        = aws_iam_role.loginsights2metrics.arn
  memory_size = 128
  timeout     = 300

  vpc_config {
    subnet_ids         = var.vpc_subnets
    security_group_ids = var.security_groups
  }

  environment {
    variables = {
      "SENTRY_DSN"     = "https://XXXXXX:@sentry.io/YYYYYY",
      "DYNAMODB_TABLE" = aws_dynamodb_table.loginsights2metrics.name
    }
  }

  lifecycle {
    ignore_changes = [last_modified, filename]
  }
}

resource "aws_lambda_permission" "loginsights2metrics" {
  statement_id  = "AllowExecutionFromCloudWatch"
  action        = "lambda:InvokeFunction"
  function_name = aws_lambda_function.loginsights2metrics.function_name
  principal     = "events.amazonaws.com"
  source_arn    = aws_cloudwatch_event_rule.rate.arn
}

data "aws_iam_policy_document" "loginsights2metrics_exec" {
  statement {
    actions = [
      "lambda:InvokeAsync",
      "lambda:InvokeFunction"
    ]

    resources = [aws_lambda_function.loginsights2metrics.arn]
  }
}

resource "aws_iam_role_policy" "loginsights2metrics_exec" {
  name   = "${var.prefix}-exec"
  role   = aws_iam_role.loginsights2metrics.name
  policy = data.aws_iam_policy_document.loginsights2metrics_exec.json
}

resource "aws_cloudwatch_event_rule" "rate" {
  # Zappa requires the name to match the processing function
  name                = "main.loginsights2metrics"
  description         = "Trigger Lambda ${var.prefix}"
  schedule_expression = "rate(2 minutes)"
}

resource "aws_cloudwatch_event_target" "lambda" {
  rule      = aws_cloudwatch_event_rule.rate.name
  target_id = "${var.prefix}-target"
  arn       = aws_lambda_function.loginsights2metrics.arn
}

locals {
  items = [
    {
      "SlotName" : {
        "S" : "Celery Logs submitted tasks"
      },
      "LogGroup" : {
        "S" : "mycluster/celery",
      },
      "ClusterName" : {
        "S" : "mycluster"
      },
      "Query" : {
        "S" : "parse @message /\\[celery\\.(?<source>[a-z.]+)\\].*Received task: (?<task>[a-z._]+)\\[/ | filter not isblank(source) | stats count(*) as Value by bin(1m)",
      },
      "Namespace" : {
        "S" : "Custom"
      },
      "MetricName" : {
        "S" : "Submitted tasks"
      }
    },
    {
      "SlotName" : {
        "S" : "Celery Logs succeeded tasks"
      },
      "LogGroup" : {
        "S" : "mycluster/celery",
      },
      "ClusterName" : {
        "S" : "mycluster"
      },
      "Query" : {
        "S" : "parse @message /\\[celery.(?<source>[a-z\\._]+)].*Task (?<task>[a-z\\._]+)\\[.*\\] (?<event>[a-z]+)/ | filter source = \"app.trace\" | filter event = \"succeeded\" | stats count(*) as Value by bin(1m)",
      },
      "Namespace" : {
        "S" : "Custom"
      },
      "MetricName" : {
        "S" : "Succeeded tasks"
      }
    },
    {
      "SlotName" : {
        "S" : "Celery Logs retried tasks"
      },
      "LogGroup" : {
        "S" : "mycluster/celery",
      },
      "ClusterName" : {
        "S" : "mycluster"
      },
      "Query" : {
        "S" : "parse @message /\\[celery.(?<source>[a-z\\._]+)].*Task (?<task>[a-z\\._]+)\\[.*\\] (?<event>[a-z]+)/ | filter source = \"app.trace\" | filter event = \"retry\" | stats count(*) as Value by bin(1m)",
      },
      "Namespace" : {
        "S" : "Custom"
      },
      "MetricName" : {
        "S" : "Retried tasks"
      }
    }
  ]
}

data "aws_security_group" "default" {
  name   = "default"
  vpc_id = var.vpc_id
}

module "loginsights2metrics" {
  source = "../../common/lambda-loginsights2metrics"

  items           = local.items
  security_groups = [data.aws_security_group.default.id]
  vpc_subnets     = var.vpc_private_subnets
}

import os
import time
import json
from datetime import datetime, timedelta

import boto3
from zappa.asynchronous import task

# CONFIG
logs = boto3.client("logs", region_name="eu-west-1")
cw = boto3.client("cloudwatch", region_name="eu-west-1")
dynamodb = boto3.resource("dynamodb", region_name="eu-west-1")


@task
def put_metric_data(item):
    slot_name = item["SlotName"]
    log_group = item["LogGroup"]
    cluster_name = item["ClusterName"]
    query = item["Query"]
    namespace = item["Namespace"]
    metric_name = item["MetricName"]

    # This runs the Log Insights query fetching data
    # for the last 15 minutes.
    # As we deal with logs processing it's entirely possible
    # for the metric to be updated, for example because
    # a log was received a bit later.
    # When we put multiple values for the same timestamp
    # in the metric CW can show max, min, avg, and percentiles.
    # Since this is an update of a count we should then always
    # use "max".
    start_query_response = logs.start_query(
        logGroupName=log_group,
        startTime=int((datetime.now() - timedelta(minutes=15)).timestamp()),
        endTime=int(datetime.now().timestamp()),
        queryString=query,
    )

    query_id = start_query_response["queryId"]

    # Just polling the API. 5 seconds seems to be a good
    # compromise between not pestering the API and not paying
    # too much for the Lambda.
    response = None
    while response is None or response["status"] == "Running":
        print(f"{slot_name}: waiting for query to complete ...")
        time.sleep(5)
        response = logs.get_query_results(queryId=query_id)

    # Data comes in a strange format, a dictionary of
    # {"field":name,"value":actual_value}, so this converts
    # it into something that can be accessed through keys
    data = []
    for d in response["results"]:
        sample = {}
        for i in d:
            field = i["field"]
            value = i["value"]

            sample[field] = value

        data.append(sample)

    # Now that we have the data, let's put them into a metric.
    for d in data:
        timestamp = datetime.strptime(d["bin(1m)"], "%Y-%m-%d %H:%M:%S.000")
        value = int(d["Value"])

        print(f"{slot_name}: putting {value} on {timestamp}")

        cw.put_metric_data(
            Namespace=namespace,
            MetricData=[
                {
                    "MetricName": metric_name,
                    "Dimensions": [{"Name": "Cluster", "Value": cluster_name}],
                    "Timestamp": timestamp,
                    "Value": value,
                    "Unit": "None",
                }
            ],
        )


def loginsights2metrics(event, context):
    with open("package_info.json", "r") as f:
        package_info = json.load(f)
        build_timestamp = int(package_info["build_time"])
        build_datetime = datetime.fromtimestamp(build_timestamp)

    print("###################################")
    print(
        "LogInsights2Metrics - Build date: "
        f'{build_datetime.strftime("%Y/%m/%d %H:%M:%S")}'
    )
    print("###################################")

    print(f'Reading task from DynamoDB table {os.environ["DYNAMODB_TABLE"]}')

    table = dynamodb.Table(os.environ["DYNAMODB_TABLE"])

    # This is the simplest way to get all entries in the table
    # The next loop will asynchronously call `put_metric_data`
    # on each entry.
    response = table.scan(Select="ALL_ATTRIBUTES")

    for i in response["Items"]:
        print(f"* Processing item {i['SlotName']}")
        put_metric_data(i)

{
  "main": {
    "app_module": "main",
    "app_function": "main.loginsights2metrics",
    "runtime": "python3.8",
    "log_level": "WARNING",
    "xray_tracing": true,
    "exception_handler": "zappa_sentry.unhandled_exceptions"
  }
}

zappa
zappa-sentry

#!/bin/bash

VENV_DIRECTORY=venv
LAMBDA_PACKAGE=lambda.zip
REGION=eu-west-1
FUNCTION_NAME=loginsights2metrics

if [[ -d ${VENV_DIRECTORY} ]]; then rm -fR ${VENV_DIRECTORY}; fi
if [[ -f ${LAMBDA_PACKAGE} ]]; then rm -fR ${LAMBDA_PACKAGE}; fi

python -m venv ${VENV_DIRECTORY}
source ${VENV_DIRECTORY}/bin/activate

pip install -r requirements.txt

zappa package main -o ${LAMBDA_PACKAGE}

rm -fR ${VENV_DIRECTORY}

aws --region=${REGION} lambda update-function-code --function-name ${FUNCTION_NAME}  --zip-file "fileb://${LAMBDA_PACKAGE}"