Springboot集成syslog+logstash收集日志到ES

🕗 发布于 2024-11-08 14:10 spring boot elasticsearch 后端

Springboot集成syslog+logstash收集日志到ES

1、背景

Logstash 是一个实时数据收集引擎，可收集各类型数据并对其进行分析，过滤和归纳。按照自己条件分析过滤出符合的数据，导入到可视化界面。它可以实现多样化的数据源数据全量或增量传输，数据标准格式处理，数据格式化输出等的功能，常用于日志处理。工作流程分为三个阶段：

input数据输入阶段，可接收oracle、mysql、postgresql、file等多种数据源；
filter数据标准格式化阶段，可过滤、格式化数据，如格式化时间、字符串等；
output数据输出阶段，可输出到elasticsearch、mongodb、kafka等接收终端。

架构原理：springboot发出syslog日志，通过系统的rsyslog服务进行数据转发，logstash监听rsyslog端口过滤数据并发到es进行存储

2、springboot集成syslog

maven依赖：

<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-api</artifactId>
    <version>1.7.7</version>
</dependency>
<dependency>
    <groupId>ch.qos.logback</groupId>
    <artifactId>logback-core</artifactId>
    <version>1.1.7</version>
</dependency>
<dependency>
    <groupId>ch.qos.logback</groupId>
    <artifactId>logback-classic</artifactId>
    <version>1.1.7</version>
</dependency>

logback.xml文件配置

配置好日志之后，在root标签中添加appender才能生效；

<?xml version="1.0" encoding="UTF-8"?>
<configuration debug="false">
  <!-- 控制台输出 -->
  <appender name="consoleLogAppender" class="ch.qos.logback.core.ConsoleAppender">
    <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
      <level>INFO</level>
    </filter>
    <encoder>
      <pattern>%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{20} - %cyan(%.-3072msg %n)</pattern>
    </encoder>
  </appender>

  <appender name="infoFileAppender" class="ch.qos.logback.core.rolling.RollingFileAppender">
    <File>./logs/service.log</File>
    <filter class="ch.qos.logback.classic.filter.LevelFilter">
      <level>INFO</level>
      <onMatch>ACCEPT</onMatch>
      <onMismatch>DENY</onMismatch>
    </filter>
    <encoder>
      <pattern>%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{20} - %cyan(%.-3072msg %n)</pattern>
    </encoder>
    <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
      <fileNamePattern>./logs/service-log-%d{yyyy-MM-dd}.log</fileNamePattern>
      <maxHistory>15</maxHistory>
      <totalSizeCap>5GB</totalSizeCap>
    </rollingPolicy>
  </appender>
  <appender name="errorFileAppender" class="ch.qos.logback.core.rolling.RollingFileAppender">
    <File>./logs/service-error.log</File>
    <filter class="ch.qos.logback.classic.filter.LevelFilter">
      <level>ERROR</level>
      <onMatch>ACCEPT</onMatch>
      <onMismatch>DENY</onMismatch>
    </filter>
    <encoder>
      <pattern>%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{20} - %cyan(%.-3072msg %n)</pattern>
    </encoder>
    <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
      <fileNamePattern>./logs/service-error.log.%d{yyyy-MM-dd}.log</fileNamePattern>
      <maxHistory>15</maxHistory>
      <totalSizeCap>5GB</totalSizeCap>
    </rollingPolicy>
  </appender>

  <appender name="msgAppender" class="ch.qos.logback.core.rolling.RollingFileAppender">
    <File>./logs/service-msg.log</File>
    <filter class="ch.qos.logback.classic.filter.LevelFilter">
      <level>INFO</level>
      <onMatch>ACCEPT</onMatch>
      <onMismatch>DENY</onMismatch>
    </filter>
    <encoder>
      <pattern>%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{20} - %cyan(%.-3072msg %n)</pattern>
    </encoder>
    <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
      <fileNamePattern>./logs/service-msg-%d{yyyy-MM-dd}.log</fileNamePattern>
      <maxHistory>5</maxHistory>
      <totalSizeCap>5GB</totalSizeCap>
    </rollingPolicy>
  </appender>

  <appender name="taskAppender" class="ch.qos.logback.core.rolling.RollingFileAppender">
    <File>./logs/service-task.log</File>
    <filter class="ch.qos.logback.classic.filter.LevelFilter">
      <level>INFO</level>
      <onMatch>ACCEPT</onMatch>
      <onMismatch>DENY</onMismatch>
    </filter>
    <encoder>
      <pattern>%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{20} - %cyan(%.-3072msg %n)</pattern>
    </encoder>
    <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
      <fileNamePattern>./logs/service-task-%d{yyyy-MM-dd}.log</fileNamePattern>
      <maxHistory>5</maxHistory>
      <totalSizeCap>5GB</totalSizeCap>
    </rollingPolicy>
  </appender>

  <appender name="mybatisplus" class="ch.qos.logback.core.rolling.RollingFileAppender">
    <File>./logs/service-mybatisplus.log</File>
    <filter class="ch.qos.logback.classic.filter.LevelFilter">
      <level>DEBUG</level>
      <onMatch>ACCEPT</onMatch>
      <onMismatch>DENY</onMismatch>
    </filter>
    <encoder>
      <pattern>%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{20} - %cyan(%.-3072msg %n)</pattern>
    </encoder>
    <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
      <fileNamePattern>./logs/service-mybatisplus-%d{yyyy-MM-dd}.log</fileNamePattern>
      <maxHistory>5</maxHistory>
      <totalSizeCap>5GB</totalSizeCap>
    </rollingPolicy>
  </appender>
  
  <!-- 定义一个 SyslogAppender -->
  <appender name="SYSLOG" class="ch.qos.logback.classic.net.SyslogAppender">
    <syslogHost>localhost</syslogHost>
    <port>12525</port>
    <facility>LOCAL0</facility> <!-- 设置 Syslog 设施，这意味着服务发送到 Syslog 服务器的所有日志都将被标记为来源于 LOCAL0 -->
    <filter class="ch.qos.logback.classic.filter.LevelFilter">
      <level>WARN</level>
      <onMatch>ACCEPT</onMatch>
      <onMismatch>DENY</onMismatch>
    </filter>
    <suffixPattern>
      [%d{yyyy-MM-dd HH:mm:ss.SSS}] - [%p] - [%X{app:-${app}}] - [%thread] - [%logger{36}.%M] - %msg%n
    </suffixPattern>
  </appender>

  <logger name="msgLogger" level="info"  additivity="false">
    <appender-ref ref="msgAppender" />
  </logger>

  <logger name="taskLogger" level="info"  additivity="false">
    <appender-ref ref="taskAppender" />
  </logger>

