Filebeat+Kafka场景下无冒号日志未入Elasticsearch，求排查方向

我有一个写入日志文件的应用，通过Filebeat的Kafka模块将日志数据发送至运行Elasticsearch 7.17.18的central_logging中央日志集群。测试时发现：

• 执行命令echo 2024-05-17 21:35:50,394 - daemon - INFO: - TEST LOG MESSAGE >> my_daemon.log写入的日志可正常进入中央日志索引；
• 执行命令echo 2024-05-17 21:35:50,394 - daemon - INFO - TEST LOG MESSAGE >> my_daemon.log（仅省略INFO后的冒号）写入的日志无法进入索引。

已确认该日志已到达Kafka服务器（通过Python的confluent-kafka消费者可接收），但未从Kafka同步至Elasticsearch索引。当前使用Filebeat 7.12.1，相关配置文件如下：

Filebeat主配置（filebeat.yml） #

filebeat.inputs:
- type: log
  paths:
    - /opt/my/path/api/my_web.log
    - /opt/my/path/api/my_daemon.log
    - /opt/my/path/api/my_daemon.error.log

filebeat.config.modules:
  path: ${path.config}/modules.d
  reload.enabled: false

setup.template.settings:
  index.number_of_shards: 3

fields: {app_name: "Application Name",app_campus: "US",app_env: "PROD",app_client: "MyApp"}

output.kafka:
  enabled: true
  version: '0.10.0.1'
  hosts: ["server01.domain.com:9092", "server02.domain.com:9092", "server03.domain.com:9092"]
  topic: 'MyTopic'
  partition.round_robin:
    reachable_only: true
  required_acks: 1
  compression: gzip
  max_message_bytes: 1000000000

Kafka模块配置（kafka.yml） #

# Module: kafka
# Docs: https://www.elastic.co/guide/en/beats/filebeat/7.x/filebeat-module-kafka.html

- module: kafka
  log:
    enabled: true

请问应从哪些环节查找对日志格式进行验证并拒绝该消息的实体？

• 中央日志集群的Ingest Pipeline/Logstash解析规则：日志已到达Kafka但未进入ES，核心原因大概率在集群侧的数据处理环节。检查是否存在针对日志内容的Grok解析或正则校验规则——比如规则中硬编码了INFO:的格式，无冒号的日志无法匹配，进而被标记为无效数据丢弃。
• Elasticsearch索引的映射与校验约束：查看目标索引的模板或映射配置，确认是否对解析后的字段（比如level）设置了严格格式要求，或开启了文档校验规则，不符合格式的文档会被ES拒绝写入。
• Kafka消费组件的过滤逻辑：确认中央集群中从Kafka拉取数据的组件（如Filebeat、Logstash）是否配置了条件过滤，比如通过drop处理器将不符合格式的日志直接丢弃。
• Elasticsearch错误日志：直接查看ES集群的日志文件，搜索mapper_parsing_exception、validation_exception等关键词，这类报错会明确说明文档被拒绝的具体原因。
• 中间处理组件运行日志：如果使用了Logstash等中间件，查看其运行日志，检查是否存在解析失败、数据过滤的记录，定位格式不兼容的具体规则。

内容的提问来源于stack exchange，提问作者Jacob Quisenberry