当前位置： 首页 > news >正文

ELK整合实战，filebeat和logstash采集SpringBoot项目日志发送至ES

news 2025/6/28 13:31:45

文章目录

    • ELK整合实战
      • 使用FileBeats将日志发送到Logstash
      • 配置Logstash接收FileBeat收集的数据并打印
      • Logstash输出数据到Elasticsearch
      • 利用Logstash过滤器解析日志
        • Grok插件
        • Grok语法
        • 用法
      • 输出到Elasticsearch指定索引

前文：FileBeats详解


前文：logstash详解


ELK整合实战

案例：采集SpringBoot应用日志

一个springboot应用，打了一个jar包，使用nohup java -jar XXX.jar &运行

日志内容如下所示

2024-08-16 16:04:04.605  INFO 9164 --- [nio-8081-exec-1] com.hs.single.controller.UserController  : 请求参数为：1
Creating a new SqlSession
SqlSession [org.apache.ibatis.session.defaults.DefaultSqlSession@4196ffc2] was not registered for synchronization because synchronization is not active
JDBC Connection [com.alibaba.druid.proxy.jdbc.ConnectionProxyImpl@790df42b] will not be managed by Spring
==>  Preparing: SELECT id,username,password,name,description,status,create_time,update_time FROM sys_user WHERE id=?
==> Parameters: 1(Long)
<==      Total: 0
Closing non transactional SqlSession [org.apache.ibatis.session.defaults.DefaultSqlSession@4196ffc2]



使用FileBeats将日志发送到Logstash

创建配置文件filebeat-logstash.yml，配置FileBeats将数据发送到Logstash

vim filebeat-logstash.yml#因为Tomcat的web log日志都是以IP地址开头的，我们应用日志一般是以日期时间开头。所以我们需要修改下匹配字段。
# 不以日期时间开头的行追加到上一行
filebeat.inputs:
- type: logenabled: truepaths:- /root/application/nohup.outmultiline.pattern: '^\d{4}-\d{2}-\d{2}'multiline.negate: truemultiline.match: afteroutput.logstash:enabled: truehosts: ["192.168.75.65:5044"]
  • pattern：正则表达式
  • negate：true 或 false；默认是false，匹配pattern的行合并到上一行；true，不匹配pattern的行合并到上一行
  • match：after 或 before，合并到上一行的末尾或开头
# 启动可能会报错 因为安全原因不要其他用户写的权限，去掉写的权限就可以了，
chmod 644 filebeat-logstash.yml

启动FileBeat，并指定使用指定的配置文件

# -e 前台运行   -c  指定配置文件
./filebeat -e -c filebeat-logstash.yml



可能出现的异常：

Exiting: error loading config file: config file ("filebeat-logstash.yml") can only be writable by the owner but the permissions are "-rw-rw-r--" (to fix the permissions use: 'chmod go-w /home/es/filebeat-7.17.3-linux-x86_64/filebeat-logstash.yml')

因为安全原因不要其他用户写的权限，去掉写的权限就可以了



