elk的logstash怎么过滤出报错的模块

Posted 2023-04-23

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了elk的logstash怎么过滤出报错的模块相关的知识，希望对你有一定的参考价值。

2.1 日志的采集灵活性是我们选择日志采集方案更看重的因素，所以logstash属于首先方案，它可以兼顾多种不同系统和应用类型等因素的差异，从源头上进行一些初步的日志预处理。 logstash唯一的小缺憾是它的不轻便，因为它是使用jruby开发... 参考技术A 不知道.........................

ELK 之数据收集传输过滤 Filebeat+Logstash 部署

本文与前文是有关联的，之前的两篇文章客官可以抬腿出门右转导读，ELK 之前端，ELK 之分布式发
#前端和消息队列搞定之后，我们需要安装数据采集工具filebeats和数据过滤机运输工具Logstash，一般情况我们都使用filebeats 用来收集日志文件,我自定义了一个log文件，文件内容如下：
55.3.244.1 GET /index.html 15824 0.043
55.3.244.1 GET /index.html 15824 0.043
文件位置放到/tmp/test.log

cd /opt/elk/filebeat-6.2.2-linux-x86_64
#filebeat 的配置文件大体上分为两部分，输入部分和输出部分，输入指定输入源即可，输出可以选择直接到elasticsearch，logstash或者是消息队列（redis&kafka）等，本文采用kafka 作为消息队列；
vi filebeat.yml
filebeat.prospectors:
#日志类型

type: log
enabled: True
日志路径可以

写多个，支持通配符
paths:
- /tmp/test.log
  #设置字符集编码
  encoding: utf-8
  #文档类型
  document_type: my-nginx-log
  #每十秒扫描一次
  scan_frequency: 10s
  实际读取文件时，每次读取 16384 字节
  
  harverster_buffer_size: 16384
  #The maximum number of bytes that a single log message can have. All bytes after max_bytes are discarded and not sent. This setting is especially useful for multiline log messages, which can get large. The default is 10MB (10485760). 一次发生的log大小值；
  max_bytes: 10485760
  
  是否从文件末尾开始读取
  
  tail_files: true
  #给日志加上tags方便在logstash过滤日志的时候做判断
  tags: ["nginx-access"]
  #剔除以.gz 结尾的文件
  exclude_files: [".gz$"]
  #output 部分的配置文档
  #配置输出为kafka，无论你是tar包还是rpm安装，目录里边会看到官方提供的filebeat.full.yml OR filebeat.reference.yml 这里边有filebeat 所有的input 方法和output 方法供你参考；
  output.kafka:
  enabled: true
  #kafka 的server，可以配置集群，例子如下：
  hosts:["ip:9092","ip2:9092","ip3:9092"]
  #这个非常重要，filebeat作为provider，把数据输入到kafka里边，logstash 作为消费者去消费这些信息，logstash的input 中需要这个topic，不然logstash没有办法取到数据。
  topic: elk-%{[type]}
  
  The number of concurrent load-balanced Kafka output workers. kafka 的并发运行进程
  
  worker: 2
  #当传输给kafka 有问题的时候，重试的次数；
  max_retries: 3
  #单个kafka请求里面的最大事件数，默认2048
  bulk_max_size: 2048
  #等待kafka broker响应的时间，默认30s
  timeout: 30s
  #kafka broker等待请求的最大时长，默认10s
  broker_timeout: 10s
  #每个kafka broker在输出管道中的消息缓存数，默认256
  channel_buffer_size: 256
  #网络连接的保活时间，默认为0，不开启保活机制
  keep_alive: 60
  #输出压缩码，可选项有none, snappy, lz4 and gzip，默认为gzip （kafka支持的压缩，数据会先被压缩，然后被生产者发送，并且在服务端也是保持压缩状态，只有在最终的消费者端才会被解压缩）
  compression: gzip
  #允许的最大json消息大小，默认为1000000，超出的会被丢弃，应该小于broker的??message.max.bytes（broker能接收消息的最大字节数）
  max_message_bytes: 1000000
  #kafka的响应返回值，0位无等待响应返回，继续发送下一条消息；1表示等待本地提交（leader broker已经成功写入，但follower未写入），-1表示等待所有副本的提交，默认为1
  required_acks: 0
  #The configurable ClientID used for logging, debugging, and auditing purposes. The default is "beats"。客户端ID 用于日志怕错，审计等，默认是beats。
  client_id: beats
  #测试配置文件：/opt/elk/filebeat/filebeat -c /opt/elk/filebeat/filebeat.yml test config
  #如果配置文件没有问题的话，会出现config ok ，如果有问题会提示具体问题在哪里。
  #启动filebeat
  可以先通过 /opt/elk/filebeat-6.2.2-linux-x86_64/filebeat -c -e /opt/elk/filebeat-6.2.2-linux-x86_64/filebeat.yml 查看一下输入filebeat是否工作正常，会有很多信息打印到屏幕上；
  nohup /opt/elk/filebeat-6.2.2-linux-x86_64/filebeat -c /opt/elk/filebeat-6.2.2-linux-x86_64/filebeat.yml >>/dev/null 2>&1&

logstash 安装配置：
input {
#数据来源
kafka {
#这个对应filebeat的output 的index
topics_pattern => "elk-.*"
#kafka 的配置
bootstrap_servers => "IP1:9092,IP2:9092,IP3:9092"
kafka 中的group ID
group_id => "logstash-g1"
}
}
#过滤器，如果你想要日志按照你的需求输出的话，需要在logstash中过滤并且给与相应的key，比如以下的是nginx 日志，用的是logstash内置好的过滤器，%{IP:client} 这个里边包含了两个信息：ip地址，client 是在kibana 中显示的自定义键值，最终显示成这样，http://grokdebug.herokuapp.com/ 是在线调试logstash filter的工具，你可以在线调试你的过滤规则。

filter {
grok {
match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}"}
#因为我们取出了字段，所以不需要原来这个message字段，这个字段里边包含之前beat 输入的所有字段。
remove_field => ["message"]
}
}
output {
elasticsearch {
hosts => ["IP:9200"]
#这个是在kibana上的index Patterns 的索引，建议什么服务就用干什么名字，因为beat 提供了些kibana的模板可以导入，导入的模板匹配的时候用的索引是对应服务的名称开头。
index => "nginx-%{+YYYY.MM.dd}"
document_type => "nginx"
#每次20000 发送一次数据到elasticsearch
flush_size => 20000
如果不够20000，没10秒会发送一次数据；、
idle_flush_time =>10
}
}