标签：ELK compose elk mdc elasticsearch docker logstash 日志

参考地址：https://www.cnblogs.com/zys2019/p/14816634.html

1. 概述

笔者在学习docker的相关知识，本着talk is cheap show me the code 原则，实际操作下，正好上面的博主分享了文档，我照着做成功了，只是有些细节上面的问题，需要解决下

2. ELK架构

Elasticsearch是个开源分布式搜索引擎，提供搜集、分析、存储数据三大功能。
Logstash 主要是用来日志的搜集、分析、过滤日志的工具，支持大量的数据获取方式。
Kibana 也是一个开源和免费的工具，Kibana可以为 Logstash 和 ElasticSearch 提供的日志分析友好的 Web 界面，可以帮助汇总、分析和搜索重要数据日志。

2.1. 部署ELK

创建docker-elk目录，在此目录创建文件和其他目录

mkdir /opt/docker_elk

创建logstash配置文件

mkdir /opt/docker_elk/logstash
touch  /opt/docker_elk/logstash/logstash.conf

配置logstash.conf

input {
  tcp {
    mode => "server"
    host => "0.0.0.0"
    port => 4560
  }
}

output {
  elasticsearch {
    hosts => ["elasticsearch:9200"]
    index => "logstash-%{+YYYY.MM.dd}"
  }
  stdout { codec => rubydebug }
}

创建docker-compose.yml文件 , touch /opt/docker_elk/docker-compose.yml

version: '3.7'
services:
  elasticsearch:
    image: elasticsearch:7.6.2
    container_name: elasticsearch
    privileged: true
    user: root
    environment:
      #设置集群名称为elasticsearch
      - cluster.name=elasticsearch 
      #以单一节点模式启动
      - discovery.type=single-node 
      #设置使用jvm内存大小
      - ES_JAVA_OPTS=-Xms512m -Xmx512m 
    volumes:
      - /opt/docker_elk/elasticsearch/plugins:/usr/share/elasticsearch/plugins
      - /opt/docker_elk/elasticsearch/data:/usr/share/elasticsearch/data
    ports:
      - 9200:9200
      - 9300:9300
    networks:
      - elk-network

  logstash:
    image: logstash:7.6.2
    container_name: logstash
    ports:
       - 4560:4560
    privileged: true
    environment:
      - TZ=Asia/Shanghai
    volumes:
      #挂载logstash的配置文件
      - /opt/docker_elk/logstash/logstash.conf:/usr/share/logstash/pipeline/logstash.conf 
    depends_on:
      - elasticsearch 
    links:
      #可以用es这个域名访问elasticsearch服务
      - elasticsearch:es 
    networks:
      - elk-network

  kibana:
    image: kibana:7.6.2
    container_name: kibana
    ports:
        - 5601:5601
    privileged: true
    links:
      #可以用es这个域名访问elasticsearch服务
      - elasticsearch:es 
    depends_on:
      - elasticsearch 
    environment:
      #设置访问elasticsearch的地址
      - elasticsearch.hosts=http://es:9200
    networks:
      - elk-network

networks:
  elk-network:
    driver: bridge

进入docker_elk目录启动docker-compose

#启动
docker-compose up -d

#关闭
docker-compose down

#重启某个容器
docker-compose restart logstash

2.2. kibana

输入http://192.168.56.188:5601/，访问Kibana web界面。点击左侧设置，进入Management界面

点击index pattern,船舰logstash-*索引

2.3. log4j记录日志

要使用log4j2，则必须排除SpringBoot自带的日志。

排除logback并导入依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter</artifactId>
    <exclusions>
        <!-- 引入log4j日志时需去掉默认的logback -->
        <exclusion>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-logging</artifactId>
        </exclusion>
    </exclusions>
</dependency>

<!-- 日志管理log4j2 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-log4j2</artifactId>
    <version>2.1.0.RELEASE</version>
</dependency>

<?xml version="1.0" encoding="UTF-8"?>
<!--日志级别以及优先级排序: OFF > FATAL > ERROR > WARN > INFO > DEBUG > TRACE > ALL -->
<!--
    status="warn" Configuration后面日志框架本身的输出日志级别,可以不加入
    monitorInterval="5" 自动加载配置文件的间隔时间, 不低于 5 秒
    注: 我们的配置文件修改后在生产环境下无需重启应用, 可以实现热更新的效果
