ELK+Filebeat 收集多项目日志 配置详解

2019年5月5日18:08:47 发表评论 10,574 views

本文主要针对6.2.3 版本进行实际生产演示,老版本与新版本存在一些差异,请大家在操作前,保证版本的一致性。

一、部署ELK环境

1.1 简介

ELK由Elasticsearch、Logstash和Kibana三部分组件组成;

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

Logstash是一个完全开源的工具,它可以对你的日志进行收集、分析,并将其存储供以后使用

kibana 是一个开源和免费的工具,它可以为 Logstash 和 ElasticSearch 提供的日志分析友好的 Web 界面,可以帮助您汇总、分析和搜索重要数据日志。

Elkstack 一键自动化部署

1.2 环境介绍

ELK+Filebeat 收集多项目日志 配置详解

本次使用Centos 6,不过还是建议大家使用centos7

二、ELK 配置文件

此配置文件已java环境为例

2.1 服务端 elk配置(Elasticsearch+logstach+kibana)

Elasticsearch集群主从

logstach

mkdir /etc/logstash/patterns -p   #创建存储目录
[root@Admin ~]# cat /etc/logstash/patterns/tomcat  #建立日志模板
TOMCAT_SERVICE_LOG \[%{LOGLEVEL:level}\] \[%{WORD:env}\] \[%{NOTSPACE:module}\] \[%{IP:server_ip}|\] \[%{GREEDYDATA:request_id}|\] %{TIMESTAMP_ISO8601:timestamp} %{GREEDYDATA:logmessage}
#上面的日志模板,确保你的日志中存在这些,否则匹配无效

2.2 logstach 根据不同日志来源创建不同的索引

