🥇Po lidhim ELK dhe Exchange. Pjesa 2

Po çfarë do ta vazhdoj rrëfimin tim për lidhjen mes Exchange dhe ELK (fillimi këtu). Kujtoj se kjo kombinim është në gjendje pa dyshim të përpunojë një sasi shumë të madhe logjesh. Këtë herë do të flasim se si të vendosim punën e Exchange me komponentët Logstash dhe Kibana.

Logstash në stek-un ELK përdoret për përpunimin inteligjent të logjeve dhe përgatitjen e tyre për t'u vendosur në Elastic në formë dokumentesh, të cilat ndihmojnë në ndërtimin e vizualizimeve të ndryshme në Kibana.

Instalimi

Kjo përbëhet nga dy etapa:

Instalimi dhe konfigurimi i paketës OpenJDK.
Instalimi dhe konfigurimi i paketës Logstash.

Instalimi dhe konfigurimi i paketës OpenJDK

Paketën OpenJDK duhet ta shkarkoni dhe ta shpërndani në një direktor të caktuar. Më pas, rruga deri në këtë direktori duhet të shtohet në variablat $env:Path dhe $env:JAVA_HOME në sistemin operativ Windows:

Le të kontrollojmë versionin e 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)

Instalimi dhe konfigurimi i paketës Logstash

Shkarkoni skedarin-arkiv me shpërndarjen e Logstash këtu. Arkiv duhet të shpërndahet në rrënjën e diskut. Nuk duhet të shpërndahen në një folder C:Program Files , Logstash nuk do të fillojë normalisht. Më pas duhet të bëjnë redaktime në skedarin jvm.options që janë përgjegjëse për përkushtimin e memories operative për procesin Java. Rekomandoj të vendosni gjysmën e memories operative të serverit. Nëse ai ka 16 GB RAM, atëherë çelësat që të japin rezultat të zakonshëm janë:

-Xms1g
-Xmx1g

duhet të zëvendësohen me:

-Xms8g
-Xmx8g

Përveç kësaj, është e arsyeshme të komentoni rreshtin -XX:+UseConcMarkSweepGC. Më shumë për këtë këtu. Hapi tjetër është krijimi i konfiguracionit të zakonshëm në skedarin logstash.conf:

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

Me këtë konfiguracion, Logstash lexon të dhënat nga konsola, kalon përmes një filtri bosh dhe i kthen përsëri në konsolë. Përdorimi i këtij konfiguracioni do të ndihmojë në verifikimin e funksionalitetit të Logstash. Për këtë do ta fillojmë atë në modalitetin interaktiv:

PS C:...bin> .logstash.bat -f .logstash.conf
...
[2019-12-19T11:15:27,769][INFO ][logstash.javapipeline    ][main] Pipelinë e nisur {"pipeline.id"=>"main"}
Plugini stdin tani po pret input:
[2019-12-19T11:15:27,847][INFO ][logstash.agent           ] Pipelinët që janë duke u ekzekutuar {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2019-12-19T11:15:28,113][INFO ][logstash.agent           ] Suksesshëm startoi pikën e API të Logstash {:port=>9600}

Logstash u startua me sukses në portin 9600.

Hapi përfundimtar i instalimit: fillimi i Logstash si një shërbim Windows. Këtë mund ta bëni, për shembull, me paketën NSSM:

PS C:...bin> .nssm.exe install logstash
Shërbimi "logstash" u instalua me sukses!

Qëndrueshmëri

Ruajtja e logëve gjatë transferimit nga serveri burim sigurohet nga mekanizmi i Radhave të Persevueshme.

Si funksionon

Skema e vendosjes së radhëve gjatë përpunimit të logëve: input → queue → filter + output.

Plugini input merr të dhëna nga burimi i logëve, i regjistron ato në radhë dhe i dërgon burimit konfirmimin e pranimit të të dhënave.

Mesazhet nga radha përpunohen nga Logstash, kalojnë në filtra dhe plugin output. Pasi merr konfirmimin e dërgimit nga output, Logstash e fshin logun e përpunuar nga radha. Nëse Logstash ndalet, të gjitha mesazhet e papërpunuara dhe mesazhet për të cilat nuk është marrë konfirmimi për dërgimin mbeten në radhë, dhe Logstash do të vazhdojë përpunimin e tyre në nisjen e ardhshme.

Configuration

Rregullohet me çelësat në skedarin C:Logstashconfiglogstash.yml:

queue.type: (vlerat e mundshme — persisted dhe memory (default)).
path.queue: (rruga deri në dossierin me skedarët e radhëve, që sipas marrëveshjes ruhen në C:Logstashqueue).
queue.page_capacity: (madhësia maksimale e faqes së radheve, vlera sipas marrëveshjes — 64mb).
queue.drain: (true/false — aktivizon/nuk aktivizon ndalimin e përpunimit të radhës para fikjes së Logstash. Nuk rekomandohet të aktivizohet, sepse do të ndikojë drejtpërdrejt në shpejtësinë e fikjes së serverit).
queue.max_events: (numri maksimal i ngjarjeve në radhë, sipas marrëveshjes — 0 (pa kufizim)).
queue.max_bytes: (madhësia maksimale e radheve në byte, sipas marrëveshjes — 1024mb (1gb)).

Nëse janë të konfiguruara queue.max_events dhe queue.max_bytes, atëherë mesazhet ndalen së pranuari në radhë kur arrihet vlera e ndonjërit nga këto rregullore. Më shumë rreth Radhave të Persevueshme flitet këtu.

