部署logstash

Logstash是免费且开放的服务器端数据处理管道，能够从多个来源采集数据，转换数据，然后将数据发送到您最喜欢的“存储库”中，Logstash的作用就是在于日志处理，转换操作。

1.下载logstash

1
[root@elk01:1 ~]#   wget https://artifacts.elastic.co/downloads/logstash/logstash-7.17.23-amd64.deb

2.安装logstash

1
[root@elk01:1 ~]# dpkg -i logstash-7.17.23-amd64.deb

3.添加软链接

1
2
3
4
[root@elk01:1 ~]# ln -svf /usr/share/logstash/bin/logstash /usr/local/bin/

#查看帮助
[root@elk01:1 ~]# logstash -h

4.启动测试

1
2
3
4
5
6
7
8
9
10
11
[root@elk01:1 ~]#  logstash -e "input { stdin { type => stdin } } output { stdout { codec => rubydebug } }"
···
The stdin plugin is now waiting for input:
111111
{
          "type" => "stdin",
    "@timestamp" => 2024-10-05T01:48:28.131Z,
       "message" => "111111",
          "host" => "elk01",
      "@version" => "1"
}

温馨提示：如果你出现了和我一样报错如下

[ERROR] 2024-10-05 01:25:25.226 [pool-3-thread-1] jvm - Unknown garbage collector name {:name=>“G1 Concurrent GC”}

1
2
3
4
5
6
7
8
9
表示 Logstash 不识别 "G1 Concurrent GC" 作为垃圾回收器的名称。这可能是因为在 `jvm.options` 配置文件中指定了错误的垃圾回收器名称。

解决：编辑 jvm.options 文件：
[root@elk01:1 ~]# vim /etc/logstash/jvm.options 

#修改或添加 JVM 参数：
在 jvm.options 文件中，应该设置 -XX:+UseG1GC 来启用 G1 垃圾回收器，而不是使用 "G1 Concurrent GC" 这样的名称。找到任何提及垃圾回收器的行，没有就添加，之后重启logstash

-XX:+UseG1GC

logstash组件

logstash有三个核心组件，分别为input，filter和output。其中，filter组件是可选组件。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
logstash多实例及pipeline概念
	input：
		数据从哪里来，可以是stdin,tcp,file,kafka,redis,...
		推荐阅读:
			https://www.elastic.co/guide/en/logstash/7.17/input-plugins.html
			
	filter:
		对数据进行过滤操作，常用的插件有: grok,date,mutate,user_agent,geoip,json,...
		推荐阅读:
			https://www.elastic.co/guide/en/logstash/7.17/filter-plugins.html
			
	output:
		数据到哪去，一般情况下都是写入elasticsearch或者stdout(测试)
		推荐阅读：
			https://www.elastic.co/guide/en/logstash/7.17/output-plugins.html
		

	一个节点可以部署多个Logstash实例，每个Logstash实例可以有多个pipeline，每个pipeline可以有input，filter和output插件。

logstash

logstash类型案例

yes的为必选参数

image-20241005110046458

input-tcp类型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
#1.编写logstash实例，标准输出stdout
[root@elk01:2 ~]# cat /etc/logstash/conf.d/01_tcp_stdout.conf
input {
  tcp {
    port => "8888"
  }
}
output {
  stdout {}
}

#2.启动logstash实例   （  -r   热加载，-f  后要跟实例）
[root@elk01:2 ~]# logstash -rf /etc/logstash/conf.d/01_tcp_stdout.conf 

#3.别的节点发送测试数据
[root@elk02:1 ~]# echo 999|nc 10.0.0.211 8888

#终端输出
{
       "message" => "999",
          "host" => "10.0.0.212",
    "@timestamp" => 2024-10-05T03:22:38.582Z,
      "@version" => "1",
          "port" => 46720
}
1
2
3
4
5
6
7
8
9
10
11
#标准输出的#，若不指定codec属性，则默认的解码器为rubydebug。
output {
  stdout {
   # codec => rubydebug
   # 此处指定的codec的解码器为json格式
   codec => json 
  }
}

#终端输出则为
{"@timestamp":"2024-08-29T07:10:55.639Z","@version":"1","host":"10.0.0.91","message":"9999","port":58096}

input-file类型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
#1.编写logstash实例，标准输出stdout（注意path类型是数组，详细看官网） 
[root@elk01:3 ~]# cat /etc/logstash/conf.d/02_file_stdout.yaml 
input {
  file {
    path => ["/tmp/test.conf"]
  }
}
output {
  stdout {}
}

#2.启动logstash实例
[root@elk01:2 ~]# logstash -rf /etc/logstash/conf.d/02_file_stdout.yaml 

#3.发送测试数据
[root@elk01:3 ~]# echo '666' >>/tmp/test.conf

#logstash的标准输出
{
          "host" => "elk01",
      "@version" => "1",
    "@timestamp" => 2024-10-05T03:34:23.489Z,
       "message" => "666",
          "path" => "/tmp/test.conf"
}

logstash的采集缓存

​ 当我们停掉实例2时候，再次往test.conf里写入新数据 7890 ，重新执行实例2时候，发现数据并不会重新采集，而是只采集新数据，这种现象同filebeat一样，将采集过的记录放在缓存文件里，下次采集前先读取这个文件，看看采集的偏移量，那这个文件在哪呢？

1
2
3
4
5
启动实例时候发现读取文件-->/usr/share/logstash/data/plugins/inputs/file/.sincedb_387337cd9c394c04fb5a33c176cf8715", :path=>["/tmp/test.conf"]}

#查看这个文件，里面记录了这个文件已经被采集的信息
[root@elk01:3 ~]# cat /usr/share/logstash/data/plugins/inputs/file/.sincedb_387337cd9c394c04fb5a33c176cf8715 
788369 0 64768 14 1728099475.853982 /tmp/test.conf

如何实现文件重新采集？

1.清空这个文件喽，直接清空这一级目录所有文件即可

rm -f /usr/share/logstash/data/plugins/inputs/file/.sincedb*

2.设置文件采集为从头采集

1
2
3
4
5
6
7
8
9
10
[root@elk01:3 ~]# vim /etc/logstash/conf.d/02_file_stdout.yaml 
input {
file {
 path => ["/tmp/test.conf","/tmp/test2.log"]  #path数组，支持采集多个文件
 start_position => "beginning"
 }
}
output {
 stdout {}
}

start_position 默认值end 从文件最后位置采集，有效值为 beginning end

logstash的多实例

1
2
同filebeat一样，启动时候指定不同数据目录    --path.data
logstash -rf /etc/logstash/conf.d/02-file-to-stdout.conf --path.data /tmp/logstash-data