搭建云上日志收集分析系统(三)

阅读数:603 2019 年 9 月 26 日 17:02

搭建云上日志收集分析系统(三)

背景说明

应用系统的日志收集与分析工作对运维来说至关重要。常见的系统解决方案中开源技术栈 ELK(Elastic Stack: Elasticsearch, Logstash, Kibana)是当前比较流行的选择。下面我们会讨论另一种构建于云原生设计的类似于 ELK 的解决方案 EKK(Amazon Elasticsearch Service, Amazon Kinesis, and Kibana)。

EKK 的优势在于组件是 AWS 托管服务,不必自己安装、运维,并且与 AWS 的其它服务轻松集成,可以很轻松的部署一套可靠、可扩展、安全、容错以及解耦和基于事件的解决方案。

传统的 Elasticsearch 中,日志数据的不断膨胀,对数据的生命周期管理越来越重要(应对此需求的新功能 ILM(index lifecycle management) 在 Elasticsearch 7.0 中闪亮登场)。本文不介绍 ILM,介绍另一种解决方案:使用 Lambda 配合实现数据的轮换。

使用 Lambda 在 ES 轮换数据数据(Rotate)

1. 为 Lambda 设置 IAM Role

搭建云上日志收集分析系统(三)搭建云上日志收集分析系统(三)

2. 在 Lambda 中创建个 Layer(层)

把项目需要的依赖包放到层里,方便 Lambda 的使用。

搭建云上日志收集分析系统(三)

层参考: https://docs.aws.amazon.com/zh_cn/lambda/latest/dg/configuration-layers.html

参考命令:

mkdirpythonpip3 install elasticsearch-curator -t ./python/$zip -q -r layer.zip ./python

搭建云上日志收集分析系统(三)

搭建云上日志收集分析系统(三)

3. 创建 Lambda: ES-Rotate

搭建云上日志收集分析系统(三)搭建云上日志收集分析系统(三)

设置 Lambda 运行的 Layers

保存

搭建云上日志收集分析系统(三)搭建云上日志收集分析系统(三)搭建云上日志收集分析系统(三)

搭建云上日志收集分析系统(三)

设置运行代码,点击

搭建云上日志收集分析系统(三)

在下面 Function code 区域更新代码,host 地址换成我们创建的 ES Cluster 地址

搭建云上日志收集分析系统(三)

复制代码
import json
import boto3
from requests_aws4auth import AWS4Auth
from elasticsearch import Elasticsearch, RequestsHttpConnection
import curator
host = 'search-ekk-log-vpfpqvgbxnom3ctwvz5evv2du4.cn-north-1.es.amazonaws.com.cn' # For example, search-my-domain.region.es.amazonaws.com
region = 'cn-north-1' # For example, us-west-1
service = 'es'
credentials = boto3.Session().get_credentials()
awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)
def lambda_handler(event, context):
es = Elasticsearch(
hosts = [{'host': host, 'port': 443}],
http_auth = awsauth,
use_ssl = True,
verify_certs = True,
connection_class = RequestsHttpConnection
)
index_list = curator.IndexList(es)
index_list.filter_by_age(source='name', direction='older', timestring='%Y-%m-%d', unit='days', unit_count=1)
print("Found %s indices to delete" % len(index_list.indices))
# If our filtered list contains any indices, delete them.
if index_list.indices:
curator.DeleteIndices(index_list).do_action()
# TODO implement
return {
'statusCode': 200,
'body': json.dumps('Hello from Lambda!')
}

修改 Lambda 运行时内存和超时时间

搭建云上日志收集分析系统(三)

4. 测试

定义测试用例输入参数,

搭建云上日志收集分析系统(三)

因此 Lambda 测试用例,不需输入参数,可使用默认设置,

搭建云上日志收集分析系统(三)

运行测试用例

搭建云上日志收集分析系统(三)搭建云上日志收集分析系统(三)

在 ES 中查看

搭建云上日志收集分析系统(三)

参考 https://docs.aws.amazon.com/zh_cn/elasticsearch-service/latest/developerguide/curator.html

5. 在 CloudWatch 设置定时轮换

搭建云上日志收集分析系统(三)搭建云上日志收集分析系统(三)搭建云上日志收集分析系统(三)

Load to ES from S3

1. 创建 Lambad s3-to-es-bulk-by-hour

搭建云上日志收集分析系统(三)

2. 设置 Layer

搭建云上日志收集分析系统(三)

搭建云上日志收集分析系统(三)搭建云上日志收集分析系统(三)

3. 设置 Code

搭建云上日志收集分析系统(三)搭建云上日志收集分析系统(三)

Bucket 设置为 Kinesis Firehose 中设置的 bucket

Host 设置为 ES 的 endpoint

复制代码
import boto3 import re import requests from requests_aws4auth import AWS4Auth import json from elasticsearch.helpers import bulk import time from elasticsearch import Elasticsearch, RequestsHttpConnection region = 'cn-north-1' # e.g. us-west-1 service = 'es' credentials = boto3.Session().get_credentials() awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token) bucket = 'zhnc-ekk-full-log' host = 'search-ekk-log-vpfpqvgbxnom3ctwvz5evv2du4.cn-north-1.es.amazonaws.com.cn' type = 'log' headers = { "Content-Type": "application/json" } s3 = boto3.client('s3', region_name=region) def get_all_s3_keys(bucket, prefix): keys = [] kwargs = {'Bucket': bucket, 'Prefix':prefix} while True: resp = s3.list_objects_v2(**kwargs) for obj in resp['Contents']: keys.append(obj['Key']) try: kwargs['ContinuationToken'] = resp['NextContinuationToken'] except KeyError: break return keys # Lambda execution starts here def lambda_handler(event, context): print(event) for record in event['Records']: msg = json.loads(json.dumps(eval(record['Sns']['Message']))) year=msg['year'] month=msg['month'] day=msg['day'] hour=msg['hour'] index = 'apachelog-{}-{}-{}-{}'.format(year,month,day,hour) print(index) keys = get_all_s3_keys(bucket,"apachelog/{}/{}/{}/{}".format(year,month,day,hour)) print(len(keys)); es = Elasticsearch( hosts = [{'host': host, 'port': 443}], http_auth = awsauth, use_ssl = True, verify_certs = True, connection_class = RequestsHttpConnection ) for key in keys: ACTIONS = [] obj = s3.get_object(Bucket=bucket, Key=key) body = obj['Body'].read() lines = body.splitlines() for line in lines: document = json.loads(line) action = { "_index": index, "_type": type, "_source": document } ACTIONS.append(action) success, _ = bulk(es, ACTIONS, index=index, raise_on_error=True) print('Performed %d actions' % success)

4. 设置运行内存,超时时间

搭建云上日志收集分析系统(三)

5. 测试

这个 Lambda 将被 SNS 触发,创建一个模拟 SNS 的事件,

搭建云上日志收集分析系统(三)搭建云上日志收集分析系统(三)

消息 Message 设置为将要测试的事件,Lambda 会读取对应的 S3 文件到 ES 中,

消息内容为,

复制代码
{
"Records": [
{
"EventSource": "aws:sns",
"EventVersion": "1.0",
"EventSubscriptionArn": "arn:aws-cn:sns:cn-north-1: 725362542198:s3-to-es-by-day:caf9a3b1-679c-4604-9b65-f15dca3b5b18",
"Sns": {
"Type": "Notification",
"MessageId": "71c6da65-49bf-5301-8270-8ff199faaa1b",
"TopicArn": "arn:aws-cn:sns:cn-north-1: 725362542198:s3-to-es-by-day",
"Subject": "None",
"Message": "{'year':'2019','month':'06','day':'30','hour':'02'}",
"Timestamp": "2019-07-02T03: 22: 45.020Z",
"SignatureVersion": "1",
"Signature": "ihaGN/JL8u/v57xEY1RTFekpUpgVukM9Ebj9IIM9Rr9KGkUMe6dO7hze7estD0yM9K0QRQAreQ5XiB0Tfj/jOCvyjL9IrRcTplQcWPzMHmVqd4C3942gduFkHyul2+lYa0DJZM46J/Yy7mihe9EfXUySf2Eyok4NsUC6WtnbyJPN17FG1t4fnEWpRwU2Yg+MLM4bJWr3sK5/6xRnUVerLlMm5tCsynybW6FQCYsVgl7SJLW6nBmbCe3v6jRMuKCNW8xptVyEAnII4h5uPVElts0IWhnE+EQG3FNFmOZmj8OLZutRadSrNFexRMZebmKwRZRD5dTaCoD5E6v6TTYGbQ==",
"SigningCertUrl": "https: //sns.cn-north-1.amazonaws.com.cn/SimpleNotificationService-3250158c6506d40f628c21ed8dad1787.pem",
"UnsubscribeUrl": "https://sns.cn-north-1.amazonaws.com.cn/?Action=Unsubscribe&SubscriptionArn=arn:aws-cn:sns:cn-north-1:725362542198:s3-to-es-by-day:caf9a3b1-679c-4604-9b65-f15dca3b5b18",
"MessageAttributes": {}
}
}
]
}

此事例将读取以下位置的日志文件

搭建云上日志收集分析系统(三)

点击测试

搭建云上日志收集分析系统(三)

ES Index 已经导入

搭建云上日志收集分析系统(三)

作者简介

陈朕,AWS 解决方案架构师,负责基于 AWS 云计算方案架构的咨询和设计,在国内推广 AWS 云平台技术和各种解决方案。十余年分布式应用、大数据的分布式处理经验。

本文转载自 AWS 技术博客

原文链接:

https://amazonaws-china.com/cn/blogs/china/ekk-amazon-elasticsearch-service-amazon-kinesis-and-kibana/

欲了解 AWS 的更多信息,请访问【AWS 技术专区】

评论

发布