[toc]

Logstash环境准备与安装

Logstash环境准备

主机名 wanIP LanIP 角色 应用
elk01 10.0.0.81 172.16.1.81 ES日志存储数据库 JDK elasticsearch
elk02 10.0.0.82 172.16.1.82 ES日志存储数据库 JDK elasticsearch
elk03 10.0.0.83 172.16.1.83 日志数据转发工具 JDK logstash

关闭防火墙

1
2
3
4
5
# CentOS6 关闭防火墙
/etc/init.d/iptables stop

# CentOS7 关闭防火墙
systemctl stop firewalld

关闭SELINUX

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 临时关闭
setenforce 0
setenforce: SELinux is disabled

# 永久关闭
vim /etc/sysconfig/selinux
# This file controls the state of SELinux on the system.
# SELINUX= can take one of these three values:
#     enforcing - SELinux security policy is enforced.
#     permissive - SELinux prints warnings instead of enforcing.
#     disabled - No SELinux policy is loaded.
SELINUX=disabled    ==>      //原来是enforcing 改成disabled
# SELINUXTYPE= can take one of these two values:
#     targeted - Targeted processes are protected,
#     mls - Multi Level Security protection.
SELINUXTYPE=targeted

设置epel源

1
2
3
4
5
# CentOS6 下载epel源
wget -O /etc/yum.repos.d/epel.repo http://mirrors.aliyun.com/repo/epel-6.repo

# CentOS7 下载epel源
wget -O /etc/yum.repos.d/epel.repo http://mirrors.aliyun.com/repo/epel-7.repo

修改时区

1
2
3
# 将时区修改为上海时区
cp /usr/share/zoneinfo/Asia/Shanghai /etc/localtime
cp：是否覆盖"/etc/localtime"？ y

设置时间同步

1
2
3
# 同步服务器时间（切记保证集群之间时间一致非常重要）
ntpdate time1.aliyun.com
28 Feb 14:11:28 ntpdate[8904]: step time server 203.107.6.88 offset 3168820.831817 sec

部署Logstash

Logstash是一个开源的数据收集引擎，可以水平伸缩，而且logstash整个ELK当中拥有最多插件的一个组件，其可以接收来自不同来源的数据并统一输出到指定的且可以是多个不同目的地。

安装JDK环境

下载地址: http://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html

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
# 解压JDK安装包
tar xf jdk-8u121-linux-x64.tar.gz

# 也可以yum直接安装
yum install -y java

# 将JDK安装包移动到安装目录下
mv jdk1.8.0_121 /usr/local/

# 做软链接(方便日后升级)
ln -s /usr/local/jdk1.8.0_121 /usr/local/jdk1.8

# 添加环境变量
vim /etc/profile.d/jdk1.8.sh
export JAVA_HOME=/usr/local/jdk1.8
export CLASSPATH=.:$JAVA_HOME/jre/lib/rt.jar:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
export PATH=$PATH:$JAVA_HOME/bin

# 加载环境变量
source /etc/profile

# 检查是否加载成功
java -version
java version "1.8.0_121"
Java(TM) SE Runtime Environment (build 1.8.0_121-b13)
Java HotSpot(TM) 64-Bit Server VM (build 25.121-b13, mixed mode)

安装Logstash

下载地址:

https://www.elastic.co/downloads/past-releases/logstash-5-3-0

https://artifacts.elastic.co/downloads/logstash/logstash-5.3.0.rpm

1
2
3
4
5
6
7
8
# 下载logstash的rpm包
wget https://artifacts.elastic.co/downloads/logstash/logstash-5.3.0.rpm

# 安装Logstash使用yum localinstall 自动安装依赖包
yum localinstall -y logstash-5.3.0.rpm

# 给Logstash目录授权
chown -R logstash.logstash /usr/share/logstash/

Logstash输入输出插件介绍

INPUT、OUTPUT插件

INPUT：插件使Logstash能够读取特定的事件源。

OUTPUT：插件将事件数据发送到特定的目的地，OUTPUT是事件流水线中的最后阶段。

INPUT支持事件源 OUTPUT支持输出源 CODEC编解码器支持编码
azure_event_hubs(微软云事件中心) elasticsearch(搜索引擎数据库) avro(数据序列化)
beats(filebeat日志收集工具) email(邮件) CEF(嵌入式框架)
elasticsearch(搜索引擎数据库) file(文件) es_bulk(ES中的bulk api)
file(文件) http(超文本传输协议) Json(数据序列化、格式化)
generator(生成器) kafka(基于java的消息队列) Json_lines(便于存储结构化)
heartbeat(高可用软件) rabbitmq(消息队列 OpenStack) line(行)
http_poller(http api) redis(缓存、消息队列、NoSQL) multiline(多行匹配)
jdbc(java连接数据库的驱动) s3*(存储) plain(纯文本，事件间无间隔)
kafka(基于java的消息队列) stdout(标准输出) rubydebug(ruby语法格式)
rabbitmq(消息队列 OpenStack) tcp(传输控制协议)
redis(缓存、消息队列、NoSQL) udp(用户数据报协议)
s3*(存储)
stdin(标准输入)
syslog(系统日志)
tcp(传输控制协议)
udp(用户数据报协议)

Logstash输入输出插件测试

测试logstash标准输入到标准输出

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# 从标准输入到标准输出
/usr/share/logstash/bin/logstash -e 'input { stdin{} } output { stdout{ codec => rubydebug }}'
## 标准输入nginx log及返回结果
nginx log
{
	## 时间戳
	"@timestamp" => 2023-05-05T08:02:13.138Z,
		## 版本信息
		"@version" => "1",
			## 主机信息
			"host" => "elkstack03",
		## 接收信息
		"message" => "nginx log" # 标准输出
}

# 示例：
xxx
{
    "@timestamp" => 2023-05-06T02:42:03.282Z,
      "@version" => "1",
          "host" => "elk03",
       "message" => "xxx"
}

结果展示如下图所示：

image-20230506104251208

测试logstash标准输入到文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 从标准输入到文件中
/usr/share/logstash/bin/logstash -e 'input { stdin{} } output { file { path => "/tmp/test_%{+YYYY.MM.dd}.log"}}'

# 输入
xxx
nginx

# 返回结果
10:45:00.340 [[main]>worker0] INFO  logstash.outputs.file - Opening file {:path=>"/tmp/test_2023.05.06.log"}

# 查看tmp目录下是否有日志文件存在
ll /tmp/
-rw-r--r-- 1 root root 178 May  6 10:45 test_2023.05.06.log

# 查看日志内容
cat /tmp/test_2023.05.06.log 
{"@timestamp":"2023-05-06T02:45:00.246Z","@version":"1","host":"elk03","message":"xxx"}
{"@timestamp":"2023-05-06T02:45:00.247Z","@version":"1","host":"elk03","message":"nginx"}

结果展示如下图所示：

image-20230506105223221

image-20230506105154786

测试logstash标准输入到ES

1
2
3
4
5
# 从标准输入到Elasticsearch
/usr/share/logstash/bin/logstash -e 'input { stdin{} } output { elasticsearch {hosts => ["10.0.0.81:9200"] index => "hcl_%{+YYYY.MM.dd}" }}'

# 输入内容
clbb

结果展示如下图所示：

打开浏览器，访问：http://10.0.0.51:9100

image-20230506110005032

Logstash收集系统日志

Logstash收集单个日志到文件中

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
# 编辑Logstash收集日志的配置文件
vim /etc/logstash/conf.d/message.conf

## 输入插件
input {
	## 文件模块
	file {
		## 日志类型
		type => "message-log"
		## 日志路径
		path => "/var/log/messages"
		## 第一次收集日志从头开始
		start_position => "beginning"
	}
}

## 输出插件
output {
	## 文件模块
	file {
		## 输出路径
		path => "/tmp/message_%{+yyyy.MM.dd}.log"
	}
}

# 检测语法
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/message.conf -t

image-20230506110948245

收集多个日志至文件之中

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
# 编辑配置文件
vim /etc/logstash/conf.d/system_log.conf

## 输入插件
input {
	## 文件模块
	file {
		## 日志路径
		path => "/var/log/messages"
		## 日志类型
		type => "system_log"
		## 第一次收集从头收集
		start_position => "beginning"
		## 收集日志间隔时间3秒
		stat_interval => "3"
	}
	## 文件模块
	file {
		## 日志路径
		path => "/var/log/secure"
		## 日志类型
		type => "secure_log"
		## 第一次收集从头收集
		start_position => "beginning"
		## 收集日志间隔时间3秒
		stat_interval => "3"
	}
}
## 输出插件
output {
	## 判断如果类型是system_log则输出到指定路径
	if [type] == "system_log" {
		## 文件模块
		file {
			## 日志输出路径
			path => "/tmp/message2_%{+yyyy.MM.dd}.log"
		}
	}
	## 判断如果类型是secure_log则输出到指定路径
	if [type] == "secure_log" {
		## 文件模块
		file {
			#日志输出路径
			path => "/tmp/secure_%{+yyyy.MM.dd}.log"
		}
	}
}

# 启动Logstash
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/system_log.conf &

将日志输出到ES

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 编辑.conf的配置文件
vim /etc/logstash/conf.d/message_file.es.conf
input{
	file{
		type => 'msg_log'
		path => "/tmp/test_2023.05.05.log"
		start_position => 'beginning'
	}
}

output{
	elasticsearch{
		hosts => ["10.0.0.81:9200"]
		index => 'xxx_msg_log-2023-05-05'
	}
}

# -f指定配置文件
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/message_file.es.conf