🥇Zaprzyjaźniamy się z ELK i Exchange. Część 2

Kontynuuję moją opowieść o tym, jak zaprzyjaźnić się z Exchange i ELK (początek tutaj). Przypomnę, że ta kombinacja jest w stanie bez wahania przetworzyć bardzo dużą liczbę kłód. Tym razem porozmawiamy o tym, jak zmusić Exchange do współpracy z komponentami Logstash i Kibana.

Logstash w stosie ELK służy do inteligentnego przetwarzania logów i przygotowania ich do umieszczenia w Elastic w postaci dokumentów, na podstawie których wygodnie jest budować różne wizualizacje w Kibanie.

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 należy dodać ścieżkę do tego katalogu do zmiennych $env:Path i $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 plik archiwum z dystrybucją Logstash stąd. Archiwum należy rozpakować do katalogu głównego dysku. Rozpakuj do folderu C:Program Files Nie warto, Logstash odmówi normalnego uruchomienia. Następnie musisz wejść do pliku jvm.options poprawki odpowiedzialne za przydzielanie pamięci RAM dla procesu Java. Zalecam określenie połowy pamięci RAM serwera. Jeśli ma na pokładzie 16 GB RAM-u, to domyślne klawisze to:

-Xms1g
-Xmx1g

należy zastąpić:

-Xms8g
-Xmx8g

Ponadto wskazane jest skomentowanie wiersza -XX:+UseConcMarkSweepGC. Więcej o tym tutaj. Następnym krokiem jest utworzenie domyślnej konfiguracji w pliku logstash.conf:

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

W tej konfiguracji Logstash odczytuje dane z konsoli, przepuszcza je przez pusty filtr i wysyła z powrotem do konsoli. Użycie tej konfiguracji przetestuje funkcjonalność Logstash. Aby to zrobić, uruchommy 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"}
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 został pomyślnie uruchomiony na porcie 9600.

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

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

tolerancja błędów

Bezpieczeństwo logów przesyłanych z serwera źródłowego zapewnia mechanizm Persistent Queues.

Jak działa

Układ kolejek podczas przetwarzania logów jest następujący: wejście → kolejka → filtr + wyjście.

Wtyczka wejściowa odbiera dane ze źródła dziennika, zapisuje je do kolejki i wysyła potwierdzenie odebrania danych do źródła.

Wiadomości z kolejki są przetwarzane przez Logstash, przepuszczane przez filtr i wtyczkę wyjściową. Po otrzymaniu potwierdzenia z wyjścia, że log został wysłany, Logstash usuwa przetworzony log z kolejki. Jeśli Logstash się zatrzyma, wszystkie nieprzetworzone wiadomości i wiadomości, dla których nie otrzymano potwierdzenia, pozostaną w kolejce, a Logstash będzie je kontynuował przy następnym uruchomieniu.

regulacja

Regulowane za pomocą kluczy w pliku C:Logstashconfiglogstash.yml:

queue.type: (możliwe wartości - persisted и memory (default)).
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 to 64 MB).
queue.drain: (true/false - włącza/wyłącza zatrzymanie przetwarzania kolejki przed zamknięciem Logstasha. Nie polecam go włączać, ponieważ wpłynie to bezpośrednio na szybkość zamykania serwera).
queue.max_events: (maksymalna liczba zdarzeń w kolejce, domyślnie 0 (nieograniczona)).
queue.max_bytes: (maksymalny rozmiar kolejki w bajtach, domyślnie - 1024mb (1gb)).

Jeśli skonfigurowano queue.max_events и queue.max_bytes, wiadomości przestaną być akceptowane w kolejce po osiągnięciu wartości któregokolwiek z tych ustawień. Dowiedz się więcej o kolejkach trwałych tutaj.

Przykład części pliku logstash.yml odpowiedzialnej za ustawienie kolejki:

queue.type: persisted
queue.max_bytes: 10gb

regulacja

Konfiguracja Logstash zazwyczaj składa się z trzech części, odpowiedzialnych za różne fazy przetwarzania przychodzących logów: odbieranie (sekcja wejściowa), analizowanie (sekcja filtrowania) i wysyłanie do Elastic (sekcja wyjściowa). Poniżej przyjrzymy się bliżej każdemu z nich.

