0x01 前言

最近在试用腾讯云的CDN,我CN的域名已经解析到腾讯云的CDN。如果你正在访问enginx\.cn,那么你目前连接的应该是腾讯云CDN的某个节点;如果目前访问的域名是enginx\.net,则是我博客的源服务器。

为了方便了解访客的具体情况,我习惯用elasticstack分析日志并生成实时图表。正好藉此机会,我详细地描述如何编写logstash的配置文件与grok的语法。

0x02 准备

首先是腾讯云CDN的日志,我在这几天写了一个python脚本,实现自动下载日志的功能,这部分内容将在日后的文章中详细说明。

腾讯云CDN的日志格式如下:

各个字段的意义可以在腾讯云的知识库中找到:

在实际的日志中我并没有找到“地域映射”的相关字段,这可能是产品迭代的时候没有更新知识库导致的,这部分先忽略。

然后是“省份映射”与“运营商映射”这两个字段的内容需要做文本匹配与替换,这在我之前分析的日志里没有遇到的。

最后需要一个配置好的elasticstack服务,具体配置方法可以参考以下文章:

0x03 输入与输出

0x03.1 输入

因为腾讯云CDN的日志是以文本文档保存的,所以我用filebeat读取并传送至logstash,那么输入的配置文件如下:

以上配置文件监听TCP 5016端口作为best的输入途径。如果需要加密,则配置文件如下:

input的方法有许多种,具体请参考官方说明:

为了方便区分和修改配置文件,建议将logstash的配置文件分配一个固定的命名格式,例如我的:

0x03.2 输出

输出的配置也很简单,例如我的:

如果logstash集群同时处理许多种日志,这时候就需要通过if..else..对日志进行分门别类地输出至elasticsearch。

如上面的配置文件,我分别对tags标签里的内容进行判断,如果包含我设定的内容,则投放至指定的索引中;而hosts中则填写了3个elasticsearch节点,可以实现负载均衡。

logstash output插件也支持许多种输出方式,具体可以参考官方文档:

0x04 过滤器

filter plugin是logstash的核心,包含许多过滤器,而分析腾讯CDN需要用到以下4种:

  1. grok:grok通过匹配模板将非结构化的数据转化为结构化的数据
  2. geoip:通过Maxmind GeoLite2数据库分析IP数据
  3. date:用于处理时间内容的字段
  4. useragent:用于分析user agent的数据
  5. mutate:用于修改、替换、重命名和修改字段或字段中的内容

其他的过滤器可以参考官方文档:

0x04.1 grok

首先通过grok分析腾讯CDN的日志:

在编写grok规则的时候,可以通过以下页面参考相关语法:

同时需要打开以下页面进行debug:

还需要了解日志各个字段的意义:

接下来先理解grok的格式,例如需要匹配日志中的第一个字段“请求时间”:

以%{}标识为一个字段,花括号内双引号左侧为匹配模板,右侧为自定义的名称。将相关内容放置到Grok Debugger中进行测试,如果没有错误,则会生成json格式的内容:

再看看匹配GitHub匹配模板中关于NUMBER的描述:

可以发现,NUMBER匹配的是10进制的数字,它可以匹配正负数、以及小数。

再来看一个例子:

以上是一个UA字段,因为UA字段内一般会有空格将各部分的内容拆分开,所以会有双引号将UA字段包裹起来,这时候的grok是这样的:

再来看看GitHub的描述:

QS匹配的是被引号包裹的字符串,那么在Grok Debugger中是这样的:

这个QS其实是一种快捷方式,其实还有一种方法:

用转义符标识出双引号,然后用GREEDYDATA匹配双引号中的内容,这个GREEDYDATA在GitHub中的描述为:

这个正则是匹配一切字符无限次。不过因为UA的两侧各有一个双引号,所以可以使用该匹配模板。

在实际应用中还有可能遇到同一个字段,可能会出现不同类型的内容,例如nginx反向代理upstream可能为空的情况,这时候可以写成如下结构:

用一个括号将规则包裹起来,然后在多个规则之间用“|”符号进行分割,匹配将从左到右进行。

IPORHOST可以匹配IP与hostname,而IP又包含IPV4和IPV6,很明显,这些内容的结构都是不同的,但他们又是层层包裹,所以仅适用IPORHOST即可匹配上我们需要的内容。而分隔符右侧的“-”则匹配无upstream的情况,在无upstream的时候,nginx日志中这个字段为空,所以是“-”。

其实grok语法非常简单,模板的存在省去了我们编写正则的麻烦,编写的时候要注意考虑字段中可能产生的文本类型,以便完全匹配。

匹配以上例子日志的完整grok规则如下:

这里需要提醒!冒号右侧的内容为自定义的名称,可以根据实际情况进行定义,而左侧的匹配模板则需要参考GitHub上的解释进行使用。

完成grok规则的编写后就可以完成grok配置文件了:

默认情况下,通过best传入的原始内容都放置在message标签内,所以通过match设定,告诉grok使用编写好的规则对message中的内容进行解析。

0x04.2 geoip

geoip是非常有用的一个插件,他可以通过Maxmind GeoLite2数据库中的数据分析IP的具体信息,虽然Maxmind GeoLite2数据库不太准确,但在判断IP所属地还是有很大用途的。

这个插件的使用很简单:

geoip插件中只需要指定分析数据的来源“source ”就好了,在这里的来源为通过grok分析后的数据,而这个数据的名为“client_ip”。

在这里有一个特别的情况,一般防火墙或路由器的日志都会有来源IP与目标IP,如果使用上述配置,则仅能分析一个IP,这时候需要为json key自定义一个名称:

那么分析出来的json如下:

因为SrcGeo中的IP为内网IP,所以没有任何信息,如果是公网IP,则与DstGeo中的内容类似。

0x04.3 date

date这个插件是非常重要的!例如腾讯云CDN的日志,统计周期为一小时,而且会延迟10分钟左右,也就是说10:15分前后才能下载9:00-10:00间的数据。

那么会有一个问题,logstash并不知道这日志是一小时前的,如果直接存入elasticsearch,elasticsearch会将接收到该日志的时间作为该日志的生成时间。这个时间差在运维工作中肯定是不允许的。

另外还有一种情况,日志是实时传输的,但是网络传输和分析都需要时间,当存入elasticsearch时可能会有半秒甚至1秒钟的误差,这也是不允许的。

不过我们的日志都是有时间戳,就是为了避免这种问题的产生,而date这个插件也是为了处理时间而存在的。

我们先看看时间在elasticsearch内的格式:

然后再看看我们日志中的格式:20180203174659,在实际应用中,日志的内容肯定是尽可能精简,哪怕是一个字符。所以需要对时间进行手动匹配:

和geoip一样,需要使用match提取出经过grok处理后的,自定义名称为“timestamp”的内容,并匹配格式:“YYYYMMddHHmmss”

以下是我nginx日志中的时间格式与匹配规则:

如果想查找时间格式的语法,请打开以下页面并搜索“More details on the syntax”:

0x04.4 useragent

对于web app来说,UA的解析非常重要,可以为产品后续的发展方向提供有意义的参考价值。

logstash useragent插件的匹配正则请参考以下页面:

而这个插件也无需过多设置:

依然是通过source指定来源,而target则是指定输出json的key名称:

0x04.5 mutate

回顾日志,发现日志中包含省份映射和运营商映射,这是以数据为id,不同的id代表不同的内容:

可是logstash并不知道对应的关系,这时候就需要mutate这个插件出手了。

这里需要用到if判断,例如grok中isp字段的内容为2,则需要将isp字段中的内容替换为“中国电信”:

那么多个映射,则需要用到if…else…:

省份的映射配置和运营商的类似。这个插件还有其他功能,具体请参考官方说明:

0x05 完整规则

为了防止类似结构的日志误用该过滤器中的规则,我使用if判断传入日志中的tags标签内是否存在我指定的内容,如果是,则应用该规则,如果不是则pass:

那么完整的规则如下:

0x06 结语

logstash是一款很重要的软件,担任着分析日志的重任,可配置的内容非常多,几乎与elasticsearch一样复杂,需要花点心思去学习。