如何检测LogStash输入连接错误

发布于 2025-01-25 09:15:08 字数 260 浏览 2 评论 0原文

将KAFKA连接到Logstash时，如何监视和检测错误。

例如，我的Kafka经纪人倒闭了，Kafka和Logstash之间没有建立任何联系。

有没有办法让监视器与logstash和kafka之间的连接状态？我可以查询logstash日志（但我认为这不是适当的方法），我尝试使用logstash Monitoring API（例如Localhost：9600/_node/stats/pipelines？Pretty），但是没有API给我连接状态是提前

谢谢你

原文

分享到QQ

分享到微博

如果你对这篇内容有疑问，欢迎到本站社区发帖提问参与讨论，获取更多帮助，或者扫码二维码加入 Web 技术交流群。

发布评论

需要登录才能够评论，你可以免费注册一个本站的账号。

╰◇生如夏花灿烂 2025-02-01 09:15:08

如果您有弹性代理或metricbeat代理安装在Kafka节点上，则可以使用其KAFKA特定模块配置代理来监视它们。

如您所述，从logstash获取连接状态，您也可以配置您的连接状态logstash配置以从日志消息中获取状态。

Elasticsearch中的示例文档：

{
    "_index": "topicname",
    "_type": "_doc",
    "_id": "ulF8uH0BK9MbBSR7DPEw",
    "_version": 1,
    "_score": null,
    "fields": {
      "@timestamp": [
        "2022-05-09T10:27:56.956Z"
            ],
        "@version": [
            "1"
        ],
        "@version.keyword": [
            "1"
        ],
        "message": [
            "{\"requestMethod\":\"GET\",\"headers\":{\"content-type\":\"application/json\",\"user-agent\":\"PostmanRuntime/7.XX.XX\",\"accept\":\"*/*\",\"postman-token\":\"11224442345223\",\"host\":\"localhost:2300\",\"accept-encoding\":\"gzip, deflate, br\",\"connection\":\"keep-alive\",\"content-length\":\"44\"},\"body\":{\"category\":\"CAT\",\"noise\":\"purr\"},\"query\":{},\"requestUrl\":\"http://localhost:2300/kafka\",\"protocol\":\"HTTP/1.1\",\"remoteIp\":\"1\",\"requestSize\":302,\"userAgent\":\"PostmanRuntime/7.XX.X\",\"statusCode\":200,\"response\":{\"success\":true,\"message\":\"Kafka Details are added\",\"data\":{\"kafkaData\":{\"_id\":\"12gvsddwqbwrfteacr313rcet5\",\"category\":\"DOG\",\"noise\":\"bark\",\"__v\":0},\"postData\":{\"category\":\"DOG\",\"noise\":\"bark\"}}},\"latency\":{\"seconds\":0,\"nanos\":61000000},\"responseSize\":193}"
        ]
    } }

可以添加以下配置以获取状态：

  input {
    kafka {
      topics => ["topicname"]
      bootstrap_servers => "11.11.11.11:1111"
    }
  }
  
  filter{
        mutate { add_field => { "StatusCode" => "%{[message][0][status]}" } }
  }
  
  output {
   elasticsearch {
      hosts => ["11.11.11.12:9200"]
      index => "topic-name-index"
   }
  }

If you have an elastic agent or a metricbeat agent installed on the Kafka node, you can configure the agent to monitor them using their Kafka specific module.

For getting the connection status from logstash as you mentioned, you can also configure your logstash config to grab the status from the log message.

Sample document in elasticsearch :

{
    "_index": "topicname",
    "_type": "_doc",
    "_id": "ulF8uH0BK9MbBSR7DPEw",
    "_version": 1,
    "_score": null,
    "fields": {
      "@timestamp": [
        "2022-05-09T10:27:56.956Z"
            ],
        "@version": [
            "1"
        ],
        "@version.keyword": [
            "1"
        ],
        "message": [
            "{\"requestMethod\":\"GET\",\"headers\":{\"content-type\":\"application/json\",\"user-agent\":\"PostmanRuntime/7.XX.XX\",\"accept\":\"*/*\",\"postman-token\":\"11224442345223\",\"host\":\"localhost:2300\",\"accept-encoding\":\"gzip, deflate, br\",\"connection\":\"keep-alive\",\"content-length\":\"44\"},\"body\":{\"category\":\"CAT\",\"noise\":\"purr\"},\"query\":{},\"requestUrl\":\"http://localhost:2300/kafka\",\"protocol\":\"HTTP/1.1\",\"remoteIp\":\"1\",\"requestSize\":302,\"userAgent\":\"PostmanRuntime/7.XX.X\",\"statusCode\":200,\"response\":{\"success\":true,\"message\":\"Kafka Details are added\",\"data\":{\"kafkaData\":{\"_id\":\"12gvsddwqbwrfteacr313rcet5\",\"category\":\"DOG\",\"noise\":\"bark\",\"__v\":0},\"postData\":{\"category\":\"DOG\",\"noise\":\"bark\"}}},\"latency\":{\"seconds\":0,\"nanos\":61000000},\"responseSize\":193}"
        ]
    } }

Below configuration can be added to fetch the status:

  input {
    kafka {
      topics => ["topicname"]
      bootstrap_servers => "11.11.11.11:1111"
    }
  }
  
  filter{
        mutate { add_field => { "StatusCode" => "%{[message][0][status]}" } }
  }
  
  output {
   elasticsearch {
      hosts => ["11.11.11.12:9200"]
      index => "topic-name-index"
   }
  }

回复收藏 0 原文

~没有更多了~