Nastavljam svoju priču o tome kako steći prijatelje Exchange i ELK (početak ). Dopustite mi da vas podsjetim da je ova kombinacija sposobna obraditi vrlo velik broj trupaca bez oklijevanja. Ovaj put ćemo govoriti o tome kako natjerati Exchange da radi s Logstash i Kibana komponentama.
Logstash u ELK stogu služi za inteligentnu obradu logova i njihovu pripremu za smještaj u Elastic u obliku dokumenata, na temelju kojih je zgodno graditi razne vizualizacije u Kibani.
Instalacija
Sastoji se od dvije faze:
- Instalacija i konfiguracija OpenJDK paketa.
- Instalacija i konfiguracija paketa Logstash.
Instalacija i konfiguracija OpenJDK paketa
Paket OpenJDK mora se preuzeti i raspakirati u određeni direktorij. Zatim se put do ovog direktorija mora unijeti u varijable $env:Path i $env:JAVA_HOME operacijskog 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 . Arhiva mora biti raspakirana u korijen diska. Raspakirajte u mapu C:Program Files Ne isplati se, Logstash će odbiti pokrenuti se normalno. Zatim morate unijeti u datoteku jvm.options popravke odgovorne za dodjelu RAM-a za Java proces. Preporučujem da navedete polovicu RAM-a poslužitelja. Ako ima 16 GB RAM-a, tada su zadani ključevi:
-Xms1g
-Xmx1g
treba zamijeniti sa:
-Xms8g
-Xmx8g
Osim toga, preporučljivo je komentirati redak -XX:+UseConcMarkSweepGC. Više o ovome . Sljedeći korak je stvaranje zadane konfiguracije u datoteci logstash.conf:
input {
stdin{}
}
filter {
}
output {
stdout {
codec => "rubydebug"
}
}
S ovom konfiguracijom, Logstash čita podatke s konzole, propušta ih kroz prazan filtar i šalje ih natrag u konzolu. Korištenje ove konfiguracije testirat će funkcionalnost Logstasha. Da bismo to učinili, pokrenimo ga u interaktivnom načinu rada:
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 je uspješno pokrenut na portu 9600.
Posljednji korak instalacije: pokrenite Logstash kao Windows uslugu. To se može učiniti, na primjer, pomoću paketa :
PS C:...bin> .nssm.exe install logstash
Service "logstash" installed successfully!
tolerancija kvarova
Sigurnost zapisa prilikom prijenosa s izvornog poslužitelja osigurava mehanizam Persistent Queues.
Kako to radi
Raspored redova tijekom obrade dnevnika je: ulaz → red → filter + izlaz.
Dodatak za unos prima podatke iz izvora dnevnika, zapisuje ih u red čekanja i šalje potvrdu da su podaci primljeni izvoru.
Poruke iz reda čekanja obrađuju Logstash, prolaze kroz filter i izlazni dodatak. Kada dobije potvrdu iz izlaza da je zapisnik poslan, Logstash uklanja obrađeni zapisnik iz reda čekanja. Ako se Logstash zaustavi, sve neobrađene poruke i poruke za koje nije primljena potvrda ostaju u redu čekanja, a Logstash će ih nastaviti obrađivati sljedeći put kada se pokrene.
podešavanje
Podesivo ključevima u datoteci C:Logstashconfiglogstash.yml:
queue.type: (moguće vrijednosti -persistedиmemory (default)).path.queue: (put do mape s datotekama reda čekanja, koje su prema zadanim postavkama pohranjene u C:Logstashqueue).queue.page_capacity: (maksimalna veličina stranice čekanja, zadana vrijednost je 64mb).queue.drain: (true/false - omogućuje/onemogućuje zaustavljanje obrade čekanja prije gašenja Logstasha. Ne preporučujem da ga uključite jer će to izravno utjecati na brzinu gašenja poslužitelja).queue.max_events: (maksimalni broj događaja u redu čekanja, zadana vrijednost je 0 (neograničeno)).queue.max_bytes: (maksimalna veličina reda u bajtovima, zadana - 1024mb (1gb)).
Ako je konfigurirano queue.max_events и queue.max_bytes, tada se poruke prestaju prihvaćati u red kada se dosegne vrijednost bilo koje od ovih postavki. Saznajte više o trajnim redovima čekanja .
Primjer dijela logstash.yml koji je odgovoran za postavljanje reda:
queue.type: persisted
queue.max_bytes: 10gb
podešavanje
Logstash konfiguracija obično se sastoji od tri dijela, odgovorna za različite faze obrade dolaznih zapisa: primanje (odjeljak za unos), parsiranje (odjeljak za filtriranje) i slanje Elasticu (odjeljak za izlaz). U nastavku ćemo pobliže pogledati svaku od njih.
Ulazni
Primamo dolazni tok s neobrađenim zapisnicima od filebeat agenata. Ovo je dodatak koji označavamo u odjeljku za unos:
input {
beats {
port => 5044
}
}
Nakon ove konfiguracije, Logstash počinje osluškivati port 5044, a kada prima zapise, obrađuje ih prema postavkama odjeljka filtera. Ako je potrebno, možete omotati kanal za primanje zapisa iz filebita u SSL. Pročitajte više o postavkama beats dodatka .
Filter
Svi tekstualni zapisnici zanimljivi za obradu koje Exchange generira su u csv formatu s poljima opisanim u samoj datoteci zapisa. Za parsiranje csv zapisa, Logstash nam nudi tri dodatka: , csv i grok. Prvi je najviše , ali se nosi s parsiranjem samo najjednostavnijih zapisa.
Na primjer, podijelit će sljedeći zapis na dva (zbog prisutnosti zareza unutar polja), zbog čega će zapisnik biti pogrešno analiziran:
…,"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 dnevnika, na primjer, IIS. U ovom slučaju odjeljak filtra 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" }
}
}
}
Logstash konfiguracija vam omogućuje korištenje , tako da samo zapise koji su označeni filebeat oznakom možemo poslati dodatku za disektiranje IIS. Unutar dodatka povezujemo vrijednosti polja s njihovim nazivima, brišemo izvorno polje message, koji je sadržavao unos iz dnevnika, a možemo dodati prilagođeno polje koje će npr. sadržavati naziv aplikacije iz koje prikupljamo zapise.
U slučaju zapisa praćenja, bolje je koristiti csv dodatak; on može 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 dodatka povezujemo vrijednosti polja s njihovim nazivima, brišemo izvorno polje message (i također polja tenant-id и schema-version), koji je sadržavao unos iz dnevnika, a možemo dodati i prilagođeno polje, koje će npr. sadržavati naziv aplikacije iz koje prikupljamo zapise.
Na izlazu iz faze filtriranja, dobit ćemo dokumente u prvoj aproksimaciji, spremne za vizualizaciju u Kibani. Nedostajat će nam sljedeće:
- Numerička polja bit će prepoznata kao tekst, što onemogućuje operacije na njima. Naime, polja
time-takenIIS dnevnik, kao i poljarecipient-countиtotal-bitesPraćenje dnevnika. - Standardna vremenska oznaka dokumenta sadržavat će vrijeme obrade dnevnika, a ne vrijeme kada je zapisan na strani poslužitelja.
- Polje
recipient-addressće izgledati kao jedno gradilište, što ne dopušta analizom brojanje primatelja pisama.
Vrijeme je da dodate malo čarolije procesu obrade dnevnika.
Pretvaranje numeričkih polja
Dodatak dissect ima opciju convert_datatype, koji se može koristiti za pretvaranje tekstualnog polja u digitalni format. Na primjer, ovako:
dissect {
…
convert_datatype => { "time-taken" => "int" }
…
}
Vrijedno je zapamtiti da je ova metoda prikladna samo ako će polje sigurno sadržavati niz. Opcija ne obrađuje Null vrijednosti iz polja i izbacuje iznimku.
Za zapisnike praćenja bolje je ne koristiti sličnu metodu pretvorbe jer polja recipient-count и total-bites može biti prazan. Za pretvaranje ovih polja bolje je koristiti dodatak :
mutate {
convert => [ "total-bytes", "integer" ]
convert => [ "recipient-count", "integer" ]
}
Dijeljenje recipient_address na pojedinačne primatelje
Ovaj se problem također može riješiti pomoću dodatka za mutaciju:
mutate {
split => ["recipient_address", ";"]
}
Promjena vremenske oznake
U slučaju zapisa praćenja, problem se vrlo jednostavno rješava pomoću dodatka , koji će vam pomoći u pisanju na terenu timestamp datum i vrijeme u traženom formatu iz polja date-time:
date {
match => [ "date-time", "ISO8601" ]
timezone => "Europe/Moscow"
remove_field => [ "date-time" ]
}
U slučaju IIS zapisa, trebat ćemo kombinirati podatke polja date и time koristeći dodatak za mutaciju, registrirajte vremensku zonu koja nam je potrebna i u nju stavite ovu vremensku oznaku timestamp pomoću dodatka za datum:
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
Izlazna sekcija koristi se za slanje obrađenih zapisa do primatelja evidencije. U slučaju slanja izravno na Elastic, koristi se dodatak , koji navodi adresu poslužitelja i predložak imena indeksa za slanje generiranog 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}"
}
}
Korisni linkovi:
Izvor: www.habr.com