<!--  <logger name="com.zbnsec.opera.project.simulator.framework.task" level="DEBUG">
    <appender-ref ref="mybatisplus" />
  </logger>-->

  <root level="INFO" additivity="false">
    <appender-ref ref="consoleLogAppender"/>
    <appender-ref ref="infoFileAppender"/>
    <appender-ref ref="errorFileAppender"/>
    <appender-ref ref="SYSLOG"/>
  </root>
</configuration>

SyslogAppender是syslog的配置：
syslogHost：指的是syslog服务器的主机名/IP地址
port：syslog服务器的监听端口，默认为514 udp
facility：标识消息的来源
suffixPattern：描述日志的格式

3、rsyslog接收springboot应用的日志

1、服务器安装rsyslog服务

apt install rsyslog 安装
systemctl start rsyslog 启动服务
systemctl status rsyslog 查看服务状态
systemctl enable rsyslog  设置rsyslog服务器在启动时自动运行

2、配置rsyslog.conf

rsyslog的配置文件位于：/etc/rsyslog.conf

global(workDirectory="/var/lib/rsyslog")
module(load="builtin:omfile" Template="RSYSLOG_TraditionalFileFormat")
include(file="/etc/rsyslog.d/*.conf" mode="optional")

*.* @@localhost:12515

*.info;mail.none;authpriv.none;cron.none                /var/log/messages
authpriv.*                                              /var/log/secure
mail.*                                                  -/var/log/maillog
cron.*                                                  /var/log/cron
*.emerg                                                 :omusrmsg:*
uucp,news.crit                                          /var/log/spooler
local7.*                                                /var/log/boot.log

以上配置转发了12525端口的syslog，@@代表udp；
如果此时需要系统日志，则需要以下配置：tail -500f /var/log/messages 则会看到系统日志一直在刷新保存

module(load="imuxsock"  SysSock.Use="off") 
module(load="imjournal"  StateFile="imjournal.state") 
module(load="imklog") 
module(load="immark") 
$imjournalRatelimitInterval 0

如果需要将sprigboot日志同时也存储在messages文件，则需要以下配置：
注意：这里监听12525端口，则在logstash启动时，同时监听12525，会出现端口占用，则logstash不会接收到springboot日志数据；

# 监听 UDP 端口
module(load="imudp")
input(type="imudp" port="12525")