Wkład

Otrzymujemy przychodzący strumień z nieprzetworzonymi logami od agentów filebeat. To właśnie tę wtyczkę wskazujemy w sekcji wejściowej:

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 filtrów. W razie potrzeby możesz owinąć kanał odbioru logów z filebit w SSL. Przeczytaj więcej o ustawieniach wtyczki beats tutaj.

Filtruj

Wszystkie logi tekstowe, które są interesujące do przetwarzania, generowane przez Exchange, są w formacie CSV z polami opisanymi w samym pliku dziennika. Do analizowania rekordów CSV Logstash oferuje nam trzy wtyczki: analizować wnikliwie, CSV i grok. Najbardziej jest ten pierwszy szybki, ale radzi sobie z analizowaniem tylko najprostszych logów.
Przykładowo podzieli następujący rekord na dwa (ze względu na obecność przecinka w polu), dlatego log będzie analizowany 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ć podczas analizowania dzienników, na przykład IIS. W tym przypadku sekcja filtra 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 Instrukcje warunkowe, więc do wtyczki dissect możemy wysyłać tylko logi oznaczone tagiem filebeat IIS. Wewnątrz wtyczki dopasowujemy wartości pól do ich nazw, usuwamy oryginalne pole message, który zawierał wpis z logu, a dodatkowo możemy dodać niestandardowe pole, które będzie zawierało np. nazwę aplikacji, z której zbieramy logi.

W przypadku logów śledzących lepiej jest skorzystać z wtyczki csv, która potrafi poprawnie przetwarzać 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 dopasowujemy wartości pól do ich nazw, usuwamy oryginalne pole message (a także pola tenant-id и schema-version), w którym znajdował się wpis z logu, a dodatkowo możemy dodać niestandardowe pole, które będzie zawierało np. nazwę aplikacji, z której zbieramy logi.

Na wyjściu z etapu filtrowania otrzymamy dokumenty w pierwszym przybliżeniu, gotowe do wizualizacji w Kibanie. Będzie nam brakować:

Pola numeryczne będą rozpoznawane jako tekst, co uniemożliwia wykonywanie na nich operacji. Mianowicie pola time-taken Dziennik IIS, a także pola recipient-count и total-bites Śledzenie dziennika.
Standardowy znacznik czasu dokumentu będzie zawierał czas przetworzenia dziennika, a nie czas jego zapisania po stronie serwera.
Pole recipient-address będzie wyglądać jak jeden plac budowy, co nie pozwala na analizę zliczania odbiorców listów.

Czas dodać trochę magii do procesu przetwarzania logów.

Konwersja pól numerycznych

Wtyczka Dissect ma opcję convert_datatype, którego można użyć do konwersji pola tekstowego na format cyfrowy. Na przykład tak:

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

Warto pamiętać, że ta metoda jest odpowiednia tylko wtedy, gdy pole na pewno będzie zawierało ciąg znaków. Opcja nie przetwarza wartości Null z pól i zgłasza wyjątek.

W przypadku dzienników śledzenia lepiej nie używać podobnej metody konwersji, ponieważ pola recipient-count и total-bites może być pusty. Do konwersji tych pól lepiej jest użyć wtyczki zmutować:

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

Podział adresu_odbiorcy na poszczególnych odbiorców

Ten problem można również rozwiązać za pomocą wtyczki mutującej:

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

Zmiana znacznika czasu

W przypadku logów śledzenia problem bardzo łatwo rozwiązuje wtyczka dane, które ułatwią Ci pisanie w terenie timestamp datę i godzinę w wymaganym formacie z pola date-time:

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

W przypadku logów IIS konieczne będzie połączenie danych z pól date и time za pomocą wtyczki mutuj zarejestruj potrzebną nam strefę czasową i umieść w niej ten znacznik czasu timestamp za pomocą wtyczki daty:

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

Wydajność

Sekcja wyjściowa służy do wysyłania przetworzonych logów do odbiornika logów. W przypadku wysyłki bezpośrednio do Elastica wykorzystywana jest wtyczka wyszukiwanie elastyczne, który określa adres serwera i szablon nazwy indeksu do przesłania wygenerowanego dokumentu:

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

Konfiguracja końcowa

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: