实验环境：关闭防火墙，完成java环境

yum -y install wget

wget https://d6.injdk.cn/oraclejdk/8/jdk-8u341-linux-x64.rpm

yum localinstall jdk-8u341-linux-x64.rpm -y

java -version

1.安装logstash

tar xf logstash-6.4.1.tar.gz -C /usr/local

ln -s /usr/local/logstash-6.4.1 /usr/local/logstash

2.修改配置文件

cd /usr/local/logstash/config/

vim logstash.yml

http.host: "0.0.0.0"

3.编写规则文件

cd /usr/local/logstash/config/

cp logstash-sample.conf logstash-ipput-output.conf

vim logstash-ipput-output.conf

input {
   stdin {}
}
output {
   stdout {}
}

或者

input {
   stdin {}
}
output {
  elasticsearch {
    hosts => ["http://192.168.148.132:9200"]    #es的主机IP
    index => "test-logstash-%{+YYYY.MM.dd}"
  }
}

4.测试

ln -s /usr/local/logstash/bin/* /usr/local/bin/

logstash -f logstash-input-output.conf

hello

去192.168.148.132:9200的head插件里就可以看到：

5.filter简单实验：

cd /usr/local/logstash/config/

cp logstash-sample.conf logstash-test.conf

vim logstash-test.conf

logstash -f logstash-input-output.conf

192.168.10.11 - - [22/Oct/2019:22:49:53 -0400] "GET / HTTP/1.1" 200 5 "-" "curl/7.29.0"

input {
   stdin {}
}

filter {
  grok  {
    pattern_definitions => {
      "IP" => "([0-9]+\.){3}[0-9]+"
      "TIME" => ".*"
      "METHOD" => "[A-Z]+"
      "URL" => "/.*"
      "VERSION" => "\d.\d"
      "CODE" => "[1-5]\d\d"
      "SEND" => "[0-9]+"
      "REF" => ".*"
      "AGENT" => ".*"
    }
    match => {
      "message" => "%{IP:ip}.*\[%{TIME:time}\] \"%{METHOD:method} %{URL:url} HTTP/%{VERSION:version}\" %{CODE:code} %{SEND:send} \"%{REF:referer}\" \"%{AGENT:agent}\""
    }
    remove_field => ["message","@timestamp","@version"]           #不显示该选项内容
  }
}

output {
   stdout {}
}

6.filter引用文件：

vim /tmp/logstash_test.sh

IP  ([0-9]+\.){3}[0-9]+
TIME  .*
METHOD  [A-Z]+
URL  /.*
VERSION  \d.\d
CODE  [1-5]\d\d
SEND  [0-9]+
REF  .*
AGENT  .*
TEST  %{IP:ip}.*\[%{TIME:time}\] \"%{METHOD:method} %{URL:url} HTTP/%{VERSION:version}\" %{CODE:code} %{SEND:send} \"%{REF:referer}\" \"%{AGENT:agent}\"

使用patterns_dir参数指定文件

vim /usr/local/logstash/config/logstash-test.conf

input {
   stdin {}
}
filter {
  grok  {
    patterns_dir => ["/tmp/logstash_test.sh"]
    match => {
      "message" => "%{TEST}"
    }
    remove_field => ["message","@timestamp","@version"]
  }
}

output {
   stdout {}
}