Как настроить рабочий процесс 1: N с фильтрами в Logstash?

Question

Я пытаюсь настроить один рабочий стол, который принимает все сообщения из одной очереди amqp/rabbitmq, фильтрует некоторые сообщения для отправки в statsD, а также отправляет ВСЕ сообщения в список поиска. Следующая реализация не отправляет ЛЮБЫЕ сообщения в ElasticSearch.

input { 
    rabbitmq { 
    host => "amqp-host" 
    queue => "elasticsearch" 
    key => "elasticsearch" 
    exchange => "elasticsearch" 
    type => "all" 
    durable => true 
    auto_delete => false 
    exclusive => false 
    format => "json_event" 
    debug => false 
    } 
} 

filter { 
    grep { 
     add_tag => "grepped" 
     match => ["@message", "Execution of .*? took .* sec"] 
    } 

    grok { 
     tags => ["grepped"] 
     add_tag => "grokked" 
     pattern => "Execution of %{DATA:command_name} took %{DATA:response_time} sec" 
    } 

    mutate { 
     tags => ["grepped", "grokked"] 
     lowercase => [ "command_name" ] 
     add_tag => ["mutated"] 
    } 
} 

output { 
    elasticsearch_river { 
    type => "all" 
    rabbitmq_host => "amqp-host" 
    debug => false 
    durable => true 
    persistent => true 
    es_host => "es-host" 
    exchange => "logstash-elasticsearch" 
    exchange_type => "direct" 
    index => "logs-%{+YYYY.MM.dd}" 
    index_type => "%{@type}" 
    queue => "logstash-elasticsearch" 
    } 

statsd { 
    type => "command-filter" 
    tags => ["grepped", "grokked", "mutated"] 
    host => "some.domain.local" 
    port => 1234 
    sender => "" 
    namespace => "" 
    timing => ["prefix.%{command_name}.suffix", "%{response_time}"] 
    increment => ["prefix.%{command_name}.suffix"] 
    } 
} 

Есть ли какой-нибудь фильтр? Или способ упорядочить теги, чтобы некоторые сообщения были отфильтрованы, а ВСЕ перенаправлены в ES?

Answer 1

Фильтр clone пригодится. Ниже приведен мой конфигурационный файл.

input { 
    rabbitmq { 
    host => "amqp-host" 
    queue => "elasticsearch" 
    key => "elasticsearch" 
    exchange => "elasticsearch" 
    type => "all" 
    durable => true 
    auto_delete => false 
    exclusive => false 
    format => "json_event" 
    debug => false 
    } 
} 

filter { 
    clone { 
     exclude_tags => ["cloned"] 
     clones => ["statsd", "elastic-search"] 
     add_tag => ["cloned"] 
    } 

    grep { 
     type => "statsd" 
     add_tag => "grepped" 
     match => ["@message", "Execution of .*Command took .* sec"] 
    } 

    grok { 
     type => "statsd" 
     tags => ["grepped"] 
     add_tag => "grokked" 
     pattern => "Execution of %{DATA:command_name}Command took %{DATA:response_time} sec" 
    } 

    mutate { 
     type => "statsd" 
     tags => ["grepped", "grokked"] 
     lowercase => [ "command_name" ] 
     add_tag => ["mutated"] 
    } 
} 

output { 
    elasticsearch_river { 
    type => "all" 
    rabbitmq_host => "amqp-host" 
    debug => false 
    durable => true 
    persistent => true 
    es_host => "es-host" 
    exchange => "logstash-elasticsearch" 
    exchange_type => "direct" 
    index => "logs-%{+YYYY.MM.dd}" 
    index_type => "%{@type}" 
    queue => "logstash-elasticsearch" 
    } 

    statsd { 
    type => "statsd" 
    tags => ["grepped", "grokked", "mutated"] 
    host => "some.host.local" 
    port => 1234 
    sender => "" 
    namespace => "" 
    timing => ["commands.%{command_name}.responsetime", "%{response_time}"] 
    increment => ["commands.%{command_name}.requests"] 
    } 
} 

Answer 2

Фильтр clone на самом деле не нужен в вашей ситуации.

Есть несколько вещей, которые я бы рекомендовал:

• Пожалуйста упростить конфигурации, пока вы не получите «базу» работает. Не устанавливать дополнительные параметры теперь
• Убедитесь type совпадала
• Используйте теги, они спасатель
• Когда «оглавлению» Вы должны оставаться последовательным и всегда использовать grok или всегда использовать grep. Я предпочитаю grok

Некоторые другие советы ... Выберите то, что вы хотите сделать с вашими сообщениями на основе того, что теги они установили. Ниже приведен пример моего конфигурационного фрагмента:

grok{ 
    type => "company" 
    pattern => ["((status:new))"] 
    add_tag => ["company-vprod-status-new", "company-vprod-status-new-%{company}", "company-vprod-status-new-%{candidate_username}"] 
    tags=> "company-vprod-status-change" 
} 
output { 
elasticsearch { 
    host => "127.0.0.1" 
    type => "company" 
    index => "logstash-syslog-%{+YYYY.MM.dd}" 
} 
statsd { 
    host => "graphite.test.company.com" 
    increment => ["vprod.statuses.new.all", "vprod.statuses.new.%{company}.all"] 
    tags => ["company-vprod-status-new"] 
} 
} 

Также обратите особое внимание на ваш тип. Если для атрибута type установлено значение, которое не существует, то этот блок никогда не будет запущен. Вот почему я предпочитаю использовать теги, если нет явной причины использовать типы.

Answer 3

Вы можете также добавить:

drop => false 

в конце Grep строфы (если вы все еще используете Grep, что есть)