# 监听 TCP 端口
module(load="imtcp")
input(type="imtcp" port="12525")

修改完配置之后，执行 systemctl restart rsyslog 重新启动服务

4、集成logstash

1、拉取logstash镜像

logstash的版本要和ES的版本一致，否则可能出现其他问题

docker pull docker.elastic.co/logstash/logstash:7.4.0

2、配置logstash

除了以下配置，其他的都使用logstash容器中的默认配置，可以起一个空的容器，把这些默认配置(config目录和pipeline目录)复制出来
logstash.yaml：

config.reload.automatic: true
config.reload.interval: 3s
http.host: "0.0.0.0"
path.logs: /usr/share/logstash/logs/

logstash.conf：

status = error
name = LogstashPropertiesConfig
appender.console.type = Console
appender.console.name = plain_console
appender.console.layout.type = PatternLayout
appender.console.layout.pattern = [%d{ISO8601}][%-5p][%-25c] %m%n
appender.json_console.type = Console
appender.json_console.name = json_console
appender.json_console.layout.type = JSONLayout
appender.json_console.layout.compact = true
appender.json_console.layout.eventEol = true
# Define Rolling File Appender
appender.rolling.type = RollingFile
appender.rolling.name = rolling
appender.rolling.fileName = ${sys:ls.logs}/logstash-plain.log
appender.rolling.filePattern = ${sys:ls.logs}/logstash-%d{yyyy-MM-dd}-%i.log.gz
appender.rolling.layout.type = PatternLayout
appender.rolling.layout.pattern = [%d{ISO8601}][%-5p][%-25c] %m%n
appender.rolling.policies.type = Policies
appender.rolling.policies.size.type = SizeBasedTriggeringPolicy
appender.rolling.policies.size.size = 100MB
appender.rolling.strategy.type = DefaultRolloverStrategy
appender.rolling.strategy.max = 20
rootLogger.level = ${sys:ls.log.level}
rootLogger.appenderRef.console.ref = ${sys:ls.log.format}_console
rootLogger.appenderRef.rolling.ref = rolling

pipelines.yml: 在pipeline目录中配置几个管道，则在这里对应配置

- pipeline.id: system-syslog
  path.config: "/usr/share/logstash/pipeline/fscr-syslog.conf"

fscr-syslog.conf：

