如果输入相同，如何区分logstash过滤器中的普通日志和审核日志？

Question

我在Kubernetes中运行了一个应用程序。我有两种原木，

1. 正常日志
2. 审计日志

例如,

下面是单个日志文件的示例，该文件具有类型的数据、普通日志和审计。

[INFO] [] [%{TraceId}] 2021-10-02_18:37:01.601 com.MyExample.Test Getting app config from Config server with Base URL BASE_URL for application.
[INFO] [] [%{TraceId}] 2021-10-02_18:25:11.807 Audit Hello world,test file,123454,123Addb,true

在上面的示例中，第一行是普通日志，第二行是审计数据。

，我希望这两个日志都转到不同的索引(2个不同的索引)，并且我想为这两个日志应用不同类型的logstash过滤器。

对于如何解决上述问题，有什么建议吗？

还添加了一个需求流程图：在这里输入图像描述

我当前的logstash.conf文件仅用于审核目的，并且我使用的是一个.csv文件。

input {
  file {
    path => "<path_to_dir>/audit-*.csv"
    start_position => "beginning"
    sincedb_path => "/dev/null"
  }
}
filter {
  csv {
      separator => ","
      skip_header => "false"
      columns => ["TraceId","timestamp","fcId","Ean","source","Quantity","updateSent","message"]
  }
}
output {
   elasticsearch {
     hosts => "http://localhost:9200"
     index => "audit"
  }
stdout {}
} 

提前感谢！

Answer 1

好吧，这不是很干净，但是现在我看不到其他的方法，如果没有标签写着audit或log。

您可以尝试将JAVACLASS、它是logstash核心的一部分与grok相匹配，可能您以后将使用grok来分析您的日志。Grok可以使用标签使用_grokparsefailure事件，以防审计日志不匹配，而且从您所说的情况来看，不应该存在。在grok之后，只需使用if并检查标记是否存在，如果存在，则为审计日志，否则为基本日志，然后根据您的意愿处理事件。

要使JAVACLASS模式在grok中工作，您需要将ecs_compatibility设置为v1，如解释的这里。

免责声明，这是基于你提供的两个例子，这可能是完全错误的，在这种情况下，请提供更多的例子，因为没有明确的模式，是不可能找到解决方案。

具体例子：

测井线

[INFO] [] [123trace456] 2021-10-02_18:37:01.601 com.MyExample.Test Getting app config from Config server with Base URL BASE_URL for application.

Grok模式

\[%{LOGLEVEL:loglevel}\] \[\] \[(?<traceId>[0-9a-z]*)\] (?<timestamp>[0-9\-_:\.]*) %{JAVACLASS:class} %{GREEDYDATA:message}

结果

{
  "loglevel": [
    [
      "INFO"
    ]
  ],
  "traceId": [
    [
      "123trace456"
    ]
  ],
  "timestamp": [
    [
      "2021-10-02_18:37:01.601"
    ]
  ],
  "class": [
    [
      "com.MyExample.Test"
    ]
  ],
  "message": [
    [
      "Getting app config from Config server with Base URL BASE_URL for application."
    ]
  ]
}

我不知道它是什么类型的时间戳，也不知道您的跟踪ID是什么样子，所以我只是做了什么，重点是JAVACLASS模式的使用。我在谷歌上找到了很多例子，请不要忘了，谷歌是你的朋友！