Analysieren von API-Gateway-Zugriffsprotokollen mit AWS Elasticsearch Service

In meinem Blog AWS Elasticsearch Service with Firehose Delivery Stream haben wir gesehen, wie einfach es ist, einen Elasticsearch-Cluster einzurichten, Daten aus Firehose aufzunehmen und Dashboards in Kibana zu erstellen. Dieses Mal verwenden wir die gleiche Architektur, um Zugriffsprotokolle von AWS ApiGateway aufzunehmen und die Daten in Kibana zu analysieren.

API-Gateway-Konto

In jedem AWS-Konto gibt es einen einzigen API Gateway (APIGW) Service. APIGW kann mehrere RestApi-Instanzen hosten. Jede RestApi-Instanz enthält mehrere Stufen wie dev, test oder prod. Um die Protokollierung vorzubereiten, muss die einzelne APIGW Service-Instanz über die Berechtigung zum Zugriff auf CloudWatch-Protokolle verfügen. Die Konfiguration ist einfach. Erstellen Sie eine Rolle, die der APIGW-Service übernehmen kann, und erstellen Sie ein 'AWS::ApiGateway::Konto', das auf die Rolle verweist.

  CloudWatchLogRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
        - Effect: Allow
          Principal:
            Service:
            - apigateway.amazonaws.com
          Action: sts:AssumeRole
      Path: /
      ManagedPolicyArns:
      - arn:aws:iam::aws:policy/service-role/AmazonAPIGatewayPushToCloudWatchLogs

  Account:
    Type: AWS::ApiGateway::Account
    Properties:
      CloudWatchRoleArn: !GetAtt 'CloudWatchLogRole.Arn'

API-Gateway-Bereitstellung

