ProHoster > Blog > Administrimi > Miqtë ELK dhe Exchange. Pjesa 2

Miqtë ELK dhe Exchange. Pjesa 2

Miqtë ELK dhe Exchange. Pjesa 2

Po vazhdoj historinë time për të bashkuar Exchange dhe ELK (fillimi këtu). Dua të kujtoj se kjo kombinacion është në gjendje të procesojë pa hezitim një sasi shumë të madhe logësh. Këtë herë do të flasim për se si të krijojmë një lidhje ndërmjet Exchange dhe komponentëve Logstash dhe Kibana.

Logstash në stekun ELK përdoret për përpunimin inteligjent të logëve dhe përgatitjen e tyre për publikim në Elastic në formën e dokumenteve, mbi të cilat është e lehtë të ndërtohen visualizime të ndryshme në Kibana.

Instalimi

Kjo përfshin dy etapa:

  • 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ë krijohet një dosje në një drejtori të caktuar. Pastaj, rruga e kësaj drejtorie duhet të futet në variablat $env:Path dhe $env:JAVA_HOME të sistemit operativ Windows:

Miqtë ELK dhe Exchange. Pjesa 2

Miqtë ELK dhe Exchange. Pjesa 2

Le të verifikojmë 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

Shkarko skedarin e arkivuar me shpërndarjen e Logstash këtu. Arkivi duhet të krijohet në rrënjën e disku. Nuk duhet të krijohet në një dosje C:Program Files , pasi Logstash do të refuzojë të fillojë normalisht. Më pas, duhen bërë ndryshime në skedarin jvm.options , që lidhen me alokimin e memorie për procesin Java. Rekomandoj të përcaktoni gjysmën e memories së serverit. Nëse serveri ka 16 GB RAM, cilësimet e paracaktuara janë:

-Xms1g
-Xmx1g

duhen zëvendësuar me:

-Xms8g
-Xmx8g

Përveç kësaj, është e arsyeshme të komentohet rreshti -XX:+UseConcMarkSweepGC. Më shumë rreth kësaj këtu. Hapi tjetër është krijimi i konfigurimit të paracaktuar në skedarin logstash.conf:

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

Me këtë konfigurim, Logstash lexon të dhënat nga konsola, i kalon përmes një filtri bosh dhe i kthen prapa në konsolë. Përdorimi i këtij konfigurimi do të lejojë testimin e funksionalitetit të Logstash. Për këtë, do ta fillojmë atë në mënyrë interaktive:

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

Hapi përfundimtar i instalimit: aktivizimi i Logstash si një shërbim Windows. Kjo mund të bëhet, për shembull, me ndihmën e paketës NSSM:

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

Qëndrueshmëria

Ruajtja e logëve gjatë transferimit nga serveri fillestar sigurohet nga mekanizmi i Queues së Përhershme.

Si funksionon

Diagrami i vendosjes sĂ« radhĂ«ve nĂ« procesin e pĂ«rpunimit tĂ« logĂ«ve: input → queue → filter + output.

Plugini input merr të dhënat nga burimi i logëve, i regjistron ato në radhë dhe i dërgon burimit një konfirmim të marrjes së të dhënave.

Mesazhet nga radhĂ«t pĂ«rpunohen nga Logstash, kalojnĂ« nĂ«pĂ«r filtrin dhe plugini output. Pas marrjes sĂ« konfirmimit nga output pĂ«r dĂ«rgimin e logut, Logstash e heq logun e pĂ«rpunuar nga radhĂ«. NĂ«se Logstash ndalon, tĂ« gjitha mesazhet e pa pĂ«rpunuara dhe mesazhet pĂ«r tĂ« cilat nuk Ă«shtĂ« marrĂ« konfirmimi i dĂ«rgimit mbeten nĂ« radhĂ«, dhe Logstash do tĂ« vazhdojĂ« t’i pĂ«rpunojĂ« ato gjatĂ« aktivizimit tĂ« ardhshĂ«m.

Konfigurimi

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

  • queue.type: (vlerat e mundshme janĂ« persisted dhe memory (default)).
  • path.queue: (rruga deri nĂ« dosjen me skedarĂ«t e radhĂ«ve, qĂ« nĂ« mĂ«nyrĂ« tĂ« paracaktuar ruhen nĂ« C:Logstashqueue).
  • queue.page_capacity: (masa maksimale e faqes sĂ« radhes, vlera e paracaktuar Ă«shtĂ« 64mb).
  • queue.drain: (true/false — aktivizon/ndalon ndalimin e pĂ«rpunimit tĂ« radhĂ«ve pĂ«rpara se Logstash tĂ« ndalohet. Nuk e rekomandoj ta aktivizoni, sepse kjo do tĂ« ndikojĂ« drejtpĂ«rdrejt nĂ« shpejtĂ«sinĂ« e ndaljes sĂ« serverit).
  • queue.max_events: (numri maksimal i ngjarjeve nĂ« radhĂ«, nĂ« mĂ«nyrĂ« tĂ« paracaktuar Ă«shtĂ« 0 (pa limit)).
  • queue.max_bytes: (masa maksimale e radhĂ«s nĂ« bajta, nĂ« mĂ«nyrĂ« tĂ« paracaktuar Ă«shtĂ« 1024mb (1gb)).

Nëse janë të konfiguruara queue.max_events dhe queue.max_bytes, atëherë mesazhet ndalojnë së pranuari në radhë kur arrijnë vlerën e çfarëdo prej këtyre cilësimeve. Më gjerësisht për Queues e Përhershme është folur këtu.

Një shembull i pjesës logstash.yml që është përgjegjës për konfigurimin e radhëve:

queue.type: persisted
queue.max_bytes: 10gb

Konfigurimi

Konfigurimi i Logstash zakonisht përbëhet nga tre pjesë, që përgjigjen për faza të ndryshme të përpunimit të logëve të ardhshëm: pranimi (seksioni input), parsing (seksioni filter) dhe dërgimi në Elastic (seksioni output). Më poshtë do të shqyrtojmë çdo një prej tyre më në hollësi.

Input

Ne marrim rrjedhën e ardhshme me logë të papërpunuara nga agjentët filebeat. Ky plugin është ai që specifikojmë në seksionin input:

input {
  beats {
    port => 5044
  }
}

Pas këtij konfigurimi, Logstash fillon të dëgjojë portin 5044, dhe kur merr logë, i përpunon ato sipas konfigurimit në seksionin filter. Nëse është e nevojshme, kanali i marrjes së logëve nga filebeat mund të enkriptohet me SSL. Më shumë rreth konfigurimeve të plugin-it beats janë përshkruar këtu.

Filter

Të gjitha logjet e interesit për përpunimin e teksteve që gjeneron Exchange, janë në formatin csv me fushat e përshkruara në vetë skedarin e logut. Për analizimin e regjistrimeve csv, Logstash na ofron tre plugina: dissect, csv dhe grok. I pari është më i shpejtë, por merret me analizimin e vetë logjeve më të thjeshta.
Për shembull, regjistrimi në vijim ai do ta ndajë në dy (për shkak të pranishmërisë së një vijëzimi brenda fushës), duke bërë që logu të analizohet gabim:


,"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 analizimin e logjeve, për shembull, IIS. Në këtë rast, seksioni filtër 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ë kushtit, ndaj mund të dërgojmë vetëm logjet që janë shënuar me tagun e filebeat IIS. Brenda plugin-it ne matim 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ë rastit, e cila do të përmbajë, për shembull, emrin e aplikacionit prej të cilit po mblidhim logjet.

Në rastin e logjeve të gjurmimit, është më mirë të përdorim plugin-in csv, ai di të përpunojë saktësisht 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 plugin-it ne matim vlerat e fushave me emrat e tyre, heqim fushën origjinale message (përveç fushave tenant-id dhe schema-version), e cila përmbante regjistrimin nga logu, dhe mund të shtojmë një fushë të rastit, e cila do të përmbajë, për shembull, emrin e aplikacionit prej të cilit po mblidhim logjet.

Në përfundim të fazës së filtrimit, ne do të marrim dokumente që janë afërsisht të gatshme për vizualizim në Kibana. Na mungon kjo:

  • Fushat numerike do tĂ« njihen si tekst, gjĂ« qĂ« nuk lejon kryerjen e operacioneve mbi to. Konkretisht, fushat time-taken e logut IIS, si dhe fushat recipient-count dhe total-bytes e logut Tracking.
  • Timestamp i standardit tĂ« dokumentit do tĂ« pĂ«rmbajĂ« kohĂ«n e pĂ«rpunimit tĂ« logut, dhe jo kohĂ«n e regjistrimit tĂ« tij nĂ« anĂ«n e serverit.
  • Fusha recipient-address do tĂ« duket si njĂ« strukturĂ« e vetme, gjĂ« qĂ« nuk lejon kryerjen e analizave me numĂ«rimin e marrĂ«sve tĂ« e-maileve.

Ka ardhur koha të shtojmë pak magji në procesin e përpunimit të logjeve.

Konvertimi i fushave numerike

Plugin-i dissect ka një opsion convert_datatype, e cila mund të përdoret për të konvertuar fushën tekst nga fusha në formatin numerik. Për shembull, kështu:

dissect {
  

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

}

ËshtĂ« e rĂ«ndĂ«sishme tĂ« mbahet mend se ky metod Ă«shtĂ« i pĂ«rshtatshĂ«m vetĂ«m nĂ«se fusha do tĂ« pĂ«rmbajĂ« patjetĂ«r njĂ« varg. Vlerat Null nga fusha nuk pĂ«rpunohen nga opsioni dhe pĂ«rfundon nĂ« njĂ« pĂ«rjashtim.

Për logjet e gjurmimit, është më mirë të mos përdorim metodën e ngjashme të konvertimit, pasi fushat recipient-count dhe total-bytes mund të jenë bosh. Për konvertimin e këtyre fushave, është më mirë të përdorim plugin-in mutate:

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

Saktësimi i recipient_address në marrës të veçantë

Këtë detyrë mund ta zgjidhim gjithashtu me anë të plugin-it mutate:

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

Ndryshimi i timestamp

Në rastin e logjeve të gjurmimit, detyra zgjidhet shumë lehtë me anë të plugin-it date, i cili do të ndihmojë të shkruhet në fushën timestamp data dhe koha në formatin e nevojshëm nga fusha date-time:

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

Në rastin e logjeve IIS do të duhet të bashkosh të dhënat e fushave date dhe time nëpërmjet plugin-it mutate, të specifikosh zonën e nevojshme dhe të vendosësh këtë timestamp në timestamp nëpërmjet plugin-it 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

Seksioni output përdoret për dërgimin e logjeve të përpunuara në marrësin e logjeve. Në rastin e dërgimit të drejtpërdrejtë në Elastic përdoret plugin-i elasticsearch, ku specifikohet adresa e serverit dhe modeli i 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

Konfigurimi përfundimtar do të duket si më poshtë:

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

Linke të dobishme:

Burimi: habr.com

Bleni hostim tĂ« besueshĂ«m pĂ«r faqe me mbrojtje nga DDoS, serverĂ« VPS VDS đŸ”„ Bleni hostim tĂ« besueshĂ«m pĂ«r faqe me mbrojtje nga DDoS, serverĂ« VPS VDS | ProHoster