利用 ELK 处理 Percona 审计日志

共计 2447 个字符，预计需要花费 7 分钟才能阅读完成。

1、有强烈的审计需求。

2、能允许 10%-15% 左右的性能损失。

3、有强烈的对数据库操作实时查看需求 (一般都是为了领导要求)。

input {
    file {path => ["/u02/backup/audit.log"]
        codec => json
    }
}
output {
    elasticsearch {hosts  => ["192.168.1.233"]
    }
}

上面的配置看上去是没有问题的，如果是一般的 json 数据哪就能解析成功了，

但是在 Percona audit plugin 中应用程序应用程序生成的 SQL 是五花八门，各种字符都有其中有。

如下审计的日志被 python 读取后的字符串展现 (红色标记)：

从上图可以看到有一些换行后 tab 的字符，这些字符使用 json.load 的时候会报错, 不能解析成 json

使用 python json 模块解析相关日志文件报错：

>>> json.loads(json_line)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/lib64/python2.7/site-packages/simplejson/__init__.py", line 516, in loads
    return _default_decoder.decode(s)
  File "/usr/lib64/python2.7/site-packages/simplejson/decoder.py", line 370, in decode
    obj, end = self.raw_decode(s)
  File "/usr/lib64/python2.7/site-packages/simplejson/decoder.py", line 400, in raw_decode
    return self.scan_once(s, idx=_w(s, idx).end())
simplejson.scanner.JSONDecodeError: Invalid control character '\t' at: line 1 column 232 (char 231)

所以在使用 logstash 的时候也就解析不了这样的 json 数据了，

最终 logstash 只能报错并将整个 message 记录到 Elasticsearch 中

解决办法就是把这些字符替换掉。如下 Logstash 配置文件

input {
    file {path => ["/u02/backup/audit.log"]
    }
}
 
filter {
    mutate {gsub => ["message", "\\\\n", " "]
        gsub => ["message", "\t", " "]
        replace => ["message", "%{message}" ]
    }
    json{source => "message"}
    mutate {remove_field => [ "message"]
    }
}
 
output {
    elasticsearch {hosts  => ["192.168.1.233"]
    }
}

该配置文件是投机取巧的办法, 把 (换行 /tab) 字符替换成空格，要注意的一点最终显示的 SQL 和原来的有所差别。

这种方法有点不灵活如果 sql 语句中还有遇到一些 json 不能解析的字符又要进行处理。

阿里云 2 核 2G 服务器 3M 带宽 61 元 1 年，有高配

腾讯云新客低至 82 元 / 年，老客户 99 元 / 年

代金券：在阿里云专用满减优惠券

刚开始以为这一切都万事大吉了。其实还有个一坑就是在使用 Kibana 查看的时候，这时候问题就来了。

有是有过 Percona audit 插件的估计都有这样的问题，就是他记录的是时间是国际形式的 (如上图黄色标记)，不像我们习惯了北京时间。因此在页面显示的时间会比我们平时的少 8 小时。

一般来说在 ELK 中使用国际的标准格式是合理的。因为在使用 Kibana 查看的时候会帮你自动转化成本地时间格式。也就是如果我们在中国他会自动把 timezone 转化为 Asia/Shanghai(东 8 区) 的。所以显示的时间应该是正确的才对。可是实际情况并没有。

是应为 Elasticsearch 将 “2016-08-30T01:45:30 UTC” 这串字符解析成了 String 类型。按道理应该解析成和 @timestamp 一样的 date 类型。

将 “2016-08-30T01:45:30 UTC” 格式转化成和 @timestamp 一样的格式 (“2016-08-30T01:45:30Z”)

input {
    file {path => ["/u02/backup/audit.log"]
    }
}
 
filter {
    mutate {gsub => ["message", "\\\\n", " "]
        gsub => ["message", "\t", " "]
        replace => ["message", "%{message}" ]
    }
 
    json{source => "message"}
 
    mutate {remove_field => [ "message"]
        gsub => ["[audit_record][timestamp]", "UTC", "Z"]
        replace => ["[audit_record][timestamp]", "%{[audit_record][timestamp]}" ]
    }
}
 
output {
    elasticsearch {hosts  => ["192.168.1.233"]
    }
}

使用上面配置就能顺利的将 时间格式 转化成 Elasticsearch 想要的时间格式，并且能在 Kibana 中正确显示。

祝大家好运。

阿里云 2 核 2G 服务器 3M 带宽 61 元 1 年，有高配

腾讯云新客低至 82 元 / 年，老客户 99 元 / 年

代金券：在阿里云专用满减优惠券