logstash-input-mongodb: управление выходом?

Question

Я пытаюсь настроить плагин logstash-input-mongodb, чтобы читать аудиты из моей базы данных, но все стратегии анализа, похоже, имеют проблемы, и я не вижу, как настроить что-либо.

«Сгладить» parse_method работает довольно хорошо, но он игнорирует идентификаторы объектов mongodb и не выводит их нигде, кроме поля log_entry.

«Простой» метод parse_method включает идентификаторы объектов, но выводит даты таким образом, что я не могу понять, как анализировать фильтр даты (например, «2017-02-12 16:30:00 UTC»). Затем, при отсутствии правильной метки времени, плагин, по-видимому, сам генерирует временные метки, которые не имеют никакого отношения к текущему времени (например, в 2022 году).

Метод «копания» Я еще не совсем понял.

Так мои вопросы:

• Есть ли способ для анализа данных из log_entry поля (смотрите пример ниже), что плагин выводит? Я пробовал фильтр json, но это не json, потому что он был отформатирован в рубине.
• Или, есть ли способ получить метод «сгладить», чтобы включить идентификаторы объектов?
• Или, есть anyw ay, чтобы получить «простой» метод для правильного форматирования полей mododb 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]')))" 
    } 
} 

, но это бросает rubyexception: ERROR logstash.filters.ruby - произошла рубин исключение: (Eval): 1: ошибка синтаксиса, неожиданный tINTEGER

Answer 1

Если вы используете метод parse_method simple, вы можете легко проанализировать метку времени с помощью следующего шаблона yyyy-MM-dd HH:mm:ss ZZZ, который вы можете добавить в свой фильтр даты.

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

Что касается последнего пункта, то я предлагаю проверять since_* настройки, которые позволяют держать курсор то, что было уже обработано, и только начиная с этого курсора на следующей перезагрузке logstash.