🥇Połączenie ELK i Exchange. Część 2

Kontynuuję swoją opowieść o tym, jak zintegrować Exchange z ELK (część pierwsza tutaj). Przypominam, że ta kombinacja potrafi bez problemu przetwarzać bardzo dużą ilość logów. Tym razem porozmawiamy o tym, jak skonfigurować współpracę Exchange z komponentami Logstash i Kibana.

Logstash w stosie ELK służy do inteligentnego przetwarzania logów i ich przygotowywania do umieszczenia w Elastic w postaci dokumentów, na podstawie których wygodnie buduje się różne wizualizacje w Kibana.

Instalacja

Składa się z dwóch etapów:

Instalacja i konfiguracja pakietu OpenJDK.
Instalacja i konfiguracja pakietu Logstash.

Instalacja i konfiguracja pakietu OpenJDK

Pakiet OpenJDK należy pobrać i rozpakować do określonego katalogu. Następnie ścieżkę do tego katalogu należy dodać do zmiennych $env:Path oraz $env:JAVA_HOME systemu operacyjnego Windows:

Sprawdźmy wersję 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)

Instalacja i konfiguracja pakietu Logstash

Pobierz archiwum z dystrybucją Logstash stąd. Archiwum należy rozpakować na korzeń dysku. Nie warto rozpakowywać go do folderu C:Program Files , ponieważ Logstash nie uruchomi się poprawnie. Następnie należy wprowadzić zmiany w pliku jvm.options , dotyczące przydzielania pamięci RAM dla procesu Java. Zalecam ustawienie połowy pamięci RAM serwera. Jeśli ma on 16 GB pamięci, to domyślne klucze będą następujące:

-Xms1g
-Xmx1g

, które należy zastąpić na:

-Xms8g
-Xmx8g

Ponadto sensowne jest zakomentowanie linii -XX:+UseConcMarkSweepGC. Więcej informacji na ten temat tutaj. Następnym krokiem jest stworzenie domyślnej konfiguracji w pliku logstash.conf:

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

Przy użyciu tej konfiguracji Logstash odczytuje dane z konsoli, przetwarza przez pusty filtr i wyprowadza z powrotem do konsoli. Zastosowanie tej konfiguracji pozwoli na sprawdzenie działania Logstash. Aby to zrobić, uruchomimy go w trybie interaktywnym:

PS C:...bin> .logstash.bat -f .logstash.conf
...
[2019-12-19T11:15:27,769][INFO ][logstash.javapipeline    ][main] Pipeline started {"pipeline.id"=>"main"}
Wtyczka stdin czeka teraz na dane wejściowe:
[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           ] Z powodzeniem uruchomiono punkt końcowy API Logstash {:port=>9600}

Logstash pomyślnie uruchomił się na porcie 9600.

Ostatni krok instalacji: uruchomienie Logstash jako usługi Windows. Można to zrobić na przykład za pomocą pakietu NSSM:

PS C:...bin> .nssm.exe install logstash
Usługa "logstash" została pomyślnie zainstalowana!

Odporność na awarie

Zachowanie logów podczas przesyłania z serwera źródłowego zapewnia mechanizm Persistent Queues.

Jak to działa

Schemat rozmieszczenia kolejek w procesie przetwarzania logów: input → queue → filter + output.

Plugin input odbiera dane z źródła logów, zapisuje je w kolejce i wysyła potwierdzenie otrzymania danych do źródła.

Wiadomości z kolejki są przetwarzane przez Logstash, przechodzą przez filtr i plugin output. Po otrzymaniu od output potwierdzenia wysłania loga, Logstash usuwa przetworzony log z kolejki. Jeśli Logstash zostanie zatrzymany, wszystkie nieprzetworzone wiadomości i wiadomości, dla których nie otrzymano potwierdzenia wysyłki, pozostają w kolejce, a Logstash wznowi ich przetwarzanie przy następnym uruchomieniu.

Konfiguracja

Regulowane kluczami w pliku C:Logstashconfiglogstash.yml:

queue.type: (możliwe wartości — persisted i memory (domyślnie)).
path.queue: (ścieżka do folderu z plikami kolejek, które domyślnie przechowywane są w C:Logstashqueue).
queue.page_capacity: (maksymalny rozmiar strony kolejki, wartość domyślna — 64mb).
queue.drain: (true/false — włącza/wyłącza zatrzymywanie przetwarzania kolejki przed wyłączeniem Logstash. Nie polecam włączać, ponieważ bezpośrednio wpłynie to na szybkość wyłączenia serwera).
queue.max_events: (maksymalna liczba wydarzeń w kolejce, domyślnie — 0 (bez ograniczeń)).
queue.max_bytes: (maksymalny rozmiar kolejki w bajtach, domyślnie — 1024mb (1gb)).

Jeśli ustawione są queue.max_events i queue.max_bytes, to wiadomości przestają być przyjmowane do kolejki po osiągnięciu wartości któregokolwiek z tych ustawień. Więcej o Persistent Queues opisano tutaj.

Przykład części logstash.yml, odpowiedzialnej za konfigurację kolejki:

queue.type: persisted
queue.max_bytes: 10gb

Konfiguracja

Konfiguracja Logstash zwykle składa się z trzech części, odpowiadających za różne etapy przetwarzania przychodzących logów: odbiór (sekcja input), parsowanie (sekcja filter) i wysyłanie do Elastic (sekcja output). Poniżej szczegółowo omówimy każdą z nich.

Input

Otrzymujemy przychodzący strumień surowych logów od agentów filebeat. To właśnie ten plugin wskazujemy w sekcji input:

input {
  beats {
    port => 5044
  }
}

Po takiej konfiguracji Logstash zaczyna nasłuchiwać na porcie 5044 i po otrzymaniu logów przetwarza je zgodnie z ustawieniami sekcji filter. W razie potrzeby można kanał otrzymywania logów od filebit zaszyfrować w SSL. Więcej o ustawieniach pluginu beats opisano tutaj.

Filter

Wszystkie interesujące do przetwarzania dzienniki tekstowe generowane przez Exchange mają format csv z polami opisanymi w samych plikach dziennika. Do analizowania wpisów csv Logstash oferuje nam trzy wtyczki: dissect, csv i grok. Pierwsza z nich jest najszybsza, ale radzi sobie tylko z analizowaniem najprostszych dzienników.Na przykład, poniższy wpis zostanie rozdzielony na dwa (z powodu występowania przecinka wewnątrz pola), przez co dziennik zostanie zanalizowany niepoprawnie:
…,"MDB:GUID1, Mailbox:GUID2, Event:526545791, MessageClass:IPM.Note, CreationTime:2020-05-15T12:01:56.457Z, ClientType:MOMT, SubmissionAssistant:MailboxTransportSubmissionEmailAssistant",…

Można go używać przy analizowaniu dzienników, na przykład, IIS. W tym przypadku sekcja filter może wyglądać następująco:

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

Konfiguracja Logstash pozwala na użycie

instrukcji warunkowych, dlatego wtyczkę dissect możemy skierować tylko do logów, które zostały oznaczone tagiem filebeatIIS . Wewnątrz wtyczki mapujemy wartości pól z ich nazwami, usuwamy pole źródłowe,które zawierało zapis z dziennika, i możemy dodać dowolne pole, które na przykład będzie zawierało nazwę aplikacji, z której zbieramy logi. messageW przypadku logów śledzenia lepiej użyć wtyczki csv, która potrafi prawidłowo obsługiwać złożone pola:

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

Wewnątrz wtyczki mapujemy wartości pól z ich nazwami, usuwamy pole źródłowe

(a także pola message tenant-id schema-version i ), które zawierało zapis z dziennika, i możemy dodać dowolne pole, które na przykład będzie zawierało nazwę aplikacji, z której zbieramy logi.Na wyjściu z etapu filtrowania otrzymamy dokumenty wstępnie gotowe do wizualizacji w Kibana. Będzie nam brakować następującego:

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

Pola numeryczne będą rozpoznawane jako tekst, co uniemożliwia wykonywanie operacji na nich. A mianowicie, pola time-taken logu IIS, a także pola recipient-count i total-bites logu Tracking.
Standardowy znacznik czasowy dokumentu będzie zawierał czas przetwarzania logu, a nie czas zapisu po stronie serwera.
Pole recipient-address będzie wyglądać jako jedna linia, co uniemożliwia przeprowadzenie analizy z liczeniem odbiorców wiadomości.

Nadszedł czas, aby dodać trochę magii do procesu przetwarzania logów.

Konwersja pól numerycznych

Wtyczka dissect ma opcję convert_datatype, którą można wykorzystać do konwersji pola tekstowego na format liczbowy. Na przykład tak:

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

Należy pamiętać, że ta metoda pasuje tylko wtedy, gdy pole na pewno będzie zawierać ciąg. Wartości Null z pól nie są obsługiwane przez opcję i powodują wyjątek.

W przypadku logów śledzenia lepiej nie używać podobnej metody konwersji, ponieważ pola recipient-count i total-bites mogą być puste. Do konwersji tych pól lepiej użyć wtyczki mutate:

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

Podział recipient_address na poszczególnych odbiorców

Tę sprawę można również rozwiązać za pomocą wtyczki mutate:

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

Zmiana znacznika czasu

W przypadku logów śledzenia zadanie to bardzo łatwo rozwiązuje wtyczka date, która pomoże wpisać w pole timestamp datę i czas w odpowiednim formacie z pola date-time:

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

W przypadku logów IIS będziemy musieli połączyć dane pól date i time za pomocą wtyczki mutate, ustawić odpowiednią strefę czasową i umieścić ten znacznik czasu w timestamp za pomocą wtyczki 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

Sekcja output służy do wysyłania przetworzonych logów do odbiornika logów. W przypadku bezpośredniego wysyłania do Elastic używana jest wtyczka elasticsearch, w której podaje się adres serwera i wzór nazwy indeksu do wysyłania utworzonego dokumentu:

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

Ostateczna konfiguracja

Ostateczna konfiguracja będzie wyglądać następująco:

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

Przydatne linki: