🥇ELK eta Exchange-ren lagunak gara. 2. zatia

Lagunak Truke eta ELK nola egin buruzko nire istorioa jarraitzen dut (hasiera Hemen). Gogorarazten dizut konbinazio hau erregistro kopuru handi bat prozesatzeko gai dela zalantzarik gabe. Oraingoan Exchange Logstash eta Kibana osagaiekin nola funtzionatzeari buruz hitz egingo dugu.

ELK pilako Logstash erregistroak modu adimentsuan prozesatzeko eta dokumentu moduan Elastic-en jartzeko prestatzeko erabiltzen da, eta horren arabera Kibana-n hainbat bistaratze eraikitzea komeni da.

Instalazio-

Bi fasez osatuta dago:

OpenJDK paketea instalatzea eta konfiguratzea.
Logstash paketea instalatzea eta konfiguratzea.

OpenJDK paketea instalatzea eta konfiguratzea

OpenJDK paketea deskargatu eta despakitu egin behar da direktorio zehatz batean. Ondoren, direktorio honetarako bidea Windows sistema eragilearen $env:Path eta $env:JAVA_HOME aldagaietan sartu behar da:

Ikus dezagun Java bertsioa:

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 paketea instalatzea eta konfiguratzea

Deskargatu artxibo-fitxategia Logstash banaketarekin beraz,. Artxiboa diskoaren errotik deskonpaktatu behar da. Despakitu karpetara C:Program Files Ez du merezi, Logstash-ek normaltasunez hasteari uko egingo dio. Ondoren, fitxategian sartu behar duzu jvm.options konponketak Java prozesurako RAM esleitzeaz arduratzen dira. Zerbitzariaren RAMaren erdia zehaztea gomendatzen dut. Plakoan 16 GB RAM baditu, tekla lehenetsiak hauek dira:

-Xms1g
-Xmx1g

ordezkatu behar da:

-Xms8g
-Xmx8g

Horrez gain, lerroa iruzkintzea komeni da -XX:+UseConcMarkSweepGC. Honi buruz gehiago Hemen. Hurrengo urratsa logstash.conf fitxategian konfigurazio lehenetsia sortzea da:

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

Konfigurazio honekin, Logstash-ek kontsolako datuak irakurtzen ditu, iragazki huts batetik igarotzen ditu eta kontsolara itzultzen ditu. Konfigurazio hau erabiliz Logstash-en funtzionaltasuna probatuko da. Horretarako, exekutatu dezagun modu interaktiboan:

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 arrakastaz abiarazi zen 9600 atakan.

Azken instalazioaren urratsa: abiarazi Logstash Windows zerbitzu gisa. Hori egin daiteke, adibidez, paketea erabiliz NSSM:

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

akatsen tolerantzia

Iturburuko zerbitzaritik transferitzen diren erregistroen segurtasuna Ilara iraunkorrak mekanismoak bermatzen du.

Funtzionamendua

Erregistroen prozesatzean ilaren diseinua hau da: sarrera → ilara → iragazkia + irteera.

Sarrerako pluginak erregistro-iturburu bateko datuak jasotzen ditu, ilara batean idazten ditu eta datuak iturburura jaso direla berrespena bidaltzen du.

Ilararen mezuak Logstash-ek prozesatzen ditu, iragazkitik eta irteerako pluginetik pasatzen dira. Erregistroa bidali dela dioen irteeratik berrespena jasotzen duenean, Logstash-ek prozesatutako erregistroa ilaratik kentzen du. Logstash gelditzen bada, prozesatu gabeko mezu eta berrespenik jaso ez duten mezu guztiak ilaran geratzen dira, eta Logstash-ek prozesatzen jarraituko du abiarazten den hurrengoan.

doikuntza

Fitxategiko teklen bidez doi daiteke C:Logstashconfiglogstash.yml:

queue.type: (balio posibleak - persisted и memory (default)).
path.queue: (ilara-fitxategiak dituen karpetarako bidea, C:Logstashqueue-n gordetzen direnak lehenespenez).
queue.page_capacity: (ilararen orriaren gehienezko tamaina, balio lehenetsia 64 MB da).
queue.drain: (egia/gezurra - Logstash itzali aurretik ilararen prozesamendua gelditzea gaitzen/desgaitzen du. Ez dut gaitzea gomendatzen, horrek zerbitzaria ixteko abiaduran zuzenean eragingo duelako).
queue.max_events: (ilaran dauden gertaera kopuru maximoa, lehenetsia 0 da (mugagabea)).
queue.max_bytes: (ilararen gehienezko tamaina bytetan, lehenetsia - 1024mb (1gb)).

Konfiguratuta badago queue.max_events и queue.max_bytes, orduan mezuak ez dira ilaran onartzen ezarpen hauetako baten baliora iristen denean. Lortu informazio gehiago ilara iraunkorrei buruz Hemen.

Ilara konfiguratzeaz arduratzen den logstash.yml atalaren adibide bat:

queue.type: persisted
queue.max_bytes: 10gb

doikuntza

Logstash konfigurazioak hiru zati ditu normalean, sarrerako erregistroak prozesatzeko fase ezberdinez arduratzen direnak: jasotzea (sarrera-atala), analizatzea (iragazki-atala) eta Elastic-era bidaltzea (irteera-atala). Jarraian, zehatzago aztertuko dugu horietako bakoitza.

Sarrerako

Filebeat agenteen erregistro gordinarekin jasotzen dugu sarrerako korrontea. Sarrera atalean adierazten dugun plugin hau da:

input {
  beats {
    port => 5044
  }
}

Konfigurazio honen ondoren, Logstash 5044 ataka entzuten hasten da, eta erregistroak jasotzean, iragazkien ataleko ezarpenen arabera prozesatzen ditu. Beharrezkoa izanez gero, filebit-eko erregistroak jasotzeko kanala SSL-n bil dezakezu. Irakurri gehiago beats plugin-en ezarpenei buruz Hemen.

iragazkiak

Exchange-k sortzen dituen prozesatzeko interesgarriak diren testu-erregistro guztiak csv formatuan daude erregistro-fitxategian bertan deskribatutako eremuekin. Csv erregistroak analizatzeko, Logstash-ek hiru plugin eskaintzen dizkigu: disekatu, csv eta grok. Lehenengoa da gehien azkarra, baina erregistro sinpleenak soilik analizatzeari aurre egiten dio.
Adibidez, hurrengo erregistroa bitan banatuko du (eremu barruan koma bat dagoelako), eta horregatik erregistroa gaizki analizatuko da:

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

Erregistroak analizatzerakoan erabil daiteke, adibidez, IIS. Kasu honetan, iragazkien atalak honelako itxura izan dezake:

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

Logstash konfigurazioak erabiltzeko aukera ematen dizu baldintzazko adierazpenak, beraz, filebeat etiketarekin etiketatu ziren erregistroak soilik bidal ditzakegu dissect pluginera IIS. Pluginaren barruan eremuen balioak beren izenekin lotzen ditugu, ezabatu jatorrizko eremua message, erregistroko sarrera bat zuena, eta, adibidez, erregistroak biltzen ditugun aplikazioaren izena izango duen eremu pertsonalizatu bat gehi dezakegu.

Jarraipen erregistroen kasuan, hobe da csv plugina erabiltzea; eremu konplexuak behar bezala prozesatu ditzake:

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

Pluginaren barruan eremuen balioak beren izenekin lotzen ditugu, ezabatu jatorrizko eremua message (eta eremuak ere bai tenant-id и schema-version), erregistroko sarrera bat zuen, eta eremu pertsonalizatu bat gehi dezakegu, zeinak, adibidez, erregistroak biltzen ditugun aplikazioaren izena izango du.

Iragazte fasetik irteeran, lehen hurbilketa batean dokumentuak jasoko ditugu, Kibanan bistaratzeko prest. Honako hauek faltako zaizkigu:

Zenbakizko eremuak testu gisa ezagutuko dira, eta horrek eragiketak eragozten ditu. Alegia, eremuak time-taken IIS erregistroa, baita eremuak ere recipient-count и total-bites Erregistroen jarraipena.
Dokumentuaren denbora-zigilu estandarrak erregistroa prozesatu den ordua izango du, ez zerbitzariaren aldean idatzi den ordua.
Field recipient-address eraikuntza gune baten itxura izango du, eta horrek ez du aztertzeko aukera ematen gutunen hartzaileak zenbatzea.

Erregistroak prozesatzeko prozesuari magia apur bat gehitzeko garaia da.

Zenbakizko eremuak bihurtzea

Disect pluginak aukera bat du convert_datatype, testu-eremu bat formatu digitalera bihurtzeko erabil daitekeena. Adibidez, honela:

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

Gogoratu behar da metodo hau soilik egokia dela eremuak behin betiko kate bat edukiko badu. Aukerak ez ditu eremuetako balio nuluak prozesatzen eta salbuespen bat botatzen du.

Erregistroak jarraitzeko, hobe da antzeko bihurketa metodorik ez erabiltzea, eremuak baitira recipient-count и total-bites hutsik egon daiteke. Eremu hauek bihurtzeko hobe da plugin bat erabiltzea mutatu:

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

destinatario_helbidea hartzaile indibidualetan banatzen

Arazo hau mutate plugina erabiliz ere konpondu daiteke:

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

Denbora-zigilua aldatzea

Jarraipen erregistroen kasuan, arazoa oso erraz konpontzen da pluginak data, eremuan idazten lagunduko dizuna timestamp data eta ordua eremutik behar den formatuan date-time:

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

IIS erregistroen kasuan, eremuko datuak konbinatu beharko ditugu date и time mutate plugina erabiliz, erregistratu behar dugun ordu-eremua eta jarri ordu-zigilu hau timestamp data plugina erabiliz:

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

Irteera-atala prozesatutako erregistroak erregistro-hartzaileari bidaltzeko erabiltzen da. Elastic-era zuzenean bidaliz gero, plugin bat erabiltzen da elastiko bilaketa, sortutako dokumentua bidaltzeko zerbitzariaren helbidea eta indize-izena txantiloia zehazten dituena:

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

Azken konfigurazioa

Azken konfigurazioa honela izango da:

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

Esteka erabilgarriak:

Iturria: www.habr.com

ELK eta Exchange-ren lagunak gara. 2. zatia

Instalazio-

Logstash paketea instalatzea eta konfiguratzea

akatsen tolerantzia

Funtzionamendua

doikuntza

doikuntza

Sarrerako

iragazkiak

Zenbakizko eremuak bihurtzea

destinatario_helbidea hartzaile indibidualetan banatzen

Denbora-zigilua aldatzea

Output

Azken konfigurazioa

Gehitu iruzkin berria Idatzi