input {
  syslog {
    port => 12525
    type => "system-syslog"
  }
}
filter {
  if [type] == "system-syslog" {
    mutate {
      # Remove ANSI escape sequences
      gsub => [
        "message", "\e\[\d+(;\d+)*m", ""
      ]
    }
    if [message] =~ /^\[/ {
  dissect {
    mapping => {
      "message" => "[%{timestamp}] - [%{loglevel}] - [%{app}] - [%{thread_info}] - [%{source_class}] - %{log_message}"
      }
    }
  }
    mutate {
      # Convert "WARN" to "WARNING"
      gsub => [
        "loglevel", "^WARN$", "WARNING"
      ]
      add_field => [ "received_at", "%{@timestamp}" ]
      add_field => [ "received_from", "%{host}" ]
      add_field => [ "syslog_hostname", "%{logsource}" ]
      add_field => [ "syslog_severity", "%{loglevel}" ]
      add_field => [ "syslog_program", "%{app}" ]
      add_field => [ "syslog_message", "%{message}" ]
      add_field => [ "syslog_timestamp", "%{timestamp}" ]
      remove_field => ["severity_label", "facility_label", "facility", "priority"]
    }
    date {
      match => ["adjusted_received_at", "ISO8601"]
      timezone => "Asia/Shanghai"
      target => "@timestamp"
    }
  }
}

output {
  if [loglevel] == "WARNING" or [loglevel] == "ERROR" {
    elasticsearch {
      hosts => ["http://esHost:9200"]
      index => "logstash-%{+YYYY.MM.dd}"
      template_name => "logstash"   # 指定模板(该模板已经存在于es中)
      template_overwrite => false
    }
  }
if [loglevel] == "WARNING" or [loglevel] == "ERROR" {
    stdout {
    codec => rubydebug
  }
  }
}

logstash.json索引文件：

{
  "name": "logstash",
  "order": 0,
  "version": 60001,
  "index_patterns": [
    "logstash-*"
  ],
  "settings": {
    "index": {
      "number_of_shards": "1",
      "refresh_interval": "5s"
    }
  },
  "mappings": {
    "dynamic_templates": [
      {
        "message_field": {
          "path_match": "message",
          "mapping": {
            "norms": false,
            "type": "text"
          },
          "match_mapping_type": "string"
        }
      },
      {
        "string_fields": {
          "mapping": {
            "norms": false,
            "type": "text",
            "fields": {
              "keyword": {
                "ignore_above": 256,
                "type": "keyword"
              }
            }
          },
          "match_mapping_type": "string",
          "match": "*"
        }
      }
    ],
    "properties": {
      "@timestamp": {
        "type": "date"
      },
      "geoip": {
        "dynamic": true,
        "properties": {
          "ip": {
            "type": "ip"
          },
          "latitude": {
            "type": "half_float"
          },
          "location": {
            "type": "geo_point"
          },
          "longitude": {
            "type": "half_float"
          }
        }
      },
      "@version": {
        "type": "keyword"
      }
    }
  },
  "aliases": {}
}

启动容器：

docker run --name logstash -itd --net=host \
-v /opt/fscr/middleware/logstash/logstash/config:/usr/share/logstash/config \
-v /opt/fscr/middleware/logstash/logstash/pipeline:/usr/share/logstash/pipeline \
-p 5044:5044 -p 9600:9600 \
logstash:8.8.0

容器启动后，无error日志，可以看到打印的日志信息，为正常启动；

原文地址：https://blog.csdn.net/magic_guo/article/details/143598892

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：设置 Linux 云服务器进入单用户模式
下一篇：海外云手机在出海业务中的优势有哪些？

基于RK3588的移动充电机器人应用解决方案
伴随着国内新能源汽车保有量的持续增长，充电难题愈发凸显，配套充电设施的建设需求进一步扩大。可外接4G/5G、WiFi模块，用于传感数据、视频数据的上传，同时，通过网络的连接，能够远程监控机器人的状态和
阅读更多2024-11-16
开源，一天200star，解锁视频字幕生成新方式——一款轻量级开源字幕工具，免费，支持花字，剪映最新会员模式吃相太难看了
是一款基于开源的轻量级视频字幕生成工具，由开发者 @WEIFENG2333 精心打造。其主要目的是为用户提供一套免费、易用的字幕生成方案，支持各种视频格式，并结合现代化的自然语言处理技术，实现精准、高
阅读更多2024-11-16
docker：基于Dockerfile镜像制作完整案例
docker：基于Dockerfile镜像制作完整案例
阅读更多2024-11-16
【数据库系列】 Spring Boot 集成 Neo4j 的详细介绍
Spring Boot 提供了对 Neo4j 的良好支持，使得开发者可以更方便地使用图数据库。通过使用 Spring Data Neo4j，开发者可以轻松地进行数据访问、操作以及管理。本文将详细介绍如
阅读更多2024-11-16
微信小程序02-页面制作
微信小程序提供了丰富的组件，如view、image等，用于构建具有微信风格的UI界面。视口单位用于移动端页面适配，如vw和vh。表单组件用于收集用户信息，如form、button、input等。
阅读更多2024-11-16
2024 CCF中国开源大会“开源科学计算与系统建模openSCS”分论坛成功举办
目前MWORKS用户已遍及国内外295所高校，仅深圳就有41家应用验证企业，覆盖新能源、芯片、车辆、低空经济、医疗、制造、自动化、电力、航天、金融、船舶等优势行业单位，已经具备良好的生态基础。该委员会
阅读更多2024-11-16
C&C++内存管理
pChar3是指针变量，存在栈中，*pChar3是由const修饰的，pChar3指向的内容不能修改，内容具有常属性，因此pChae3存在代码段中，也就是常量区。内存泄漏并不是指内存在物理上的消失，而
阅读更多2024-11-16
单片机面试
答案：宏定义是在编译预处理阶段被处理的。预处理包含：头文件包含，宏替换，条件编译，去除注释，添加行号。
阅读更多2024-11-16
Ubuntu23.10下解决C语言调用mysql.h问题
在学习C语言和MySQL的调用的时候遇到包和版本的问题，由于使用的书很老（10年的），因此很多MySQL的包已经过时，在查找很多资料和询问gpt之后得到了解决方案
阅读更多2024-11-16
高级java每日一道面试题-2024年11月07日-Redis篇-Redis有哪些功能?
Redis 是一个功能强大且灵活的 NoSQL 数据库，广泛应用于缓存、消息队列、实时分析等多种场景。在高级 Java 面试中，了解 Redis 的数据类型、持久化、发布/订阅、事务、Lua 脚本、主
阅读更多2024-11-16

Springboot集成syslog+logstash收集日志到ES