[root@Admin ~]# vim /usr/local/logstash/config/01-syslog.conf
input {
  beats {
    port => 5044
  }
}
filter {
    grok {  #如果不想匹配,直接把grok 字段 删除
      patterns_dir => ["/etc/logstash/patterns"] #patterns_dir设置告诉logstash目录是你的自定义模式
      match => {
          "message" => "%{TOMCAT_SERVICE_LOG}"  #这里的日志,必须要与你要收集的日志格式一样
      }
      remove_field => "message"  #然后移除掉默认的meassage

    }
    date {
      match => ["timestamp","yyyy-MM-dd HH:mm:ss"]  #这里的日志,必须要与你要收集的日志格式一样
      remove_field => "timestamp"
    }
![](https://www.dgstack.cn/wp-content/uploads/2019/05/2353a144176adc43b89bfc4bf9181284.png)


}
output {
if [fields][log_source] == "client-api" {
    elasticsearch {
        hosts => "172.17.92.43:9200"
        index => "client_api-%{+YYYY.MM.dd}"
      }
}
if [fields][log_source] == "bank-xw" {
    elasticsearch {
        hosts => "172.17.92.43:9200"
        index => "bank-xw-%{+YYYY.MM.dd}"
      }
}
}

logstash 排除无效字段

ELK+Filebeat 收集多项目日志 配置详解
想要排错字段,首先你要匹配上字段,然后在message里定义模板,最后remove 你不想看到的字段。
或者 每次都手动点击想要的字段
ELK+Filebeat 收集多项目日志 配置详解

2.2 客户端-filebeat 收集不同项目日志

filebeat.prospectors:
- input_type: log  #项目一
  paths:
  - /root/.gosuv/log/qiyuetao1/output.log
  fields:
      log_source: client-api
      logtype: client-api
      tail_files: true

- input_type: log #项目二
  paths:
  - /root/.gosuv/log/qiuyueta2/output.log
  fields:
      log_source: bank-xw
      logtype: bank-xw
      tail_files: true

output.logstash:
  hosts: ["172.17.92.43:5044"]
  enabled: true
  compression_level: 3

kibana

[root@Admin ~]# vim /usr/local/kibana/config/kibana.yml
server.port: 5601
server.host: "0.0.0.0"
elasticsearch.url: "http://172.17.92.43:9200"

在web添加索引 即可
ELK+Filebeat 收集多项目日志 配置详解

三、filebeat 配置详解

filebeat 是基于原先 logstash-forwarder 的源码改造出来的。换句话说:filebeat 就是新版的 logstash-forwarder,也会是 日志收集器的第一选择。

3.1 input 配置详解

# 日志类型
- input_type: log

  # 日志路径 可以写多个 ,支持通配符
  paths:
    - /home/logs/shiqc_log/*access.log
  #设置字符编码
  encoding: utf-8
  # 添加字段信息
  fields:
    logsource: nginx_access_log
    logtype: nginx_dev
  #文档类型
  document_type: shiqc_dev_123_nginxlog
  #每 10 秒钟扫描一次
  scan_frequency: 10s
  # 实际读取文件时,每次读取 16384 字节
  harvester_buffer_size: 16384
  # 是否从文件末尾开始读取
  tail_files: true
  # 多行日志合并
  # 匹配的正则表达式. 例如匹配所有以 [ 开头的行
  multiline.pattern: ^\[
  # 设置符合上面匹配条件的的行,是否应该被合并成一条日志,false为合并; true为不合并
  multiline.negate: false
  # 设置 符合pattern行 应该被合并到之前不符合pattern的行,还是之后的行
  multiline.match: after

3.2 filter 配置详解

5.0 版本后,beats 新增了简单的 filter 功能,用来完成事件过滤和字段删减:

filters:
    - drop_event:
        regexp:
            message: "^DBG:"
    - drop_fields:
        contains:
            source: "test"
        fields: ["message"]
    - include_fields:
        fields: ["http.code", "http.host"]
        equals:
            http.code: 200
        range:
            gte:
                cpu.user_p: 0.5
            lt:
                cpu.user_p: 0.8

可用的条件判断还包括

equals  #等于
contains  ##包含
regexp   ##正则表达式
range   ##范围
or    ##或者
and   ##并且或和
not    ##不

3.3 output 配置详解

目前 beat 可以发送数据给 ElasticsearchLogstashFileKafkaRedisConsole 六种目的地址。

这里以logstash为例:
beats 发送到 Elasticsearch 也是走 HTTP 接口。示例配置段如下

#----------------------------- Logstash output --------------------------------
output.logstash:
  # The Logstash hosts
  hosts: ["127.0.0.1:5044"]
  worker: 2   #默认进程1,是 beat连到每个host的线程数,对应es的个数
  compression_level: 3  ##发送数据压缩级别
  loadbalance: true # TRUE是配置了多台logstash主机输出插件将负载均衡的发布事件到所有logstash主机。如果设置为false输出插件发送所有事件到随机的一台主机上如果选择的不可达将切换到另一台主机。默认是false。

Filebeat output会包含以下字段:

beat.hostname beat 运行的主机名
beat.name shipper 配置段设置的 name,如果没设置,等于 beat.hostname
@timestamp 读取到该行内容的时间
type 通过 document_type 设定的内容
input_type 来自 "log" 还是 "stdin"
source 具体的文件名全路径
offset 该行日志的起始偏移量
message 日志内容
fields 添加的其他固定字段都存在这个对象里面

filebeat 官网文档重点中文解释:

filebeat:
    spool_size: 1024                         # 最大可以攒够 1024 条数据一起发送出去
    idle_timeout: "5s"                       # 否则每 5 秒钟也得发送一次
    registry_file: ".filebeat"               #文件读取位置记录文件,会放在当前工作目录下。所以如果你换一个工作目录执行 filebeat 会导致重复传输!
    config_dir: "path/to/configs/contains/many/yaml" # 如果配置过长,可以通过目录加载方式拆分配置
    prospectors:                             # 有相同配置参数的可以归类为一个 prospector
        - input_type: log  除了 "log",还有 "stdin"
            fields:
                # 类似 logstash 的 add_fields
                logsource: YII_INFO
                logtype: YIIdev              
            paths:
                # 指明读取文件的位置
                - /home/www/temp/logs/qiuyuetao1.log
                - /home/www/temp/logs/qiuyuetao2.log
                - /home/www/temp/logs/qiuyuetao3.log
                - /home/www/temp/logs/qiuyuetao4.log
                - /home/www/temp/logs/qiuyuetao5.log
            include_lines: ["^ERR", "^WARN"] # 只发送包含这些字样的日志
            exclude_lines: ["^OK"]           # 不发送包含这些字样的日志
            document_type: "YII_dev_123_log" # 定义写入 ES 时的 _type 值
            ignore_older: "24h"              # 超过 24 小时没更新内容的文件不再监听。在 windows 上另外有一个配置叫 force_close_files,只要文件名一变化立刻关闭文件句柄,保证文件可以被删除,缺陷是可能会有日志还没读完
            scan_frequency: "10s"           # 每 10 秒钟扫描一次目录,更新通配符匹配上的文件列表
            tail_files: false               # 是否从文件末尾开始读取
            harvester_buffer_size: 16384    # 实际读取文件时,每次读取 16384 字节
            backoff: "1s"                   # 每 1 秒检测一次文件是否有新的一行内容需要读取
  # 多行日志合并
  # 匹配的正则表达式. 例如匹配所有以 [ 开头的行
  multiline.pattern: ^\[
  # 设置符合上面匹配条件的的行,是否应该被合并成一条日志,false为合并; true为不合并
  multiline.negate: false
  # 设置 符合pattern行 应该被合并到之前不符合pattern的行,还是之后的行
  multiline.match: after
            ... 下面可以有多个- input_type: 甚至多个prospectors

问题总结

参考文章:http://www.xiaot123.com/post/elk_filebeat1

如果logstash 无法启动,可以手动测试排查问题

 /usr/local/logstash/bin/logstash -f /usr/local/logstash/config/01-syslog.conf --config.test_and_exit   #用来手动测试 logstash文件

ELK+Filebeat 收集多项目日志 配置详解

  • QQ精品交流群
  • weinxin
  • 微信公众号
  • weinxin
admin

发表评论

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen: