🥇Дружим ELK и Exchange. Част 2

Продължавам разказа си за това как да свържа Exchange и ELK (начало тук). Напомням, че тази комбинация без колебание може да обработва много голямо количество логове. Този път ще говорим за това как да настроим работата на Exchange с компонентите Logstash и Kibana.

Logstash в стека ELK се използва за интелигентна обработка на логове и тяхната подготовка за разполагане в Elastic под формата на документи, на база на които удобно се изграждат различни визуализации в Kibana.

Инсталиране

Състои се от два етапа:

Инсталиране и настройка на пакета OpenJDK.
Инсталиране и настройка на пакета Logstash.

Инсталиране и настройка на пакета OpenJDK

Пакетът OpenJDK трябва да бъде изтеглен и разархивиран в определена директория. След това пътят до тази директория трябва да бъде добавен в променливите $env:Path и $env:JAVA_HOME на операционната система Windows:

Нека проверим версията на Java:

PS C:<; java -version
openjdk version "13.0.1" 2019-10-15
OpenJDK Runtime Environment (build 13.0.1+9)
OpenJDK 64-Bit Server VM (build 13.0.1+9, mixed mode, sharing)

Инсталиране и настройка на пакета Logstash

Файл-архив с дистрибутива на Logstash изтеглете оттук. Архивът трябва да бъде разархивиран в корена на диска. Не е добре да се разархивира в папка C:Program Files , Logstash няма да може да стартира нормално. След това е необходимо да се направят промени в файла jvm.options , отговарящи за разпределението на оперативната памет за Java процеса. Препоръчвам да зададете половината от оперативната памет на сървъра. Ако на него има 16 ГБ оперативна памет, то ключовете по подразбиране:

-Xms1g
-Xmx1g

трябва да се заменят с:

-Xms8g
-Xmx8g

Освен това, уместно е да се коментира редът -XX:+UseConcMarkSweepGC. Повече информация за това тук. Следващата стъпка е създаване на конфигурация по подразбиране в файла logstash.conf:

input {
 stdin{}
}
 
filter {
}
 
output {
 stdout {
 codec => "rubydebug"
 }
}

С използването на тази конфигурация Logstash чете данни от конзолата, пропуска ги през празен филтър и ги извежда обратно в конзолата. Приложението на тази конфигурация ще провери работоспособността на Logstash. За целта ще го стартираме в интерактивен режим:

PS C:...bin> .logstash.bat -f .logstash.conf
...
[2019-12-19T11:15:27,769][INFO ][logstash.javapipeline    ][main] Pipeline started {"pipeline.id"=>"main"}
The stdin plugin is now waiting for input:
[2019-12-19T11:15:27,847][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2019-12-19T11:15:28,113][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}

Logstash успешно стартира на порт 9600.

Финалната стъпка на инсталацията: стартиране на Logstash като Windows услуга. Това може да бъде направено, например, с помощта на пакета NSSM:

PS C:...bin> .nssm.exe install logstash
Служба "logstash" успешно установлена!

Отказоустойчивост

Запазването на логове при предаване от изходния сървър се осигурява от механизма за Персистентни Опашки.

Как работи

Схемата за разположение на опашките в процеса на обработка на логове: input → queue → filter + output.

Input плъгинът получава данни от източника на логовете, записва ги в опашката и изпраща потвърждение на източника за получаване на данните.

Съобщенията от опашката се обработват от Logstash, преминават през филтрите и output плъгина. При получаване на потвърждение от output за изпращането на лог, Logstash изтрива обработения лог от опашката. Ако Logstash бъде спрян, всички необработени съобщения и съобщения, за които не е получено потвърждение за изпращане, остават в опашката и Logstash ще продължи тяхната обработка при следващото стартиране.

Настройка

Регулира се с ключове в файла C:Logstashconfiglogstash.yml:

queue.type: (възможни стойности — persisted и memory (по подразбиране)).
path.queue: (пътя до папката с файловете на опашките, които по подразбиране се съхраняват в C:Logstashqueue).
queue.page_capacity: (максималният размер на страница в опашката, стойността по подразбиране — 64mb).
queue.drain: (true/false — включва/изключва спирането на обработката на опашката преди изключване на Logstash. Не препоръчвам да бъде включвано, тъй като това пряко ще се отрази на скоростта на изключване на сървъра).
queue.max_events: (максималното число събития в опашката, по подразбиране — 0 (неограничено)).
queue.max_bytes: (максималният размер на опашката в байтове, по подразбиране — 1024mb (1gb)).

Ако са настроени queue.max_events и queue.max_bytes, то съобщенията престават да се приемат в опашката при достигане на стойността на всяка от тези настройки. Подробности за Персистентните Опашки са представени тук.

Пример за част от logstash.yml, отговаряща за настройката на опашката:

queue.type: persisted
queue.max_bytes: 10gb

Настройка

Конфигурацията на Logstash обикновено се състои от три части, отговарящи за различни етапи на обработка на входящите логове: получаване (секция input), парсинг (секция filter) и изпращане в Elastic (секция output). По-долу ще разгледаме всяка от тях по-подробно.

Вход

Входящият поток с необработени логове приемаме от агенти filebeat. Този плъгин указваме в секция input:

input {
  beats {
    port => 5044
  }
}

След такава настройка Logstash започва да слуша порта 5044, и при получаване на логове ги обработва в съответствие с настройките на секция filter. При необходимост можете да завиете канала за получаване на логове от filebeat в SSL. Подробности за настройките на плъгина beats са описани тук.

Филтър

Всички интересни текстови логове, генерирани от Exchange, имат csv формат с описаните в самия файл логове полета. За парсинг на csv записи Logstash предлага три плъгина: dissect, csv и grok. Първият е най- бърз, но се справя само с парсинга на най-простите логове.
Например, следната запис ще бъде разбит на две (заради наличието на запетая в полето), заради което логът ще бъде разгледан неправилно:

…,"MDB:GUID1, Mailbox:GUID2, Event:526545791, MessageClass:IPM.Note, CreationTime:2020-05-15T12:01:56.457Z, ClientType:MOMT, SubmissionAssistant:MailboxTransportSubmissionEmailAssistant",…

Той може да се използва при парсинга на логове, например, IIS. В този случай секцията filter може да изглежда по следния начин:

filter {
  if "IIS" in [tags] {
    dissect {
      mapping => {
        "message" => "%{date} %{time} %{s-ip} %{cs-method} %{cs-uri-stem} %{cs-uri-query} %{s-port} %{cs-username} %{c-ip} %{cs(User-Agent)} %{cs(Referer)} %{sc-status} %{sc-substatus} %{sc-win32-status} %{time-taken}"
      }
      remove_field => ["message"]
      add_field => { "application" => "exchange" }
    }
  }
}

Конфигурацията на Logstash позволява използването на условни оператори, затова можем да насочим плъгина dissect единствено към логовете, които са били маркирани с filebeat тага IIS. Вътре в плъгина съпоставяме стойностите на полетата с техните названия, премахваме изходното поле message, което е съдържало записа от лога, и можем да добавим произволно поле, което ще съдържа, например, името на приложението, от което събираме логовете.

В случай с логовете за проследяване е по-добре да се използва плъгинът csv, той може да обработва сложни полета коректно:

filter {
  if "Tracking" in [tags] {
    csv {
      columns => ["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","transport-traffic-type","log-id","schema-version"]
      remove_field => ["message", "tenant-id", "schema-version"]
      add_field => { "application" => "exchange" }
    }
}

Вътре в плъгина съпоставяме стойностите на полетата с техните названия, премахваме изходното поле message (както и полетата tenant-id и schema-version), което е съдържало записа от лога, и можем да добавим произволно поле, което ще съдържа, например, името на приложението, от което събираме логовете.

На изхода от етапа на филтриране ще получим документи, от първоначално готови за визуализация в Kibana. Ще ни липсва следното:

Числовите полета ще бъдат разпознати като текст, което не позволява извършването на операции с тях. А именно, полетата time-taken на IIS логовете, както и полетата recipient-count и total-bites на Tracking.
Стандартният времеви печат на документа ще съдържа времето за обработка на логовете, а не времето на запис на сървъра.
Поле recipient-address ще изглежда като единни структури, което не позволява извършването на анализ с преброяване на получателите на писмата.

Време е да добавим малко магия в процеса на обработка на логовете.

Конвертиране на числови полета

Плагинът dissect има опция convert_datatype, която може да се използва за конвертиране на текстово поле в цифров формат. Например, така:

dissect {
  …
  convert_datatype => { "time-taken" => "int" }
  …
}

Трябва да се помни, че този метод е подходящ само ако полето ще съдържа точна стойност. Null-стойностите от полето не се обработват от опцията и водят до изключение.

За логовете за тракинг е по-добре да не се използва аналогичният метод convert, тъй като полетата recipient-count и total-bites могат да бъдат празни. За конвертиране на тези полета е по-добре да се използва плагинът mutate:

mutate {
  convert => [ "total-bytes", "integer" ]
  convert => [ "recipient-count", "integer" ]
}

Разделяне на recipient_address на отделни получатели

Тази задача може да бъде решена и с помощта на плагина mutate:

mutate {
  split => ["recipient_address", ";"]
}

Промяна на времевия печат

При логовете за тракинг, задачата може да бъде решена лесно с плагина date, който ще помогне да се запише в полето timestamp датата и часа в необходимия формат от полето date-time:

date {
  match => [ "date-time", "ISO8601" ]
  timezone => "Europe/Moscow"
  remove_field => [ "date-time" ]
}

При логовете на IIS ще трябва да обединим данните от полетата date и time чрез плагина mutate, да зададем необходимата времева зона и да поставим този времеви печат в timestamp чрез плагина date:

mutate { 
  add_field => { "data-time" => "%{date} %{time}" }
  remove_field => [ "date", "time" ]
}
date { 
  match => [ "data-time", "YYYY-MM-dd HH:mm:ss" ]
  timezone => "UTC"
  remove_field => [ "data-time" ]
}

Регистри за посока, регистра за изход. В текущия проект те няма да са ни нужни.

Секцията output се използва за изпращане на обработените логове към приемника на логовете. При изпращане директно в Elastic се използва плагин elasticsearch, в който се посочва адреса на сървъра и шаблонът на името на индекса за изпращане на формирования документ:

output {
  elasticsearch {
    hosts => ["127.0.0.1:9200", "127.0.0.2:9200"]
    manage_template => false
    index => "Exchange-%{+YYYY.MM.dd}"
  }
}

Крайна конфигурация

Крайната конфигурация ще изглежда по следния начин:

input {
  beats {
    port => 5044
  }
}
 
filter {
  if "IIS" in [tags] {
    dissect {
      mapping => {
        "message" => "%{date} %{time} %{s-ip} %{cs-method} %{cs-uri-stem} %{cs-uri-query} %{s-port} %{cs-username} %{c-ip} %{cs(User-Agent)} %{cs(Referer)} %{sc-status} %{sc-substatus} %{sc-win32-status} %{time-taken}"
      }
      remove_field => ["message"]
      add_field => { "application" => "exchange" }
      convert_datatype => { "time-taken" => "int" }
    }
    mutate { 
      add_field => { "data-time" => "%{date} %{time}" }
      remove_field => [ "date", "time" ]
    }
    date { 
      match => [ "data-time", "YYYY-MM-dd HH:mm:ss" ]
      timezone => "UTC"
      remove_field => [ "data-time" ]
    }
  }
  if "Tracking" in [tags] {
    csv {
      columns => ["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","transport-traffic-type","log-id","schema-version"]
      remove_field => ["message", "tenant-id", "schema-version"]
      add_field => { "application" => "exchange" }
    }
    mutate {
      convert => [ "total-bytes", "integer" ]
      convert => [ "recipient-count", "integer" ]
      split => ["recipient_address", ";"]
    }
    date {
      match => [ "date-time", "ISO8601" ]
      timezone => "Europe/Moscow"
      remove_field => [ "date-time" ]
    }
  }
}
 
output {
  elasticsearch {
    hosts => ["127.0.0.1:9200", "127.0.0.2:9200"]
    manage_template => false
    index => "Exchange-%{+YYYY.MM.dd}"
  }
}

Полезни връзки: