ElasticStack：使用FileBeat、Logstash、Elasticsearch、Kibana收集清洗存储查看分析数据

2022-10-23 77

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

本文涉及的产品

日志服务 SLS，月写入数据量 50GB 1个月

Elasticsearch Serverless通用抵扣包，测试体验金 200元

简介： ElasticStack：使用FileBeat、Logstash、Elasticsearch、Kibana收集清洗存储查看分析数据

借用ElasticStack的一张图，很好的阐述了LEK的在数据处理中的位置

一、环境：

版本均未5.2.0

https://www.elastic.co/cn/downloads/past-releases

1、filebeat:

https://www.elastic.co/cn/downloads/past-releases/filebeat-5-2-0

2、logstash

https://www.elastic.co/cn/downloads/past-releases/logstash-5-2-0

3、elasticsearch:

https://www.elastic.co/cn/downloads/past-releases/elasticsearch-5-2-0

4、kibana:

https://www.elastic.co/cn/downloads/past-releases/kibana-5-2-0

二、日志准备

使用python脚本定时生成模拟日志

generator_log.py

# -*- encoding:utf-8 -*-

import time
from chinesename import ChineseName

cn = ChineseName()

while True:
    now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    message = "{} {}\n".format(now, cn.getName())
    print(message)

    with open("demo.log", "a", encoding="utf-8") as f:
        f.write(message)

    # 每3秒生成一条日志     
    time.sleep(3)

日志示例（日期姓名）：

2019-06-13 18:01:31 容休

三、filebeat

1、配置

修改配置文件filebeat.yml

可以选择直接将数据传入Elasticsearch，也可以传入Logstash处理

filebeat.prospectors:

- input_type: log
paths:
# 配置需要收集的文件地址
- /var/log/*.log

#-------------------------- Elasticsearch output ------------------------------
# output.elasticsearch:
# hosts: ["localhost:9200"]

#----------------------------- Logstash output --------------------------------
output.logstash:
hosts: ["localhost:5044"]

2、启动：

./filebeat -e -c filebeat.yml -d "publish"

参考:开始使用Filebeat

四、logstash

1、匹配说明

（1）内置匹配

%{SYNTAX:SEMANTIC}

（2）ruby正则

(?<name>pattern)

关于Ruby的正则：

Ruby 正则表达式： https://www.runoob.com/ruby/ruby-regular-expressions.html

Ruby 正则匹配测试： https://rubular.com/

2、配置

新建一个文件夹存放自定义匹配模式

$ mkdir ./patterns
$ cat ./patterns/datetime.re
DATETIME \d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}

es-pipeline.conf

input {
    beats {
        port => "5044"
    }
}
filter {
    grok {
        patterns_dir => ["./patterns"]
        match => { 
            "message" => "%{DATETIME:logdate} (?<text>(.*))"
        }
        remove_field  => "message"
   }
   date {
        match => ["logdate", "yyyy-MM-dd HH:mm:ss"]
    }
}
output {
    stdout { codec => rubydebug }
    elasticsearch {
        hosts => [ "localhost:9200" ]
    }
}

3、启动logstash

# 解析配置文件并报告任何错误
$ ./bin/logstash -f es-pipeline.conf --config.test_and_exit

# 启用自动配置加载
$ ./bin/logstash -f es-pipeline.conf --config.reload.automatic