logstash-输入-mongoDB：控制输出？

Question

我正在尝试设置logstash input mongodb插件，以便从我的数据库中读取审计，但所有解析策略似乎都有问题，我不知道如何定制任何内容。

“flatte”parse_方法工作得很好，但它忽略了mongodb对象ID，并且除了在log_输入字段中之外，不会在任何地方输出它们。

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

“我还没弄明白方法。”。

所以我的问题是：

• 是否有办法解析插件输出的log_entry（见下面的示例）字段中的数据？我尝试过json过滤器，但它不是json，因为它是ruby格式的。
• 或者，有没有办法让"flatten"方法包含对象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_输入字段中提取组织？

我试过以下方法：

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

但这带来了一个rubyexception：错误日志。过滤器。ruby-发生ruby异常：（eval）：1:语法错误，意外的tINTEGER

Answer 1

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

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

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