logstash-input-mongodb：控制输出？

Question

我正在尝试设置 logstash-input-mongodb 插件以从我的数据库中读取审核，但所有解析策略似乎都有问题，我看不到如何自定义任何内容。

“扁平化”的 parse_method 工作得很好，但它忽略了 mongodb 对象 ID，并且不会在除 log_entry 字段之外的任何地方输出它们。

“简单”的 parse_method 包括对象 ID，但输出日期的方式我无法弄清楚如何使用日期过滤器进行解析（例如，“2017-02-12 16:30:00 UTC”）。然后，在没有适当时间戳的情况下，插件似乎会自行生成与当前时间无关的时间戳（例如，2022 年）。

“挖”的方法我还没搞清楚。

所以我的问题：

• 有没有办法从插件输出的 log_entry（参见下面的示例）字段中解析数据？我试过 json 过滤器，但它不是 json，因为它是 ruby​​ 格式的。
• 或者，有没有办法让“扁平化”方法包含对象 ID？
• 或者，有没有办法获得“简单”的方法来正确格式化 mongodb ISODate 字段？
• 有什么办法可以防止插件从一开始就读取数据（我只想把最后一天左右的时间推送到logstash）？

可以用任何配置复制，这是我的基本配置：

input {
  mongodb {
    uri => 'mongodb://localhost:27017/test'
    placeholder_db_dir => '/elk/logstash-mongodb/'
    placeholder_db_name => 'logstash_sqlite.db'
    collection => 'auditcommunications'
    batch_size => 1000
    parse_method => "flatten"
  }
}

filter {
  date {
    match => [ "timestamp", "ISO8601" ]
  }
}

output {
    stdout { codec => rubydebug }
}

包括log_entry的示例数据：

{
        "audit-id" => "58a2edc916e057270065fa74",
         "created" => "2017-02-14T11:45:13Z",
            "type" => "mongodb-audit",
      "audit-type" => "PaymentAudit",
        "mongo_id" => "58a2edc916e057270065fa74",
       "expiresAt" => "2017-05-15T11:45:13Z",
     "lastUpdated" => "2017-02-14T11:45:13Z",
      "@timestamp" => 2017-02-14T11:45:13.000Z,
       "log_entry" => "{\"_id\"=>BSON::ObjectId('58a2edc916e057270065fa74'), \"order\"=>BSON::ObjectId('a8a2f205790858970046aa59'), \"_type\"=>\"PaymentAudit\", \"lastUpdated\"=>2017-02-14 11:45:13 UTC, \"created\"=>2017-02-14 11:45:13 UTC, \"payment\"=>BSON::ObjectId('58a2edc02eafcd560101ee5f'), \"organization\"=>BSON::ObjectId('56edde0ba33e1c03ff54a5ec'), \"status\"=>\"succeeded\", \"context\"=>{\"type\"=>\"order\", \"id\"=>BSON::ObjectId('58a2e205790852270046ab59')}, \"expiresAt\"=>2017-05-15 11:45:13 UTC, \"__v\"=>0}",
         "logdate" => "2017-02-14T11:45:13+00:00",
             "__v" => 0,
        "@version" => "1",
    "context_type" => "order",
          "status" => "succeeded",
       "timestamp" => "2017-02-14T11:45:13Z"
}

如何从上面的 log_entry 字段中提取组织？

我尝试了以下方法：

filter {
  ruby {
    code => "event.set('organization', eval(event.get('[log_entry]')))"
  }
}

但这会引发 ruby​​exception：ERROR logstash.filters.ruby - 发生 Ruby 异常：(eval):1: syntax error, unexpected tINTEGER

Answer 1

如果您使用simple parse_method，那么您可以使用以下模式轻松解析时间戳yyyy-MM-dd HH:mm:ss ZZZ，您可以将其添加到日期过滤器中。

filter {
  date {
    match => [ "timestamp", "yyyy-MM-dd HH:mm:ss ZZZ" ]
  }
}

关于最后一点，我建议检查since_* 设置，该设置允许您保留已处理内容的光标，并且仅在下次重新启动logstash 时从该光标开始。