Ние сме приятели с 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 поправки, отговорни за разпределянето на RAM за Java процеса. Препоръчвам да посочите половината от RAM на сървъра. Ако има 16 GB RAM на борда, тогава ключовете по подразбиране са:

-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. Това може да стане например с помощта на пакета НСМС:

PS C:...bin> .nssm.exe install logstash
Service "logstash" installed successfully!

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

Безопасността на регистрационните файлове при прехвърляне от изходния сървър се гарантира от механизма за постоянни опашки.

Как работи

Оформлението на опашките по време на обработката на журнала е: вход → опашка → филтър + изход.

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

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

регулиране

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

• queue.type: (възможни стойности - persisted и memory (default)).
• 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 обикновено се състои от три части, отговорни за различните фази на обработка на входящи регистрационни файлове: получаване (секция за въвеждане), анализиране (секция за филтриране) и изпращане до Elastic (секция за изход). По-долу ще разгледаме по-отблизо всеки от тях.

Вход

Получаваме входящия поток със сурови регистрационни файлове от filebeat агенти. Това е този плъгин, който посочваме в секцията за въвеждане:

input {
  beats {
    port => 5044
  }
}

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

филтър

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

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

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

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 ви позволява да използвате условни изявления, така че можем да изпращаме само регистрационни файлове, които са маркирани с маркера 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 Проследяване на регистрационни файлове.
• Времето на стандартния документ ще съдържа времето, когато регистрационният файл е бил обработен, а не времето, когато е бил записан от страната на сървъра.
• Област recipient-address ще изглежда като една строителна площадка, което не позволява анализ за преброяване на получателите на писма.

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

Преобразуване на числови полета

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

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

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

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

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

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

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

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

Промяна на клеймото за време

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

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

В случай на регистрационни файлове на IIS ще трябва да комбинираме данни от полета date и time използвайки плъгина за мутиране, регистрирайте часовата зона, от която се нуждаем, и поставете този времеви печат в нея timestamp използвайки приставката за дата:

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" ]
}

Продукция

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

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}"
  }
}

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

• Как да инсталирам OpenJDK 11 на Windows?
• Изтеглете Logstash
• Elastic използва оттеглена опция UseConcMarkSweepGC #36828
• НСМС
• Постоянни опашки
• Плъгин за въвеждане на Beats
• Logstash Пич, къде ми е моторният трион? Трябва да дисектирам моите трупи
• Плъгин за филтър Dissect
• условностите
• Плъгин за мутиране на филтър
• Плъгин за филтър за дата
• Плъгин за извеждане на Elasticsearch

Източник: www.habr.com