input 插件用于指定输入源，一个 pipeline 可以有多个 input 插件，我们主要围绕下面几个 input插件进行介绍
stdin
file
beat
kafka
http

从标准输入读取数据，从标准输出中输出内容
cat /etc/logstash/conf.d/stdin_logstash.conf 
# 从终端中输入，输出到中端
input {
	stdin {
	    type => "stdin" # 自定义事件类型，可用于后续判断
	    tags => "stdin_type" # 自定义tag，用于后续事件判断
	}
}
output {
    stdout {
	codec => "rubydebug"
	}
}

# 执行 -r 表示不停止logstash下修改配置文件
/usr/share/logstash/bin/logstash  -f /etc/logstash/conf.d/stdin_logstash.conf

终端端中输入 test logstash，返回结果如下
          "type" => "stdin",
       "message" => "test logstash",
          "host" => "logstash-node1",
          "tags" => [
        [0] "stdin_type"
    ],
      "@version" => "1",
    "@timestamp" => 2021-10-30T07:34:57.440Z
}

从 file 文件中读取数据，然后输入至标准输入
cat file_logstash.conf 
input {
	file {
		path => "/var/log/test.log"
		type => "syslog"
		exclude => "*.gz" # 不监听的文件 
		start_position => "beginning" # 第一次从头开始读取文件 beginning or end
		stat_interval => "3" #定时检查文件是否更新，默认1s
   }
}
output {
	stdout {
		codec => rubydebug
	}
}

/usr/share/logstash/bin/logstash  -f /etc/logstash/conf.d/file_logstash.conf

echo "file logstash" > /var/log/test.log
{
      "@version" => "1",
          "path" => "/var/log/test.log",
          "type" => "syslog",
    "@timestamp" => 2021-10-30T07:58:42.699Z,
          "host" => "logstash-node1",
       "message" => "file logstash"
}

从filebeat文件中读取数据，然后输入至标准输入
cat beats_logstash.conf 
input {
	beats {
		port => 5044 # filebeat 发送数据到logstash主机的5044端口
	}
}
output {
	stdout {
		codec => rubydebug
	}
}

/usr/share/logstash/bin/logstash  -f /etc/logstash/conf.d/beats_logstash.conf

从kafka文件中读取数据，然后输出至标准输出
input {
	kafka {
		zk_connect =>
			"kafka1:2181,kafka2:2181,kafka3:2181"
			 group_id => "logstash"
			 topic_id => "apache_logs"
			 consumer_threads => 16
	}
}