测试启动

标准输入显示到终端

test.yml

  1. filebeat.inputs:
  2. - type: stdin
  3.   enabled: true
  4. output.console:
  5.   pretty: true
  6.   enable: true

filebeat -e -c test.yml

读取文件显示到终端

nginx.yml

  1. filebeat.inputs:
  2. - type: log
  3.   enabled: true
  4.   paths:
  5.   - /data/nginx/logs/default*
  6. output.console:
  7.   pretty: true
  8.   enable: true

filebeat -e -c nginx.yml

启动进程日志

logging.level: XXX
debug, info(默认), warning, or error

path.logs: /xxx/xxx
默认 安装目录/logs/filebeat
启动或重启生成新的，旧的命名为 filebeat.1 类推

image.png

读取文件状态保持

保存每个文件的状态并经常将状态刷新到磁盘上的注册文件中
该状态用于记住harvester正在读取的最后偏移量，并确保发送所有日志行
如果输出（例如Elasticsearch或Logstash）无法访问，Filebeat会跟踪最后发送的行，并在输出再次可用时继续读取文件
数据目录/registry/filebeat/

在Filebeat运行时，每个prospector内存中也会保存文件状态信息，当重新启动Filebeat时，将使用注册文件的数据来重建文件状态，Filebeat将每个harvester在从保存的最后偏移量继续读取。

每个prospector为它找到的每个文件保留一个状态。由于文件可以被重命名或移动，因此文件名和路径不足以识别文件。对于每个文件，Filebeat存储唯一标识符以检测文件是否先前已被采集过。

专用日志搜集模块

以 nginx 为例

m.yml

filebeat.config.modules:
path: ${path.config}/modules.d/*.yml

filebeat -c m.yml modules list # 列出支持的
image.png

filebeat -c m.yml modules enable nginx # 启用、禁用
filebeat -c m.yml modules disable nginx # 启用、禁用

  1. - module: nginx
  2.   # Access logs
  3.   access:
  4.     enabled: true
  5.     var.paths: /var/log/nginx/access.log* # 默认访问日志路径
  7.   # Error logs
  8.   error:
  9.     enabled: true
  10.     var.paths: /var/log/nginx/error.log* # 默认错误日志路径

var.paths 接收一个数组

  1. var.paths: ["xx1.log","xx2.log"]
  3. 或
  5. var.paths:
  6.  -  "xx1.log"
  7.  -  "xx2.log"

input

行过滤

  1. filebeat.inputs:
  2. - type: log
  3.   enabled: true
  4.   paths:
  5.   - /var/log/messages
  6.   - /var/log/secure
  8.   fields:
  9.     type: oslog
  11.   exclude_lines: ["user1","admin"]
  12.   include_lines: ["Failed"]

单条日志有多行处理

multiline:
pattern: ‘^[‘
negate: true
match: after

  1. filebeat.inputs:
  2. - type: log
  3.   enabled: true
  4.   paths:
  5.   - /usr/local/tomcat/logs/catalina.out
  6.   fields:
  7.     type: tomcat
  8.   fields_under_root: true
  9.   multiline:
  10.     pattern: '^\['
  11.     negate: true
  12.     match: after

自定义字段

tags 添加值

tags：[“aaa”,”bbb”]
image.png

fields 添加键值

fields:
name1: “aaa”
name2: “啊啊啊”
image.png

fields_under_root: true # 删除 fields父级字段，置于顶级字段
image.png

output

console

  1. output.console:
  2.   pretty: true
  3.   enable: true
  4.   codec.format:        # 只输出指定字段拼成的字符串
  5.     string: '%{[@timestamp]} %{[message]}'

es

指定索引名

  1. output.elasticsearch:
  2.   hosts: ["http://es:9200"]
  3.   index: "%{[fields.type]}-%{+yyyy.MM.dd}"
  4. setup.template.name: "filebeat"
  5. setup.template.pattern: "filebeat-*"

kafka

  1. filebeat.inputs:
  2. - type: log
  3.   enabled: true
  4.   paths:
  5.   - /var/log/message
  6.   - /var/log/secure
  7. fields:
  8.   type: oslog
  9.   name: "192.168.80.191"
  12. output.kafka:
  13.   enabled: true
  14.   hosts: ["192.168.80.81:9092","192.168.80.82:9092","192.168.80.83:9092"]
  15.   topic: "%{[fields][type]}"
  16.   partition.round_robin:
  17.     reachable_only: true
  18.   required_acks: 1
  19.   compression: gzip
  20.   max_message_bytes: 1000000
  21. logging.level: warning

logstash

  1. output.logstash:
  2.   hosts: ["localhost:5044"]

processors 处理

丢弃符合正则的行

  1. processors:
  2. - drop_event:
  3.     when:
  4.       regexp:
  5.         message: "^ab"    # 丢弃 message 中 ab 开头的行

过略字段

  1. processors:
  2. - drop_fields:
  3.     fields:
  4.     - "log"
  5.     - "input"
  6.     - "ecs.version"    # 子级字段
  7.     - "agent"

添加字段

  1. processors:
  2. - add_fields:
  3.     target: a1
  4.     fields:
  5.       k1: v1
  6.       k2: v2

image.png

添加元数据信息

  1. processors:
  2. - add_host_metadata:
  3.     when.not.contains.tags: forwarded
  4. - add_cloud_metadata: ~
  5. - add_docker_metadata: ~
  6. - add_kubernetes_metadata: ~