自定义 TCP 日志包初始化一个侦听 TCP 套接字,该套接字收集接收到的任何 TCP 流量并将每一行作为文档发送到 Elasticsearch。 可以通过将 ingest pipeline 的名字添加到管道配置选项来添加自定义摄取管道,可以通过 API 或摄取节点管道 UI创建自定义摄取管道。
在进行下面的练习之前,请先阅读我之前的文章 “Observability:使用 Elastic Agent 来进行 Uptime 监控” 来搭建自己的测试环境。我们按照那篇文章的配置来进行,直到我们添加 integration 那一步。
为了能够把 TCP 的日志包写入到 Elasticsearch 中,我们可以添加 Custom TCP Logs 集成:
我们保存当前的配置:
从上面的配置中,我们可以看出来,我们的 tcp-1 集成收集 localhost:9900 端口的 TCP 日志并传入到 Elasticsearch 中。
经过上面的配置后,我们可以使用如下的命令来检查当前 Ubuntu 机器上的 9900 端口的使用情况:
sudo netstat -tulpn | grep LISTEN | grep 9900
- liuxg@liuxgu:~$ sudo netstat -tulpn | grep LISTEN | grep 9900
- [sudo] password for liuxg:
- tcp 0 0 127.0.0.1:9900 0.0.0.0:* LISTEN 10368/filebeat
我们可以看到当前的 9900 是正在使用的。
为了能够测试这个集成,我们在 Ubuntu 机器的一个目录中创建如下的一个 logs 文件。它是一个典型的 Syslog 日志:
- liuxg@liuxgu:~/data/customtcplogs$ pwd
- /home/liuxg/data/customtcplogs
- liuxg@liuxgu:~/data/customtcplogs$ ls
- logs
- liuxg@liuxgu:~/data/customtcplogs$ cat logs
- May 4 00:10:36 liuxg xpcproxy[69746]: libcoreservices: _dirhelper_userdir: 557: bootstrap_look_up returned (ipc/send) invalid destination port
我们使用如下的命令来发送日志到 localhost:9900 的 TCP 端口中:
head -n 1 logs | nc localhost 9900
- liuxg@liuxgu:~/data/customtcplogs$ pwd
- /home/liuxg/data/customtcplogs
- liuxg@liuxgu:~/data/customtcplogs$ head -n 1 logs | nc localhost 9900
我们接下去 Kibana 查看一下:
我们可以看到在过去 15分钟里有一个文档。点击详情:
从上面的 message 字段中,我们可以看出来我们发送来的一个日志信息。它就是我们之前发送的。
我们也可以通过如下的命令来发送信息:
在上面,我们使用 telnet 来发送一个信息到 TCP 端口 9900。同样,我们再去 Kibana 进行查看:
这次,我们可以看到有两条信息发送进来了。
从上面显示的信息,我们看出来 message 字段还是一个非结构化的字段。它不便于我们对这个信息的分析。为此,我们希望使用 ingest pipeline 来对这个信息进行结构化。为了下面的步骤的进行,我们先使用如下的命令来删除已经写入的文档:
- POST .ds-logs-tcp.generic-*/_delete_by_query
- {
- "query": {
- "match": {
- "data_stream.dataset": "tcp.generic"
- }
- }
- }
首先,我们使用 ingest pipeline API 来测试我们的 pipelines:
- POST _ingest/pipeline/_simulate
- {
- "pipeline": {
- "processors": [
- {
- "grok": {
- "field": "message",
- "patterns": [
- "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:[%{POSINT:syslog_pid}])?:%{GREEDYDATA:syslog_message}"
- ]
- }
- }
- ]
- },
- "docs": [
- {
- "_source": {
- "message": "May 4 00:10:36 liuxg xpcproxy[69746]: libcoreservices: _dirhelper_userdir: 557: bootstrap_look_up returned (ipc/send) invalid destination port"
- }
- }
- ]
- }
我们使用上面的测试命令,可以看到如下的结果:
- {
- "docs": [
- {
- "doc": {
- "_index": "_index",
- "_id": "_id",
- "_version": "-3",
- "_source": {
- "syslog_program": "xpcproxy[69746]",
- "message": "May 4 00:10:36 liuxg xpcproxy[69746]: libcoreservices: _dirhelper_userdir: 557: bootstrap_look_up returned (ipc/send) invalid destination port",
- "syslog_hostname": "liuxg",
- "syslog_message": " libcoreservices: _dirhelper_userdir: 557: bootstrap_look_up returned (ipc/send) invalid destination port",
- "syslog_timestamp": "May 4 00:10:36"
- },
- "_ingest": {
- "timestamp": "2022-09-21T09:23:14.824894Z"
- }
- }
- }
- ]
- }
很显然,我们可以看到结构化的字段,比如 syslog_program, syslog_hostname 等。当然,我们还可以使用另外一个 processor 来删除 message 这个字段。如果你对 ingest pipeline 还不是很熟的话,请阅读我之前的文章 “Elastic:开发者上手指南” 中的 “Ingest pipeline” 章节。
我们接下来创建一个叫做 structure_message 的 pipeline:
- PUT _ingest/pipeline/structure_message
- {
- "description": "This is used to structure messages",
- "processors": [
- {
- "grok": {
- "field": "message",
- "patterns": [
- "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:[%{POSINT:syslog_pid}])?:%{GREEDYDATA:syslog_message}"
- ]
- }
- }
- ]
- }
注意:在我一开始做展示时,是使用 Apache 的一个日志来做练习的。在展示时,使用了像在文章 “Logstash:Logstash 入门教程 (二)” 中的方法来解析 Apache 日志。结果发现在解析的过程中,它生成了一个叫做 agent 的字段, 而这个字段在 Elastic Agent 的摄入中也创建了一个。也就造成了冲突。查看 Elastic Agent 的错误信息,你可以在 /opt/Elastic/Agent/data/elastic-agent-xxxxxx/logs/default 里进行查看。这个是针对 Linux 的安装。针对其它操作系统来说,这个路径会有所不同。
我们接下来重新来编辑之前的 integration:
这样,我们就更新成功了。
接下来,我们再次使用上面的命令来写入一个文档:
head -n 1 logs| nc localhost 9900
- liuxg@liuxgu:~/data/customtcplogs$ pwd
- /home/liuxg/data/customtcplogs
- liuxg@liuxgu:~/data/customtcplogs$ head -n 1 logs | nc localhost 9900
我们回到 Kibana 的界面:
我们可以看到有一个文档被写入进来了。我们点击上面文档的详情:
从上面,我们可以看出来,我们这次看到的数据是分析好的结构化的数据。这个便于我们对日志数据进行分析和统计。