ELK技术-Logstash

 1.背景

1.1 简介

Logstash 是一个功能强大的工具，可与各种部署集成。 它提供了大量插件，可帮助业务做解析，丰富，转换和缓冲来自各种来源的数据。

Logstash 是一个数据流引擎

• 它是用于数据物流的开源流式 ETL（Extract-Transform-Load）引擎

• 在几分钟内建立数据流管道

• 具有水平可扩展及韧性且具有自适应缓冲

• 不可知的数据源

• 具有 200 多个集成和处理器的插件生态系统

• 使用 Elastic Stack 监视和管理部署

Logstash 几乎可以摄入各种类别的数据

它可以摄入日志，文件，指标或者网路真实数据。经过 Logstash 的处理，变为可以使用的 Web Apps 可以消耗的数据，也可以存储于数据中心，或变为其它的流式数据。

Logstash 相关概念

• Logstash 实例是一个正在运行的 Logstash 进程。建议在 Elasticsearch 的单独主机上运行 Logstash，以确保两个组件有足够的计算资源可用。

• 管道（pipeline）是配置为处理给定工作负载的插件集合。一个 Logstash 实例可以运行多个管道。（彼此独立）

• 输入插件（input plugins）用于从给定的源系统中提取或接收数据。 Logstash 参考指南中提供了支持的输入插件列表：https://www.elastic.co/guide/en/logstash/current/input-plugins.html

• 过滤器插件（filter plugin）用于对传入事件应用转换和丰富。 Logstash 参考指南中提供了支持的过滤器插件列表：Filter plugins | Logstash Reference [8.3] | Elastic

• 输出插件（output plugin）用于将数据加载或发送到给定的目标系统。 Logstash 参考指南中提供了支持的输出插件列表：https://www.elastic.co/guide/en/logstash/current/output-plugins.html

Logstash 包含3个主要部分： 输入（inputs），过滤器（filters）和输出（outputs）。 你必须定义这些过程的配置才能使用 Logstash，尽管不是每一个都必须的。在有些情况下，可以甚至没有过滤器。在过滤器的部分，它可以对数据源的数据进行分析，丰富，处理等。

1.2 学习参考

• Logstash官方文档：《Logstash官方文档》

• 中国社区官方博客：《Logstash入门教程》

• 其他参考技术博客：《通过Logstash实现mysql数据定时增量同步到ES》

• Logstash解析：《解析插件-Grok》

1.3 本例测试版本

[root@dev1613 study]# sudo -u logstash ../bin/logstash  --versionUsing bundled JDK: /opt/logstash/jdklogstash 7.12.1

2.功能应用

2.1 基础测试

输入测试命令，../bin为当前执行命令所在文件夹，与logstash安装后bin的相对目录位置。

sudo -u logstash ../bin/logstash -e 'input { stdin { } } output { stdout {} }'

执行命令后，输出结果如图：

2.2 Logstash解析日志文件

最原始的 Log 数据，经过 Logstash 的处理，可以把非结构化的数据变成结构化的数据。甚至可以使用 Logstash 强大的 Filter 来对数据继续加工。最终将加工后的数据存储下来，用于分析和搜索。

日志原始内容

2022-07-06 18:48:37.453 ERROR 14677 --- [ dispatcher 108] c.a.c.s.dashboard.metric.MetricFetcher   : Failed to fetch metric from <http://10.32.4.230:8719/metric?startTime=1657104506000&endTime=1657104512000&refetch=false>: socket timeout2022-07-06 18:48:44.439 ERROR 14677 --- [ dispatcher 109] c.a.c.s.dashboard.metric.MetricFetcher   : Failed to fetch metric from <http://10.32.4.230:8719/metric?startTime=1657104513000&endTime=1657104519000&refetch=false>: socket timeout2022-07-06 18:48:51.514 ERROR 14677 --- [ dispatcher 110] c.a.c.s.dashboard.metric.MetricFetcher   : Failed to fetch metric from <http://10.32.4.230:8719/metric?startTime=1657104520000&endTime=1657104526000&refetch=false>: socket timeout

