mongodb - 如何为 Logstash 提供 MongoDB 类型的输入

Question

我知道我们可以输入文件，然后输出到 mongo 数据库。但是我的 mongodb 中有一个集合，我想将其作为输入，以便我可以将其与 ES 一起使用。这可能吗？谢谢。

Answer 1

我也遇到过类似的问题，logstash-input-mongodb插件很好，但是功能很有限，好像也不再维护了，所以，我选择了logstash-integration-jdbc 插件。

我按照以下步骤将 MongoDB 集合与 ES 同步:

首先，我下载了DBSchema开发的MongoDB的JDBC驱动，你可以找到here .

我已经准备了一个自定义 Dockerfile 来集成驱动程序和插件，如下所示:

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

RUN mkdir /usr/share/logstash/drivers
COPY ./drivers/* /usr/share/logstash/drivers/

RUN logstash-plugin install logstash-integration-jdbc
RUN logstash-plugin install logstash-output-elasticsearch

我配置了一个每 30 秒执行一次的查询，它会查找插入时间戳晚于上次查询时间戳的文档(由参数 :sql_last_value 提供)

input {
  jdbc {
    jdbc_driver_library => "/usr/share/logstash/drivers/mongojdbc2.3.jar"
    jdbc_driver_class => "com.dbschema.MongoJdbcDriver"
    jdbc_connection_string => "jdbc:mongodb://devroot:devroot@mongo:27017/files?authSource=admin"
    jdbc_user => "devroot"
    jdbc_password => "devroot"
    schedule => "*/30 * * * * *"
    statement => "db.processed_files.find({ 'document.processed_at' : {'$gte': :sql_last_value}},{'_id': false});"
  }
}

output {
  stdout {
    codec => rubydebug
  }
  elasticsearch {
    action => "create"
    index => "processed_files"
    hosts => ["elasticsearch:9200"]
    user => "elastic"
    password => "password"
    ssl => true
    ssl_certificate_verification => false
    cacert => "/etc/logstash/keys/certificate.pem"
  }
}

希望对大家有帮助，问候