云搜索服务
云搜索服务
文档指南
请输入
- 文档首页
云搜索服务最佳实践数据集成通过 Logstash 读取日志服务主题数据写入云搜索服务
复制全文
下载 pdf通过 Logstash 读取日志服务主题数据写入云搜索服务
文档反馈
问问助手本文介绍如何通过 Logstash 管道，实现读取 TLS 主题中的日志数据，并写入到云搜索服务索引中。
背景信息
日志服务 TLS 提供 Kafka 协议消费功能，可以将一个日志主题当作一个 Kafka Topic 来消费，每条日志对应一条 Kafka 消息。您可以使用 Logstash Kafka 连接器连接日志服务，通过 Logstash 管道将 TLS 中采集的日志数据消费到下游的大数据组件或者数据仓库。
流程介绍
- 准备数据源 TLS 主题。
- 您需要在 TLS 控制台创建一个日志项目，然后创建一个日志主题，并开通 Kafka 协议消费。
- 还需要获取项目的访问地址、项目 ID、主题 ID，Kafka 协议主题 ID，以便在 Logstash 管道中填入信息实现与 TLS 的连通。
- 准备数据目标端云搜索服务索引。
- 您需要在云搜索服务控制台创建实例并获取实例的访问地址。
- 创建 Logstash 管道。
- 当您准备好数据源和数据目标端后，便可以在云搜索服务控制台创建 Logstash 管道。
注意事项
为保证网络访问安全，本文所使用的云产品服务均使用内网访问方式，因此要求 Logstash 与 Elasticsearch 或 OpenSearch 实例均处于相同地域的同一个 VPC 内。
步骤一：准备数据源 TLS 主题
通过 Logstash 管道消费 TLS 日志前，首先需要先为 TLS 完成如下准备工作：
- 创建日志项目。具体操作，请参见创建日志项目。
- 创建日志主题。具体操作，请参见创建日志主题。
- 开启 Kafka 协议消费并获取 Kafka 协议消费主题 ID。具体操作，请参见开启 Kafka 消费功能。
说明
Kafka 协议消费 ID 用于在消费消息的 Logstash 管道中指明读取消息的目标主题。
- 获取接入点地址，包括日志项目的 ID、私网地址，以及日志主题 ID。具体操作，请参见查看私网地址等信息。
说明
- 日志项目 ID：在通过 Kafka 协议消费 TLS 日志时，作为 Kafka SASL 用户名。
- 日志项目私网地址：在 Logstash 管道中，作为 TLS 日志项目的连接地址。
- 日志主题 ID：在生产消息的 Logstash 管道中，用于指定消息写入的目标主题。
步骤二：准备数据目标端云搜索服务索引
使用 Logstash 管道消费 TLS 日志，需要准备一个数据写入的目的源，本文使用的是云搜索服务索引。请按照以下步骤创建实例，并获取实例的访问地址。
- 创建云搜索服务实例。具体操作，请参见创建实例。
- 获取实例的私网地址。具体操作，请参见查看实例详情。
说明
实例的私网地址用于在 Logstash 管道中访问实例。
步骤三：创建 Logstash 管道
- 在步骤二的云搜索服务实例的同一地域和 VPC 中创建 Logstash 实例。具体操作，请参见创建 Logstash 实例。
- 创建 Logstash 管道，引用 Input-kafka 和 Output-elasticsearch/opensearch 模板，并根据实际情况修改模板中的内容。具体操作，请参见创建管道。配置示例如下。
- input {kafka {# 注意域名需要替换成项目的域名bootstrap_servers => ["tls-cn-beijing.ivolces.com:9093"]auto_offset_reset => "latest"consumer_threads => 5decorate_events => true# 获得的 Kafka 协议消费 IDtopics => ["out-**"]security_protocol => "SASL_SSL"sasl_mechanism => "PLAIN"sasl_jaas_config => "org.apache.kafka.common.security.plain.PlainLoginModule required username='***' password='***';"# 主题中存储 JSON 格式内容，会解析后按字段写入云搜索服务索引codec => json {ecs_compatibility => "disabled"}}}output {opensearch {# 目标实例 endpointhosts => "https://elasticsearch-o-dev-******.escloud-dev.ivolces.com:****"# 云搜索实例访问链接# 目标实例 basicAuth 鉴权信息，如果存在user => "admin"password => "********"# 索引名称index => "test_orders"# 这里的 id 是某个输入中有的字段。如果没有指定 ES 会自动生成# document_id => "%{id}"# 是否仅用 `hosts` 配置项数组中的 HTTPS 链接进行请求ssl => false# 当使用 HTTPS 链接时，是否验证云搜索服务端证书。必须为 falsessl_certificate_verification => false}}
- 保存并部署管道，等待 Logstash 实例状态再次变成运行中时，在日志中查看管道是否正常运行。具体操作，请参见查询日志。
结果验证
GET test_orders/_search
{
"size": 10
}
示例输出如下。
{
"took": 2,
"timed_out": false,
"_shards": {
"total": 1,
"successful": 1,
"skipped": 0,
"failed": 0
},
"hits": {
"total": {
"value": 1,
"relation": "eq"
},
"max_score": 1,
"hits": [
{
"_index": "test_orders",
"_type": "_doc",
"_id": "test_id_1",
"_score": 1,
"_source": {
"id": "test_id_1",
"i_field": 12,
"k_field": "test_1",
"@version": "1",
"@timestamp": "2025-11-03T09:05:23.444042342Z"
}
}
]
}
}
最近更新时间：2026.07.31 15:55:01
这个页面对您有帮助吗？
有用
有用
无用
无用