写在前面

在 ELK日志分析体系中，Filebeat 扮演着“日志采集器”的角色，是整个数据流动链条的起点。它专注于从服务器本地读取日志文件，并将其可靠地转发至 Logstash 或 Elasticsearch，相当于 ELK 体系中的“前哨”。

相比传统的 Logstash agent，Filebeat 更加轻量，占用资源更少，适合部署在大规模分布式节点上。它以高性能、低延迟的方式，将系统、服务或应用产生的日志源源不断地采集并发送，为后续的日志过滤、分析与可视化打下坚实基础。

可以说，没有稳定的日志采集，就没有可靠的日志分析；而 Filebeat 正是 ELK 中实现这一目标的关键一环。

1. 安装和部署

获取软件包

下载有限制的小伙伴可以直接使用下载好的：filebeat-7.17.5

https://www.elastic.co/downloads/elasticsearch

这里我们要下载tar.gz包

--- 1. 上传软件包至服务器上
--- 2. 解压
tar xf filebeat-7.17.5-linux-x86_64.tar.gz
--- 3. 创建软连接
ln -s /software/filebeat-7.17.5-linux-x86_64/filebeat   /usr/local/sbin/
--- 4. 验证 出现信息表示成功
filebeat -h

2. 轻度体验filebeat

参考官网文档

https://www.elastic.co/guide/en/beats/filebeat/7.17/console-output.html

1. filbeat的input类型之sdtin

--- 1. 创建工作目录
mkdir -p config 
--- 2. 编写配置文件
cat config/01-stdin-to-console.yml
## 配置filebeat的输入端
filebeat.inputs:
  ## 指定输入端的类型为标准输入
- type: stdin

## 指定filebeat的输出端为console
output.console:
  ## 表示输出的内容以漂亮格式显示
  pretty: true
--- 3. 启动filebeat实例
filebeat -e -c config/01-stdin-to-console.yaml

2. filbeat的input类型之tcp

--- 1. 编写配置文件
cat 02-tcp-to-console.yaml 
filebeat.inputs:
  ## 指定类型
- type: tcp
  ## 指定tcp监听的端口
  host: "0.0.0.0:8888"

output.console:
  pretty: true

--- 2. 启动实例
filebeat -e -c 02-tcp-to-console.yaml

--- 3. 测试 观察是否有日志打出来
echo 'AAAA' |nc 192.168.0.160 8888

3. filebeat的input类型之通用字段（common options）

--- 1. 编写配置文件
cat 03-common-to-console.yaml
filebeat.inputs:
- type: log
  paths:
    - /tmp/filebeat/*.log
  ## 是否启用该类型,默认是true
  enabled: true 
    #- type: tcp
    #enabled: false
    #host: "0.0.0.0:8888"
  ## 给数据打标签,会在顶级字段多出来多个标签
  tags: ["test","xingzhibang","elk"]
  ## 给数据添加key-value 类型的字段,默认是放在了"fields"字段下
  fields:
    function: test
    name: xingzhibang
    soft: elk 
  ## 若设置为true时，则将fields添加的自定义字段放在顶级字段中，默认值为false。
  fields_under_root: true
  # 定义处理器，过滤指定的数据
  processors:
    ## 删除消息是以linux开头的时间
  - drop_event:
      when:
        regexp:
          message: "^linux"
    ## 消息包含error内容事件就可以删除自定义字段或者tags
  - drop_fields:
      when:
        contains: 
          message: "error"
      fields: ["function", "tags"]
      ignore_missing: false
    ## 修改字段名称
  - rename:
     fields:
         # 源字段
       - from: "function"
         # 目标字段
         to: "环境"
       - from: "name"
         to: "名称"
output.console:
  pretty: true

### 包含指定数据采集，排除指定数据采集及json格式数据采集案例
--- 编写配置文件
cat 05-log-to-console.yam
filebeat.inputs:
- type: log
  paths:
    - /tmp/filebeat/*
  # 排除以log结尾的文件
  #exclude_files: ['\.log$']
  # 只采集包含指定信息的数据
  #include_lines: ['linux']
  # 只要包含特定的数据就不采集该事件(event)
  #exclude_lines: ['^linux']
  # 将message字段的json数据格式进行解析，并将解析的结果放在顶级字段中
  json.keys_under_root: true
  # 如果解析json格式失败，则会将错误信息添加为一个"error"字段输出
  json.add_error_key: true

output.console:
  pretty: true

4. filebeat采集nginx 的json格式日志

--- 1. 安装nginx环境
--- 2. 编写配置文件
## vim /etc/nginx/nginx.conf
http{
 ....
  log_format main  '{"@timestamp":"$time_iso8601",'
		                     '"host":"$server_addr",'
		                     '"clientip":"$remote_addr",'
		                     '"SendBytes":$body_bytes_sent,'
		                     '"responsetime":$request_time,'
		                     '"upstreamtime":"$upstream_response_time",'
		                     '"upstreamhost":"$upstream_addr",'
		                     '"http_host":"$host",'
		                     '"uri":"$uri",'
		                     '"domain":"$host",'
		                     '"xff":"$http_x_forwarded_for",'
		                     '"referer":"$http_referer",'
		                     '"tcp_xff":"$proxy_protocol_addr",'
		                     '"http_user_agent":"$http_user_agent",'
		                     '"status":"$status"}';
	access_log /var/log/nginx/access.log main;
....
}
--- 3. 热加载nginx
nginx -t 
systemctl reload nginx 

--- 4. 测试访问nginx并查看日志
--- 5. filebeat采集nginx的json格式日志
filebeat.inputs:
- type: log
  paths:
    - /var/log/nginx/access.log*
  json.keys_under_root: true
  json.add_error_key: true
output.elasticsearch:
  hosts: ["http://192.168.0.160:9200"]
--- 6. 这里我将数据直接打到了es集群中，可以通过kibana来展示数据

5. filebeat采集tomcat访问日志:

--- 1. 前去tomcat官网获取tomcat软件包
--- 2. 解压软件包
tar xf apache-tomcat-9.0.108.tar.gz -C /software
--- 3. 修改tomcat配置文件
## /software/apache-tomcat-9.0.108/conf/server.xml  
## 大概在文件末的位置
<Host name="192.168.0.160"  appBase="webapps"
  unpackWARs="true" autoDeploy="true">
   <Valve className="org.apache.catalina.valves.AccessLogValve" directory="logs"
     prefix="192.168.0.160_access_log" suffix=".txt"
	  pattern="{&quot;clientip&quot;:&quot;%h&quot;,&quot;ClientUser&quot;:&quot;%l&quot;,&quot;authenticated&quot;:&quot;%u&quot;,&quot;AccessTime&quot;:&quot;%t&quot;,&quot;request&quot;:&quot;%r&quot;,&quot;status&quot;:&quot;%s&quot;,&quot;SendBytes&quot;:&quot;%b&quot;,&quot;Query?string&quot;:&quot;%q&quot;,&quot;partner&quot;:&quot;%{Referer}i&quot;,&quot;http_user_agent&quot;:&quot;%{User-Agent}i&quot;}"/>
--- 4. 配置环境变量并启动tomcat
## cat ~/.bashrc
export TOMCAT_HOME=/software/apache-tomcat-9.0.108
export PATH=$PATH:$TOMCAT_HOME/bin
source ~/.bashrc
/software/apache-tomcat-9.0.108/bincatalina.sh start
--- 5. 8080端口访问测试
--- 6. 使用filebeat采集tomcat访问日志
## 编写配置文件
filebeat.inputs:
- type: log
  paths:
    - /software/apache-tomcat-9.0.108/logs/*.txt
  json.keys_under_root: true
  json.add_error_key: true

output.elasticsearch:
  hosts: ["http://192.168.0.160:9200"]
--- 6. 这里我将数据直接打到了es集群中，可以通过kibana来展示数据

6. filebeat采集tomcat的错误日志多行匹配

## 编写配置文件
filebeat.inputs:
- type: log
  paths:
    - /software/apache-tomcat-9.0.108/logs/catalina*
  multiline.type: pattern
  multiline.pattern: '^\d{2}'
  multiline.negate: true
  multiline.match: after 
output.elasticsearch:
  hosts: ["http://192.168.0.160:9200"]
## 这里我将数据直接打到了es集群中，可以通过kibana来展示数据

7. filebeat采集docker日志

--- 1. 安装docker 环境
--- 2. 启动容器
## 查看容器完成id
docker ps --no-trunc
-- 3. 编写配置文件
filebeat.inputs:
- type: container
  paths: 
   ## 一般容器日志的存放位置
    - '/var/lib/docker/containers/**/*.log'
output.console:
  pretty: true

3. filebeat的input类型之filestream

## 编写配置文件
filebeat.inputs:
  # 指定类型为filestream，在7.16版本中已经弃用log类型
- type: filestream
  enabled: false
  paths:
    - /linux85.log
- type: filestream
  enabled: false
  paths:
    - /tmp/filebeat/docker.json
  # 配置解析
  parsers:
    # 配置json格式解析
    - ndjson:
       # 将错误消息记录到error字段中
       add_error_key: true
       # 如果解析的json格式字段和filebeat内置的顶级字段冲突，则覆盖，默认是不覆盖的。
       overwrite_keys: true
       # 将message解析的字段放入一个自定义的字段下。若不指定该字段，则默认解析的键值对会在顶级字段.
       target: linux
- type: filestream
  enabled: false
  paths:
    - /tmp/filebeat/test.log
  # 配置解析
  parsers:
    - multiline:
        type: count
        count_lines: 3

## 写入数据到ES集群
output.elasticsearch:
  # 指定ES集群地址
  hosts: 
  - "http://192.168.0.160:9200"
  - "http://192.168.0.161:9200"
  - "http://192.168.0.162:9200"
  # 指定索引
  index: "shopping-%{+yyyy.MM.dd}"

# 禁用索引声明管理周期，若不禁用则自动忽略自定义索引名称
setup.ilm.enabled: false
# 设置索引模板的名称
setup.template.name: "shopping"
# 指定索引模板的匹配模式
setup.template.pattern: "shopping-*"
# 是否覆盖原有的索引模板
setup.template.overwrite: true
# 设置索引模板
setup.template.settings:
 # 指定分片数量为8
 index.number_of_shards: 8
 # 指定副本数量为0
 index.number_of_replicas: 0

## 将多个数据写入到不同的集群中
output.elasticsearch:
  hosts: 
  - "http://192.168.0.160:9200"
  - "http://192.168.0.161:9200"
  - "http://192.168.0.162:9200"
  indices:
     - index: "docker-%{+yyyy.MM.dd}"
       when.contains:
         ## 这里的tags里的内容 需要在上面input 中指定呦
         tags: "docker"
     - index: "linux-%{+yyyy.MM.dd}"
       when.contains:
         tags: "linux"
     - index: "demo-%{+yyyy.MM.dd}"
       when.contains:
         tags: "demo"
setup.ilm.enabled: false
setup.template.name: "shoppig"
setup.template.pattern: "shopping-*"
setup.template.overwrite: true