Sõbrune ELK ja Exchange. Osa 2

Jätkan oma juttu sellest, kuidas sõbrustada Exchange'i ja ELK-i (algus siin). Tuletan meelde, et see kombinatsioon suudab kindlalt töödelda väga suurt hulka logisid. Sel korral räägime, kuidas seadistada Exchange'i koostööd Logstash'i ja Kibana komponentidega.

Logstash'i kasutatakse ELK-hunnikus logide intelligentseks töötlemiseks ja nende ettevalmistamiseks Elasticus dokumentide kujul, mille põhjal on mugav luua erinevaid visualiseeringuid Kibanas.

Installeerimine

Koosneb kahest etapist:

• OpenJDK paketi installimine ja seadistamine.
• Logstash paketi installimine ja seadistamine.

OpenJDK paketi installimine ja seadistamine

OpenJDK paketi tuleb alla laadida ja lahti pakkida kindlasse katalooge. Seejärel tuleb selle katalooge tee lisada Windowsi operatsioonisüsteemi muutujatesse $env:Path ja $env:JAVA_HOME:

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)

Logstash paketi installimine ja seadistamine

Laadige alla Logstash'i jaotise arhiivifail siit. Arhiiv tuleb lahti pakkida ketta juurtesse. Ei tasu lahti pakkida kausta C:Program Files , Logstash keeldub õigesti käivitumast. Seejärel tuleb faili lisada jvm.options Java protsessi mälu määramiseks vajalikud parameetrid. Soovitan määrata pool serveri RAM-ist. Kui serveril on 16 GB RAM-i, siis vaikimisi parameetrid:

-Xms1g
-Xmx1g

tuleb asendada:

-Xms8g
-Xmx8g

Lisaks on mõistlik kommentaariks märkida rida -XX:+UseConcMarkSweepGC. Rohkem selle kohta siin. Järgmine samm on vaikimisi konfigureerimine failis logstash.conf:

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

Selle konfiguratsiooni kasutamisel loeb Logstash andmeid konsoolist, läbib tühja filtri ja tagastab andmed konsooli. Selle konfiguratsiooniga saab kontrollida Logstashi töövõimet. 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] 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 käivitati edukalt pordil 9600.

Installeerimise viimane samm: Logstashi käitamine Windowsi teenusena. Seda saab teha näiteks paketi abil NSSM:

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

Katastroofitaluvus

Logide säilitamine edastamise ajal algserverist on tagatud Pikaajaliste Ridade mehhanismiga.

Kuidas see töötab

Ridade asukoha skeem logide töötlemise protsessi jooksul: input → queue → filter + output.

Input plugin saab andmeid logi allikast, salvestab need järjekorda ja saadab allikatele andmete vastuvõtmise kinnituse.

Järjekorrast saadetud sõnumid töötleb Logstash, need läbivad filtri ja output plugina. Kui output annab logi edastamise kinnituse, kustutab Logstash töötlenud logi järjekorrast. Kui Logstash peatub, jäävad kõik töötlemata sõnumid ja sõnumid, mille kohta ei ole saadud kinnitust edastamisest, järjekorda ning Logstash jätkab nende töötlemist järgmise käivitamise ajal.

Seadistamine

Reguleeritakse failis C:Logstashconfiglogstash.yml:

• queue.type: (võimalikud väärtused — persisted ja memory (vaikimisi)).
• path.queue: (tee kausta, kus asuvad järjekorra failid, mis vaikimisi asuvad C:Logstashqueue-s).
• queue.page_capacity: (järjekorra maksimaalne lehekülje suurus; vaikimisi väärtus — 64mb).
• queue.drain: (true/false — aktiveerib/deaktiveerib järjekorra töötlemise peatamise Logstash'i sulgemisel. Soovitame mitte aktiveerida, kuna see mõjutab serveri sulgemise kiirus).
• queue.max_events: (maximaalne sündmuste arv järjekorras, vaikimisi — 0 (piiramatu)).
• queue.max_bytes: (maximaalne järjekorra suurus baitides, vaikimisi — 1024mb (1gb)).

Kui on seadistatud queue.max_events ja queue.max_bytes, siis sõnumite vastuvõtt järjekorda lõpetatakse, kui saavutatakse ükskõik millise nende seadistuste väärtus. Rohkem teavet Persistent Queues kohta on kirjutatud siin.

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

queue.type: persisted
queue.max_bytes: 10gb

Seadistamine

Logstash'i konfiguratsioon koosneb tavaliselt kolmest osast, mis vastutavad erinevate etappide eest sissetulevate logide töötlemisel: vastuvõtt (input sektsioon), parsimine (filter sektsioon) ja saatmine Elasticusse (output sektsioon). Allpool vaatleme neid igaüht lähemalt.

Sisend

Vastuvõtame toorandmeid filebeat'i agentidelt. Just selle plugina määrame input sektsioonis:

input {
  beats {
    port => 5044
  }
}

Pärast sellist seadistust hakkab Logstash kuulama porti 5044 ning töötleb logisid vastavalt filtri sektsiooni seadistustele. Vajadusel saab filebiti logide vastuvõtukanali ümber pakkida SSL-iga. Plugin beats seadistuste kohta on rohkem teavet. siin.

Filtreerimine

Kõik Exchange'i genereeritud huvitavad tekstilogid on csv-formaadis ning logifailis on kirjeldatud väljad. Csv-kandjate parsimiseks pakub Logstash kolme pluginat: dissect, csv ja grok. Esimene on kõige kiirem, kuid suudab parsimisega hakkama saada vaid kõige lihtsamate logidega.
Näiteks jagab see järgmise kirje kaheks (sisevälja koma tõttu), 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 näiteks IIS-i logide parsimiseks. Sel juhul võib filtri 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 operaatoreid, seetõttu saame dissect pluginasse suunata ainult logid, mis on märgistatud filebeat sildiga IIS. Pluginas seome väljaannete väärtused nende nimedega, eemaldame algse välja message, mis sisaldas logi kirjet, ning saame lisada omavolilise väljundi, mis võib näiteks sisaldada rakenduse nime, mille logisid kogume.

Juhul, kui tegeleme jälgimislogidega, on parim kasutada csv pluginat, mis oskab korrektselt töödelda keerulisi välju:

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

Pluginas seome väljaannete väärtused nende nimedega, eemaldame algse välja message (ja ka väljad tenant-id ja schema-version), mis sisaldas logi sissekannet ning saame lisada suvalise välja, mis näiteks sisaldab rakenduse nime, millest kogume logisid.

Filtreerimise etapi väljundis saame dokumendid, mis on esimeses lähenemises valmis visualiseerimiseks Kibanas. Meil puudub veel järgnev:

• Numbrilised väljad tuvastatakse tekstina, mis ei võimalda nendega tehingute tegemist. Eelkõige väljadel time-taken IIS logist ning ka väljadel recipient-count ja total-bites Tracking logis.
• Dokumendi standardne ajatempleeritud ei sisalda logi töötlemise aega, vaid serveripoolse salvestamise aega.
• Väli recipient-address näeb välja ühe ahelana, mis ei võimalda meil analüüsi läbi viia e-kirjade saajate arvu peale.

On aeg lisada logitöötlusprotsessi veidi võlu.

Numbriliste väljade konverteerimine

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

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

Oluline on meeles pidada, et see meetod sobib ainult juhul, kui väli sisaldab kindlasti stringi. Null-väärtusi väljadest see meetod ei töötle ja põhjustab erandi.

Jälgimislogide jaoks ei ole sarnast konverteerimismeetodit parem kasutada, kuna väljad võivad olla tühjad. recipient-count ja total-bites Nende väljade konverteerimiseks on parem kasutada pistikprogrammi. mutate:

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

recipient_address jagamine eraldi saajateks.

Seda ülesannet saab samuti lahendada pistikprogrammi mutate abil:

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

Muudame timestamp'i.

Jälgimislogide puhul lahendab ülesande väga lihtsalt pistikprogramm date, mis aitab sisestada vajalikku kuupäeva ja aega timestamp väljast. date-time:

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

IIS logide puhul peame ühendama väljade andmed date ja aeg pistikprogrammi mutate abil, määrama soovitud ajatsooni ja paigutama selle ajatempli timestamp pistikprogrammi date abil:

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

Väljund sektsioon on mõeldud töödeldud logide saatmiseks logide vastuvõtjana. Otsese saatmise korral Elasticusse kasutatakse pistikprogrammi elasticsearch, kus määratakse serveri aadress ja indeksi nime mall saatmiseks genereeritud dokumendile:

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
• NSSM
• Püsivad järjekorrad
• Beatsi sisendi plugin
• Logstash Dude, kus mu mootorsaag on? Mul on vaja oma logisid dissekteerida
• Dissect filter plugin
• Tingimused
• Mutate filter plugin
• Kuupäeva filter plugin
• Elasticsearchi väljundplugin

Allikas: habr.com