-->
<Configuration monitorInterval="5">

    <!--全局属性-->
    <Properties>
        <Property name="APP_NAME">coding</Property>
        <Property name="LOG_FILE_PATH">/Users/A-Study/logs/${APP_NAME}</Property>
        <Property name="PATTERN_FORMAT">%date{yyyy-MM-dd HH:mm:ss.SSS}$$%mdc{logId}$$%mdc{hostName}$$%mdc{ip}$$%level$$%mdc{module}$$%class$$%method$$%msg%n</Property>
    </Properties>

    <!--输出源-->
    <Appenders>

        <!--输出到控制台-->
        <Console name="Console" target="SYSTEM_OUT"><!--输出的类型SYSTEM_ERR-->
            <PatternLayout pattern="${PATTERN_FORMAT}"/>
        </Console>

        <!--输出到logstash的appender-->
        <Socket name="Socket" host="192.168.56.188" port="4560" protocol="TCP">
            <!--输出到logstash的日志格式-->
            <PatternLayout pattern="${PATTERN_FORMAT}"/>
        </Socket>

        <!--输出info信息日志到文件 用来定义超过指定大小自动删除旧的创建新的的Appender.-->
        <RollingFile name="RollingInfoFile" fileName="${LOG_FILE_PATH}/info.log"
                filePattern="${LOG_FILE_PATH}/$${date:yyyyMM}/info-%d{yyyyMMdd}-%i.log.gz">
            <!--控制台只输出level及以上级别的信息（onMatch），其他的直接拒绝（onMismatch）-->
            <Filters>
                <ThresholdFilter level="warn" onMatch="DENY"
                        onMismatch="NEUTRAL"/>  <!--高于warn级别就放行,低于这个级别就拦截-->
                <ThresholdFilter level="info" onMatch="ACCEPT" onMismatch="DENY"/>
            </Filters>

            <PatternLayout>
                <pattern>${PATTERN_FORMAT}</pattern> <!--文件路径-->
            </PatternLayout>

            <!--设置文件具体拆分规则-->
            <Policies>
                <!--在系统启动时, 触发拆分规则，生产一个新的日志文件-->
                <OnStartupTriggeringPolicy/>
                <!--按照文件大小拆分, 30 MB -->
                <SizeBasedTriggeringPolicy size="30 MB"/>
                <!--按照时间节点拆分, 规则根据filePattern定义的-->
                <TimeBasedTriggeringPolicy/>
            </Policies>
            <!--在同一个目录下，文件的个数限定为 30 个, 超过进行覆盖-->
            <DefaultRolloverStrategy max="30"/>
        </RollingFile>

        <!--输出错误日志到文件-->
        <RollingFile name="RollingErrorFile" fileName="${LOG_FILE_PATH}/error.log"
                filePattern="${LOG_FILE_PATH}/$${date:yyyyMM}/error-%d{yyyyMMdd}-%i.log.gz">
            <!--控制台只输出level及以上级别的信息（onMatch），其他的直接拒绝（onMismatch）-->
            <ThresholdFilter level="error" onMatch="ACCEPT" onMismatch="DENY"/>

            <PatternLayout>
                <pattern>${PATTERN_FORMAT}</pattern>
            </PatternLayout>

            <Policies>
                <!--在系统启动时, 触发拆分规则，生产一个新的日志文件-->
                <OnStartupTriggeringPolicy/>
                <!--按照文件大小拆分, 30 MB -->
                <SizeBasedTriggeringPolicy size="30 MB"/>
                <!--按照时间节点拆分, 规则根据filePattern定义的-->
                <TimeBasedTriggeringPolicy/>
            </Policies>
            <!--在同一个目录下，文件的个数限定为 30 个, 超过进行覆盖-->
            <DefaultRolloverStrategy max="30"/>
        </RollingFile>
    </Appenders>


    <!--定义logger,日志记录器配置-->
    <Loggers>
        <!--过滤掉spring和mybatis的一些无用的DEBUG信息-->
        <Logger name="org.springframework" level="INFO"/>
        <Logger name="org.mybatis" level="INFO"/>

        <!-- LOG "com.luis*" at TRACE level -->
        <Logger name="com.luis" level="INFO"/>
        <!--使用 rootLogger 配置 日志级别 level="trace"-->
        <Root level="INFO">
            <!--指定日志使用的处理器-->
            <AppenderRef ref="Console"/>
            <AppenderRef ref="RollingInfoFile"/>
            <AppenderRef ref="RollingErrorFile"/>
            <AppenderRef ref="Async"/>
            <AppenderRef ref="Socket"/>
        </Root>
    </Loggers>
</Configuration>

主要是Socket 那部分配置，笔者的mdc里面的字段有自定义，在项目日志切面部分做了MDC输入值的处理，非原始配置

日志能够上传到es中，kibana里面可以查看到

2.4. logstash文件信息切割

对应logstash，默认是格式是使用的logback的默认格式，一旦使用log4j2或其他日志框架来指定日志的输出格式时，那么logstash便无法解析，需要进行自定义过滤。

2.4.1. 自定义过滤

笔者使用的是

<Property name="PATTERN_FORMAT">%date{yyyy-MM-dd HH:mm:ss.SSS}$$%mdc{logId}$$%mdc{hostName}$$%mdc{ip}$$%level$$%mdc{module}$$%class$$%method$$%msg%n</Property>

在logstash.conf中添加的filter过滤规则如下

filter {
  grok {
    match => {
    "message" => "^%{TIMESTAMP_ISO8601:timestamp}\$\$(?<logId>[a-fA-F0-9]{32})?\$\$(?<hostName>[^\$]+)\$\$(?<ip>%{IP})\$\$%{WORD:level}\$\$(?<module>[^\$]*)?\$\$(?<class>.*?)\$\$(?<method>[^\\$]+)\$\$(?<msg>.*)"
    } 
  }
  date {
        match => ["timestamp", "yyyy-MM-dd HH:mm:ss.SSS"]
        target => "@timestamp"
  }
  mutate {
    remove_field => ["message","port"]
  }
}

显示结果如下：

2.4.2. 多行问题

有时输出的是异常日志，存在多行问题，并不是按照这个格式打印，需要修改input配置，支持多行的日志

input {
  tcp {
    mode => "server"
    host => "0.0.0.0"
    port => 4560
    codec => multiline {
      pattern => "^\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}\.\d{3}" # 确保这与您的日志格式匹配
      negate => true
      what => "previous"
      auto_flush_interval => 1
    }
  }
}

这样打印出来的支持多行合并

2.4.3. 遗留问题

目前只是部署了ELK，而且还是单机模式，当前的日志搜集使用的是tcp，由于多行的日志，搜集不知道下一次什么时候结束，所以当前的部署方式存在当前请求的接口日志只打印了部分，下一次请求来了时，上一次的日志才会完全打印出

解决这个问题，一个是自己部署虚拟机的配置尽量高一点

再一个可以部署filebeat，来进行优化日志收集过滤

标签：ELK,compose,elk,mdc,elasticsearch,docker,logstash,日志
From： https://www.cnblogs.com/live2learn/p/18171137

docker-compose部署ELK