Logstash学习_基础

任务

1.理解logstash在ELK中的“位置”：数据清洗与格式化，也可以作为收集器，使用java编写，较为重量级，丰富的插件，具有较强的处理能力（https://www.elastic.co/guide/en/logstash/6.8/introduction.html）

2.logstash安装（使用tar压缩包安装，不需要安装java环境）https://www.elastic.co/guide/en/logstash/7.10/installing-logstash.html#installing-binary

3.认识config目录下的jvm.options的内容：修改Xmx和Xms值适合自己的虚拟机或服务器；了解logstash.yml和pipeline.yml，两个文件分别是logstash的主配置文件和加载配置文件的配置（类似filebeat，但logstash正式的启动方式原本就具备配置文件的加载文件，但启动动态加载需要在logstash.yml里面配置）；pipeline.yml里面写的是每一个外部的配置文件和指定多少个线程，当这两个配置文件完成配置后，启动logstash就只需要bin/logstash，测试时采用 bin/logstash -f xxx.conf 来指定配置文件

4.了解logstash的pipeline的工作流程（每一个配置文件就是一个管线，即pipeline）https://www.elastic.co/guide/en/logstash/7.10/pipeline.html，其配置文件也遵循这个编写顺序：

i: input与output，使用beats插件，从filebeat获取输入数据（https://www.elastic.co/guide/en/logstash/7.10/plugins-inputs-beats.html）， stdout作为输出（需检验）（https://www.elastic.co/guide/en/logstash/7.10/plugins-outputs-stdout.html）

ii: filter，对数据进行切割、格式化、字段重命名计算等

a. 了解mutate插件：使用stdin作为input，用mutate去给输入的信息增加字段：add_field，以及删除字段：remove_field

b. 了解date插件：使用stdin作为input，用date插件处理输入的时间字符串，例如“2022-05-01 00:00:00”，输出到logdatetime字段（需检验）

c. 了解drop插件：在上一步的基础上，输入不匹配的时间字符串，导致date插件失败，实现手动丢弃丢弃该条数据（需检验）

Logstash概述

简单来说logstash就是一根具备实时数据传输能力的管道，负责将数据信息从管道的输入端传输到管道的输出端；与此同时这根管道还可以让你根据自己的需求在中间加上滤网，Logstash提供里很多功能强大的滤网以满足你的各种应用场景。

logstash常用于日志系统中做日志采集设备，最常用于ELK中作为日志收集器使用

1

集中、转换和存储你的数据，是一个开源的服务器端数据处理管道，可以同时从多个数据源获取数据，并对其进行转换，然后将其发送到你最喜欢的“存储

Logstash基本架构

logstash的基本流程架构：input | filter | output 如需对数据进行额外处理，filter可省略。

image-20220511153509543

安装Logstash

官方下载zip包直接解压到机器里面, 不做赘述

由于不可抗力, 此处我用的是7.6.1版本

目录结构如下

image-20220511154215799

简单配置Logstash

jvm.options

config/jvm.options文件, 配置内存大小优化 可根据服务器的性能进行配置

1
2
3
4
5
## JVM配置
# Xms 表示内存的初始大小
# Xmx 表示内存的最大大小
-Xms1g
-Xmx1g

logstash.yml

配置文件config/logstash.yml是Logstash的主配置文件，详解如下(7.x版本)

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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
#使用分层表单来设置管道的批处理大小和批处理延迟
   pipeline:
     batch:
       size: 125        #管道批处理大小
       delay: 5             #管道批处理延迟
 
#若要表示与平面键相同的值：
   pipeline.batch.size: 125
   pipeline.batch.delay: 5
 
 
#节点名称，在集群中具备唯一性，默认为logstash主机的主机名
node.name: logstast-node1
 
#logstash及其插件所使用的数据路径，默认路径为logstash家目录下的data目录
path.data: /usr/local/logstash-7.0.0/data/
 
#管道的ID，默认为main
pipeline.id: main
 
#输入、输出及过滤器的总工作数量，也就是logstash的工作进程，此工作进程默认为主机的cpu核心数量
pipeline.workers: 16 
 
#在输入阶段，单个工作线程将从输入中收集的最大事件数，此事件数堆内存开销较大，内存开销可在jvm.options中设置堆内存大小来优化此选项
pipeline.batch.size: 125
 
#在将一个较小的批发送到filters+output之前,轮询下一个事件时等待的时间(以毫秒为单位)
pipeline.batch.delay: 50
 
#设置为true时，在强制关闭logstash期间，即使内存中还有事件，那么为true将会强制关闭，导致数据丢失；默认为false，false在强制关闭logstash期间，将拒绝退出，直到所有在管道中的事件被安全输出，再关闭。
pipeline.unsafe_shutdown: false
 
#指定管道配置的目录，在此目录下的所有管道配置文件都将被logstash读取，除管道配置外，不要放任何文件
path.config: /usr/local/logstash-7.0.0/conf.d/
 
#在启动时，测试配置是否有效并退出，检测配置文件是否正确，包括检测管道配置文件，默认为false
config.test_and_exit: true
 
#定期检查配置是否更改并重新加载管道，默认为false
config.reload.automatic: true
 
#logstash间隔多久检查一次配置中的更改，默认为3秒
config.reload.interval: 600s
 
#设置为true时，将完全编译的配置显示为调试日志消息
config.debug: false
 
#用于事件缓冲的内部排队模型;可以指定内存memory或者磁盘persisted，内存处理速度相对磁盘来说效率要高，默认为内存
queue.type: memory
 
#启用持久队列时将存储数据文件的目录路径,默认为logstash路径下的queue
path.queue: /usr/local/logstash-7.0.0/queue/
 
