Logstash Pipeline未执行数据转换直接推送至Elasticsearch排查

问题背景 #

通过Docker Compose部署ELK 7.3.2栈，所有服务运行正常，但Logstash的aggregate过滤器未生效，Elasticsearch索引中存储的是JDBC拉取的原始数据，未按预期完成聚合转换。

相关配置如下：

docker-compose.yml配置 #

version: '3.8'
services: 
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.3.2
    ports:
      - 9300:9300
      - 9200:9200
    environment:
      - http.cors.enabled=true
      - http.cors.allow-origin=*
      - http.cors.allow-methods=OPTIONS,HEAD,GET,POST,PUT,DELETE
      - http.cors.allow-headers=X-Requested-With,X-Auth-Token,Content-Type,Content-Length,Authorization
      - transport.host=127.0.0.1
      - cluster.name=docker-cluster
      - discovery.type=single-node
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    volumes:
      - elasticsearch_data:/usr/share/elasticsearch/data
    networks:
      - share-network
  kibana:
    image: docker.elastic.co/kibana/kibana:7.3.2
    ports:
      - 5601:5601
    networks:
      - share-network
    depends_on:
      - elasticsearch
  logstash:
    build: 
      dockerfile: Dockerfile
      context: .
    env_file:
      - .local.env
    volumes: 
      - ./pipelines/provider_scores.conf:/usr/share/logstash/pipeline/logstash.conf
    ports:
      - 9600:9600
      - 5044:5044
    networks:
      - share-network
    depends_on:
      - elasticsearch
      - kibana
volumes:
  elasticsearch_data:
networks:
  share-network:

Logstash Dockerfile配置 #

FROM docker.elastic.co/logstash/logstash:7.3.2

# install dependency
RUN /usr/share/logstash/bin/logstash-plugin install logstash-input-jdbc
RUN /usr/share/logstash/bin/logstash-plugin install logstash-filter-aggregate
RUN /usr/share/logstash/bin/logstash-plugin install logstash-filter-jdbc_streaming
RUN /usr/share/logstash/bin/logstash-plugin install logstash-filter-mutate

# copy lib database jdbc jars
COPY ./drivers/mysql/mysql-connector-java-8.0.11.jar /usr/share/logstash/logstash-core/lib/jars/mysql-connector-java.jar
COPY ./drivers/sql-server/mssql-jdbc-7.4.1.jre11.jar /usr/share/logstash/logstash-core/lib/jars/mssql-jdbc.jar
COPY ./drivers/oracle/ojdbc6-11.2.0.4.jar /usr/share/logstash/logstash-core/lib/jars/ojdbc6.jar
COPY ./drivers/postgres/postgresql-42.2.8.jar /usr/share/logstash/logstash-core/lib/jars/postgresql.jar

Logstash管道配置provider_scores.conf #

input {
    jdbc {
        jdbc_driver_library => "${LOGSTASH_JDBC_DRIVER_JAR_LOCATION}"
        jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"
        jdbc_connection_string => "jdbc:sqlserver://${DbServer};database=${DataDbName}"
        jdbc_user => "${DataUserName}"
        jdbc_password => "${DataPassword}"
        schedule => "${CronSchedule_Metrics}"
        statement => "
            select pws.ProviderID,
                pws.SpeedScore,
                pws.QualityScore
            from ProviderWeightedOverallScore pws
            order by pws.ProviderID
            "
    }
}
filter {
    aggregate {
        task_id => "%{ProviderID}"
        code => "
             map['providerid'] ||= event.get('ProviderID')
             map['kpi'] ||= []
             map['kpi'] << {
                'speedscore' => event.get('SpeedScore'),
                'qualityscore' => event.get('QualityScore')
                }
             event.cancel()
        "      
        push_previous_map_as_event => true
        timeout => 3
    }
}
output {
    elasticsearch {
        hosts => ["${LOGSTASH_ELASTICSEARCH_HOST}"]
        document_id => "%{providerid}"
        index => "testing-%{+YYYY.MM.dd.HH.mm.ss}"
        action => "update"
        doc_as_upsert => true
    }
    stdout {  }

}

故障原因及解决办法 #

1. JDBC输入字段名自动小写导致聚合逻辑失效 #

这是最可能的原因：Logstash的jdbc输入插件默认开启lowercase_column_names => true，会将数据库返回的列名（如ProviderID、SpeedScore）自动转换为小写的providerid、speedscore、qualityscore。

而你的aggregate配置中，使用的是大写开头的字段名（event.get('ProviderID')），这些字段实际不存在，导致：

• task_id => "%{ProviderID}"无法正确替换为实际的ProviderID值，所有事件会被分配到同一个无效的task_id下，聚合逻辑完全错误；
• 代码中无法获取到SpeedScore、QualityScore字段值，聚合后的kpi数组为空；
• 若代码执行出现异常，aggregate过滤器可能直接跳过，原始事件未被event.cancel()取消，直接流入Elasticsearch。

解决办法：
修改aggregate配置中的字段名为小写，或者在JDBC输入中关闭字段名小写转换：

• 方法一（推荐）：调整aggregate逻辑中的字段名：

filter {
    aggregate {
        task_id => "%{providerid}"
        code => "
             map['providerid'] ||= event.get('providerid')
             map['kpi'] ||= []
             map['kpi'] << {
                'speedscore' => event.get('speedscore'),
                'qualityscore' => event.get('qualityscore')
                }
             event.cancel()
        "      
        push_previous_map_as_event => true
        timeout => 30  # 同时建议延长超时时间
    }
}

• 方法二：在JDBC输入中添加lowercase_column_names => false，保留原始列名：

input {
    jdbc {
        # 其他配置不变
        lowercase_column_names => false
        statement => "
            select pws.ProviderID,
                pws.SpeedScore,
                pws.QualityScore
            from ProviderWeightedOverallScore pws
            order by pws.ProviderID
            "
    }
}

2. aggregate超时时间设置过短 #

当前配置中timeout => 3（单位：秒），若JDBC输入拉取数据的时间超过3秒，或者同ProviderID的多条记录未能在3秒内全部到达aggregate过滤器，会导致聚合未完成就提前推送事件，甚至触发部分原始数据流入输出。

解决办法：
根据数据量大小延长超时时间，比如设置为timeout => 30或更长：

aggregate {
    # 其他配置不变
    timeout => 30
}

3. 管道配置加载异常 #

检查Logstash容器日志，确认是否存在配置文件加载失败的错误：

• 执行命令查看日志：docker logs <logstash-container-name>
• 若发现配置文件语法错误或路径问题，修正挂载路径或配置语法。

4. aggregate插件兼容性问题 #

虽然Dockerfile中通过官方命令安装插件，但仍需确认插件版本与Logstash 7.3.2兼容。可通过以下命令查看插件版本：

docker exec <logstash-container-name> /usr/share/logstash/bin/logstash-plugin list --verbose logstash-filter-aggregate

若版本不兼容，指定兼容版本重新安装：

RUN /usr/share/logstash/bin/logstash-plugin install logstash-filter-aggregate==7.3.2

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