一、我的痛點

1、傳統(tǒng)一般是使用filebeat的模塊收集日志，而當(dāng)一個linux服務(wù)器上有多種日志要收時，只能全部吐到logstash，因為7.X版本不再支持多個輸出源。
2、這里以system模塊為典型例子，通常收集以后，在kibana上會經(jīng)常找不到想要的日志。
3、一是因為存在es中的時間不是日志本身時間，而是filebeat收集到的時間。
4、二是因為如果使用其他輸出源，其filebeat模塊不會正常解析日志。

二、思路

1、在linux上運行多個filebeat實例，一個直接輸出到es，一個輸出到logstash
2、同時建議es使用專門的負載均衡節(jié)點來承受輸出，不要直接輸出到數(shù)據(jù)節(jié)點

靈感來源于https://zh.codepre.com/how-to-19067.html
這個問題困擾了我很久，百度翻完了也找不到合適的方案，大部分講的都比較淺顯，在bing才找到這篇文章，實在感謝大神。

我這里采用systemd方式

三、實現(xiàn)收集模塊日志(以system為例，輸出到es)

1、編輯多filebeat實例腳本

#復(fù)制filebeat配置目錄
cp -r /etc/filebeat{,-elasticsearch}

#編輯systemd服務(wù)
cat > /etc/systemd/system/filebeat-elasticsearch.service << "EOF"
[Unit]
Description=Filebeat sends log files to directly to Elasticsearch.
Documentation=https://www.elastic.co/products/beats/filebeat
Wants=network-online.target
After=network-online.target

[Service]

Environment="BEAT_LOG_OPTS="
Environment="BEAT_CONFIG_OPTS=-c /etc/filebeat-elasticsearch/filebeat.yml"
Environment="BEAT_PATH_OPTS=--path.home /usr/share/filebeat --path.config /etc/filebeat-elasticsearch --path.data /var/lib/filebeat-elasticsearch --path.logs /var/log/filebeat-elasticsearch"
ExecStart=/usr/share/filebeat/bin/filebeat --environment systemd $BEAT_LOG_OPTS $BEAT_CONFIG_OPTS $BEAT_PATH_OPTS
Restart=always

[Install]
WantedBy=multi-user.target
EOF

2、編輯/etc/filebeat-elasticsearch/filebeat.yml

filebeat.inputs:
- type: log
  enabled: false
  paths:
    - /var/log/*.log

- type: filestream
  enabled: false
  paths:
    - /var/log/*.log

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

setup.template.settings:
  index.number_of_shards: 1

setup.kibana:
  host: "192.168.18.13:5601"
setup.template.overwrite: true
setup.template.enabled: true
setup.ilm.enabled: false

output.elasticsearch:
  hosts: ["192.168.18.13:9200"]
  indices:
    - index: "os-linux-auth-%{+yyyy.MM.dd}"
      when.equals:
        event:
          dataset: "system.auth"
    - index: "os-linux-syslog-%{+yyyy.MM.dd}"
      when.equals:
        event:
          dataset: "system.syslog"

processors:
  - add_host_metadata:
      when.not.contains.tags: forwarded

3、啟動

mv /etc/filebeat-ealsticsearch/modules/system.yml.disable /etc/filebeat-ealsticsearch/modules/system.yml
systemctl start filebeat-ealsticsearch

4、在kibana上查看日志

四、收集非模塊日志（以tomcat為例，輸出到logstash）

1、準(zhǔn)備日志樣本

2、復(fù)制filebeat-logstash

#復(fù)制filebeat配置目錄
cp -r /etc/filebeat{,-logstash}

#編輯systemd服務(wù)
cat > /etc/systemd/system/filebeat-logstash.service << "EOF"
[Unit]
Description=Filebeat sends log files to Logstash.
Documentation=https://www.elastic.co/products/beats/filebeat
Wants=network-online.target
After=network-online.target

[Service]

Environment="BEAT_LOG_OPTS="
Environment="BEAT_CONFIG_OPTS=-c /etc/filebeat-logstash/filebeat.yml"
Environment="BEAT_PATH_OPTS=--path.home /usr/share/filebeat --path.config /etc/filebeat-logstash --path.data /var/lib/filebeat-logstash --path.logs /var/log/filebeat-logstash"
ExecStart=/usr/share/filebeat/bin/filebeat --environment systemd $BEAT_LOG_OPTS $BEAT_CONFIG_OPTS $BEAT_PATH_OPTS
Restart=always

[Install]
WantedBy=multi-user.target
EOF

3、編輯/etc/filebeat-logstash/filebeat.yml

filebeat.inputs:
- type: log
  enabled: false
  paths:
    - /var/log/*.log

- type: log
  enabled: true
  paths:
    - /etc/filebeat-logstash/testlog/localhost_access_log.*.txt
  fields:
    type_name: "web-tomcat-access"
  #將fieds添加的字段置于文檔根路徑
  fields_under_root: true

- type: filestream
  enabled: false
  paths:
    - /var/log/*.log

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


setup.template.settings:
  index.number_of_shards: 1

#setup.kibana:
#  host: "192.168.18.13:5601"
setup.template.overwrite: true
setup.template.enabled: true
setup.ilm.enabled: false

output.kafka:
  hosts: ["192.168.18.15:9092","192.168.18.16:9092"]
  topics:
    - topic: "web-tomcat-access"
      when.equals:
        type_name: "web-tomcat-access"
  

processors:
  - add_host_metadata:
      when.not.contains.tags: forwarded

4、編輯logstash
4.1、input.conf

input {
    kafka {
                bootstrap_servers => "192.168.18.15:9092,192.168.18.16:9092"
                topics => ["web-tomcat-access"]
                consumer_threads => 5
                codec => json
        }
}

4.2、tomcat_out.conf

filter {
    if "web-tomcat-access" in [type_name] {
        grok {
            match => ["message", "%{IPORHOST:client_ip} (%{USER:ident}|-) (%{USER:auth}|-) \[%{HTTPDATE:log_time}\] \"(?:%{WORD:verb} %{NOTSPACE:request_url}(?: HTTP/%{NUMBER:http_version})?|-)\" %{NUMBER:response} (%{NUMBER:bytes}|-)" ]
        }
                date {
                        match => [ "log_time" ,"dd/MMM/YYYY:HH:mm:ss Z" ]
                        target => "@timestamp"
                }
                geoip {
            source =>"client_ip"
            target =>"geoip"
            database =>"/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-filter-geoip-6.0.3-java/vendor/GeoLite2-City.mmdb"            
            add_field => ["[geoip][coordinates]", "%{[geoip][longitude]}" ]
            add_field => ["[geoip][coordinates]", "%{[geoip][latitude]}" ]
        }
    }
}

output {
    if "web-tomcat-access" in [type_name] {
                elasticsearch {
                        hosts => ["192.168.18.13:9200"]
                        index => "web-tomcat-%{+YYYY.MM.dd}"
                }
    }
}

5、啟動logstash，在kibana查看日志