#启用持久队列时使用的页面数据文件的大小(queue.type: persisted)队列数据由分成页面的仅附加数据文件组成
queue.page_capacity: 64mb
 
#启用持久队列时队列中未读事件的最大数量(queue.type: persisted)，默认为0，0为无限制
queue.max_events: 0
 
#队列的总容量，以字节数表示，默认为1G，根据业务需求而定
queue.max_bytes: 1024mb
 
#启用持久队列时强制检查点之前最大的ACK事件数量(queue.type: persisted)，设置为0，表示无限制，默认为1024
queue.checkpoint.acks: 1024
 
#启用持久队列时强制检查点之前写入事件的最大数量(queue,type: persisted)，设置为0，表示无限制，默认为1024
queue.checkpoint.writes: 1024
 
#启用持久队列(queue,type: persisted)，强制在头部页面上设置检查点的间隔(以毫秒为单位)，有周期性检查点的默认值是1000毫秒
queue.checkpoint.interval: 1000
 
#用于指示logstast启用插件支持DLQ功能的标志，默认为false
dead_letter_queue.enable: false
 
#每个死信队列的最大大小，如果条目超过此设置会增加死信队列的大小，则会删除条目，默认为1024mb
dead_letter_queue.max_bytes: 1024mb
 
#为死信队列存储数据文件的目录路径
path.dead_letter_queue: /usr/local/logstash-7.0.0/letter-queue
 
#度量标准REST端点的绑定地址，默认为127.0.0.1
http.host: "127.0.0.1"
 
#度量标准REST端点的绑定端口，默认为9600
http.port: 9600
 
#日志级别，可以设置为以下几种级别，默认为info
log.level: info
           fatal
           error
           warn
           info (default)
           debug
           trace
 
#logstash日志目录位置，默认为logstash路径下的logs
path.logs: /usr/local/logstash-7.0.0/logs
 
 
#logstash插件路径
path.plugins: []

pipeline.yml

一个 logstash 实例中可以同时进行多个独立数据流程的处理工作

之前用户只能通过在单机运行多个 logstash 实例或者在配置文件中增加大量 if-else 条件判断语句来解决

使用 multiple pipeline 只需要将不同的 pipeline 在 config/pipeline.yml中定义好即可

例子

1
2
3
4
5
6
- pipeline.id: apache
  pipeline.batch.size: 125
  queue.type: persisted
  path.config: "/path/to/config/apache.cfg"
- pipeline.id: nginx
  path.config: "/path/to/config/nginx.cfg"

其中apache和nginx作为独立的 pipeline 执行，而且配置也可以独立设置，互不干扰。

pipeline.yml的引入极大地简化了 logstash 的配置管理工作

简单input与output

在/config新建一个filebeat_std.conf

1
2
3
4
5
6
7
8
input {
    beats {
        port => 5044
    }
}
output {
    stdout { codec => json_lines }
}

然后在filebeat的根目录下再新建一个test07_filebeat.yml\

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 输入
filebeat.inputs:
- type: log
  paths:
    - /var/log/messages
    - /var/log/*.log

fields:
  instance_id: i-efun_test02

# 表示字段加在根节点下
fields_under_root: true


# 输出
# 输出到5044端口
output.logstash:
  # The Logstash hosts
  hosts: ["localhost:5044"]

当时这里碰到了一个小问题 由于手滑在写output.logstash的时候上面的output.file:一行忘记删了

根目录就出现了filebeat.1 filebeat.2这种文件, 询问过何老师后发现这是由于它输出的文件是默认没改的，可能就是输出到相对路径./filebeat，然后，它发现里面有一个filebeat（可执行文件）；接着它就把这个filebeat当做日志，给滚动命名了

先跑logstash再跑filebeat 顺序不能颠倒

分别再二者的根目录执行

1
./bin/logstash -f ./config/filebeat_std.conf
1
./filebeat -e -c test07_filebeat.yml

image-20220512164019074

初始化logstash成功

image-20220512163841396

和以前一样filebeat 正常启动并收集

image-20220512163938313

这时候logstash里面已经出现数据, 就证明成功了

插件简单使用

mutate插件

在/config新建一个std_std.conf

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
input {
	stdin {
		type => "demo-stdin"
		add_field => {"我是stdin里直接增加的字段01" => "对 是的"}
		codec => "plain"
		tags => ["stdin-test01", "stdin-test02"]
			}
}

filter {
    mutate {
	add_field => [ "我是mutate里增加的字段02", "楼上说的对" ]
    }
}
output{
stdout{  }
}

第一次跑起来

image-20220512170748793

我们加一个 remove_field => [ "type" ]之后再跑一次

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
input {
	stdin {
		type => "demo-stdin"
		add_field => {"我是stdin里直接增加的字段01" => "对 是的"}
		codec => "plain"
		tags => ["stdin-test01", "stdin-test02"]
			}
}

filter {
    mutate {
	remove_field => [ "type" ]
	add_field => [ "我是mutate里增加的字段02", "楼上说的对" ]
    }
}
output{
stdout{  }
}

image-20220512171032469

如预期, type字段已经被删除掉了

date插件

建 std_std_date.conf后再跑

1
2
3
4
5
6
7
8
9
10
11
12
13
input {
    stdin {}
}
filter {
    date {
        match => ["message", "yyyy-MM-dd HH:mm:ss"]
        target => "timestamp"
    }
}
output {
    stdout {}
}

跑

image-20220513151912742

这里由于我本人理解不到位, 刚开始把 target => "timestamp"漏掉了, 导致未能成功跑出结果

drop插件

去公司了和何老师好好交流一下再补