分布式架构技术
ElasticSearch

分布式-logstash配置文件编写

贴一下配置好的logstash配置文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
input{
	tcp{
		port=>4675
		codec=>json
	}
}

filter{
	mutate{
		gsub=>["message","[\\]",""]
	}
	json{
		source=>"message"
	}
}
output{
	elasticsearch{
		action=>index
		host=>localhost:9092
		index=>"dmp_audit_logs_%{[+YYYY-MM-dd]}"
	}
}

配置中主要分为三部分input、filter、output，这三部分是logstash pipeline中的三个元素，其中input和output是必须的，filter是可选的。

input

配置数据的输入源

控制台输入

文件输入

插件输入（beat）

中间件输入（kafka）

应用日志输入（日志appender）

最简单的输入源配置：

1
2
input{ stdin{} }
output{ stdout{} }  //控制台输出

这样就配置了一个控制台输入和输出，启动logstash在控制台输入一条消息：hello logstash

控制台输出如下：

1
2
3
4
5
6
{
    "@timestamp" => 2020-06-10T00:29:46.056Z,
       "message" => "hello logstash",
      "@version" => "1",
          "host" => "zzk-redis.novalocal"
}

logstash会自动给消息加上时间戳和版本信息，以及消息来源的ip，这里由于是本地发送的消息，所以host显示的是主机名

需求：将系统日志以json格式直接发送到logstash服务器

项目中配置logAppender，将日志直接发送给logshatsh的10514端口，

1
2
3
4
5
6
input{
	tcp{
		port=>10514    // 配置logstash监听4675端口
		codec=>json   // 配置数据解析方式为json
	}
}

在生产环境中，当服务器访问量较大，连续执行日志写出动作，出现了tcp拆包的问题，见下图：

可以看到，一条日志被拆成了两份，导致json解析失败。

所以高并发场景下，直接将日志写出到logstash是不可取的。

在中间加一层kafka就能解决上述问题，logAppender直接将日志文件写出到kafka中，logstash配置输入源为kafka，kafka作为消息中间件，一定不存在tcp拆包问题。

output

配置数据输出目的地：

标准输出

es输出

中间件输出

需求：将日志数据输出到es存储

1
2
3
4
5
6
7
output{
	elasticsearch{
		action=>index
		host=>localhost:9092
		index=>"dmp_audit_logs_%{[+YYYY-MM-dd]}"
	}
}

action=>index表示是插入索引操作

host=>localhost:9092 es的host

index=>”dmp_audit_logs_%{[+YYYY-MM-dd]}” 表示插入的索引名称

配置完input和output就可以收集数据上传到es啦，那么logstash肯定不是只有这么简单的收集&上传操作，logstash主要解决的问题是将乱七八糟各式各样的日志文件，整理成统一的格式上传到es，方便es进行统计分析。最主要的功能是在可选的filter模块实现的。

filter

我在上传审计日志的过程中，遇到一些问题：

审计日志主要是采集了用户的一些请求信息，将这些信息转换成json字符串之后，字串携带了大量的转义反斜杠，将这个字串直接上传到logstash，发现审计信息不能解析出来，还是以json字串形式存放在message字段中，我的需求是将所有的信息都解析成独立的字段。

字串去反斜杠

去反斜杠的操作不能再在java中执行，因为java并不知道反斜杠的存在

因此需要在logshtash中进行配置

1
2
3
4
5
6
7
8
filter{
	mutate{
		gsub=>["message","[\\]",""] // 去掉反斜杠
	}
	json{
		source=>"message"  
	}
}

gsub 操作：将制定字段中的 某字串 替换成 另一个字符（串）

json操作： 对制定字段的信息执行反json操作，获取字段信息

以上就解决了将java审计日志上传至es的配置

filter可以对数据字段进行很多的操作，本业务场景中并没有用到，可以参考一以下链接

一文快速上手Logstash

logstash配置文件详解

Logstash输出到Elasticsearch笔记

Logstash替换字符串