"message":"firewall,info input_ip input: in:pppoe-1000M-4 out:(unknown 0), proto TCP (SYN), 111.7.96.136:28927->113.116.194.10:58000, len 52" "message":"firewall,info input: in:pppoe-out4 out:(unknown 0), src-mac da:86:8e:97:00:36, proto ICMP (type 8, code 0), 13.244.104.42->119.122.115.239, len 36" 使用logstash + grok 对上面两段message兼顾提取里面的全部ip和端口

可以使用以下的 grok 表达式来提取：

%{GREEDYDATA:message}, %{IPV4:input_ip}:%{NUMBER:input_port}->%{IPV4:output_ip}:%{NUMBER:output_port}, len %{NUMBER:length}

这个表达式可以匹配第一行的 message，提取出 input_ip、input_port、output_ip、output_port 和 length。

同样的，对于第二行的 message，可以使用以下的 grok 表达式来提取：

%{GREEDYDATA:message}, %{IPV4:source_ip}->%{IPV4:destination_ip}, len %{NUMBER:length}

这个表达式可以匹配第二行的 message，提取出 source_ip、destination_ip 和 length。

在 logstash 配置文件中，可以针对每条日志分别使用不同的 grok 表达式进行匹配。完整的配置文件示例如下：

input {
  beats {
    port => 5044
  }
}

filter {
  if "firewall,info input_ip input" in [message] {
    grok {
      match => { "message" => "%{GREEDYDATA:message}, %{IPV4:input_ip}:%{NUMBER:input_port}->%{IPV4:output_ip}:%{NUMBER:output_port}, len %{NUMBER:length}" }
    }
  } else if "firewall,info input: in:pppoe-out4" in [message] {
    grok {
      match => { "message" => "%{GREEDYDATA:message}, %{IPV4:source_ip}->%{IPV4:destination_ip}, len %{NUMBER:length}" }
    }
  }
}

output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "firewall-%{+YYYY.MM.dd}"
  }
}

这个配置文件中，使用了两个条件判断语句来选择不同的 grok 表达式。第一个条件判断语句匹配第一条日志，第二个条件判断语句匹配第二条日志。在匹配成功之后，将会提取出相应的字段，并将数据输出到 Elasticsearch 中。