自定义 TCP 日志包初始化一个侦听 TCP 套接字，该套接字收集接收到的任何 TCP 流量并将每一行作为文档发送到 Elasticsearch。 可以通过将 ingest pipeline 的名字添加到管道配置选项来添加自定义摄取管道，可以通过 API 或摄取节点管道 UI创建自定义摄取管道。

前提条件

在进行下面的练习之前，请先阅读我之前的文章 “Observability：使用 Elastic Agent 来进行 Uptime 监控” 来搭建自己的测试环境。我们按照那篇文章的配置来进行，直到我们添加 integration 那一步。

添加 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}"
        ]
      }
    }
  ]
}

我们接下来重新来编辑之前的 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 的界面：

 我们可以看到有一个文档被写入进来了。我们点击上面文档的详情：

从上面，我们可以看出来，我们这次看到的数据是分析好的结构化的数据。这个便于我们对日志数据进行分析和统计。