Eine RestApi-Instanz enthält ein Deployment, in dem u.a. die Konfiguration der Protokollierung und der Ablaufverfolgung festgelegt ist. Um Zugriffsprotokolle zu konfigurieren, müssen wir die AccessLogs-Einstellungen konfigurieren und ein Protokollzeilenformat definieren. Die RestApi wird dieses Format verwenden, um Zugriffsprotokollzeilen in CloudWatch-Protokollen zu erstellen. Es stehen mehrere Standardformate zur Auswahl, aber Sie können auch ein Standardformat wie das unten stehende JSON angeben.

  ApiGatewayDeployment:
    Type: AWS::ApiGateway::Deployment
    Properties:
      RestApiId: !Ref RestAPIv1
      StageName: dev
      Stagecontent:
        DataTraceEnabled: true
        LoggingLevel: INFO
        MetricsEnabled: true
        TracingEnabled: true
        MethodSettings:
        - LoggingLevel: INFO
          ResourcePath: /*
          HttpMethod: '*'
        AccessLogSetting:
          DestinationArn: !GetAtt 'CloudWatchAccessLogGroup.Arn'
          Format: >-
            {
            "requestId":"$context.requestId",
            "ip": "$context.identity.sourceIp",
            "caller":"$context.identity.caller",
            "user":"$context.identity.user",
            "requestTime":"$context.requestTime",
            "httpMethod":"$context.httpMethod",
            "resourcePath":"$context.resourcePath",
            "status":"$context.status",
            "protocol":"$context.protocol",
            "responseLength":"$context.responseLength"
            }

Wenn das Gateway durch Eingabe von make hello oder make error aufgerufen wird, wird die RestApi aufgerufen und die folgenden Zugriffsprotokolle werden angezeigt:

Fehler:

{
    "requestId": "3b864b4b-ea50-11e8-b859-9bd3a7d2b23c",
    "ip": "217.19.26.243",
    "caller": "-",
    "user": "-",
    "requestTime": "17/Nov/2018:10:04:55 +0000",
    "httpMethod": "GET",
    "resourcePath": "/error",
    "status": "502",
    "protocol": "HTTP/1.1",
    "responseLength": "36"
}

Hallo:

{
    "requestId": "3a71f4fa-ea50-11e8-b4f5-8116b21e9431",
    "ip": "217.19.26.243",
    "caller": "-",
    "user": "-",
    "requestTime": "17/Nov/2018:10:04:53 +0000",
    "httpMethod": "GET",
    "resourcePath": "/hello",
    "status": "200",
    "protocol": "HTTP/1.1",
    "responseLength": "14"
}

CloudWatch Logs Abonnement-Filter

Ein CloudWatch Logs Subscription Filter (CLSF) sendet Protokollereignisse an den Kinesis Stream, den Kinesis Data Firehose Delivery Stream oder die Lambda-Funktion. Um auf diese Ressourcen zugreifen zu können, muss der CLSF über Berechtigungen verfügen. Die Konfiguration ist einfach. Erstellen Sie eine Rolle, die CloudWatch übernehmen kann, und konfigurieren Sie diese Rolle für die Ressource 'AWS::Logs::SubscriptionFilter'. Der Filter ist leer, so dass alle Protokollzeilen verarbeitet werden.

 CloudWatchLogSubscriptionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
        - Effect: Allow
          Principal:
            Service:
            - logs.eu-west-1.amazonaws.com
          Action: sts:AssumeRole
      Path: /
      Policies:
      - PolicyName: Allow
        PolicyDocument:
          Statement:
          - Effect: Allow
            Action:
            - firehose:*
            Resource:
            - '*'
      ManagedPolicyArns:
      - arn:aws:iam::aws:policy/service-role/AmazonAPIGatewayPushToCloudWatchLogs

  CloudWatchLogSubscription:
    Type: AWS::Logs::SubscriptionFilter
    Properties:
      DestinationArn: !GetAtt Deliverystream.Arn
      FilterPattern: ''
      LogGroupName: !Ref CloudWatchAccessLogGroup
      RoleArn: !GetAtt CloudWatchLogSubscriptionRole.Arn

Dekomprimieren, Dekodieren und Auswählen

Wenn wir uns ein empfangenes CloudWatch-Ereignis ansehen, sehen wir, dass das Feld logEvents einen einzigen Eintrag enthält, nämlich das Zugriffsprotokoll. Das Zugriffsprotokoll ist in einem JSON-Feld als String gespeichert, so dass wir das Feld in JSON dekodieren müssen. Nach der Dekodierung des Nachrichtenfeldes wählen wir nur das erste Element des logEvents-Feldes aus, nämlich das Zugriffsprotokoll.

{
   "messageType": "DATA_MESSAGE",
   "owner": "612483924670",
   "logGroup": "api-gateway-access-logs-dev",
   "logStream": "8d5e957f297893487bd98fa830fa6413",
   "subscriptionFilters": [
       "blog-aws-elasticsearch-firehose-api-gw-example-elasticsearch-CloudWatchLogSubscription-95E57LT45IU0"
   ],
   "logEvents": [
       {
           "id": "34397996112421660129451470773032598202872829922794405888",
           "timestamp": 1542459492102,
           "message": "{ "requestId":"70939afd-ea68-11e8-add9-17a2f8edb26e", "ip": "217.19.26.243", "caller":"-", "user":"-", "requestTime":"17/Nov/2018:12:58:12 +0000", "httpMethod":"GET", "resourcePath":"/error", "status":"502", "protocol":"HTTP/1.1", "responseLength":"36" }"
       }
   ]
}

Feuerschlauch-Prozessor

Um Protokollzeilen korrekt zu indizieren, müssen wir die Protokollzeilen nachbearbeiten, bevor sie in Elasticsearch veröffentlicht werden. Firehose unterstützt die Verarbeitung von Nachrichten vor der Auslieferung mit Hilfe von AWS Lambda. CloudWatch-Ereignisse veröffentlichen die Protokollzeilen im Gzip-Format in Elasticsearch. Das Cloudwatch-Ereignis enthält auch Informationen, an denen wir nicht interessiert sind. Wir sind nur an den Feldern der Zugriffsprotokolle interessiert. Wir verwenden den folgenden Lambda, um die Protokolle zu verarbeiten:

Prozessor:

from base64 import b64encode, b64decode
import json
import gzip

def decompress(data):
    return gzip.decompress(data)

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

def handler(event, context):
    records = event['records']
    for record in records:
        record.pop('approximateArrivalTimestamp', None)
        decoded = decode_record(record)
        if decoded['messageType'] == "DATA_MESSAGE":
            event = decoded['logEvents'][0]
            event.update({'message': json.loads(event['message'])})
            msg = b64encode(bytes(json.dumps(event), 'utf-8')).decode('ascii')
            record.update({'data': msg})
            record.update({'result': 'Ok'}) # Ok, Dropped, ProcessingFailed
        else:
            record.update({'result': 'Dropped'}) # Ok, Dropped, ProcessingFailed

    return {'records': records}

Beispiel

Das Beispielprojekt zeigt, wie Sie ein Projekt konfigurieren, um einen elasticsearch-Cluster zu erstellen und API Gateway-Zugriffsprotokolle aufzunehmen. Das Beispiel kann mit bereitgestellt und mit entfernt werden. Um den Zugriff auf den API Gateway zu veröffentlichen, geben Sie make hello und make error ein, um einige Einträge in den Zugriffsprotokollen zu erhalten.

Kibana

Melden Sie sich bei der 'AWS Console' an, dann beim 'Elasticsearch Service Dashboard' und klicken Sie auf die Kibana URL. Sobald Sie eingeloggt sind, klicken Sie auf "discover" und erstellen ein neues Indexmuster mit dem Namen example-*. Klicken Sie ein weiteres Mal auf 'entdecken' und Sie sollten Daten sehen. Falls nicht, geben Sie und ein paar Mal in die Konsole ein, damit die Daten in ES verfügbar sind. Um nach Daten zu suchen, geben Sie oder in die Suchleiste ein.
Versuchen Sie, eine Visualisierung und Dashboards mit den Ihnen zur Verfügung stehenden Zugriffsprotokollen zu erstellen. Wenn Sie meinen vorherigen Blog-Beitrag über AWS Elasticsearch Service mit Firehose Delivery Stream gelesen haben, sollte das nicht schwierig sein

Fazit

In diesem Beispiel haben wir einen Elasticsearch-Dienst implementiert, Zugriffsprotokolle aufgenommen und ein Dashboard erstellt. Elasticsearch eignet sich hervorragend für die Analyse von Zugriffsprotokollen, um in Echtzeit Informationen über die Leistung einer API zu erhalten. Dashboards liefern aggregierte Daten und bieten Einblicke, die für Änderungen an der Plattform genutzt werden können. Nächstes Mal werden wir uns mit dem Ingesting von CloudTrail-Protokollen befassen und Einblicke erhalten, wer was im AWS-Konto tut.