Ne jemi miq me ELK dhe Exchange. Pjesa 2

Unë vazhdoj historinë time se si të bëj miq Exchange dhe ELK (fillimi këtu). Më lejoni t'ju kujtoj se ky kombinim është i aftë të përpunojë një numër shumë të madh shkrimesh pa hezitim. Këtë herë ne do të flasim për mënyrën se si të punojmë Exchange me komponentët Logstash dhe Kibana.

Logstash në pirgun ELK përdoret për të përpunuar në mënyrë inteligjente shkrime dhe për t'i përgatitur ato për vendosje në Elastic në formën e dokumenteve, mbi bazën e të cilave është i përshtatshëm për të ndërtuar vizualizime të ndryshme në Kibana.

Instalim

Përbëhet nga dy faza:

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

Instalimi dhe konfigurimi i paketës OpenJDK

Paketa OpenJDK duhet të shkarkohet dhe të shpaketohet në një direktori specifike. Pastaj shtegu për në këtë direktori duhet të futet në variablat $env:Path dhe $env:JAVA_HOME të sistemit operativ Windows:

Le të kontrollojmë versionin 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 e arkivit me shpërndarjen Logstash prandaj. Arkivi duhet të zbërthehet në rrënjën e diskut. Shpaketojeni në dosje C:Program Files Nuk ia vlen, Logstash do të refuzojë të fillojë normalisht. Pastaj duhet të futeni në skedar jvm.options rregullime përgjegjëse për ndarjen e RAM-it për procesin Java. Unë rekomandoj të specifikoni gjysmën e RAM-it të serverit. Nëse ka 16 GB RAM në bord, atëherë çelësat e paracaktuar janë:

-Xms1g
-Xmx1g

duhet të zëvendësohet me:

-Xms8g
-Xmx8g

Për më tepër, këshillohet të komentoni linjën -XX:+UseConcMarkSweepGC. Më shumë rreth kësaj këtu. Hapi tjetër është krijimi i një konfigurimi të paracaktuar në skedarin logstash.conf:

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

Me këtë konfigurim, Logstash lexon të dhënat nga tastiera, i kalon ato përmes një filtri bosh dhe i nxjerr përsëri në tastierë. Përdorimi i këtij konfigurimi do të testojë funksionalitetin e Logstash. Për ta bërë këtë, le ta ekzekutojmë atë në modalitetin interaktiv:

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 u nis me sukses në portin 9600.

Hapi i fundit i instalimit: nisni Logstash si një shërbim Windows. Kjo mund të bëhet, për shembull, duke përdorur paketën NSSM:

PS C:...bin> .nssm.exe install logstash
Service "logstash" installed successfully!

toleranca ndaj gabimeve

Siguria e regjistrave kur transferohen nga serveri burim sigurohet nga mekanizmi Persistent Queues.

Si punon kjo

Paraqitja e radhëve gjatë përpunimit të regjistrave është: hyrje → radhë → filtër + dalje.

Shtojca e hyrjes merr të dhëna nga një burim regjistri, i shkruan ato në një radhë dhe dërgon konfirmimin që të dhënat janë marrë në burim.

Mesazhet nga radha përpunohen nga Logstash, kalohen përmes filtrit dhe shtojcës së daljes. Kur merr konfirmimin nga dalja se regjistri është dërguar, Logstash heq regjistrin e përpunuar nga radha. Nëse Logstash ndalon, të gjitha mesazhet dhe mesazhet e papërpunuara për të cilat nuk është marrë asnjë konfirmim mbeten në radhë dhe Logstash do të vazhdojë t'i përpunojë ato herën tjetër që të fillojë.

rregullim

E rregullueshme nga çelësat në skedar C:Logstashconfiglogstash.yml:

• queue.type: (vlerat e mundshme - persisted и memory (default)).
• path.queue: (rruga drejt dosjes me skedarët e radhës, të cilat ruhen në C:Logstashqueue si parazgjedhje).
• queue.page_capacity: (madhësia maksimale e faqes së radhës, vlera e paracaktuar është 64mb).
• queue.drain: (true/false - aktivizon/çaktivizon ndalimin e përpunimit të radhës përpara mbylljes së Logstash. Nuk rekomandoj aktivizimin e tij, sepse kjo do të ndikojë drejtpërdrejt në shpejtësinë e mbylljes së serverit).
• queue.max_events: (numri maksimal i ngjarjeve në radhë, parazgjedhja është 0 (i pakufizuar)).
• queue.max_bytes: (madhësia maksimale e radhës në bajt, e paracaktuar - 1024 mb (1 GB)).

Nëse konfigurohet queue.max_events и queue.max_bytes, atëherë mesazhet ndalojnë së pranuari në radhë kur të arrihet vlera e ndonjërit prej këtyre cilësimeve. Mësoni më shumë rreth Radhëve të Përhershme këtu.

Një shembull i pjesës së logstash.yml përgjegjëse për vendosjen e radhës:

queue.type: persisted
queue.max_bytes: 10gb

rregullim

Konfigurimi Logstash zakonisht përbëhet nga tre pjesë, përgjegjëse për faza të ndryshme të përpunimit të regjistrave hyrës: marrja (seksioni i hyrjes), analizimi (seksioni i filtrit) dhe dërgimi në Elastic (seksioni i daljes). Më poshtë do të hedhim një vështrim më të afërt në secilën prej tyre.

të dhëna

Ne e marrim transmetimin në hyrje me regjistra të papërpunuar nga agjentët e filebeat. Është kjo shtojcë që ne tregojmë në seksionin e hyrjes:

