Lambda CloudWatch Logs Abonnement

AWS CloudWatch Logs (CW) ist ein Service, der unter anderem Protokolldateien von AWS-Services an einem zentralen Ort speichert. CW unterstützt Abonnements, mit denen Protokollereignisse von CloudWatch Logs gesendet und an andere Dienste wie einen Amazon Kinesis Stream, Amazon Kinesis Data Firehose Stream oder AWS Lambda zur benutzerdefinierten Verarbeitung, Analyse oder zum Laden in andere Systeme übermittelt werden. Um mit dem Abonnieren von Protokollereignissen zu beginnen, erstellen Sie die Empfangsquelle, z. B. AWS Lambda, an die die Ereignisse geliefert werden sollen. Ein Abonnementfilter definiert das Filtermuster, mit dem gefiltert wird, welche Protokollereignisse an Ihre AWS-Ressource geliefert werden, sowie Informationen darüber, wohin passende Protokollereignisse gesendet werden sollen. In diesem Blog werden wir ein Lambda-Abonnement für ein CloudWatch-Protokoll einrichten.

Architektur

Wir werden die folgende Anwendung erstellen. Eine CloudWatch Event Rule plant ein Ereignis für TriggerFunction. TriggerFunction schreibt eine Protokollnachricht in die CloudWatch-Protokolle. CloudWatchSubscriptionLambda ist auf das Protokoll von TriggerFunction abonniert. Wenn es einen Protokolleintrag gibt, wird CloudWatchSubscriptionLambda mit der Protokollzeile aufgerufen. CloudWatchSubscriptionLambda schreibt das empfangene Ereignis in sein Protokoll.

Geplante Ereignisse

Um Ereignisse für einen Lambda zu planen, benötigen wir die folgende Konfiguration:

  LambdaBasicExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Statement:
        - Effect: Allow
          Principal:
            Service: lambda.amazonaws.com
          Action: sts:AssumeRole
          Condition: {}
      Path: /
      ManagedPolicyArns:
      - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

  TriggerFunction:
    Type: AWS::Lambda::Function
    Properties:
      Handler: index.handler
      Runtime: python3.6
      Role: !GetAtt 'LambdaBasicExecutionRole.Arn'
      MemorySize: 128
      Timeout: 30
      Code:
        ZipFile: |-
          def handler(event, ctx):
              print(event)

  TriggerFunctionLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub '/aws/lambda/${TriggerFunction}'
      RetentionInDays: 30

  CloudWatchEventsRule:
    Type: AWS::Events::Rule
    Properties:
      ScheduleExpression: rate(1 minute)
      State: ENABLED
      Targets:
      - Arn: !GetAtt TriggerFunction.Arn
        Id: scheduled-event

  InvokeTriggerFunctionPermission:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !GetAtt TriggerFunction.Arn
      Action: lambda:InvokeFunction
      Principal: events.amazonaws.com
      SourceArn: !GetAtt CloudWatchEventsRule.Arn

CloudWatch Abonnement

Für das CloudWatch Lambda-Abonnement benötigen wir die folgende Konfiguration:

  CloudWatchSubscriptionLambda:
    Type: AWS::Lambda::Function
    Properties:
      Handler: index.handler
      Runtime: python3.6
      Role: !GetAtt 'LambdaBasicExecutionRole.Arn'
      MemorySize: 128
      Timeout: 30
      Code:
        ZipFile: |-
          def handler(event, ctx):
              print(event)

  CloudWatchSubscriptionLambdaLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub '/aws/lambda/${CloudWatchSubscriptionLambda}'
      RetentionInDays: 30

  CloudWatchLogSubscription:
    Type: AWS::Logs::SubscriptionFilter
    DependsOn: CloudWatchSubscriptionFunctionPermission
    Properties:
      DestinationArn: !GetAtt CloudWatchSubscriptionLambda.Arn
      FilterPattern: ''
      LogGroupName: !Ref TriggerFunctionLogGroup

  CloudWatchSubscriptionFunctionPermission:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !GetAtt CloudWatchSubscriptionLambda.Arn
      Action: 'lambda:InvokeFunction'
      Principal: 'logs.eu-west-1.amazonaws.com'
      SourceArn: !GetAtt TriggerFunctionLogGroup.Arn

Rohes Protokoll Ereignis

Das Protokollereignis, das vom CloudWatchSubscriptionLambda empfangen wird, hat das folgende Format. Der Protokollsatz ist base64-kodiert und mit gzip komprimiert.

{"awslogs": {"data": "H4sIAAAAAAAAAIVRy27bMBD8FYPoMYSWFJ+6qa1tBIlbwFKbQ2oUlEQpBGRJpaS6RZB/z7oPoLdiL8vhYnZ25pmc/Ty7zpc/J08y8j4v86+HbVHk+y25IeNl8BFhxbgwqeVCaUC4H7t9HNcJfxJ3mZPenavGJRXiFN/U925eQj17F+sn2obon8bZUzfRMoau83G3DvUSxoGyWzgc3vLy0/HD3fE3c7FE785IzYGZhLGE2eTxzX1ebovyJHiTShCqraARjXOWs1ZVDUArGu10ihTzWs11DNOVfxf6xceZZI/k/+IC7S7U/3Dnqfe07se1ubgFR95d24dre4/q/mGnD9zsgB0+3u13lpx+qd9+98NyXfhMQoNHpEIA56A5CCxpjbESpJESLbUcuGFKM6YYgAapU2VNiqU5HrIEzGZBOSRjUnCFCfDUGnbzNzOkL8r8WG6O/tuKo7dNtmGmNRWTmvqaA2XMG1q1WlJoK6vrSijg9eYzeoL6s80fW78M5OX08grvTTW4DAIAAA=="}}

Verarbeitung von Protokollzeilen

Das rohe Protokollereignis sollte mit einem Lambda verarbeitet werden:

import gzip
import json
from base64 import b64decode

def decompress(data) -> bytes:
    return gzip.decompress(data)

def decode_record(data: dict) -> dict:
    x = decompress(b64decode(data['data']))
    return json.loads(x.decode('utf8'))

def decode_event(event: dict) -> dict:
    return decode_record(event['awslogs'])

def handler(event, ctx) -> None:
    print(json.dumps(decode_event(event)))

Die Protokollzeile nach der Verarbeitung lautet wie folgt:

{
    "messageType": "DATA_MESSAGE",
    "owner": "612483924670",
    "logGroup": "/aws/lambda/blog-aws-elasticsearch-firehose-ap-TriggerFunction-1I0MMB2TURNKR",
    "logStream": "2018/11/19/[$LATEST]42d35046fb0d4daa921f6bd00f4d7a73",
    "subscriptionFilters": [
        "blog-aws-elasticsearch-firehose-api-gw-example-cloudwatch-CloudWatchLogSubscription-W28F01MOKGF9"
    ],
    "logEvents": [
        {
            "id": "34402224585553617062605662326569493632889831387406598144",
            "timestamp": 1542649103395,
            "message": "{'version': '0', 'id': '8d73f860-d1d1-d867-bb38-0093a2e3790d', 'detail-type': 'Scheduled Event', 'source': 'aws.events', 'account': '612483924670', 'time': '2018-11-19T17:37:38Z', 'region': 'eu-west-1', 'resources': ['arn:aws:events:eu-west-1:612483924670:rule/blog-aws-elasticsearch-fireho-CloudWatchEventsRule-127RVH0C949OJ'], 'detail': {}}n"
        },
        {
            "id": "34402224585575917807804192949711029351162479748912578561",
            "timestamp": 1542649103396,
            "message": "END RequestId: e9a9b88a-ec21-11e8-bc46-871711881058n"
        },
        {
            "id": "34402224585575917807804192949711029351162479748912578562",
            "timestamp": 1542649103396,
            "message": "REPORT RequestId: e9a9b88a-ec21-11e8-bc46-871711881058tDuration: 1.19 mstBilled Duration: 100 ms tMemory Size: 128 MBtMax Memory Used: 21 MBtn"
        }
    ]
}

Beispiel

Das Beispielprojekt zeigt, wie Sie das Projekt einrichten. Das Beispiel kann mit make create bereitgestellt und mit make delete entfernt werden.

Beispiel mit privatem Netzwerk

Um auszuprobieren, ob CloudWatch Zugriff auf die Lambdas hat, wenn diese über eine Netzwerkverbindung mit einem privaten Subnetz verfügen, habe ich das folgende Beispielprojekt erstellt, das eine VPC einsetzt und die Lambdas in das private Subnetz platziert. Erfreulicherweise kann CloudWatch immer noch Protokolle von einem Lambda empfangen. Außerdem kann CloudWatch den Lambda, der ein Protokoll abonniert hat, immer noch auslösen, selbst wenn der Lambda über eine private Netzwerkverbindung verfügt.

Fazit

AWS CloudWatch Logs unterstützt die Protokollverarbeitung in Echtzeit, indem es Abonnements anbietet. Unterstützte Ziele sind Kinesis, Firehose, SNS und Lambda. In diesem Blog richten wir ein Lambda als Protokollabonnement ein, das ein Lambda dekodiert und die Ausgabe in CloudWatch-Protokolle schreibt. Wir haben auch ein Lambda konfiguriert, das auf der Grundlage eines Zeitplans mit CloudWatch-Ereignissen ausgelöst wird.