Blog
Analysieren von CloudFront mit AWS Elasticsearch Service
In meinem Blog AWS Elasticsearch Service with Firehose Delivery Stream und Analyzing API Gateway Access Logs with AWS Elasticsearch Service haben wir gesehen, wie einfach es ist, einen Elasticsearch-Cluster einzurichten, Daten aus Firehose aufzunehmen und Dashboards in Kibana zu erstellen. Dieses Mal werden wir Daten von AWS CloudFront CDN aufnehmen. CloudFront speichert seine Daten in einem S3-Bucket. Wir werden CloudFront simulieren, indem wir eine CloudFront-Protokolldatei in den Bucket hochladen. Mithilfe von
CloudFront Log Format
CloudFront schreibt Protokollzeilen in S3. Das Protokoll beginnt mit zwei Zeilen, die mit einer Raute beginnen und das Schema und die Version der Protokolldatei definieren. Ein Beispiel ist unten abgebildet.
CloudFront-Protokoll:
#Version: 1.0
#Fields: date time x-edge-location sc-bytes c-ip cs-method cs(Host) cs-uri-stem sc-status cs(Referer) cs(User-Agent) cs-uri-query cs(Cookie) x-edge-result-type x-edge-request-id x-host-header cs-protocol cs-bytes time-taken x-forwarded-for ssl-protocol ssl-cipher x-edge-response-result-type cs-protocol-version fle-status fle-encrypted-fields
2018-11-10 00:16:26 DEN50-C1 887 65.154.226.109 GET d313bttlbsfoy.cloudfront.net / 403 - Mozilla/4.0%2520(compatible;%2520MSIE%25208.0;%2520Windows%2520NT%25206.1;%2520WOW64;%2520Trident/4.0;%2520SLCC2;%2520.NET%2520CLR%25202.0.50727;%2520.NET%2520CLR%25203.5.30729;%2520.NET%2520CLR%25203.0.30729;%2520Media%2520Center%2520PC%25206.0;%2520.NET4.0C;%2520.NET4.0E;%2520InfoPath.2) - - Error OuhTpC1iLajIPb2jeR9_rUEFEyboP9LI2YECBzkUrbefHicnvFIM5A== www.example.com http 479 0.001 - - - Error HTTP/1.1 - -
Die Protokollzeilen werden im tabulatorgetrennten Format geschrieben. NotificationLambda parst die Protokollzeilen, erstellt eine JSON-Nachricht und veröffentlicht die Protokollzeilen in Stapeln von 500 Zeilen an Firehose. Firehose liefert die Protokollzeilen an den Elasticsearch Service, damit sie indiziert werden.
NotificationLambda:
import boto3
import json
import os
s3 = boto3.resource('s3')
firehose = boto3.client('firehose')
delivery_stream_name = os.environ['DELIVERY_STREAM_NAME']
def process_line(line: str):
record = line.strip('n').split('t')
return {
'date': record[0],
'time': record[1],
'x-edge-location': record[2],
'sc-bytes': record[3],
'c-ip': record[4],
'cs-method': record[5],
'cs-host': record[6],
'cs-uri-stem': record[7],
'sc-status': record[8],
'cs-referer': record[9],
'cs-user-agent': record[10],
'cs-uri-query': record[11],
'cs-cookie': record[12],
'x-edge-result-type': record[13],
'x-edge-request-id': record[14],
'x-host-header': record[15],
'cs-protocol': record[16],
'cs-bytes': record[17],
'time-taken': record[18],
'x-forwarded-for': record[19],
'ssl-protocol': record[20],
'ssl-cipher': record[21],
'x-edge-response-result-type': record[22],
'cs-protocol-version': record[23],
'fle-status': record[24],
'fle-encrypted-fields': record[25]
}
def filter_lines(lines: [str]) -> [str]:
return list(filter(lambda line: not line.startswith('#'), lines))
def publish(batch: [dict]) -> None:
assert len(batch) <= 500
data = []
for log in batch:
encoded = bytes(json.dumps(log), 'utf-8')
data.append({'Data': encoded})
print(f'Publishing to {delivery_stream_name} data: {data}')
firehose.put_record_batch(DeliveryStreamName=delivery_stream_name, Records=data)
def handler(event, ctx):
counter = 0
batch: [dict] = []
bucket_event = event['Records'][0]
if bucket_event['eventName'] == 'ObjectCreated:Put':
bucket = bucket_event['s3']['bucket']['name']
key = bucket_event['s3']['object']['key']
obj = s3.Object(bucket, key)
for line in obj.get()['Body'].read().splitlines():
text: str = line.decode('utf-8')
if not text.startswith('#'):
if counter < 500:
batch.append(process_line(text))
counter += 1
else:
publish(batch)
batch = []
counter = 0
if counter is not 0:
publish(batch)
Beispiel
Das Beispielprojekt zeigt, wie Sie ein Projekt konfigurieren, um einen elasticsearch-Cluster zu erstellen und API CloudFront-Protokolle aufzunehmen. Das Beispiel kann mit make merge-lambda && make deploy bereitgestellt und mit make delete entfernt werden.
Laden Sie das CloudFront-Protokoll hoch.
Um ein CloudFront-Beispielprotokoll auf s3 hochzuladen, benötigen Sie den Namen des Log-Buckets. Geben Sie
aws s3 cp logs/cf-logs.txt s3://<name of bucket>
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, laden Sie das CloudFront-Beispielprotokoll auf s3 hoch.
Versuchen Sie, eine Visualisierung und Dashboards mit den Ihnen zur Verfügung stehenden Zugriffsprotokollen zu erstellen. Wenn Sie meinen früheren 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 von CloudFront aufgenommen und ein Dashboard erstellt. Elasticsearch eignet sich perfekt 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.
Verfasst von
Dennis Vriend
Unsere Ideen
Weitere Blogs
Contact