win10 docker-compose搭建ELK日志收集

elk的威名大家都知道，以前前司有专门的人维护，现在换了环境，实在不想上服务器看，所以就摸索下自己搭建，由于现场服务器是需要类似向日葵那样连接，我还是把日志弄回来，自己本地filebeat上传到es中

配置文件

没啥好说的，就是volume 需要根据自己的环境调整。这里的 ". "相当于你的docker-compose.yml文件所在的目录层级！

• docker-compose

version: '3.2'

services:
    elasticsearch:
        image: elasticsearch:7.17.4
        volumes:
            - ./es/plugins:/usr/share/elasticsearch/plugins #插件文件挂载
            - ./es/data:/usr/share/elasticsearch/data #数据文件挂载
        ports:
            - '9200:9200'
            - '9300:9300'
        container_name: elasticsearch
        restart: always
        environment:
            - 'cluster.name=elasticsearch' #设置集群名称为elasticsearch
            - 'discovery.type=single-node' #以单一节点模式启动
            - 'ES_JAVA_OPTS=-Xms1024m -Xmx1024m' #设置使用jvm内存大小
        networks:
            - elk
    logstash:
        image: logstash:7.17.4
        container_name: logstash
        restart: always
        volumes:
            - ./logstash/my-logstash.conf:/usr/share/logstash/config/my-logstash.conf
            - ./logstash/logs:/var/logs/demo
        ports:
            - '5044:5044/tcp'
            - '50000:50000/tcp'
            - '50000:50000/udp'
            - '9600:9600'
        environment:
            LS_JAVA_OPTS: -Xms1024m -Xmx1024m
            TZ: Asia/Shanghai
            MONITORING_ENABLED: false
        entrypoint:
            - logstash
            - -f
            - /usr/share/logstash/config/my-logstash.conf  
        links:
            - elasticsearch:es #可以用es这个域名访问elasticsearch服务
        networks:
            - elk
        depends_on:
            - elasticsearch
    kibana:
        image: kibana:7.17.4
        container_name: kibana
        restart: always
        volumes:
            - ./kibana/config/kibana.yml:/usr/share/kibana/config/kibana.yml
        ports:
            - '5601:5601'
        links:
            - elasticsearch:es #可以用es这个域名访问elasticsearch服务
        environment:
            - ELASTICSEARCH_URL=http://elasticsearch:9200 #设置访问elasticsearch的地址
            - 'elasticsearch.hosts=http://es:9200' #设置访问elasticsearch的地址
            - I18N_LOCALE=zh-CN
        networks:
            - elk
        depends_on:
            - elasticsearch
    filebeat:
        user: root
        image: elastic/filebeat:7.17.4
        container_name: filebeat
        restart: always
        volumes:
            # 将宿主机目录挂载到容器中，这个目录就是我真正放 从服务器上下载spring-boot日志的地方
            - ./filebeat/logs/elk:/var/logs
            # 指定配置文件
            - ./filebeat/filebeat.yml:/usr/share/filebeat/filebeat.yml
            - ./filebeat/logs:/usr/share/filebeat/logs
            - ./filebeat/data:/usr/share/filebeat/data
        links:
            - logstash:logstash
        command: ["--strict.perms=false"]
        networks:
            - elk  
        depends_on:
            - logstash          
networks:
    elk:
        name: elk
        driver:
            bridge

• logstash 配置

# beats传入的端口，默认5044
input {
  beats {
    port => 5044
  }
}
filter {
  grok {
    pattern_definitions => { "MyTimestamp" => "(\d\d){1,2}-(?:0?[1-9]|1[0-2])-(?:(?:0[1-9])|(?:[12][0-9])|(?:3[01])|[1-9])\s+(2[0123]|[01]?[0-9])(:[0-5][0-9])((:[0-5]?[0-9]|60)(?:[:.,][0-9]+)?)\.\d{3}" }
    match => { "message" => "(?m)%{MyTimestamp:log_create_time}\s+\[%{GREEDYDATA:thread_name}\]\s+%{LOGLEVEL:log-level}\s+%{JAVACLASS:java-class}\s+-%{JAVALOGMESSAGE:message}" }
    overwrite => ["message"]
  }  
  date {
   match => [ "log_create_time","yyyy-MM-dd HH:mm:ss.SSS"]
   target => "@timestamp"
  }
   if "_grokparsefailure" in [tags] {
          drop { }
    }
}
# 输出日志的方式
output { 
# 按照日志标签对日志进行分类处理，日志标签后续会在filebeat中定义
 
  elasticsearch {
    hosts => ["http://es:9200"]
    index => "demo"
  }
  # 这部分主要是用于本地调试的，prd可以注释掉
#	stdout{
 #       codec=> rubydebug
  #  }
}

• filebeat配置文静

# 从日志文件输入日志
filebeat.inputs:
- type: filestream
  id: my-filestream-id 
  enabled: true
  prospector.scanner.check_interval : 15
  paths:
    - /var/logs/*.log
  # 定义日志标签，注意当order服务时将该标签改为order-log
  parsers:
  - multiline:
      type: pattern
      pattern: '^(\d\d){1,2}'
      negate: true
      match: after  
 
setup.template.settings:
# 设置主分片数
  index.number_of_shards: 1
# 因为测试环境只有一个es节点，所以将副本分片设置为0，否则集群会报黄
  index.number_of_replicas: 0
# 输出到logstash
output.logstash:
# logstash所在服务器的ip和端口
    hosts: ["logstash:5044"]
# output.console:
#     pretty: true
#     enable: true  
# 默认配置，不做改动
processors:
  - add_host_metadata:
      when.not.contains.tags: forwarded
  - add_cloud_metadata: ~
  - add_docker_metadata: ~
  - add_kubernetes_metadata: ~

遇到的问题

启动filebeat 遇到的问题

查了下 需要 command: [“–strict.perms=false”] 加上这段，貌似filebeat容器就不会有权限问题了

java异常的多行处理

这个搞了好长时间，文档上说如果你用了fileBeat 那么就在fileBeat 用 multiline插件处理，而不是在logstash处理

Logstash 中对于message 匹配处理

这个也花了不少时间，首先日志一开始还是GBK的导致日志灌入es 乱码，所以我这里是调整logback的输出格式 ,指定charset 为 UTF8

 <appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <encoder>
<!--            <pattern>${FILE_LOG_PATTERN}</pattern>-->
            <!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符-->
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger -%msg%n</pattern>
            <charset>UTF-8</charset>
        </encoder>
        <file>${LOG_PATH}/spring-boot-plus.log</file>
        <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
            <fileNamePattern>${LOG_PATH}/spring-boot-plus-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
            <maxFileSize>${MAX_FILE_SIZE}</maxFileSize>
            <maxHistory>${MAX_HISTORY}</maxHistory>
        </rollingPolicy>
    </appender>

注意你的logstash message处理的格式正是 这里指定的 %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger -%msg%n 这个格式！！！

如果你有特殊的 格式要处理，可以参考 logstash 官方维护的pattern 仓库

这里是引用 logstash 官方维护的pattern 仓库

如果有自定义的格式，比如我这里的时间，折腾我好久，要去kibana debugger 来自己调试

• 如何多次触发filebeat 将日志采集灌入，只需要手动删除filebeat下面的registry 目录即可