简单实验：filebeat->logstash

filebeat配置:

# cd /usr/local/filebeat/

# cp filebeat.yml filebeat2.yml

# vim filebeat2.yml

filebeat.inputs:
- type: log
  paths:
    - /var/log/httpd/access_log

output.logstash:
  hosts: ["192.168.148.131:5044"]

# ./filebeat -c filebeat2.yml &

logstash配置:

# vim /usr/local/logstash/config/logstash-filebeat.conf

input {
    beats {
        port => 5044
        codec => json
    }
}
output {
    stdout {}
}

# logstash -f /usr/local/logstash/config/logstash-filebeat.conf

# curl 192.168.148.131:80 访问httpd测试

复杂实验：filebeat->logstash->es并采集多个日志

##用field和if、elif

filebeat配置:

# cd /usr/local/filebeat/

# cp filebeat.yml filebeat3.yml

# vim filebeat3.yml

# grep -vE "^$|^[[:space:]]*#" filebeat3.yml

filebeat.inputs:
- type: log
  paths:
    - /var/log/httpd/access_log
  fields:
    filetype: web                              # 用于区别不同的日志
  fields_under_root: true                   # 将自定义字段置于顶层

- type: log
  paths:
    - /var/log/secure
  fields:
    filetype: sys
  fields_under_root: true

output.logstash:
  hosts: ["192.168.148.131:5044"]

# ./filebeat -c filebeat3.yml &

logstash配置:

# vim /usr/local/logstash/config/logs.conf

input {
    beats {
        port => 5044
     }
}

filter {
    if [filetype] == "web" {
        grok {
            match => {
                "message" => "%{COMBINEDAPACHELOG}"
            }
            remove_field => ["message","beat","offset","tags","prospector"]
        }
    }
}

output {
    if [filetype] == "web" {
        elasticsearch {
            hosts => ["192.168.148.132:9200"]
            index => "http-%{+YYYY.MM.dd}"
        }
    } 
    else if [filetype] == "sys" {
        elasticsearch {
            hosts => ["192.168.148.132:9200"]
            index => "syslog-%{+YYYY.MM.dd}"
        }
    }
}

# logstash -f /usr/local/logstash/config/logs.conf

# curl 192.168.148.132:9200/_cat/indices? #测试查看