🥇Integrácia ELK a Exchange. Časť 2

Pokračujem vo svojom rozprávaní o tom, ako prepojiť Exchange a ELK (začiatok tu). Pripomínam, že táto kombinácia dokáže bez problémov spracovať veľmi veľké množstvo logov. Tentoraz sa porozprávame o tom, ako nastaviť spoluprácu Exchange s komponentmi Logstash a Kibana.

Logstash v streame ELK sa používa na inteligentné spracovanie logov a ich prípravu na umiestnenie do Elastic vo forme dokumentov, na základe ktorých je pohodlné vytvárať rôzne vizualizácie v Kibana.

Inštalácia

Skladá sa z dvoch etáp:

Inštalácia a nastavenie balíka OpenJDK.
Inštalácia a nastavenie balíka Logstash.

Inštalácia a nastavenie balíka OpenJDK

Balík OpenJDK je potrebné stiahnuť a rozbaliť do určitého adresára. Potom je potrebné pridať cestu k tomuto adresáru do premenných $env:Path a $env:JAVA_HOME operačného systému Windows:

Skontrolujeme verziu 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)

Inštalácia a nastavenie balíka Logstash

Súbor-archív s distribúciou Logstash si stiahnite odtiaľto. Archív je potrebné rozbaliť do koreňového adresára disku. Rozbaliť do priečinka C:Program Files nie je vhodné, Logstash sa odmietne spustiť. Potom je potrebné upraviť súbor jvm.options úpravy zodpovedajúce za pridelenie operačnej pamäte procesu Java. Odporúčam uviesť polovicu operačnej pamäte servera. Ak má 16 GB RAM, predvolené kľúče sú:

-Xms1g
-Xmx1g

je potrebné nahradiť za:

-Xms8g
-Xmx8g

Okrem toho je rozumné zakomentovať riadok -XX:+UseConcMarkSweepGC. Viac o tom tu. Ďalším krokom je vytvorenie predvolenej konfigurácie v súbore logstash.conf:

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

Pri použití tejto konfigurácie Logstash číta dáta z konzoly, prechádza prázdnym filtrom a znovu ich vypisuje do konzoly. Použitie tejto konfigurácie umožní skontrolovať funkčnosť Logstash. Na tento účel ho spustíme v interaktívnom režime:

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 sa úspešne spustil na porte 9600.

Posledný krok inštalácie: spustenie Logstash ako služby Windows. To je možné napríklad pomocou balíka NSSM:

PS C:...bin> .nssm.exe install logstash
Služba "logstash" bola úspešne nainštalovaná!

Odolnosť voči zlyhaniam

Ochrana logov počas prenosu z pôvodného servera je zabezpečená mechanizmom Persistent Queues.

Ako to funguje

Schéma umiestnenia fronty počas spracovania logov: input → queue → filter + output.

Plugin input získava dáta zo zdroja logov, zapisuje ich do fronty a odosiela potvrdenie zdroju o prijatí údajov.

Správy z fronty spracováva Logstash, prechádzajú filtrom a pluginom output. Pri prijatí potvrdenia od output o odoslaní logu Logstash odstráni spracovaný log z fronty. Ak sa Logstash zastaví, všetky nespracované správy a správy, pre ktoré nebolo prijaté potvrdenie o odoslaní, zostanú vo fronte, pričom Logstash pokračuje v ich spracovaní pri ďalšom spustení.

Nastavenie

Reguluje sa kľúčmi v súbore C:Logstashconfiglogstash.yml:

queue.type: (možné hodnoty — persisted a memory (predvolené)).
path.queue: (cesta k adresáru s súbormi fronty, ktoré sa predvolene ukladajú do C:Logstashqueue).
queue.page_capacity: (maximálna veľkosť stránky fronty, predvolené hodnota — 64mb).
queue.drain: (true/false — zapína/vypína zastavenie spracovania fronty pred vypnutím Logstash. Neodporúčam zapínať, pretože to priamo ovplyvní rýchlosť vypnutia servera).
queue.max_events: (maximálny počet udalostí vo fronte, predvolené — 0 (neobmedzené)).
queue.max_bytes: (maximálna veľkosť fronty v bajtoch, predvolené — 1024mb (1gb)).

Ak sú nastavené queue.max_events a queue.max_bytes, správy prestanú byť prijímané do fronty pri dosiahnutí hodnoty ktoréhokoľvek z týchto nastavení. Podrobnejšie informácie o Persistent Queues sú uvedené tu.

Príklad časti logstash.yml, ktorá zodpovedá za nastavenie fronty:

queue.type: persisted
queue.max_bytes: 10gb

Nastavenie

Konfigurácia Logstash sa zvyčajne skladá z troch častí, ktoré zodpovedajú za rôzne fázy spracovania prichádzajúcich logov: prijatie (sekcia input), parsovanie (sekcia filter) a odoslanie do Elastic (sekcia output). Nižšie podrobnejšie preskúmame každú z nich.

Vstup

Prichádzajúci tok so surovými logmi prijímame od agentov filebeat. Tento plugin uvádzame v sekcii input:

input {
  beats {
    port => 5044
  }
}

Po tejto konfigurácii Logstash začne počúvať port 5044 a spracováva prijaté logy podľa nastavení sekcie filter. Ak je to potrebné, môže byť kanál na prijímanie logov od filebit zabalený do SSL. Podrobnejšie informácie o nastaveniach pluginu beats sú popísané tu.

Filter

Všetky zaujímavé textové protokoly, ktoré generuje Exchange, majú formát csv s popisom polí priamo v súbore protokolu. Pre parsovanie csv záznamov Logstash nám ponúka tri pluginy: dissect, csv a grok. Prvý je naj rýchlejší, ale zvláda parsovanie iba najjednoduchších protokolov.
Napríklad nasledujúci záznam rozdelí na dva (kvôli prítomnosti čiarky v poli), čo spôsobí nesprávne parsovanie protokolu:

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

Môže sa použiť pri parsovaní protokolov, napríklad IIS. V tomto prípade môže sekcia filter vyzerať nasledovne:

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

Konfigurácia Logstash umožňuje používať podmienené operátory, takže do pluginu dissect môžeme poslať iba protokoly, ktoré boli označené tagom filebeat IIS. Vnútri pluginu priraďujeme hodnoty polí k ich názvom, odstraňujeme pôvodné pole message, ktoré obsahovalo záznam z protokolu, a môžeme pridať ľubovoľné pole, ktoré môže napríklad obsahovať názov aplikácie, z ktorej zbierame protokoly.

V prípade protokolov sledovania je lepšie použiť plugin csv, ten dokáže správne spracovať zložité polia:

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

Vnútri pluginu priraďujeme hodnoty polí k ich názvom, odstraňujeme pôvodné pole message (a taktiež polia tenant-id a schema-version), ktoré obsahovalo záznam z protokolu, a môžeme pridať ľubovoľné pole, ktoré môže napríklad obsahovať názov aplikácie, z ktorej zbierame protokoly.

Na výstupe zo štádia filtrovania získame dokumenty, ktoré sú v prvom priblížení pripravené na vizualizáciu v Kibana. Čo nám bude chýbať, bude nasledovné:

Číselné polia budú rozpoznávané ako text, čo neumožňuje vykonávať operácie s nimi. A to konkrétne, polia time-taken logu IIS, ako aj polia recipient-count a total-bites logu Tracking.
Štandardné časové pečiatky dokumentu budú obsahovať čas spracovania logu, a nie čas jeho zápisu na strane servera.
Pole recipient-address bude vyzerať ako jedna stavba, čo neumožňuje vykonávať analýzu s počtom príjemcov e-mailov.

Prišiel čas pridať trošku mágie do procesu spracovania logov.

Konverzia číselných polí

Plugin dissect má možnosť convert_datatype, ktorú je možné použiť na konverziu textového poľa do číselného formátu. Napríklad takto:

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

Treba si uvedomiť, že táto metóda sa hodí len v prípade, ak pole určite bude obsahovať reťazec. Null-hodnoty z polí táto možnosť nezpracováva a vyhadzuje výnimku.

Pre logy sledovania je podobná metóda convert lepšie nepoužívať, pretože polia recipient-count a total-bites môžu byť prázdne. Pre konverziu týchto polí je lepšie použiť plugin mutate:

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

Rozdelenie recipient_address na jednotlivých príjemcov

Túto úlohu je možné podobne vyriešiť pomocou pluginu mutate:

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

Zmeníme timestamp

V prípade logov sledovania je úloha veľmi jednoducho riešiteľná pluginom date, ktorý pomôže zapísať do poľa timestamp dátum a čas v požadovanom formáte z poľa date-time:

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

V prípade logov IIS bude potrebné spojiť údaje z polí date a time pomocou pluginu mutate, zadať požadovanú časovú zónu a umiestniť tento časový pečiatok v timestamp pomocou pluginu 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

Sekcia output sa používa na odoslanie spracovaných logov do prijímača logov. V prípade odosielania priamo do Elastic sa používa plugin elasticsearch, kde sa uvádza adresa servera a šablóna názvu indexu na odoslanie vytvoreného dokumentu:

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

Konečná konfigurácia

Konečná konfigurácia bude vyzerať nasledovne:

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

Užitočné odkazy: