[toc]

Logstash将日志写入Redis

为什么要使用Redis

在企业中，日志规模的量级远远超出我们的想象，这就是为什么会有一家公司日志易专门做日志收集，给大型金融公司收集日志，比如银行，因为你有可能看到，1秒钟好几千万的日志量，往服务器写入，那么企业中的集群，架构都不是单台的，而是多台的，一台如果是1千万，那么5台的量级，10台的量级，我们要对他们进行收集，进行分析，难免会在网络传输过程中，丢数据。

日志是什么？
日志对于企业来说，有什么作用？
用户使用我们的产品，体验如何？
用户的客诉，我们能拿出什么样的数据来说话？
…

一系列的问题，都和日志相关，如果至关重要的那个数据丢失了，那么公司的损失可不仅仅是一条日志那么简单。如果我们不知道，用户对我们产品最感兴趣的地方在哪，那么产品的寿命也就越来越短。如果被攻击了，恶意攻击的IP源我们都找不到，那么或许就不是产品的寿命越来越短，而是这个企业存在的寿命，越来越短。

一个大规模日志量级的企业想要做到数据的安全性，数据的一致性，我们需要消息队列：Redis, Kafka，在ELK5版本中，建议使用Redis来做消息队列，Kafka能不能用？也能，只不过会有一些不必要的坑，需要我们去爬。在ELK6版本中，开始使用Kafka来做消息队列。

接下来就开始将Logstash收集到的日志，输出到Redis中。

加入redis的结构流程图

image-20230510000926296

Redis部署

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
# 下载
wget http://download.redis.io/releases/redis-3.2.12.tar.gz

# 解压
tar xf redis-3.2.12.tar.gz

# 移动到指定目录
mv redis-3.2.12 /application/

# 做软链接
ln -s /application/redis-3.2.12 /application/redis

# 进入redis目录
cd /application/redis

# 编译
make

# 添加环境变量
vim /etc/profile.d/redis.sh
export PATH="/application/redis/src:$PATH"

# 读取并执行环境变量
source /etc/profile

# 创建配置文件存放目录
mkdir -p /data/6379

#编辑redis配置文件
vim /data/6379/redis.conf
port 6379
daemonize yes
pidfile /data/6379/redis.pid
logfile "/data/6379/redis.log"
dbfilename dump.rdb
dir /data/6379
protected-mode no
requirepass  hcl

# 启动redis
redis-server /data/6379/redis.conf

# 检查端口
netstat -lntup | grep 6379

Logstash收集日志输出至Redis

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
# 编辑Logstash配置文件
vim /etc/logstash/conf.d/nginx_redis.conf
input {
	file {
		type => "www.xxx.com_access"
		path => "/var/log/nginx/www.xxx.com_access_json.log"
		start_position => "end"
	}
	file {
		type => "xxx.xxx.com_access"
		path => "/var/log/nginx/xxx.xxx.com_access_json.log"
		start_position => "end"
		codec => json
	}
}
output {
	if [type] == "www.xxx.com_access" {
		redis {
			data_type => "list"
			key => "nginx_log"
			host => "10.0.0.52"
			port => "6379"
			db => "14"
			password => "hcl"
	 }
}
	if [type] == "xxx.xxx.com_access" {
		redis {
			data_type => "list"
			key => "nginx1_log"
			host => "10.0.0.52"
			port => "6379"
			db => "15"
			password => "hcl"
		}
	}
}

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

验证Redis数据

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
# 连接redis
redis-cli -a hcl

# 在0库中查看所有key
127.0.0.1:6379> KEYS *
(empty list or set)

# 查看tomcat_log的长度（日志的条数）
127.0.0.1:6379> LLEN tomcat_log

# 切换15库
127.0.0.1:6379> SELECT 15
OK

# 在15库中查看所有key
127.0.0.1:6379[15]> KEYS *
1) "nginx_log"

# 查看nginx_log的长度（日志的条数）
127.0.0.1:6379[15]> LLEN nginx_log
(integer) 6
 
# 演示Logstash如何取走一条nginx日志
127.0.0.1:6379[15]> LPOP nginx_log

# 再次查看长度
127.0.0.1:6379> Llen nginx_log
(integer) 7

Logstash从Redis中取出日志输出到ES

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
# 编辑Logstash配置文件
vim /etc/logstash/conf.d/ngx_redis.conf
input {
	redis {
		data_type => "list"
		key => "nginx_log"
		host => "10.0.0.52"
		port => "6379"
		db => "14"
		password => "hcl"
		codec => "json"
	}

	redis {
		data_type => "list"
		key => "nginx1_log"
		host => "10.0.0.52"
		port => "6379"
		db => "15"
		password => "hcl"
	}
}

output {
	if [type] == "www.xxx.com_access" {
		elasticsearch {
			hosts => ["10.0.0.51:9200"]
			index => "www.xxx.com-%{+YYYY.MM.dd}"
	}
}

	if [type] == "xxx.xxx.com_access" {
		elasticsearch {
			hosts => ["10.0.0.51:9200"]
			index => "xxx.xxx.com-%{+YYYY.MM.dd}"
		}
	}
}

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

验证Logstash中的数据是否被取出

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 连接Redis
redis-cli -a hcl

# 查看所有key
127.0.0.1:6379> KEYS *
(empty list or set)

# 切换1库
127.0.0.1:6379> SELECT 1
OK

# 查看所有key
127.0.0.1:6379[1]> KEYS *
(empty list or set)

在ES中查看数据

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

将ES索引添加到Kibana中

打开浏览器，访问：http://10.0.0.83:5601