Beats配置详解
原文链接： Beats配置详解
使用 pipelines
output.elasticsearch:
hosts: ["${hotname}:80"]
template.enabled: true
template.path: "filebeat.template.json"
template.overwrite: false
index: "cs-email"
#pipelines:
# - pipeline: critical_pipeline
# when.equals:
# type: "critical"
# - pipeline: normal_pipeline
# when.equals:
# type: "normal"
output.elasticsearch:
hosts: ["http://localhost:9200"]
username: "admin"
password: "s3cr3t"
template.enabled: true
template.fields: "fields.yml"
template.overwrite: false
index: "{beatname_lc}"
ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]
ssl.certificate: "/etc/pki/client/cert.pem"
ssl.key: "/etc/pki/client/cert.key"
./scripts/import_dashboards -es http://120.92.36.21:9200 -user elastic -pass changeme
-E <setting>=<value>
Override a specific configuration setting. For example:
sudo ./metricbeat -c metricbeat.yml -E name=mybeat
This setting is applied to the currently running Metricbeat process. The Metricbeat configuration file is not changed.
-N
Disable the publishing of events to the defined output. This option is useful only for testing the Beat.
-c <file>
Pass the location of a configuration file for the Beat.
-configtest
Test the configuration file and then exit. This option is useful for troubleshooting the configuration of a Beat.
-d <selectors>
Enable debugging for the specified selectors. For the selectors, you can specify a comma-separated list of components, or you can use -d "*" to enable debugging for all components. For example, -d "publish" displays all the "publish" related messages.
-e
Log to stderr and disable syslog/file output.
-path.config
Set the default location for configuration (e.g. the Elasticsearch template). See the Directory Layout section for details.
-path.data
Set the default location for data files. See the Directory Layout section for details.
-path.home
Set the default location for miscellaneous files. See the Directory Layout section for details.
-path.logs
Set the default location for log files. See the Directory Layout section for details.
-setup
Load the sample Kibana dashboards. By default, this downloads an archive file containing the Beats dashboards from the elastic.co website. See the Dashboards section for more details and more options.
-v
Enable verbose output to show INFO-level messages.
-version
Display the Beat version and exit.
filebeats 配置
Filebeat 发送的日志，会包含以下字段：
beat.hostname beat 运行的主机名
beat.name shipper 配置段设置的 name，如果没设置，等于 beat.hostname
@timestamp 读取到该行内容的时间
type 通过 document_type 设定的内容
input_type 来自 "log" 还是 "stdin"
source 具体的文件名全路径
offset 该行日志的起始偏移量
message 日志内容
fields 添加的其他固定字段都存在这个对象里面
paths：指定要监控的日志，目前按照Go语言的glob函数处理。没有对配置目录做递归处理，比如配置的如果是：
/var/log/* /*.log
则只会去/var/log目录的所有子目录中寻找以”.log”结尾的文件，而不会寻找/var/log目录下以”.log”结尾的文件。
encoding：指定被监控的文件的编码类型，使用plain和utf-8都是可以处理中文日志的。
input_type：指定文件的输入类型log(默认从文件中读取)或者stdin。
document_type：设定Elasticsearch输出时的document的_type字段，也可以用来给日志进行分类。 当未指定时使用input_type设定值
exclude_lines：在输入中排除符合正则表达式列表的那些行。
include_lines：包含输入中符合正则表达式列表的那些行（默认包含所有行），include_lines执行完毕之后会执行exclude_lines。
exclude_files：忽略掉符合正则表达式列表的文件（默认为每一个符合paths定义的文件都创建一个harvester）。
fields：向输出的每一条日志添加额外的信息，比如“level:debug”，方便后续对日志进行分组统计。默认情况下，会在输出信息的fields子目录下以指定的新增fields建立子目录，例如fields.level。
fields_under_root：如果该选项设置为true，则新增fields成为顶级目录，而不是将其放在fields目录下。自定义的field会覆盖filebeat默认的field。例如添加如下配置：
fields:
level: debug
fields_under_root: true
ignore_older：可以指定Filebeat忽略指定时间段以外修改的日志内容，比如2h（两个小时）或者5m(5分钟)。
close_older：如果一个文件在某个时间段内没有发生过更新，则关闭监控的文件handle。默认1h,change只会在下一次scan才会被发现
force_close_files：Filebeat会在没有到达close_older之前一直保持文件的handle，如果在这个时间窗内删除文件会有问题，所以可以把force_close_files设置为true，只要filebeat检测到文件名字发生变化，就会关掉这个handle。
scan_frequency：Filebeat以多快的频率去prospector指定的目录下面检测文件更新（比如是否有新增文件），如果设置为0s，则Filebeat会尽可能快地感知更新（占用的CPU会变高）。默认是10s。
harvester_buffer_size：每个harvester监控文件时，使用的buffer的大小。
max_bytes：单行日志最大字节,多余丢弃;日志文件中增加一行算一个日志事件，max_bytes限制在一次日志事件中最多上传的字节数，多出的字节会被丢弃。
multiline：适用于日志中每一条日志占据多行的情况，比如各种语言的报错信息调用栈。这个配置的下面包含如下配置：
multiline 合并多行日志为一行： 当然在logstash input中使用codec multiline设置是一样的
pattern：匹配规则，这里指匹配每条日志开始的年份；
negate： 是否开始一个新记录，这里指当pattern匹配后，结束之前的记录，创建一条新日志记录；默认false
match： 有before与after，这里指从该行开始向后匹配；
negate：默认是false，正则匹配的行合并到上一行；
true，不匹配pattern的行合并到上一行
match： after 或 before，匹配合并到上一行的末尾或下一行开头
negate match Result
[false before 匹配中 之前的行 新建到下一行 (用来匹配 多行开始的行)]
[false after 匹配中 之后的行 追加到上一行 (用来匹配 多行结束的行)]
true after 不匹配 之前的行 追加到上一行
true before 不匹配 之后的行 追加到下一行
tail_files：如果设置为true，Filebeat从文件尾开始监控文件新增内容，把新增的每一行文件作为一个事件依次发送，而不是从文件开始处重新发送所有内容。
backoff：Filebeat检测到某个文件到了EOF之后，每次等待多久再去检测文件是否有更新，默认为1s。
max_backoff：Filebeat检测到某个文件到了EOF之后，等待检测文件更新的最大时间，默认是10秒。
backoff_factor：定义到达max_backoff的速度，默认因子是2，到达max_backoff后，变成每次等待max_backoff那么长的时间才backoff一次，直到文件有更新才会重置为backoff。
如果设置成1，意味着去使能了退避算法，每隔backoff那么长的时间退避一次。
spool_size:spooler的大小，spooler中的事件数量超过这个阈值的时候会清空发送出去（不论是否到达超时时间）。
idle_timeout:spooler的超时时间，如果到了超时时间，spooler也会清空发送出去（不论是否到达容量的阈值）。
registry_file:记录filebeat处理日志文件的位置的文件
config_dir:如果要在本配置文件中引入其他位置的配置文件，可以写在这里（需要写完整路径），但是只处理prospector的部分。
publish_async：是否采用异步发送模式（实验功能）。
filebeat 处理json
filebeat.prospectors:
- paths:
- input.json
multiline.pattern: '^{'
multiline.negate: true
multiline.match: after
processors:
- decode_json_fields:
fields: ['message']
target: json
output.console.pretty: true
filebeat:
spool_size: 1024 # 最大可以攒够 1024 条数据一起发送出去
idle_timeout: "5s" # 否则每 5 秒钟也得发送一次
registry_file: ".filebeat" # 文件读取位置记录文件，会放在当前工作目录下。所以如果你换一个工作目录执行 filebeat 会导致重复传输！
config_dir: "path/to/configs/contains/many/yaml" # 如果配置过长，可以通过目录加载方式拆分配置
prospectors: # 有相同配置参数的可以归类为一个 prospector
-
fields:
ownfield: "mac" # 类似 logstash 的 add_fields
paths:
- /var/log/system.log # 指明读取文件的位置
- /var/log/wifi.log
include_lines: ["^ERR", "^WARN"] # 只发送包含这些字样的日志
exclude_lines: ["^OK"] # 不发送包含这些字样的日志
-
document_type: "apache" # 定义写入 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 秒检测一次文件是否有新的一行内容需要读取
paths:
- "/var/log/apache/*" # 可以使用通配符
exclude_files: ["/var/log/apache/error.log"]
-
input_type: "stdin" # 除了 "log"，还有 "stdin"
multiline: # 多行合并
pattern: '^[[:space:]]'
negate: false
match: after
filebeat.prospectors:
- input_type: log
paths: ["/var/log/**/*.log"]
multiline.pattern: '^['
multiline.match: after
filebeat.prospectors:
- document_type: tomcat
paths:
- /var/log/java/log #日志文件地址
input_type: log #从文件中读取
tail_files: true #以文件末尾开始读取数据
multiline:
pattern: ^\d{4}
match: after
negate: true
# logstash中FILTERS配置
filter {
if [type] == "tomcat" {
grok{
match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{JAVALOGMESSAGE:msg}" }
}
date {
match => [ "timestamp" , "yyyy-MM-dd HH:mm:ss,S", "ISO8601" ]
}
}
}
##################################################
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]
output
目前 beat 可以发送数据给 Elasticsearch、Logstash、File、Kafka、Redis 和 Console 六种目的地址。
Elasticsearch
beats 发送到 Elasticsearch 也是走 HTTP 接口。示例配置段如下：
output:
elasticsearch:
hosts: ["http://120.92.36.21:9200", "https://onesslip:9200/path", "anotherip"]
parameters: {pipeline: my_pipeline_id} # 仅用于 Elasticsearch 5.0 以后的 ingest 方式
username: "user"
password: "pwd"
index: "topbeat"
bulk_max_size: 20000
flush_interval: 5
tls:
certificate_authorities: ["/etc/pki/root/ca.pem"]
certificate: "/etc/pki/client/cert.pem"
certificatekey: "/etc/pki/client/cert.key"
hosts 中可以通过 URL 的不同形式，来表示 HTTP 还是 HTTPS，是否有添加代理层的 URL 路径等情况。
index 表示写入 Elasticsearch 时索引的前缀，比如示例即表示索引名为 topbeat-yyyy.MM.dd
filebeat --> logstash
下面是一个简单的Filebeat配置，采集2个文件夹下的日志并转发至Logstash。
filebeat:
prospectors:
-
paths:
- /dir1/access_log.*
input_type: log
document_type: dir1_log
-
paths:
- /dir2/ofbiz.log.*
input_type: log
document_type: dir2_log
output:
logstash:
hosts: ["10.90.4.9:5044"]
在Logstash中根据 document_type定义解析日志的正则并输出到ELasticsearch集群。
input {
beats{
host => "192.2.11.145"
port => 5044
}
}
filter {
if[type]=="dir1_log"{
grok {
match => { "message" => "%{COMBINEDAPACHELOG}"}
}
} else if ([type]=="dir2_log") {
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:time}\s*%{NUMBER:logtime} \[\s*%{JAVAFILE:class}\:%{NUMBER:lineNumber}\s*\:%{LOGLEVEL:level}\s*\]\s*(?<info>([\s\S]*))"}
}
}
}
output {
elasticsearch {
hosts => ["10.90.4.9","10.90.4.8","10.90.4.7"]
}
}