Logstash配置文件

编写日志解析配置文件，并解析时间，错误级别，错误行，错误信息。提取出来变为结构化数据。编写配置文件如下：

配置相关节点参考官方文档：《plugins-inputs-file》

input {    file {        path => "/opt/logstash/study/outlog.log"        start_position => "beginning"        stat_interval => "3"        type => "sentinel-log"            }}filter {      grok {        match => ["message","%{TIMESTAMP_ISO8601:datetime} %{LOGLEVEL:loglevel} %{NUMBER:textid} %{GREEDYDATA:errormsg}"]    }    json {        source => "request"    }}output {    stdout { codec => rubydebug }}

Grok日志解析在线测试

基于elastic在线网页，可编写解析日志测试demo。

日志解析结构化输出

运行命令：sudo -u logstash ../bin/logstash -f study-file-es.conf

运行logstash加载配置文件命令，启动测试输出结构化内容如下：

2.3 Logstash-数据库同步

本例将MySql数据表中的数据，基于修改时间同步到es数据存储中心。

基础数据内容

数据源-mysql数据表建表语句：

CREATE TABLE `study_logstash_es` (  `id` int NOT NULL AUTO_INCREMENT COMMENT '自增主键',  `study_code` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL COMMENT '编码',  `study_name` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL COMMENT '名称',  `study_tag` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL DEFAULT '' COMMENT '标签',  `study_level` smallint NOT NULL DEFAULT '0' COMMENT '等级,如1,2,3',  `is_delete` tinyint unsigned NOT NULL DEFAULT '0' COMMENT '0 未删除  1 删除',  `create_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',  `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',  `operate_user` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '操作人',  PRIMARY KEY (`id`),  UNIQUE KEY `uniq_study_code` (`study_code`)) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='学习-logstash同步msql数据到es';

目标源-es索引创建脚本：

PUT /study_logstash_es{  "settings": {    "index": {      "number_of_shards": 1,      "number_of_replicas": 1     }  },  "mappings": {      "properties": {        "id": {          "type": "integer"          },         "study_code": {          "type": "text"          },        "study_name": {          "type": "text"          },         "operate_user": {          "type": "text"          },         "study_tag": {          "type": "keyword"          },        "is_delete": {          "type": "integer"          },         "study_level": {          "type": "integer"          },         "mark_time": {          "type": "date",          "format": "epoch_millis"        },        "update_time": {          "type": "date"        }      }    }}

Logstash配置文件

本例测试的数据库地址，es地址，已经基于xxx脱敏。更多jdbc的配置，请参考官方文档：《plugins-inputs-jdbc》。

jdbc_driver_library：为mysql连接包，可在Maven上下载，下载地址参考：《mysql-connector-java.jar 包下载》。

input {  jdbc {    jdbc_driver_library => "/opt/logstash/study/mysql-connector-java-8.0.30.jar"    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"    jdbc_connection_string => "jdbc:mysql://xxx.xxx.xx.x:3306/study_database?serverTimezone=Asia/Shanghai&allowMultiQueries=true&characterEncoding=utf-8"    jdbc_user => "root"    jdbc_password => "xxxxx"    jdbc_paging_enabled => true    jdbc_page_size => "2"    use_column_value => true    tracking_column => "mark_time"    tracking_column_type => "numeric"    schedule => "* * * * *"    statement => "SELECT id,study_code,study_name,study_tag,study_level,operate_user,update_time,UNIX_TIMESTAMP(update_time) as mark_time from study_logstash_es where UNIX_TIMESTAMP(update_time)>:sql_last_value AND update_time < NOW()"  }}output{     elasticsearch{         hosts => ["xxx.xxx.16.4:9200","xxx.xxx.16.xx:9200","192.xxx.xx.xx:9200"]         index => "study_logstash_es"         timeout => 300         user => "xxx"         password => "xxxxx"     }}

数据同步es

运行命令：sudo -u logstash ../bin/logstash -f study-mysql-es.conf

运行logstash加载配置文件命令，启动运行日志，es同步的数据如下：

es数据查询如下：

2.4 Logstash-kafka消息同步

Logstash的输入项可以监听kafka消息，消费消息记录。

input {    kafka {        bootstrap_servers => "xxx.xxx.xx.4:9092,xxx.xxx.16.4:9093,xxx.xxx.16.4:9094" #kafka服务器地址        topics => "xxxlog"        # batch_size => 5        codec => "json"        group_id => "logstash"        consumer_threads => 3    }}filter {    # 丢弃所有的header请求   if [request][method] == "HEAD" {           drop { }    }    # 因为[request][querystring]这个玩意中的字段类型可能不一样，所以全部干成字符串   ruby {    code => "event.set('[request][querystring]', event.get('[request][querystring]').to_s) if event.get('[request][querystring]')"   }    if [request][uri] =~ "^/ucenter-admin-view/v3(.*)" {        mutate {            add_field => { "log_source" => "用户中心管理后台" }            add_field => { "log_source_id" => "1" }        }    }    else if [request][uri] =~ "^/ucenter-org-view/v3/(.*)" {        mutate {            add_field => { "log_source" => "用户中心工作台" }            add_field => { "log_source_id" => "2" }        }    }    else if [request][uri] =~ "^/safety-admin-api(.*)" {        mutate {                add_field => { "log_source" => "安全管理平台" }                add_field => { "log_source_id" => "3" }            }    }    else{        mutate {            add_field => { "log_source" => "其他" }            add_field => { "log_source_id" => "0" }        }    }     grok {        match => { "[request][uri]" => "%{URIPATH:[request][path]}" }        named_captures_only => false    } }output{#    stdout { #   codec => json  # }     elasticsearch{         hosts => ["xxx.xxx.xx.4:9200","xxx.xxx.16.13:9200","xxx.xxx.16.14:9200"]         index => "apisixlog"         timeout => 300         user => "elastic"         password => "HApn2xCJMuRlg0UOIV0P"     }

3.总结

1. Logstash基于 输入（inputs），过滤器（filters）和输出（outputs）可以方便快捷的处理数据，将一些非结构化数据，处理为结构化数据。Logstash支持数据中转，数据同步等场景的应用。本例只是简要测试，在实际业务使用时，可基于某一个输入插件/输出插件参考官方文档，结合项目使用。

2. 在做一些数据同步工作时，出于性能等各方面考虑，如同步数据表到es中，除了Logstash这种方案，也可以参考其他的方案，如alibaba/DataX。

3. Logstash收集大量日志时，存在耗内存的情况，建议参考官方推荐的FileBeat模式。详情参考文档：《开源日志管理方案 ELK 和 EFK 的区别》,《通过Filebeat把日志传入到Elasticsearch》。

4. Logstash在配置文件调整后，启动命令，可能出现如下报错：

删除掉Logstash/data文件下的缓存文件，即可重新启动成功。

5. Logstash启动命名常用如下：

sudo -u logstash ../bin/logstash -f study-file-es.conf  表示当前窗口启动，关闭或退出命令行时，logstash实例关闭。sudo -u logstash ../bin/logstash -f study-file-es.conf --config.reload.automatic表示当前窗口启动，配置文件变化时，不用重新启动实例，可自动加载。关闭或退出命令行时，logstash实例关闭。sudo -u logstash ../bin/logstash -f study-mysql-es.conf & test.out --config.reload.automatic表示后台启动，关闭退出命令，实例在后台一直运行。ps -ef|grep logstash  kill-9 进程号， 关闭对应的实例

6. Logstash运行日志查看

查看cat logstash-plain.log 文件，可查看Logstash运行日志记录。