Filebeat收集日志數據傳輸到Redis，通過Logstash來根據日志字段創建不同的ES索引

作者：非著名運維 2022-04-28 23:28:07

在redis中顯示的效果是都會輸出到key值nginx_log對應的列表中，根據key值是沒法進行區分的，只能根據key值列表中每一行數據中的log_source或者自己定義的屬性來判斷該行是哪一個應用的日志。

1.Filebeat.yml配置

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /usr/local/nginx/logs/access.log
  exclude_files: ['.gz$','INFO']
  multiline.pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2}'
  multiline.negate: true
  multiline.match: after
  tags: ["nginx-log-messages"]
  fields:
    log_source: messages
  fields_under_root: true


output.redis:
  hosts: ["192.168.0.111:6379"]
  key: nginx_log
  password: nginxredis
  db: 0

參數說明：

  fields:
    log_source: messages
  fields_under_root: true

使用fields表示在filebeat收集的日志中多增加一個字段log_source,其值是messages，用來在logstash的output輸出到elasticsearch中判斷日志的來源，從而建立相應的索引若fields_under_root設置為true，表示上面新增的字段是頂級參數。

頂級字段在output輸出到elasticsearch中的使用如下：

[root@es-master21 logstash]# vim config/logstash.conf
input {
  redis {
    data_type => "list"
    host => "192.168.0.111"
    db => "0"
    port => "6379"
    key => "nginx_log"
    password => "nginxredis"
  }
}

output {
  # 根據redis鍵 messages_secure 對應的列表值中，每一行數據的其中一個參數來判斷日志來源
  if [log_source] == 'messages' {  # 注意判斷條件的寫法
    elasticsearch {
      hosts => ["192.168.0.111:9200"]
      index => "nginx-message-%{+YYYY.MM.dd}"
      #user => "elastic"
      #password => "elastic123"
    }
  }

  #或者也可以根據tags判斷
  if "nginx-log-messages" in [tags] {
    elasticsearch {
        hosts => [""192.168.0.111:9200"]
        index => "nginx-message-%{+YYYY.MM.dd}"
    }
  }

}

2.多個應用的日志都輸出到redis

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /usr/local/nginx/logs/access.log
  tags: ["nginx-log-access"]
  fields:
    log_source: access
  fields_under_root: true

- type: log
  enabled: true
  paths:
    - /usr/local/nginx/logs/error.log
  tags: ["nginx-log-error"]
  fields:
    log_source: error
  fields_under_root: true

output.redis:
  hosts: ["192.168.0.111:6379"]
  key: nginx_log
  password: nginxredis
  db: 0

?在redis中顯示的效果是都會輸出到key值nginx_log對應的列表中，根據key值是沒法進行區分的，只能根據key值列表中每一行數據中的log_source或者自己定義的屬性來判斷該行是哪一個應用的日志。

3.不同的應用日志使用不同的rediskey值

使用output.redis中的keys值，官方例子：

output.redis:
  hosts: ["localhost"]
  key: "default_list"
  keys:
    - key: "error_list"   # send to info_list if `message` field contains INFO
      when.contains:
        message: "error"
    - key: "debug_list"  # send to debug_list if `message` field contains DEBUG
      when.contains:
        message: "DEBUG"
    - key: "%{[fields.list]}"

說明：默認的key值是default_list，keys的值是動態分配創建的，當redis接收到的日志中message字段的值包含有error字段，則創建key為error_list，當包含有DEBUG字段，則創建key為debug_list。

問題的解決方法是在每個應用的輸出日志中新增一個能夠區分這個日志的值，然后再在keys中設置，這樣一來就能夠把不同應用的日志輸出到不同的redis的key中。

責任編輯：武曉燕來源：非著名運維