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-syslogpath.config: "/usr/share/logstash/pipeline/fscr-syslog.conf"

fscr-syslog.conf：

input {syslog {port => 12525type => "system-syslog"}
}
filter {if [type] == "system-syslog" {mutate {# Remove ANSI escape sequencesgsub => ["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