求助：通过Logstash解析Cisco交换机NetFlow至Elasticsearch失败

先梳理下从你的配置和日志里发现的几个关键问题，一步步来修复：

1. IP地址不一致导致的连接异常 #

看你的配置里，UDP input绑定的是120.126.160.91，但Logstash日志里显示Elasticsearch连接的是120.127.160.91，UDP监听器也启动在了120.127.160.91:5556。这明显是配置里的IP写错了，得把所有IP统一成正确的那个——要么改成120.126.160.91，要么改成120.127.160.91，确保Logstash监听的IP和Elasticsearch的IP都是正确可达的。

2. 语法错误：字段引用格式不正确 #

你的filter和output里的条件判断都有语法问题：if[type]=="netflow" 应该写成 if [type] == "netflow"——方括号和字段名type之间必须有空格，否则Logstash无法正确解析这个条件，会导致过滤或输出逻辑失效。

3. Filter逻辑完全反转 #

当前filter里的条件是：

if "" not in [IPV4_SRC_ADDR] and "" not in [IPV6_SRC_ADDR] {
  drop{}
}

这意味着只要源IP（不管是IPv4还是IPv6）存在，就会丢弃这条数据，完全和你的需求相反！你应该是想丢弃那些没有源IP的无效数据，所以正确的逻辑应该是：

if [IPV4_SRC_ADDR] == "" and [IPV6_SRC_ADDR] == "" {
  drop{}
}

或者更简洁的写法：

if !([IPV4_SRC_ADDR] or [IPV6_SRC_ADDR]) {
  drop{}
}

4. NetFlow Codec未指定版本 #

默认的netflow codec可能只支持特定版本，而Cisco交换机通常发送的是NetFlow v5或v9，所以最好显式指定支持的版本，避免解析失败：

codec => netflow {
  versions => [5, 9]
}

修正后的完整配置 #

input{
  udp{
    host => "120.126.160.91" # 这里改成正确的统一IP
    port => 5556
    codec => netflow {
      versions => [5, 9]
    }
    type =>"netflow"
  }
}
filter{
  if [type] == "netflow" {
    # 丢弃没有源IP的无效数据
    if !([IPV4_SRC_ADDR] or [IPV6_SRC_ADDR]) {
      drop{}
    }
  }
}
output{
  elasticsearch {
    hosts => ["120.126.160.91:9200"] # 统一IP
    document_type => "_doc" # 适配ES 6.x+的类型要求，避免警告
  }
  if [type] == "netflow" {
    stdout{codec=> rubydebug}
  }
}

额外检查步骤 #

• 确认Cisco交换机的NetFlow配置：确保交换机发送NetFlow的目标IP是Logstash的正确IP，端口是5556，版本和你在Logstash里指定的一致（v5或v9）。
• 网络连通性验证：检查Logstash服务器的UDP 5556端口是否开放，交换机能正常发送数据包到这个端口；同时确认Logstash能访问Elasticsearch的9200端口。
• 查看Logstash运行状态：启动修正后的配置后，观察stdout的rubydebug输出，如果能看到NetFlow数据，说明解析正常，再去Elasticsearch里查看索引是否生成数据。

内容的提问来源于stack exchange，提问作者張皓翔