Kafka日志的Logstash grok模式

要给出Kafka日志的Logstash grok模式，首先需要了解Kafka的日志格式。Kafka的日志格式如下：

[2021-01-01 00:00:00,000] INFO [KafkaApi-0] Created topic my-topic (kafka.admin.TopicCommand$)

该日志包含了时间戳、日志级别、日志来源、日志消息。

下面是一个示例的Logstash grok模式，可以用于解析上述日志格式：

grok {
  match => { "message" => "\[%{TIMESTAMP_ISO8601:timestamp}\] %{WORD:level} \[%{DATA:source}\] %{GREEDYDATA:message}" }
}

这个模式使用了Logstash的grok插件，它将日志消息中的不同部分提取为字段。具体解释如下：

• \[%{TIMESTAMP_ISO8601:timestamp}\]：匹配时间戳，并将其提取为名为timestamp的字段。
• %{WORD:level}：匹配日志级别，并将其提取为名为level的字段。
• \[%{DATA:source}\]：匹配日志来源，并将其提取为名为source的字段。
• %{GREEDYDATA:message}：匹配剩余的日志消息，并将其提取为名为message的字段。

使用以上模式后，Logstash会将解析后的字段发送到Elasticsearch或其他目标。

完整的配置文件示例如下：

input {
  kafka {
    bootstrap_servers => "localhost:9092"
    topics => ["my-topic"]
    group_id => "my-group"
    auto_offset_reset => "earliest"
  }
}

filter {
  grok {
    match => { "message" => "\[%{TIMESTAMP_ISO8601:timestamp}\] %{WORD:level} \[%{DATA:source}\] %{GREEDYDATA:message}" }
  }
}

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

这个配置文件从Kafka的my-topic主题中消费消息，并使用grok模式解析日志消息，最后将解析后的日志发送到Elasticsearch中的kafka-logs-YYYY.MM.dd索引中。

使用以上的配置文件和grok模式，你可以将Kafka日志解析并存储到Elasticsearch中，以供后续搜索和分析。