🥇 Upoznajemo ELK i Exchange. Dio 2

Nastavljam svoju priču o tome kako uskladiti Exchange i ELK (početak ovdje). Podsjećam da ova kombinacija može bez problema obraditi vrlo veliki broj logova. Ovaj put razgovaramo o tome kako uspostaviti rad Exchangea s komponentama Logstash i Kibana.

Logstash u ELK stogu koristi se za inteligentnu obradu logova i pripremu za pohranu u Elastic u obliku dokumenata, na temelju kojih je lako izraditi različite vizualizacije u Kibani.

Instalacija

Sastoji se od dva koraka:

Instalacija i konfiguracija paketa OpenJDK.
Instalacija i konfiguracija paketa Logstash.

Instalacija i konfiguracija paketa OpenJDK

Paket OpenJDK potrebno je preuzeti i raspakirati u određeni direktorij. Zatim treba unijeti put do tog direktorija u varijable $env:Path i $env:JAVA_HOME operativnog sustava Windows:

Provjerimo verziju Jave:

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)

Instalacija i konfiguracija paketa Logstash

Preuzmite arhivsku datoteku s distribucijom Logstash odavde. Arhiv treba raspakirati u korijenu diska. Ne treba raspakirati u mapu C:Program Files , Logstash neće ispravno raditi. Zatim treba udatoteci jvm.options izvršiti izmjene koje se odnose na dodjelu RAM-a za Java proces. Preporučujem da navedete polovicu RAM-a servera. Ako ima 16 GB RAM-a, tada zadnje vrijednosti su:

-Xms1g
-Xmx1g

morate zamijeniti s:

-Xms8g
-Xmx8g

Osim toga, korisno je komentirati redak -XX:+UseConcMarkSweepGC. Više informacija o ovome ovdje. Sljedeći korak je izrada konfiguracije prema zadanim postavkama u datoteci logstash.conf:

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

Korištenjem ove konfiguracije Logstash čita podatke iz konzole, prolazi kroz prazni filtar i vraća ih natrag u konzolu. Ova konfiguracija će omogućiti testiranje ispravnosti rada Logstasha. Da bismo to učinili, pokrenimo ga u interaktivnom načinu:

PS C:...bin> .logstash.bat -f .logstash.conf
...
[2019-12-19T11:15:27,769][INFO ][logstash.javapipeline    ][main] Pipeline started {"pipeline.id"=>"main"}
Plugin stdin sada čeka unos:
[2019-12-19T11:15:27,847][INFO ][logstash.agent           ] Pipelini rade {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2019-12-19T11:15:28,113][INFO ][logstash.agent           ] Uspješno pokrenut Logstash API endpoint {:port=>9600}

Logstash se uspješno pokrenuo na portu 9600.

Završni korak instalacije: pokretanje Logstasha kao Windows usluge. To se može učiniti, na primjer, pomoću paketa NSSM:

PS C:...bin> .nssm.exe install logstash
Usluga "logstash" je uspješno instalirana!

Otpornost na greške

Zaštita logova pri prijenosu s izvornog servera osigurava mehanizam Persistent Queues.

Kako radi

Shema rasporeda redova u procesu obrade logova: input → queue → filter + output.

Input plugin prima podatke iz izvora logova, zapisuje ih u red i šalje izvoru potvrdu o prijemu podataka.

Poruke iz reda obrađuje Logstash, prolaze kroz filter i output plugin. Kada output primi potvrdu o slanju loga, Logstash briše obrađeni log iz reda. Ako se Logstash zaustavi, sve neobrađene poruke i poruke za koje nije primljena potvrda o slanju ostaju u redu, a Logstash će nastaviti njihovu obradu prilikom sljedećeg pokretanja.

Podešavanje

Podešava se ključevima u datoteci C:Logstashconfiglogstash.yml:

queue.type: (moguće vrijednosti — persisted i memory (zadano)).
path.queue: (put do mape s datotekama redova, koji se prema zadanim postavkama čuvaju u C:Logstashqueue).
queue.page_capacity: (maksimalna veličina stranice reda, zadana vrijednost — 64mb).
queue.drain: (true/false — uključuje/isključuje zaustavljanje obrade reda prije gašenja Logstasha. Ne preporučujem uključiti, jer će to izravno utjecati na brzinu gašenja servera).
queue.max_events: (maksimalni broj događaja u redu, prema zadanim postavkama — 0 (neograničeno)).
queue.max_bytes: (maksimalna veličina reda u bajtovima, prema zadanim postavkama — 1024mb (1gb)).

Ako su postavljene queue.max_events i queue.max_bytes, tada poruke prestaju pristizati u red kada se dostigne vrijednost bilo kojeg od ovih postavki. Detaljnije o Persistent Queues pročitajte ovdje.

Primjer djela logstash.yml koji odgovara za postavku reda:

queue.type: persisted
queue.max_bytes: 10gb

Podešavanje

Konfiguracija Logstasha obično se sastoji od tri dijela, koja odgovaraju različitim fazama obrade ulaznih logova: prijem (input sekcija), parsiranje (filter sekcija) i slanje u Elastic (output sekcija). U nastavku ćemo detaljnije razmotriti svaki od njih.

Ulaz

Ulazni tok sa sirovim logovima primamo od agenata filebeat. Upravo taj plugin navodimo u input sekciji:

input {
  beats {
    port => 5044
  }
}

Nakon ove postavke Logstash počinje slušati port 5044, i prilikom primitka logova obrađuje ih prema postavkama filter sekcije. Po potrebi se kanal za primanje logova od filebeat može omotati u SSL. Detaljnije o postavkama beats plugina napisano je. ovdje.

Filtriraj

Svi zanimljivi tekstualni logovi koje generira Exchange imaju csv-format s opisanim poljima unutar samog log datoteka. Za parsiranje csv-zapisa, Logstash nam nudi tri plugina: dissect, csv i grok. Prvi je naj brži, ali se nosi samo s parsiranjem najjednostavnijih logova.
Na primjer, sljedeći zapis će se podijeliti na dva (zbog prisutnosti zareza unutar polja), što će dovesti do pogrešne analize loga:

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

Može se koristiti prilikom parsiranja logova, na primjer, IIS. U tom slučaju, sekcija filter može izgledati ovako:

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

Konfiguracija Logstash omogućuje korištenje uvjetnih operatora, pa tako možemo usmjeriti samo logove označene filebeat tagom IIS. Unutar plugina usklađujemo vrijednosti polja s njihovim nazivima, uklanjamo izvorno polje message, koje je sadržavalo zapis iz loga, i možemo dodati proizvoljno polje koje će, na primjer, sadržavati ime aplikacije iz koje prikupljamo logove.

U slučaju logova praćenja, bolje je koristiti plugin csv, koji zna ispravno obraditi složena polja:

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

Unutar plugina usklađujemo vrijednosti polja s njihovim nazivima, uklanjamo izvorno polje message (a također polja tenant-id i schema-version), koje je sadržavalo zapis iz loga, i možemo dodati proizvoljno polje koje će, na primjer, sadržavati ime aplikacije iz koje prikupljamo logove.

Na izlazu iz faze filtriranja dobit ćemo dokumente koji su u prvom pristupu spremni za vizualizaciju u Kibana. Nedostajat će nam sljedeće:

Numerička polja će biti prepoznata kao tekst, što onemogućava izvođenje operacija nad njima. Konkretno, polja time-taken logova IIS, kao i polja recipient-count i total-bites logova praćenja.
Standardna vremenska oznaka dokumenta će sadržavati vrijeme obrade loga, a ne vrijeme njegovog zapisivanja na strani poslužitelja.
Polje recipient-address će izgledati kao jedan niz, što onemogućava analizu sa brojanjem primatelja e-pošte.

Došlo je vrijeme da dodamo malo magije u proces obrade logova.

Konverzija numeričkih polja

Plugin dissect ima opciju convert_datatype, koju je moguće koristiti za konverziju tekstualnog polja u digitalni format. Na primjer, ovako:

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

Važno je napomenuti da je ova metoda pogodna samo ako polje sigurno sadrži string. Null vrijednosti iz polja opcija ne obrađuje i dolazi do iznimke.

Za logove praćenja slična metoda konverzije bolje je ne koristiti, jer polja recipient-count i total-bites mogu biti prazna. Za konverziju tih polja najbolje koristiti plugin mutate:

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

Razdvajanje recipient_address na pojedinačne primatelje

Ovu zadaću također je moguće riješiti pomoću plugina mutate:

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

Mijenjanje timestamp

U slučaju logova praćenja, zadatak se vrlo lako rješava pomoću plugina date, koji će pomoći da se u polje timestamp upišu datum i vrijeme u željenom formatu iz polja date-time:

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

U slučaju logova IIS bit će potrebno spojiti podatke iz polja date i time uz pomoć plugina mutate, odrediti potrebnu vremensku zonu i smjestiti ovu vremensku oznaku timestamp uz pomoć plugina 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" ]
}

Izlaz

Sekcija output se koristi za slanje obrađenih logova u prijemnik logova. U slučaju slanja izravno u Elastic koristi se plugin elasticsearch, u kojem se navodi adresa poslužitelja i obrazac imena indeksu za slanje formiranog dokumenta:

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

Konačna konfiguracija

Konačna konfiguracija će izgledati ovako:

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

Korisne veze: