Logstash Grok过滤器配置求助：解析Fortigate 6.4 Syslog并输出到Kafka

需求说明 #

• 将Fortigate 6.4的Syslog日志中message字段内的date、time、devname等所有字段解析后索引到Elasticsearch
• 配置第二个Kafka输出，仅转发包含srcip、dstip、sentbyte、rcvdbyte字段的日志

日志示例 #

@timestamp:Sep 22, 2023 @ 17:08:49.500 @version:1 facility:0 facility_label:kernel host:X.X.X.X message:<45>date=2023-09-22 time=17:07:46 devname="DEVICE1" devid="GTSV5865" eventtime=1695395266695275960 tz="+0200" logid="0000000013" type="traffic" subtype="forward" level="notice" vd="root" srcip=X.X.X.X srcname="SWITCH-SOMEWHERE" identifier=125970 srcintf="vlanZZ" srcintfrole="lan" dstip=X.X.X.X dstintf="INTERNET" dstintfrole="wan" srccountry="Reserved" dstinetsvc="Google-ICMP" dstcountry="United States" dstregion="California" dstcity="Mountain View" dstreputation=5 sessionid=193890225 proto=1 action="deny" policyid=0 policytype="policy" service="Google-ICMP" trandisp="noop" duration=0 sentbyte=0 rcvdbyte=0 sentpkt=0 rcvdpkt=0 vwlid=2 vwlquality="Seq_num(12 wan2), alive, sla(0x1), gid(1), cfg_order(0), cost(0), selected" vwlname="SVC_INTERNET" appcat="unscanned" crscore=30 craction=131072 crlevel="high" srchwvendor="Cisco" devtype="Network" srcfamily="Router"

之前尝试过的无效配置 #

filter {
  if [type] == "Fortimanager" {
    grok {
      match => { "message" => "message:<%{POSINT:logid}>date=%{DATE:date} time=%{TIME:time} devname=\"%{DATA:devname}\" devid=\"%{DATA:devid}\" eventtime=%{INT:eventtime} tz=\"%{DATA:tz}\" logid=\"%{DATA:logid}\" type=\"%{DATA:type}\" subtype=\"%{DATA:subtype}\" level=\"%{DATA:level}\" vd=\"%{DATA:vd}\" srcip=%{IP:srcip} srcname=\"%{DATA:srcname}\" identifier=%{INT:identifier} srcintf=\"%{DATA:srcintf}\" srcintfrole=\"%{DATA:srcintfrole}\" dstip=%{IP:dstip} dstintf=\"%{DATA:dstintf}\" dstintfrole=\"%{DATA:dstintfrole}\" srccountry=\"%{DATA:srccountry}\" dstinetsvc=\"%{DATA:dstinetsvc}\" dstcountry=\"%{DATA:dstcountry}\" dstregion=\"%{DATA:dstregion}\" dstcity=\"%{DATA:dstcity}\" dstreputation=%{INT:dstreputation} sessionid=%{INT:sessionid} proto=%{INT:proto} action=\"%{DATA:action}\" policyid=%{INT:policyid} policytype=\"%{DATA:policytype}\" service=\"%{DATA:service}\" trandisp=\"%{DATA:trandisp}\" duration=%{INT:duration} sentbyte=%{INT:sentbyte} rcvdbyte=%{INT:rcvdbyte} sentpkt=%{INT:sentpkt} rcvdpkt=%{INT:rcvdpkt} vwlid=%{INT:vwlid} vwlquality=\"%{DATA:vwlquality}\" vwlname=\"%{DATA:vwlname}\" appcat=\"%{DATA:appcat}\" crscore=%{INT:crscore} craction=%{INT:craction} crlevel=\"%{DATA:crlevel}\" srchwvendor=\"%{DATA:srchwvendor}\" devtype=\"%{DATA:devtype}\" srcfamily=\"%{DATA:srcfamily}\"" }
    }
    mutate {
      remove_field => ["message", "facility", "facility_label", "host", "time", "devname", "devid", "eventtime", "tz", "logid", "level", "vd", "srcuuid", "dstuuid", "srccountry", "dstinetsvc", "dstinetsvc", "dstcity", "dstreputation", "sessionid", "proto", "action", "policyid", "policytype", "poluuid", "service", "trandisp", "transip", "transport", "appid", "app", "appcat", "apprisk", "applist"]
    }
  }
}

# KO
filter {
  if [type] == "Fortimanager" {
    kv {
      source => "message"
      field_split => " "
    }
  }
}

# KO
filter {
  if [type] == "fortimanager" {
    grok {
      match => {
        "message" => "message:<%{POSINT:logid}>date=%{DATE:date} time=%{TIME:time} devname=\"%{DATA:devname}\" devid=\"%{DATA:devid}\" eventtime=%{INT:eventtime} tz=\"%{DATA:tz}\" logid=\"%{DATA:logid}\" type=\"%{DATA:type}\" subtype=\"%{DATA:subtype}\" level=\"%{DATA:level}\" vd=\"%{DATA:vd}\" appid=%{INT:appid} srcip=%{IP:srcip} dstip=%{IP:dstip} srcport=%{INT:srcport} dstport=%{INT:dstport} srcintf=\"%{DATA:srcintf}\" srcintfrole=\"%{DATA:srcintfrole}\" dstintf=\"%{DATA:dstintf}\" dstintfrole=\"%{DATA:dstintfrole}\" proto=%{INT:proto} service=\"%{DATA:service}\" action=\"%{DATA:action}\" policyid=%{INT:policyid} sessionid=%{INT:sessionid} appcat=\"%{DATA:appcat}\" app=\"%{DATA:app}\" hostname=\"%{DATA:hostname}\" incidentserialno=%{INT:incidentserialno} url=\"%{DATA:url}\" msg=\"%{DATA:msg}\" apprisk=\"%{DATA:apprisk}\" scertcname=\"%{DATA:scertcname}\" scertissuer=\"%{DATA:scertissuer}\""
      }
    }
  }
}

# KO
filter {
  if [type] == "fortimanager" {
    grok {
      match => {
        "message" => "%{SYSLOG5424PRI}%{GREEDYDATA:syslog5424_sd}$|(?:%{SYSLOGTIMESTAMP:timestamp}|%{TIMESTAMP_ISO8601:timestamp8601}) (?:%{SYSLOGFACILITY} )?%{SYSLOGHOST:logsource}+(?: %{SYSLOGPROG}:|) %{GREEDYDATA:syslog5424_sd}$"
      }
    }
  }
}

# KO
filter {
  if [type] == "fortimanager" {
    grok {
      match => {
        "message" => "message:<%{POSINT:logid}> %{GREEDYDATA:UNWANTED} devname=\"%{HOSTNAME:devname}\" %{GREEDYDATA:UNWANTED}"
      }
    }
  }
}

解决方案 #

1. 日志解析配置（提取所有字段到Elasticsearch） #

不用写超长Grok规则，先剥离Syslog头部，再用kv插件解析键值对，这是处理Fortigate日志最可靠的方式：

filter {
  # 匹配你的日志类型（注意大小写，和你输入的[type]字段保持一致）
  if [type] == "Fortimanager" {
    # 第一步：剥离Syslog的PRI头部（比如<45>），提取纯键值对内容到临时字段
    grok {
      match => { "message" => "<%{POSINT:syslog_pri}>%{GREEDYDATA:temp_message}" }
      remove_field => ["message"] # 移除原始message字段
    }

    # 第二步：用kv插件自动解析所有key=value格式的字段
    kv {
      source => "temp_message"
      field_split => " " # 按空格分割键值对
      value_split => "=" # 按=分割键和值
      trim_value => "\"" # 自动去掉值两端的引号
      remove_field => ["temp_message"] # 临时字段用完删除
    }

    # 可选优化：将日志自带的时间替换Logstash默认的@timestamp
    mutate {
      add_field => { "log_full_time" => "%{date}T%{time}%{tz}" }
    }
    date {
      match => [ "log_full_time", "yyyy-MM-dd'T'HH:mm:ssZ" ]
      target => "@timestamp"
      remove_field => ["date", "time", "tz", "log_full_time"]
    }
  }
}

配置说明

• 剥离Syslog头部：避免kv插件把<45>这种优先级码当成字段解析，导致后续出错
• KV插件解析：Fortigate日志天生是标准key=value格式，用kv比硬写Grok规则更灵活，能兼容不同类型的日志（比如traffic、system、alert等，字段差异自动适配）
• 时间字段替换：让Elasticsearch里的@timestamp显示日志实际生成时间，而非Logstash接收时间，更符合分析需求

2. Kafka输出配置（仅转发指定字段的日志） #

在Logstash的output区块添加第二个Kafka输出，通过条件判断筛选符合要求的日志：

output {
  # 原有Elasticsearch输出保持不变
  elasticsearch {
    hosts => ["elasticsearch:9200"]
    index => "fortigate-logs-%{+YYYY.MM.dd}"
  }

  # 新增Kafka输出：仅转发包含指定字段的日志
  if [srcip] and [dstip] and [sentbyte] and [rcvdbyte] {
    kafka {
      bootstrap_servers => "kafka:9092" # 替换为你的Kafka集群地址
      topic_id => "fortigate-traffic-logs" # 目标Kafka主题名称
      # 可选：只发送指定字段，减少数据传输量
      fields => ["srcip", "dstip", "sentbyte", "rcvdbyte", "devname", "action", "policyid"]
    }
  }
}

配置说明

• 条件筛选：if [srcip] and [dstip] and [sentbyte] and [rcvdbyte]确保只有四个字段都存在的日志才会被转发到Kafka
• 字段精简：用fields参数指定要发送的字段，避免冗余数据占用带宽和存储

调试技巧 #

• 临时添加stdout输出：在output区块加入stdout { codec => rubydebug }，可以直接在Logstash控制台查看解析后的字段结构，快速定位问题
• Grok规则测试：如果需要调整Grok头部匹配规则，可直接用Kibana自带的Grok调试器，粘贴日志片段实时验证匹配结果

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