Shembull i një pjese të logstash.yml, që merret me konfigurimin e radhës:

queue.type: persisted
queue.max_bytes: 10gb

Configuration

Konfigurimi i Logstash zakonisht përbëhet nga tri pjesë, që i përkasin fazave të ndryshme të përpunimit të logëve të ardhshëm: pranimi (seksioni input), analizimi (seksioni filter) dhe dërgimi në Elastic (seksioni output). Më poshtë do të shqyrtojmë me detaje secilën prej tyre.

Input

Rryma e ardhshme me logë të papërpunuara pranohet nga agjentët filebeat. Ky është plugini që ne e specifikojmë në seksionin input:

input {
  beats {
    port => 5044
  }
}

Pas një konfigurimi të tillë, Logstash fillon të dëgjojë portin 5044, dhe kur merr logë, i përpunon ato sipas konfigurimeve të seksionit filter. Nëse është e nevojshme, mund të rrethoni kanal për marrjen e logëve nga filebit në SSL. Më shumë rreth konfigurimeve të pluginës beats flitet këtu.

Filter

Të gjitha logët e interesantë për t'u përpunuar, të gjeneruara nga Exchange, janë në format csv me fushat e përshkruara në vetë skedarin e logëve. Për përpunimin e të dhënave csv, Logstash na ofron tre plugins: dissect, csv dhe grok. E para është më e shpejta, por merret vetëm me përpunimin e logëve më të thjeshta.
Për shembull, ky regjistër do ta ndajë në dy (për shkak të pranishmërisë së një presjeje brenda fushës), e cila do ta bëjë logun të përpunohet gabimisht:

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

Ai mund të përdoret për përpunimin e logëve, për shembull, IIS. Në këtë rast, seksioni filter mund të duket si më poshtë:

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

Konfigurimi i Logstash lejon përdorimin e operatorëve të kushtëzuar, kështu që ne mund ta drejtojmë pluginin dissect vetëm në logët që kanë qenë të etiketuar me tagun filebeat IIS. Brenda plugin-it ne përputhim vlerat e fushave me emrat e tyre, heqim fushën origjinale message, e cila përmbante regjistrimin nga logu, dhe mund të shtojmë një fushë të rastësishme, e cila do të përmbajë, për shembull, emrin e aplikacionit nga i cili po mbledhim logët.

Në rastin e logëve të ndjekjes, është më mirë të përdorni pluginin csv, ai di të përpunojë saktë fushat e ndërlikuara:

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

Brenda plugin-it ne përputhim vlerat e fushave me emrat e tyre, heqim fushën origjinale message (dhe gjithashtu fushat tenant-id dhe schema-version), e cila përmbante regjistrimin nga logu, dhe mund të shtojmë një fushë të rastësishme, e cila do të përmbajë, për shembull, emrin e aplikacionit nga i cili po mbledhim logët.

Në daljen nga faza e filtrimit do të marrim dokumente që janë në afërsi të gatshme për vizualizim në Kibana. Një gjë që do na mungojë është si më poshtë:

Fushat numerike do të njihet si tekst, çka nuk lejon kryerjen e operacioneve me to. Konkretisht, fushat time-taken e logut IIS, si dhe fushat recipient-count dhe total-bites e logut Tracking.
Stampimi standardi i kohës së dokumentit do të përmbajë kohën e përpunimit të logut, dhe jo kohën e regjistrimit të tij nga ana e serverit.
Fusha recipient-address do të duket si një strukturë e vetme, çka nuk lejon të bëhet analiza me numërimin e marrësve të email-eve.

Ka ardhur koha për të shtuar pak magji në procesin e përpunimit të logëve.

Konvertimi i fushave numerike

Shtesa dissect ka një opsion convert_datatype, i cili mund të përdoret për të konvertuar një fushë teksti në formatin numerik. Për shembull, kështu:

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

Duhet mbajtur mend se kjo metodë funksionon vetëm nëse fusha do të përmbajë saktësisht një varg. Vlera null nga fushat, opsioni nuk e përpunon dhe kalon në një përjashtim.

Për logët e tracking, metoda e ngjashme e konvertimit është më mirë të mos përdoret, pasi fushat recipient-count dhe total-bites mund të jenë bosh. Për të konvertuar këto fusha, më mirë të përdoret shtesa mutate:

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

Shkëputja e recipient_address në marrës të veçantë

Ky problem mund të zgjidhet gjithashtu me ndihmën e shtesës mutate:

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

Ndryshimi i timestamp

Në rastin e logëve të tracking, problemi zgjidhet shumë lehtë me ndihmën e shtesës date, e cila do të ndihmojë për të regjistruar në fushën timestamp data dhe kohën në formatin e duhur nga fusha date-time:

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

Në rastin e logëve IIS do të na nevojitet të bashkojmë të dhënat e fushave date dhe time me ndihmën e shtesës mutate, të specifikojmë zonën tonë të nevojshme dhe të vendosim këtë timestamp në timestamp me ndihmën e shtesës 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

Sekcioni i output-it përdoret për dërgimin e logëve të përpunuara në marrësin e logëve. Në rastin e dërgimit direkt në Elastic, përdoret shtesa elasticsearch, e cila specifikon adresën e serverit dhe modelin e emrit të indeksit për dërgimin e dokumentit të formuar:

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

Konfigurimi përfundimtar

Konfiguracioni përfundimtar do të duket kështu:

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

Linqe të dobishme: