Логирование Squid и Exchange

Добрый день.

Возникло желание для этих двух систем настроить долговременное хранение логов. Поиск готовых решений привел к некоторым результатам, но есть некоторые проблемы:

  1. Squid. Он генерирует огромное, для нашей инфраструктуры, количество логов. Порядка 100ГБ в день в ежедневном индексе. Хотелось бы как-то их ужать, но не знаю как.
    Из пришедших в голову вариантов было попытаться писать больше информации в один документ, возможно через multiline. Либо создавать под каждого пользователя свой индекс и всю информацию по пользователю писать туда. Но есть подозрение что если это и получится реализовать, то искать информацию будет очень сложно.

  2. Exchange. Я хотел бы вести логи журнала MessageTracking для 4 серверов. Они генерируют меньше логов, хотя объем тоже получается большой. Идея была распарсить лог через фильтр grok, затем вырезать лишние поля, в надежде что лог будет меньше и искать данные в индексах будет проще. Но надежды на простое решение разбились об регулярные выражения. Через grok debugger я вывел вот такую регулярку:

(%{TIMESTAMP_ISO8601:date_time}),,(%{HOSTNAME:client_hostname}),,(%{HOSTNAME:server_hostname}),(%{GREEDYDATA:source_context}),,(%{GREEDYDATA:connector_id}),(%{WORD:source}),(%{WORD:event_id}),<(%{GREEDYDATA:internal_message_id})>,(%{UUID:message_id}),(?<recipient_adress>[a-zA-Z0-9_.+=(%{NUMBER:total-bytes}):-]+@[0-9A-Za-z][0-9A-Za-z-]{0,62}(?:\.(?:[0-9A-Za-z][0-‌​9A-Za-z-]{0,62}))*),(%{DATA:recipient_status}),(%{NUMBER:total_bytes}),(%{NUMBER:recipient_count}),,,(%{DATA:message_subject}),(?<sender_adress>[a-zA-Z0-9_.+=(%{NUMBER:total-bytes}):-]+@[0-9A-Za-z][0-9A-Za-z-]{0,62}(?:\.(?:[0-9A-Za-z][0-‌​9A-Za-z-]{0,62}))*),(?<return_path>[a-zA-Z0-9_.+=(%{NUMBER:total-bytes}):-]+@[0-9A-Za-z][0-9A-Za-z-]{0,62}(?:\.(?:[0-9A-Za-z][0-‌​9A-Za-z-]{0,62}))*),(%{GREEDYDATA:message_info})?,(%{WORD:directionality})?,(%{GREEDYDATA:tenant_id})

В дебаггере она отрабатывает идеально, но если вставлять это в конфиг фильтра logstash в таком виде

    filter {
        if "exchange" in [tags] {
      grok {
        match => {
            "message" => "(%{TIMESTAMP_ISO8601:date_time}),,и далее до конца"
         }
       }
     }
    }

То logstash просто перестает принимать логи со всех серверов вообще. В логе filebeat появляются такие сообщения

 ERROR    pipeline/output.go:100    Failed to connect to backoff(async(tcp://10.0.215.141:5044)): dial tcp 10.0.215.141:5044: connectex: No connection could be made because the target machine actively refused it.

Если менять ту часть, где выделяется email, на (%{EMAILADDRESS:email}), то в дебаггере появляется ошибка Compile ERROR (причем судя по темам с названием grok EMAILADDRESS, проблема не только у меня). А в logstash логи приходят с тегом _grokparsefailure.
В logstash_plain.log есть такое:

[ERROR][logstash.javapipeline    ] Pipeline aborted due to error {:pipeline_id=>"main", :exception=>#<RegexpError: empty range in char class
[ERROR][logstash.agent           ] Failed to execute action {:id=>:main, :action_type=>LogStash::ConvergeResult::FailedAction, :message=>"Could not execute action: PipelineAction::Create<main>, action_result: false", :backtrace=>nil}

Проблема точно связана с %{EMAILLOCALPART}, но непонятно что с этим можно сделать. Регулярка до части с почтой отрабатывает в Logstash отлично и в ES приходит правильно.

Вопросы:
Возможно ли сделать с размером логов squid/exchange то, что хотелось бы или такой рост документов норма и нужно смириться?
В чем может быть ошибка регулярного выражения? Это проблема именно с логами Exchange или EMAILLOCALPART в таком виде не работает у всех?
Адеватно ли вообще собирать эти логи через ELK или я трачу время впустую?

С гроком я так на вскидку сказать где проблема не могу, что касается размеров логов - то вопрос в том, какая у Вас цель? Что Вы в этих логах squid собираетесь найти?

Со squid хотелось бы получать данные о том, куда какой пользователь ходил за последние пол года. Для этих целей уже есть lightsquid+sqstat, но мне бы хотелось держать все логи в одном месте.
Для Exchange в идеале хотелось бы видеть аналог powershell команды "Get-MessageTrackingLog -Sender mailto:user@domain.com | ft Timestamp, sender, recipient, message, subject-AutoSize" и тоже держать эти логи вместе с остальными.

В этом случае URL - это, наверное, самая большая часть и единственно что можно порезать - это вытащить из URL домейн. Но не уверен, что это имеет смысл.

Я так понимаю сейчас объем такой большой из-за того что под каждую строку создается отдельный документ? Можно как-нибудь рассчитать насколько увеличивается объем строчки лога, когда она загружается в elasticsearch?
А если попытаться класть больше одной строки в один документ, это может помочь? Только в этом случае наверно придется отказаться от фильтров.
Или если лить как сейчас, но удалять лишние поля? У меня сейчас grok разбирает message на отдельные поля по коду, ip источника, ip хоста и т.д. Половина полей точно не нужна.

Еще немного потестировал регулярки для Exchange, понял что хорошо работает такая:

(%{TIMESTAMP_ISO8601:date_time}),,(%{HOSTNAME:client_hostname}),,(%{HOSTNAME:server_hostname}),(%{DATA:source_context}),,(%{DATA:connector_id}),(%{WORD:source}),(%{WORD:event_id}),<(%{DATA:internal_message_id})>,(%{UUID:message_id}),(%{DATA:recipient_address}),(%{DATA:recipient_status}),(%{NUMBER:total_bytes}),(%{NUMBER:recipient_count}),,,(%{DATA:message_subject}),(%{DATA:sender_address}),(%{DATA:message_info}),(%{WORD:directionality}),(%{DATA:tenant_id})

Но еще оказалось что записи в журналах Exchange выглядят по-разному. У всех одна структура:
#Fields: date-time,client-ip,client-hostname,server-ip,server-hostname,source-context,connector-id,source,event-id,internal-message-id,message-id,network-message-id,recipient-address,recipient-status,total-bytes,recipient-count,related-recipient-address,reference,message-subject,sender-address,return-path,message-info,directionality,tenant-id,original-client-ip,original-server-ip,custom-data

Но поля могут быть заполненными, а могут быть пустыми. Там, где заполненные поля совпадают с регуляркой, все супер. А где отличаются - там parse error. Пока не придумал как это победить.

А сначала разбить по запятым, а потом возиться с каждым полем отдельно не пробовали?

Попробовал, понравилось намного больше. В logstash отрезал поля, которые получил из csv, а в filebeat удалил лишние системные поля. Спасибо.
А по предыдущему вопросу про объем документов сможете подсказать?

Если вам только статистику надо знать, то можно выключить _source и индексацию полей c URL, оставить только docvalues. То есть проиндескировать только timestamp и оставить docvalues только для пользователей, timestamp и URL. Думаю, это должно уменьшить размер раза в 2-3. Но кроме статистики по тому какие сайты кто посещал, вы больше ничего не получите.

1 Like

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.