Failed to connect to backoff(async(tcp://192.168.75.65:5044)): dial tcp 192.168.75.65:5044: connect: connection refused

FileBeat将尝试建立与Logstash监听的IP和端口号进行连接。但此时，我们并没有开启并配置Logstash，所以FileBeat是无法连接到Logstash的。



配置Logstash接收FileBeat收集的数据并打印

输入的使用5044端口接收输入，输出是直接在控制台打印，查看是否能正常接收数据

vim config/filebeat-console.conf
# 配置从FileBeat接收数据
input {beats {port => 5044}
}output {stdout {codec => rubydebug}
}



启动logstash

# -t检查配置文件是否存在语法错误
[root@hs-es-node1 logstash-7.17.3]# bin/logstash -t -f config/filebeat-console.conf
[root@hs-es-node1 logstash-7.17.3]# bin/logstash -f config/filebeat-console.conf

调用接口，filebeat成功将数据发送至logstash中

在这里插入图片描述



Logstash输出数据到Elasticsearch

如果我们需要将数据输出值ES而不是控制台的话，我们修改Logstash的output配置。

vim config/filebeat-elasticSearch.conf
input {beats {port => 5044}
}output {elasticsearch {hosts => ["http://192.168.75.65:9200"]user => "elastic"password => "123456"}stdout{codec => rubydebug}
}



启动logstash

# -t检查配置文件是否存在语法错误
[root@hs-es-node1 logstash-7.17.3]# bin/logstash -t -f config/filebeat-elasticSearch.conf
[root@hs-es-node1 logstash-7.17.3]# bin/logstash -f config/filebeat-elasticSearch.conf



ES中会生成一个以logstash开头的索引，测试日志是否保存到了ES。

在这里插入图片描述



思考：日志信息都保证在message字段中，是否可以把日志进行解析一个个的字段？例如：时间、日志级别、哪个类打印的日志、日志具体内容。

在这里插入图片描述



利用Logstash过滤器解析日志

在Logstash中可以配置过滤器Filter对采集到的数据进行过滤处理，Logstash中有大量的插件可以供我们使用。

# 查看Logstash已经安装的插件,默认已经安装了grok插件
[root@hs-es-node1 logstash-7.17.3]# bin/logstash-plugin list



Grok插件

Grok是一种将非结构化日志解析为结构化的插件。这个工具非常适合用来解析系统日志、Web服务器日志、MySQL或者是任意其他的日志格式。

https://www.elastic.co/guide/en/logstash/7.17/plugins-filters-grok.html



Grok语法

Grok是通过模式匹配的方式来识别日志中的数据,可以把Grok插件简单理解为升级版本的正则表达式。它拥有更多的模式，默认Logstash拥有120个模式。如果这些模式不满足我们解析日志的需求，我们可以直接使用正则表达式来进行匹配。

grok模式的语法是：

# SYNTAX（语法）指的是Grok模式名称，SEMANTIC（语义）是给模式匹配到的文本字段名。
# 注意这中间不要有空格
%{SYNTAX:SEMANTIC}# 案例  duration表示：匹配一个数字，client表示匹配一个IP地址。
%{NUMBER:duration} %{IP:client}

默认在Grok中，所有匹配到的的数据类型都是字符串，如果要转换成int类型（目前只支持int和float），可以这样：

%{NUMBER:duration:int} %{IP:client}

常用的Grok模式

https://help.aliyun.com/document_detail/129387.html?scm=20140722.184.2.173



用法

比如，tomacat日志

192.168.65.103 - - [23/Jun/2022:22:37:23 +0800] "GET /docs/images/docs-stylesheet.css HTTP/1.1" 200 5780
字段名说明
client IP浏览器端IP
date请求的时间戳
method请求方式（GET/POST）
uri请求的链接地址
status服务器端响应状态
length响应的数据长度
%{IP:ip} - - \[%{HTTPDATE:date}\] \"%{WORD:method} %{PATH:uri} %{DATA:protocol}\" %{INT:status} %{INT:length}



为了方便测试，我们可以使用Kibana来进行Grok开发：

在这里插入图片描述



我们现在的日志和对应的grok语法：

2024-08-16 16:04:04.605  INFO 9164 --- [nio-8081-exec-1] com.hs.single.controller.UserController  : 请求参数为：1%{TIMESTAMP_ISO8601:timestamp}  %{LOGLEVEL:level} %{INT:thread} --- \[%{NOTSPACE:serverPort}\] %{JAVACLASS:class}  : %{GREEDYDATA:logResult}

在这里插入图片描述



修改Logstash配置文件

vim config/filebeat-elasticSearch.confinput {beats {port => 5044}
}# 添加下面的filter
# 因为我们的消息是保存在message字段中，所以下面主要就是匹配message字段中的值
filter {grok {match => { "message" => "%{TIMESTAMP_ISO8601:timestamp}  %{LOGLEVEL:level} %{INT:thread} --- \[%{NOTSPACE:serverPort}\] %{JAVACLASS:class}  : %{GREEDYDATA:logResult}" }}
}output {stdout {codec => rubydebug}
}



启动logstash测试

# --config.reload.automatic修改了配置文件能热加载
bin/logstash -f config/filebeat-elasticSearch.conf --config.reload.automatic

在这里插入图片描述



使用mutate插件过滤掉不需要的字段

mutate {enable_metric => "false"remove_field => ["message", "log", "tags", "input", "agent", "host", "ecs", "@version"]
}

在这里插入图片描述



如果要将日期格式进行转换，我们可以使用Date插件来实现。该插件专门用来解析字段中的日期，官方说明文档：https://www.elastic.co/guide/en/logstash/7.17/plugins-filters-date.html

用法如下：

在这里插入图片描述

将date字段转换为「年月日 时分秒」格式。默认字段经过date插件处理后，会输出到@timestamp字段，所以，我们可以通过修改target属性来重新定义输出字段。

# [字段名，当前时间格式，修改后的时间格式]    我们这里可以不用改，但是需要了解这个功能
date {match => ["date","dd/MMM/yyyy:HH:mm:ss Z","yyyy-MM-dd HH:mm:ss"]target => "date"
}



输出到Elasticsearch指定索引

index来指定索引名称，默认输出的index名称为：logstash-%{+yyyy.MM.dd}。

但注意，要在index中使用时间格式化，filter的输出必须包含 @timestamp字段，否则将无法解析日期。

output {elasticsearch {hosts => ["http://192.168.75.65:9200"]index => "app_web_log_%{+YYYY-MM-dd}"user => "elastic"password => "123456"}stdout{codec => rubydebug}
}

注意：index名称中，不能出现大写字符



完整的Logstash配置文件

input {beats {port => 5044}
}filter {grok {match => {"message" => "%{TIMESTAMP_ISO8601:timestamp}  %{LOGLEVEL:level} %{INT:thread} --- \[%{NOTSPACE:serverPort}\] %{JAVACLASS:class}  : %{GREEDYDATA:logResult}"}}mutate {enable_metric => "false"remove_field => ["message", "log", "tags", "input", "agent", "host", "ecs", "@version"]}# 我们这里可以不用改，但是需要了解这个功能#date {#  match => ["date","dd/MMM/yyyy:HH:mm:ss Z","yyyy-MM-dd HH:mm:ss"]#  target => "date"#}
}output {elasticsearch {hosts => ["http://192.168.75.65:9200"]index => "app_web_log_%{+YYYY-MM-dd}"user => "elastic"password => "123456"}stdout{codec => rubydebug}
}
http://www.lryc.cn/news/425866.html

相关文章：

  • 网络编程：OSI协议，TCP/IP协议，IP地址，UDP编程
  • QtExa001自动包装流水线的框架设计vs2019QT
  • SpringBoot拦截器的使用介绍
  • Spring Boot应用中的资源分离与高效打包实践
  • 分析 avformat_open_input 数据读取过程
  • Apache HOP (Hop Orchestration Platform) VS Data Integration (通常被称为 Kettle)
  • 如何判断一个dll/exe是32位还是64位
  • 加速网页加载，提升用户体验：HTML、JS 和 Vue 项目优化全攻略
  • LVS服务器基础环境配置
  • 【Python OpenCV】使用OpenCV实现两张图片拼接
  • springboot jar -jar centos后台运行的几种方式
  • 【GitLab】使用 Docker 安装 GitLab：配置 SSH 端口
  • 【pdf文件生成】如何将盖章的文件生成PDF文件
  • 铝壳电阻在电路中的作用和影响是什么？
  • # Python 判断入参日期是周几
  • 井字棋游戏（HTML+CSS+JavaScript）
  • HTML 列表和容器元素——WEB开发系列10
  • Java数组的高级使用技巧与性能优化
  • python spyne报No module named ‘http.cookies‘的解决
  • vmware虚拟机玩GPU显卡直通
  • Linux下Oracle 11g升级19c实录
  • haproxy实验-2
  • 動態PPTP代理IP是什麼？
  • 《全面解析 Nginx：从下载安装到高级应用与问题解决》
  • python获取视频时长
  • php-xlswriter实现数据导出excel单元格合并，内容从指定行开始写
  • 注意力模型QKV矩阵与位置向量计算
  • glm4-9B-chat，使用提示工程激活模型最大潜力
  • [Linux]在Ubuntu中如何正确安装python
  • [Vue3 + TS + Vite]文件选择器-组件