java - 过滤logstash输入

Question

我有以下logstash管道，它基本上使用logstash中的jdbc输入插件从Oracle数据库中获取记录，并将它们输出到elasticsearch索引中。

input {
    jdbc {
        id => "nouvelle-demande"

        jdbc_connection_string => "${JDBC_CONNECTION_STRING}"
        jdbc_user => "${JDBC_USER}"
        jdbc_password => "${JDBC_PASSWORD}"
        jdbc_driver_library => "${JDBC_DRIVER_LIBRARY}"
        jdbc_driver_class => "${JDBC_DRIVER_CLASS}"
        jdbc_validate_connection => true
        jdbc_fetch_size => 512
        clean_run => "${CLEAN_RUN}"

        schedule => "${NOUVELLE_DEMANDE_SCHEDULE}"
        statement_filepath => "/usr/share/logstash/query/nouvelle-demande.sql"
        use_column_value => true
        tracking_column => "id"
        last_run_metadata_path => "/usr/share/logstash/nouvelle-demande-logstash_jdbc_last_run"
        tags => "nouvelle-demande"

    }
}

output {
    if "nouvelle-demande" in [tags] { 
        elasticsearch {
            id => "nouvelle-demande-output"
            hosts => "elasticsearch:9200"
            index => "nouvelle-demande"
            document_type => "demande"
            document_id => "%{id}"
        }
    }
}

然后，Java 应用程序将elasticsearch 索引用作处理队列。一旦索引被处理，它就会被 Java 应用程序从索引中删除，并添加到另一个 elasticsearch 索引中，该索引基本上充当已处理索引的日志。该日志索引永远不会被删除，它是处理历史记录。

我想做的是让logstash忽略我的历史索引中已经存在的索引。基本上，我在 Java 应用程序中的处理循环确实会检查索引是否已在历史索引中，如果是，则会跳过它，但我的队列仍然包含所有已处理过的索引，这会导致处理膨胀，因为队列不断获取每次我的 Logstash 计划运行时都会变大。

我查看了logstash的elasticsearch filter plugin ，以及其他可用的过滤器插件，但它们似乎都无法实现我想要做的事情。还做了很多谷歌搜索，但找不到任何适合我需要的东西，我想这是非常规的。

两个索引之间的数据模型并不完全相同，但两个索引中的 id 相同。

注意:是的，我可以在 Java 应用程序本身中进行过滤，但我希望由 Logstash 自动处理此过滤，并且从应用程序本身处理它将是我最后的解决方案。

谢谢。

Answer 1

您可以使用logstash ruby​​ 过滤器来执行类似的操作。问题是，每次收到一条记录时，它都必须进行查询，这可能是一个巨大的负载。

https://www.elastic.co/guide/en/logstash/current/plugins-filters-ruby.html