Sõbrune ELK ja Exchange. Osa 2

Jätkan oma juttu sellest, kuidas ühendada Exchange ja ELK (algus siit). Tuletan meelde, et see kombinatsioon suudab murefree töödelda väga suurt logide hulka. Seekord räägime sellest, kuidas seadistada Exchange'i koostööd Logstashi ja Kibana komponente.

Logstash ELK virnas kasutatakse logide nutikaks töötlemiseks ja nende ettevalmistamiseks Elasticus dokumentide kujul, mille põhjal on mugav luua erinevaid visualiseeringuid Kibanas.

Paigaldamine

Koosneb kahest etapist:

• OpenJDK paketi installimine ja seadistamine.
• Logstashi paketi installimine ja seadistamine.

OpenJDK paketi tuleb alla laadida ja pakkida välja kindlasse kausta. Seejärel tuleb selle kausta tee lisada Windowsi operatsioonisüsteemi $env:Path ja $env:JAVA_HOME keskkonnamuutujatesse:

Kontrollime Java versiooni:

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)

Logstashi paketi installimine ja seadistamine

Laadige alla Logstashi jaotuse arhiivifail

. Arhiiv tuleb pakkida lahti ketta juurese. Lahti pakkida kausta siitei ole soovitatav, Logstash ei käivitu korralikult. Seejärel tuleb faili C:Program Files jvm.options muutmine, mis vastutab Java protsessile mälu eraldamise eest. Soovitan määrata poole serveri mälust. Kui serveril on 16 GB mälu, siis vaikeväärtused on: -Xms1g -Xmx1g

tuleb muuta järgmiseks:

-Xms8g -Xmx8g

Lisaks on mõistlik kommenteerida välja rida

-XX:+UseConcMarkSweepGC . Täiendavat teavet selle kohta. Järgmine samm on luua vaikimisi konfiguratsioon faili logstash.conf: siitinput { stdin{} } filter { } output { stdout { codec => "rubydebug" } }

Selle konfiguratsiooni kasutamisel loeb Logstash andmeid käsurealt, suunab need läbi tühja filtri ja väljastab tagasi käsureale. Selle konfiguratsiooni rakendamine võimaldab kontrollida Logstashi töökindlust. Selleks käivitame selle interaktiivses režiimis:

PS C:...bin> .logstash.bat -f .logstash.conf ... [2019-12-19T11:15:27,769][INFO ][logstash.javapipeline][main] Toru algas {"pipeline.id"=>"main"} Stdin plug-in ootab nüüd sisendit: [2019-12-19T11:15:27,847][INFO ][logstash.agent] Torud töötavad {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]} [2019-12-19T11:15:28,113][INFO ][logstash.agent] Logstashi API lõpp-punkt käivitati edukalt {:port=>9600}

Logstash käivitati edukalt pordil 9600.

Paigaldamise viimane samm: Logstashi käivitamine Windowsi teenusena. Seda saab teha näiteks paketi

NSSM abil.:

PS C:...bin> .nssm.exe install logstash
Teenus "logstash" on edukalt installitud!

Talitluskatkestustunne

Logide säilitamine edastamise ajal algselt serverilt tagatakse Persistent Queues mehhanismiga.

Kuidas see töötab

Logide töötlemise protsessi järjekordade paigutuse skeem: input → queue → filter + output.

Input plugin saab andmed logi allikast, salvestab need järjekorda ja saadab allikale kinnituse andmete saamise kohta.

Järjekorrast saadetised töödeldakse Logstashis, lähevad läbi filtri ja output plugin'i. Kui output'ilt saadakse logi saatmise kinnitus, kustutab Logstash töödeldud logi järjekorrast. Kui Logstash peatub, jäävad kõik töötlemata sõnumid ja sõnumid, mille saatmise kohta pole kinnitust saadud, järjekorda ning Logstash jätkab nende töötlemist järgmisel käivitusel.

Seadistamine

Reguleeritakse failis C:Logstashconfiglogstash.yml:

• queue.type: (võimalikud väärtused — persisted ja memory (vaikimisi)).
• path.queue: (tee järjekorrafailide kaustani, mis vaikimisi asuvad C:Logstashqueue).
• queue.page_capacity: (järjekorralehe maksimaalne suurus, vaikimisi väärtus — 64mb).
• queue.drain: (true/false — lubab/kepib järjekorra töötlemise peatamise enne Logstashi väljalülitamist. Ma ei soovita seda lubada, kuna see mõjutab otseselt serveri väljalülitamise kiirus).
• queue.max_events: (maksimaalne sündmuste arv järjekorras, vaikimisi — 0 (piiramatu)).
• queue.max_bytes: (maksimaalne järjekorra suurus baitides, vaikimisi — 1024mb (1gb)).

Kui on seadistatud queue.max_events ja queue.max_bytes, siis sõnumid lõpetavad järjekorda sisenemise, kui saavutatakse mis tahes sellise seadistuse väärtus. Rohkem teavet Persistent Queues kohta on saadaval siit.

Näidis logstash.yml osast, mis vastutab järjekorra seadistamise eest:

queue.type: persisted
queue.max_bytes: 10gb

Seadistamine

Logstashi konfigureerimine koosneb tavaliselt kolmest osast, mis vastutavad erinevate etappide eest sisenenud logide töötlemisel: vastuvõtt (input sektsioon), parsimine (filter sektsioon) ja saatmine Elastic'i (output sektsioon). Allpool vaatame igaüht neist lähemalt.

Sisend

Vastuvõtame toore logide voolu filebeat agentidelt. Just seda pluginat näitame input sektsioonis:

input {
  beats {
    port => 5044
  }
}

Pärast sellist seadistust hakkab Logstash kuulama porti 5044 ja kui logid saabuvad, töötleb see neid filter sektsiooni seadistuste kohaselt. Vajadusel saab filebeat'i logide vastuvõtukanali pakkida SSL-i. Rohkem teavet plugin-beats seadistuste kohta on kirjeldatud siit.

Filter

Kõik huvitavad tekstilised logid, mida Exchange genereerib, on csv-formaadis ja sisaldavad kirjeldusi logifailis. Csv-kirjete töötlemiseks pakub Logstash meile kolme plugina võimalust: dissect, csv ja grok. Esimene neist on kõige kiirem, kuid tegeleb ainult kõige lihtsamate logide töötlemisega.
Näiteks jagab ta järgmise kirje kaheks (selle tõttu, et väljande sees on koma), mistõttu logi tõlgendatakse valesti:

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

Seda saab kasutada logide töötlemiseks, näiteks IIS logide puhul. Sel juhul võib filter sektsioon välja näha järgmiselt:

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

Logstashi konfiguratsioon võimaldab kasutada tingimuslikke operaatorid, seega saame dissect plugina suunata ainult need logid, mis on märgistatud filebeat sildiga IIS. Plugina sees seome väärtused väljadega ja eemaldame algse välja message, mis sisaldas logikirjet, ja saame lisada suvalise välja, mis võib näiteks sisaldada rakenduse nime, mille logisid me kogume.

Juhul kui tegemist on jälgimislogidega, on parem kasutada csv plugina, mis oskab keerulisi välju õigesti töödelda:

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

Plugina sees seome väärtused väljadega ja eemaldame algse välja message (ja väljad tenant-id ja schema-version), mis sisaldas logikirjet, ja saame lisada suvalise välja, mis näiteks sisaldab rakenduse nime, mille logisid me kogume.

Filtreerimise etapis saame dokumendid, mis on esialgu valmis visualiseerimiseks Kibanas. Meil jääb vajaka järgmistest:

• Numbrilised väljad tuvastatakse tekstina, mis ei võimalda nendega tehingute tegemist. Täpsemalt, väljad time-taken IIS logidest, samuti väljad recipient-count ja total-bites Tracking logidest.
• Dokumendi standardne ajatempli formaat sisaldab logide töötlemise aega, mitte selle kirjutamise aega serveris.
• Väli recipient-address näeb välja kui üks rida, mis ei luba analüüsi koos e-kirjade saajate arvu kokku lugemisega.

On aeg lisada töötlemisprotsessi natuke maagia.

Numbriliste väljade konverteerimine

Dissect pluginil on valik convert_datatype, mida saab kasutada tekstivälja konverteerimiseks numbriliseks formaadiks. Näiteks nii:

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

Tasub meeles pidada, et see meetod sobib ainult juhul, kui väli sisaldab kindlasti stringi. Null-väärtused väljadest seda valikut ei töötle ja viskab erandi.

Trackingu logide puhul on sama meetodi konverteerimist parem mitte kasutada, kuna väljad recipient-count ja total-bites võivad olla tühjad. Nende väljade konverteerimiseks on parem kasutada pluginat mutate:

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

recipient_address jagamine eraldi saajateks

Seda ülesannet saab lahendada ka pluginaga mutate:

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

Muudame timestampi

Trackingu logide puhul lahendatakse probleem pluginaga date, mis aitab määrata väljas timestamp kuupäeva ja aega vajalikus formaadis väljast date-time:

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

IIS logide puhul tuleb meil kombineerida väljade andmed date ja time pluginaga mutate, määrata vajalik ajavöönd ja panna see ajatempleti timestamp pluginaga 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" ]
}

Väljund

Output-sektsioon kasutatakse töödeldud logide saatmiseks logide vastuvõtjasse. Otsese saatmise puhul Elasticusse kasutatakse pluginat elasticsearch, kus määratakse serveri aadress ja indeksi nime mall töödeldud dokumendi saatmiseks:

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

Lõplik konfiguratsioon

Lõplik konfiguratsioon näeb välja järgmiselt:

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

Kasulikud lingid:

• Kuidas installida OpenJDK 11 Windowsis?
• Laadi alla Logstash
• Elastic kasutab aegunud valikut UseConcMarkSweepGC #36828
• abil.
• Püsivad järjekorrad
• Beatsi sisendplugin
• Logstash Kutt, kus on mu kettsaag? Pean oma logisid dissectima
• Dissect plugin
• Tingimused
• Muuda filter plugin
• Kuupäeva filter plugin
• Elasticsearchi väljundplugin

Allikas: habr.com