输入

input {
    #标准输入
    stdin {
        codec => "plain"
    }

    http {
        host => "0.0.0.0"
        port => "8099"
    }

    #rsyslog
    syslog{
        type => "system-syslog"
        port => 514
    }

    #beats系列，如filebeat
    beats {
        port => 5044
        host => "0.0.0.0"
    }

    #从文件读取数据
    file{
        path => ['/var/log/nginx/access.log']  #要输入的文件路径
        type => 'nginx_access_log'
        start_position => "beginning"
    }
    # path  可以用/var/log/*.log,/var/log/**/*.log，如果是/var/log则是/var/log/*.log
    # type 通用选项. 用于激活过滤器
    # start_position 选择logstash开始读取文件的位置，begining或者end。
    # 还有一些常用的例如：discover_interval，exclude，sincedb_path,sincedb_write_interval等可以参考官网

    #rsyslog 通过网络将系统日志消息读取为事件
    syslog{
        port =>"514"
        type => "syslog"
    }
    # port 指定监听端口(同时建立TCP/UDP的514端口的监听)
    #从syslogs读取需要实现配置rsyslog：
    # cat /etc/rsyslog.conf   加入一行
    # *.* @172.17.128.200:514　  #指定日志输入到这个端口，然后logstash监听这个端口，如果有新日志输入则读取
    # service rsyslog restart   #重启日志服务

    #kafka 将 kafka topic 中的数据读取为事件
    kafka{
        bootstrap_servers=> "kafka01:9092,kafka02:9092,kafka03:9092"
        topics => ["access_log"]
        #group_id => "logstash-file"
        codec => "json"
    }
    # bootstrap_servers 用于建立群集初始连接的Kafka实例的URL列表。
    # topics  要订阅的主题列表，kafka topics
    # group_id 消费者所属组的标识符，默认为logstash。kafka中一个主题的消息将通过相同的方式分发到Logstash的group_id
    # codec 通用选项，用于输入数据的编解码器。
}

还有很多的input插件类型，可以参考官方文档来配置。

输出

output {
    elasticsearch {
        hosts => ["elasticsearch:9200"]
        index => "system-syslog-%{+YYYY.MM}"
    }
    file {
       path => "/var/log/nginx/%{host}/save.txt"
       codec => line { format => "%{message}" }
    }
    kafka {
        codec => json
        topic_id => "mytopic"
    }
    stdout { codec => rubydebug}
}

调试

logstash -e 'input { stdin{} } filter { grok { patterns_dir => "/usr/share/logstash/patterns" match => { "message" => "%{NGINX_ACCESS}" } }} output { stdout {} }' 

#接着输入
172.19.0.1 - - [08/Mar/2022:08:20:29 +0000] "GET / HTTP/1.1" 404 153 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:97.0) Gecko/20100101 Firefox/97.0" "-"
#响应
{
          "bytes" => "153",
           "host" => "centos8.localdomain",
       "@version" => "1",
           "verb" => "GET",
          "agent" => "\"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:97.0) Gecko/20100101 Firefox/97.0\"",
        "request" => "/",
    "httpversion" => "1.1",
     "@timestamp" => 2022-03-18T09:47:04.498Z,
        "message" => "172.19.0.1 - - [08/Mar/2022:08:20:29 +0000] \"GET / HTTP/1.1\" 404 153 \"-\" \"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:97.0) Gecko/20100101 Firefox/97.0\" \"-\"",
      "forwarder" => "\"-\"",
       "clientip" => "172.19.0.1",
          "ident" => "-",
      "timestamp" => "08/Mar/2022:08:20:29 +0000",
       "response" => "404",
       "referrer" => "\"-\""
}

#/usr/share/logstash/patterns/nginx
NGINX_ACCESS %{IPORHOST:clientip} (?:-|(%{WORD}.%{WORD})) %{USER:ident} \[%{HTTPDATE:timestamp}\] "(?:%{WORD:verb} %{NOTSPACE:request}(?: HTTP/%{NUMBER:httpversion})?|%{DATA:rawrequest})" %{NUMBER:response} (?:%{NUMBER:bytes}|-) %{QS:referrer} %{QS:agent} %{QS:forwarder}

默认patterns：/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-patterns-core-4.3.2/patterns

调试工具：http://grokdebug.herokuapp.com/

代码

https://github.com/chudaozhe/efk/tree/master/logstash

相关链接

https://www.cnblogs.com/wzxmt/p/11031110.html

https://www.jmsite.cn/blog-855.html

https://www.elastic.co/guide/en/logstash/7.17/input-plugins.html

https://www.elastic.co/guide/en/logstash/7.17/filter-plugins.html

https://www.elastic.co/guide/en/logstash/7.17/output-plugins.html