Logstash 管道推送数据而不将其转换为 Elasticsearch答案

【问题标题】：Logstash pipeline pushing the data without transforming it to ElasticsearchLogstash 管道推送数据而不将其转换为 Elasticsearch
【发布时间】：2022-08-16 21:41:13
【问题描述】：

我正在使用带有 ELK 堆栈（Elastic、Logstash、Kibana）的 docker-compose 文件。 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 只是为了在 Docker 的 Logstash 镜像中安装一些插件：

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

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 {  }

}

那是我的 docker 配置。一切正常，唯一的问题是filter->aggregate 部分不工作，弹性索引被直接数据填充，没有发生转换。

任何线索为什么过滤器部分不转换数据？

标签： elasticsearch logstash kibana logstash-configuration logstash-jdbc

【解决方案1】：

最常见的原因是 Logstash 管道由多个工作线程处理，具体取决于您拥有的 CPU 数量。默认情况下，pipeline.workers setting 设置为主机上可用的 CPU 数量。

为了正常工作，aggregate 过滤器must run with a single worker 否则事件可能不会全部通过同一个工作线程。

所以你应该确保设置pipeline.workers: 1 或确保你有environment variable PIPELINE_WORKERS 设置为1

【讨论】：