ELK 分布式日志

Elasticsearch 基于 Java，是个开源分布式搜索引擎，它的特点有：分布式，零配置，自动发现，索引自动分片，索引副本机制，restful风格接口，多数据源，自动搜索负载等。

Kibana 基于 NodeJs，也是一个开源和免费的工具，Kibana 可以为 Logstash 和 ElasticSearch 提供的日志分析友好的Web 界面，可以汇总、分析和搜索重要数据日志。

Logstash 基于 Java，是一个开源的用于收集,分析和存储日志的工具。

一、Docker 搭建ELK

1.1、安装 ELK

这里使用 docker-compose 编排安装 ELK 服务，以下镜像采用 cpu架构 arm64 版本，如需在x86环境下，请自行更换镜像。

  1. version: '3'
  2. services:
  3.   es:
  4.     container_name: blog-es
  5.     image: 'gagara/elasticsearch-oss-arm64:latest'
  6.     #restart: always
  7.     privileged: false
  8.     environment:
  9.       - "cluster.name=elasticsearch" #设置集群名称为elasticsearch
  10.       - "discovery.type=single-node" #以单一节点模式启动
  11.       - "ES_JAVA_OPTS=-Xms512m -Xmx512m" #设置使用jvm内存大小
  12.     volumes:
  13.       - ./es/plugins:/usr/share/elasticsearch/plugins
  14.       - ./es/data:/usr/share/elasticsearch/data
  15.     ports:
  16.       - 9200:9200
  17.       - 9300:9300
  18.     networks:
  19.       - blog-nextwork
  20.   kibana:
  21.     image: 'gagara/kibana-oss-arm64:latest'
  22.     container_name: blog-kibana
  23.     #restart: always
  24.     depends_on:
  25.       - es #kibana在elasticsearch启动之后再启动
  26.     environment:
  27.       - "ELASTICSEARCH_HOSTS=http://es:9200"
  28.     ports:
  29.       - 5601:5601
  30.     networks:
  31.       - blog-nextwork
  32.   logstash:
  33.     image: 'gagara/logstash-oss-arm64:latest'
  34.     container_name: blog-logstash
  35.     #restart: always
  36.     environment:
  37.       - TZ=Asia/Shanghai
  38.     volumes:
  39.       - ./logstash/logstash.conf:/usr/share/logstash/pipeline/logstash.conf #挂载logstash的配置文件
  40.     depends_on:
  41.       - es #logstash在elasticsearch启动之后再启动
  42.     ports:
  43.       - 4560:4560
  44.       - 5044:5044
  45.     networks:
  46.       - blog-nextwork
  47. networks: 
  48.   blog-nextwork:
  49.     driver: bridge

1.2、Logstash 配置文件简单了解

1.2.1、配置文件分为以下三个部分
  1. # 输入模块，负责从数据源提取数据
  2. input {
  3.   ...
  4. }
  5. # 过滤器模块，负责通过配置插件来解析、处理日志数据
  6. filter {
  7.   ...
  8. }
  9. # 输出模块，负责输出数据到指定端，如ES 
  10. output {
  11.   ...
  12. }

1.2.2、 input 插件配置

input plugins 让 logstash 可以读取特定的事件源。例如 beats input 插件

  1. input {
  2.   beats {
  3.       host => "0.0.0.0" #要监听的ip地址
  4.     port => 5044 #要监听的端口
  5.     #ssl => false  # 是否启用https 传输
  6.     #ssl_certificate_authorities => ''  
  7.     #ssl_certificate => ''
  8.     #ssl_key => ''
  9.   }
  10. }

1.2.3、filter plugin

filter plugin 过滤器插件，对事件执行中间处理，以下是简单使用

  1. filter {
  2.         mutate {
  3.            # 使用 | 分割日志
  4.            split => {"message"=>"|"}
  5.         }
  6.         mutate {
  7.             #把分割出来的各个部分，分别写入对应的字段中
  8.             add_field => {
  9.                 "date_time" => "%{[message][0]}"
  10.                 "appName" => "%{[message][1]}"
  11.                 "tlog_flag" => "%{[message][2]}"
  12.                 "log_level" => "%{[message][4]}"
  13.             }
  14.         }
  15.         mutate {
  16.             # 删除多余字段
  17.            remove_field => [ "host" ,"ecs"]
  18.         }
  19. }