input {
  beats {
    port => 5044
  }
}

Pas këtij konfigurimi, Logstash fillon të dëgjojë portin 5044 dhe kur merr regjistrat, i përpunon ato sipas cilësimeve të seksionit të filtrit. Nëse është e nevojshme, mund ta mbështillni kanalin për marrjen e regjistrave nga skedari në SSL. Lexoni më shumë rreth cilësimeve të shtojcave beats këtu.

filtra

Të gjithë regjistrat e tekstit që janë interesantë për përpunim që gjeneron Exchange janë në formatin csv me fushat e përshkruara në vetë skedarin e regjistrit. Për analizimin e të dhënave csv, Logstash na ofron tre shtojca: zbërthejnë, csv dhe grok. E para është më e shumta shpejt, por përballon analizimin vetëm të regjistrave më të thjeshtë.
Për shembull, do të ndajë rekordin e mëposhtëm në dysh (për shkak të pranisë së një presjeje brenda fushës), kjo është arsyeja pse regjistri do të analizohet gabimisht:

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

Mund të përdoret kur analizoni regjistrat, për shembull, IIS. Në këtë rast, seksioni i filtrit mund të duket si ky:

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 Logstash ju lejon të përdorni deklarata të kushtëzuara, kështu që ne mund të dërgojmë vetëm regjistrat që janë etiketuar me etiketën filebeat në shtojcën dissect IIS. Brenda shtojcës ne përputhim vlerat e fushës me emrat e tyre, fshijmë fushën origjinale message, i cili përmbante një hyrje nga regjistri dhe ne mund të shtojmë një fushë të personalizuar që, për shembull, do të përmbajë emrin e aplikacionit nga i cili mbledhim regjistrat.

Në rastin e regjistrave të gjurmimit, është më mirë të përdorni shtojcën csv; ai mund të përpunojë saktë fushat komplekse:

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 shtojcës ne përputhim vlerat e fushës me emrat e tyre, fshijmë fushën origjinale message (dhe gjithashtu fusha tenant-id и schema-version), i cili përmbante një hyrje nga regjistri, dhe ne mund të shtojmë një fushë të personalizuar, e cila, për shembull, do të përmbajë emrin e aplikacionit nga i cili mbledhim regjistrat.

Në dalje nga faza e filtrimit, do të marrim dokumentet në një përafrim të parë, gati për vizualizim në Kibana. Do të na mungojnë sa vijon:

• Fushat numerike do të njihen si tekst, gjë që parandalon veprimet në to. Domethënë, fushat time-taken Regjistri IIS, si dhe fushat recipient-count и total-bites Ndjekja e regjistrave.
• Vula standarde e dokumentit do të përmbajë kohën e përpunimit të regjistrit, jo kohën kur është shkruar në anën e serverit.
• Fushë recipient-address do të duket si një kantier ndërtimi, i cili nuk lejon analiza për të numëruar marrësit e letrave.

Është koha për të shtuar pak magji në procesin e përpunimit të regjistrave.

Konvertimi i fushave numerike

Shtojca dissect ka një opsion convert_datatype, i cili mund të përdoret për të kthyer një fushë teksti në një format dixhital. Për shembull, si kjo:

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

Vlen të kujtohet se kjo metodë është e përshtatshme vetëm nëse fusha do të përmbajë patjetër një varg. Opsioni nuk përpunon vlerat Null nga fushat dhe hedh një përjashtim.

Për gjurmimin e regjistrave, është më mirë të mos përdorni një metodë të ngjashme konvertimi, që nga fushat recipient-count и total-bites mund të jetë bosh. Për të konvertuar këto fusha është më mirë të përdorni një shtojcë ndryshohem:

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

Ndarja e adresës së marrësit në marrës individualë

Ky problem mund të zgjidhet gjithashtu duke përdorur shtojcën mutate:

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

Ndryshimi i vulës kohore

Në rastin e regjistrimeve të gjurmimit, problemi zgjidhet shumë lehtë nga shtojca data, e cila do t'ju ndihmojë të shkruani në terren timestamp data dhe ora në formatin e kërkuar nga fusha date-time:

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

Në rastin e regjistrave të IIS, do të na duhet të kombinojmë të dhënat e fushës date и time duke përdorur shtojcën mutate, regjistrojmë zonën kohore që na nevojitet dhe vendosim këtë vulë kohore timestamp duke përdorur shtojcën e datës:

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

prodhim

Seksioni i daljes përdoret për të dërguar regjistrat e përpunuar te marrësi i regjistrit. Në rast të dërgimit direkt në Elastic, përdoret një plugin kërkesë elastike, i cili specifikon adresën e serverit dhe modelin e emrit të indeksit për dërgimin e dokumentit të krijuar:

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

Konfigurimi përfundimtar

Konfigurimi përfundimtar do të duket si ky:

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

Lidhje të dobishme:

• Si të instaloni OpenJDK 11 në Windows?
• Shkarkoni Logstash
• Elastic përdor opsionin e privuar UseConcMarkSweepGC #36828
• NSSM
• Radhë të vazhdueshme
• Shtojca e hyrjes Beats
• Logstash Dude, ku është sharrë elektrike me zinxhir? Më duhet të zbërthej shkrimet e mia
• Shtojca për zbërthimin e filtrit
• conditionals
• Ndrysho shtojcën e filtrit
• Shtojca e filtrit të datës
• Shtojca e daljes Elasticsearch

Burimi: www.habr.com