如何通过Logstash Websocket输入插件从带认证与参数的Websocket服务器获取数据

要通过Logstash Websocket客户端连接需要认证的服务器，先发送认证指令再发送数据请求指令，默认配置无法自动完成这两步，可按以下方式调整：

修改后的完整配置 #

input {
  websocket {
    mode => client
    url => "wss://0.0.0.0/api"
    # 连接建立后自动发送认证指令
    init_payload => '{
      "command":"auth",
      "command_ref":"command reference",
      "args":{
        "token":"9118ce123456db123456b268e0135e3"
      }
    }'
    codec => json
  }
}

filter {
  # 根据服务器实际认证响应格式调整判断条件
  if [command] == "auth_response" and [status] == "success" {
    ruby {
      code => '
        # 发送数据请求指令到Websocket服务器
        event.set("[@metadata][websocket_send]", {
          "command":"get_calls",
          "command_ref":"command reference",
          "args":{
            "start_time":"2018-07-06T12:13:17Z",
            "end_time":"2022-07-21T12:13:17Z"
          }
        })
      '
    }
  }
}

output {
  stdout { codec => rubydebug }
  elasticsearch {
    hosts => ["0.0.0.0:9200"]
    index => "ws_cdrs"
  }
}

关键说明 #

• init_payload参数：用于Websocket连接建立后自动发送第一条认证指令，需严格匹配服务器要求的JSON格式。
• Ruby过滤器：监听服务器返回的认证成功响应（需根据实际返回字段调整if条件），触发发送第二条get_calls指令。[@metadata][websocket_send]是Logstash Websocket插件约定的字段，专用于向服务器主动发送消息。
• Codec设置：输入插件指定codec => json，自动解析服务器返回的JSON数据，便于后续过滤判断。

备选方案（批量发送） #

若服务器支持批量接收消息，可将两条指令用换行分隔后直接放入init_payload，无需过滤器：

init_payload => '{
  "command":"auth",
  "command_ref":"command reference",
  "args":{
    "token":"9118ce123456db123456b268e0135e3"
  }
}
{
  "command":"get_calls",
  "command_ref":"command reference",
  "args":{
    "start_time":"2018-07-06T12:13:17Z",
    "end_time":"2022-07-21T12:13:17Z"
  }
}'

内容的提问来源于stack exchange，提问作者FAYIS