1.2.4、output plugin

output plugin 输出插件，将事件发送到特定目标。以下以elasticsearch为例

  1. output {
  2.     #控制台输出 调试时使用
  3.     stdout {
  4.         codec => rubydebug
  5.     }
  7.     # 来自blog-cloud 的日志写入 blog_cloud_%{+YYYY.MM.dd} 索引中
  8.     if [fields][from] == 'blog-cloud' {
  9.         elasticsearch {
  10.          hosts => ["127.0.0.1:9200"]
  11.          index => "blog_cloud_%{+YYYY.MM.dd}"
  12.         }
  13.     }
  14. }

1.2.5、logstash 完整配置
  1. input {
  2.   beats {
  3.     port => 5044
  4.   }
  5. }
  6. #filter {
  7. #        mutate {
  8. #            # 使用 | 分割日志
  9. #           split => {"message"=>"|"}
  10. #        }
  11. #        mutate {
  12. #            #把分割出来的各个部分，分别写入对应的字段中
  13. #            add_field => {
  14. #                "date_time" => "%{[message][0]}"
  15. #                "appName" => "%{[message][1]}"
  16. #                "tlog_flag" => "%{[message][2]}"
  17. #                "log_level" => "%{[message][4]}"
  18. #            }
  19. #        }
  20. #        mutate {
  21. #           remove_field => [ "host" ,"ecs"]
  22. #        }
  23. #}
  24. output {
  25.     #控制台输出 调试时使用
  26.     stdout {
  27.         codec => rubydebug
  28.     }
  30.     # 来自blog-cloud 的日志写入 blog_cloud_%{+YYYY.MM.dd} 索引中
  31.     if [fields][from] == 'blog-cloud' {
  32.         elasticsearch {
  33.          hosts => ["es:9200"]
  34.          index => "blog_cloud_%{+YYYY.MM.dd}"
  35.         }
  36.     }
  37. }

1.3、安裝Filebeat

在需要采集的服务器上安装 Filebeat

1.3.1、docker 安装
  1. version: '3'
  2. services:
  3.   filebeat:
  4.     image: 'docker.elastic.co/beats/filebeat:7.6.2'
  5.     #restart: always
  6.     environment:
  7.       - TZ=Asia/Shanghai
  8.     volumes:
  9.       - ./filebeat/filebeat.yml/usr/share/filebeat/filebeat.yml
  10.       - /home/hdj/IdeaProjects/blog-cloud/logs:/var/log/blog  #监听的日志路径

1.3.2、filebeat 配置
  1. # 数据处理，如果我们的数据不存在唯一主键，则使用fingerprint否则可以使用add_id来实现
  2. processors:
  3.   # 指纹，防止同一条数据在output的es中存在多次。
  4.   #（此处为了演示使用message字段做指纹，实际情况应该根据不用的业务来选择不同的字段）
  5.   - fingerprint:
  6.       fields: ["message"]
  7.       ignore_missing: false
  8.       target_field: "@metadata._id"
  9.       method: "SHA256"
  12. # 输入配置
  13. filebeat.inputs:
  14.   - type: log
  15.     enabled: true
  16.     encoding: utf-8
  17.     # 容器内的路径，注意映射
  18.     # 映射宿路径： /home/hdj/IdeaProjects/blog-cloud/logs:/var/log/blog
  19.     paths:
  20.       # 抓取日志路径
  21.       - /var/log/blog/*/blog_*.log
  22.     # 添加自定义字段
  23.     fields:
  24.       from: 'blog-cloud'
  25.     multiline:
  26.       # 正则表达式
  27.       pattern: "^[[:space:]]+(at|\\.{3})[[:space:]]+\\b|^Caused by:"
  28.       # 是否开启正则匹配，true:开启，false:不开启
  29.       negate: false
  30.       # 不匹配正则的行是放到匹配到正则的行的after(后面)还是before(前面)
  31.       match: after
  32.       # 多行日志结束的时间，多长时间没接收到日志，如果上一个是多行日志，则认为上一个结束了
  33.       timeout: 5s
  36. # 输出到 logstash
  37. output.logstash:
  38.   hosts: ["192.168.43.178:5044"]

二、安全校验配置

参考