🥇Ons is vriende met ELK en Exchange. Deel 2

Ek gaan voort met my storie oor hoe om vriende te maak Exchange en ELK (begin hier). Laat ek jou daaraan herinner dat hierdie kombinasie in staat is om 'n baie groot aantal stompe sonder huiwering te verwerk. Hierdie keer sal ons praat oor hoe om Exchange met Logstash- en Kibana-komponente te laat werk.

Logstash in die ELK-stapel word gebruik om logs intelligent te verwerk en voor te berei vir plasing in Elastic in die vorm van dokumente, op grond waarvan dit gerieflik is om verskeie visualiserings in Kibana te bou.

installasie

Bestaan uit twee fases:

Installeer en konfigureer die OpenJDK-pakket.
Installeer en konfigureer die Logstash-pakket.

Installeer en konfigureer die OpenJDK-pakket

Die OpenJDK-pakket moet afgelaai en uitgepak word in 'n spesifieke gids. Dan moet die pad na hierdie gids in die $env:Path en $env:JAVA_HOME veranderlikes van die Windows-bedryfstelsel ingevoer word:

Kom ons kyk na die Java-weergawe:

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)

Installeer en konfigureer die Logstash-pakket

Laai die argieflêer af met die Logstash-verspreiding vandaar. Die argief moet tot by die wortel van die skyf uitgepak word. Pak uit na vouer C:Program Files Dit is nie die moeite werd nie, Logstash sal weier om normaal te begin. Dan moet jy in die lêer ingaan jvm.options regstellings wat verantwoordelik is vir die toekenning van RAM vir die Java-proses. Ek beveel aan om die helfte van die bediener se RAM te spesifiseer. As dit 16 GB RAM aan boord het, is die versteksleutels:

-Xms1g
-Xmx1g

moet vervang word met:

-Xms8g
-Xmx8g

Daarbenewens is dit raadsaam om kommentaar te lewer uit die reël -XX:+UseConcMarkSweepGC. Meer hieroor hier. Die volgende stap is om 'n verstekkonfigurasie in die logstash.conf-lêer te skep:

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

Met hierdie konfigurasie lees Logstash data vanaf die konsole, stuur dit deur 'n leë filter en voer dit terug na die konsole. Die gebruik van hierdie konfigurasie sal die funksionaliteit van Logstash toets. Om dit te doen, laat ons dit in interaktiewe modus laat loop:

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 is suksesvol op poort 9600 bekendgestel.

Die laaste installasiestap: begin Logstash as 'n Windows-diens. Dit kan gedoen word, byvoorbeeld deur die pakket te gebruik NSSM:

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

fout verdraagsaamheid

Die veiligheid van logs wanneer dit vanaf die bronbediener oorgedra word, word verseker deur die aanhoudende toue-meganisme.

Hoe werk dit?

Die uitleg van toue tydens logverwerking is: invoer → tou → filter + afvoer.

Die invoerinprop ontvang data vanaf 'n logbron, skryf dit na 'n tou en stuur bevestiging dat die data ontvang is na die bron.

Boodskappe vanaf die tou word deur Logstash verwerk, deur die filter en die uitvoerinprop gestuur. Wanneer bevestiging van afvoer ontvang word dat die logboek gestuur is, verwyder Logstash die verwerkte logboek uit die tou. As Logstash stop, bly alle onverwerkte boodskappe en boodskappe waarvoor geen bevestiging ontvang is nie in die tou, en Logstash sal voortgaan om dit te verwerk die volgende keer dat dit begin.

aanpassing

Verstelbaar deur sleutels in die lêer C:Logstashconfiglogstash.yml:

queue.type: (moontlike waardes - persisted и memory (default)).
path.queue: (pad na die gids met toulêers, wat by verstek in C:Logstashqueue gestoor word).
queue.page_capacity: (maksimum toubladsygrootte, verstekwaarde is 64mb).
queue.drain: (waar/onwaar - aktiveer/deaktiveer die stop van touverwerking voordat Logstash afgesluit word. Ek beveel nie aan om dit te aktiveer nie, want dit sal die spoed van bedienerafskakeling direk beïnvloed).
queue.max_events: (maksimum aantal gebeurtenisse in die tou, verstek is 0 (onbeperk)).
queue.max_bytes: (maksimum tougrootte in grepe, verstek - 1024mb (1gb)).

Indien gekonfigureer queue.max_events и queue.max_bytes, dan hou boodskappe op om in die tou aanvaar te word wanneer die waarde van enige van hierdie instellings bereik word. Kom meer te wete oor aanhoudende toue hier.

'n Voorbeeld van die deel van logstash.yml wat verantwoordelik is vir die opstel van die tou:

queue.type: persisted
queue.max_bytes: 10gb

aanpassing

Die Logstash-konfigurasie bestaan gewoonlik uit drie dele, verantwoordelik vir verskillende fases van die verwerking van inkomende logs: ontvang (invoerafdeling), ontleed (filterafdeling) en stuur na Elastic (afvoerafdeling). Hieronder sal ons elkeen van nader kyk.

insette

Ons ontvang die inkomende stroom met rou logs van filebeat-agente. Dit is hierdie inprop wat ons in die invoerafdeling aandui:

input {
  beats {
    port => 5044
  }
}

Na hierdie opstelling begin Logstash na poort 5044 luister, en wanneer logs ontvang word, verwerk dit volgens die instellings van die filterafdeling. Indien nodig, kan jy die kanaal vir die ontvangs van logs van filebit in SSL toedraai. Lees meer oor beats-inpropinstellings hier.

Filters

Alle tekslogboeke wat interessant is vir verwerking wat Exchange genereer, is in csv-formaat met die velde wat in die loglêer self beskryf word. Vir die ontleding van csv-rekords bied Logstash ons drie inproppe: dissekteer, csv en grok. Die eerste een is die meeste vinnig, maar hanteer slegs die eenvoudigste logboeke.
Dit sal byvoorbeeld die volgende rekord in twee verdeel (as gevolg van die teenwoordigheid van 'n komma binne die veld), en daarom sal die logboek verkeerd ontleed word:

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

Dit kan gebruik word wanneer logboeke ontleed word, byvoorbeeld IIS. In hierdie geval kan die filterafdeling soos volg lyk:

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-konfigurasie laat jou toe om te gebruik voorwaardelike verklarings, so ons kan slegs logs wat met die filebeat-merker gemerk is, na die dissect-inprop stuur IIS. Binne die inprop pas ons die veldwaardes met hul name, vee die oorspronklike veld uit message, wat 'n inskrywing uit die log bevat het, en ons kan 'n pasgemaakte veld byvoeg wat byvoorbeeld die naam van die toepassing sal bevat waaruit ons logs versamel.

In die geval van opsporingslogboeke, is dit beter om die csv-inprop te gebruik; dit kan komplekse velde korrek verwerk:

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

Binne die inprop pas ons die veldwaardes met hul name, vee die oorspronklike veld uit message (en ook velde tenant-id и schema-version), wat 'n inskrywing uit die log bevat, en ons kan 'n pasgemaakte veld byvoeg, wat byvoorbeeld die naam van die toepassing sal bevat waaruit ons logs versamel.

By die uitgang van die filterstadium sal ons dokumente in 'n eerste benadering ontvang, gereed vir visualisering in Kibana. Ons gaan die volgende mis:

Numeriese velde sal as teks herken word, wat bewerkings daarop verhoed. Naamlik die velde time-taken IIS-logboek, sowel as velde recipient-count и total-bites Log dop.
Die standaard dokument tydstempel sal die tyd bevat wat die logboek verwerk is, nie die tyd wat dit aan die bedienerkant geskryf is nie.
Veld recipient-address sal lyk soos een konstruksieterrein, wat nie toelaat vir ontleding om die ontvangers van briewe te tel nie.

Dit is tyd om 'n bietjie magie by die logverwerkingsproses te voeg.

Omskakeling van numeriese velde

Die dissekteer-inprop het 'n opsie convert_datatype, wat gebruik kan word om 'n teksveld na 'n digitale formaat om te skakel. Byvoorbeeld, soos volg:

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

Dit is die moeite werd om te onthou dat hierdie metode slegs geskik is as die veld beslis 'n string sal bevat. Die opsie verwerk nie nulwaardes uit velde nie en skep 'n uitsondering.

Vir die dop van logs is dit beter om nie 'n soortgelyke omskakelingsmetode te gebruik nie, aangesien die velde recipient-count и total-bites mag leeg wees. Om hierdie velde om te skakel, is dit beter om 'n inprop te gebruik muteren:

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

Verdeel ontvanger_adres in individuele ontvangers

Hierdie probleem kan ook opgelos word met behulp van die mutate plugin:

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

Verander die tydstempel

In die geval van doplogboeke, word die probleem baie maklik deur die inprop opgelos datum, wat jou sal help om in die veld te skryf timestamp datum en tyd in die vereiste formaat uit die veld date-time:

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

In die geval van IIS-logboeke, sal ons velddata moet kombineer date и time gebruik die mutate-inprop, registreer die tydsone wat ons benodig en plaas hierdie tydstempel in timestamp gebruik die datum-inprop:

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

Uitgawe

Die uitvoerafdeling word gebruik om verwerkte logs na die log-ontvanger te stuur. In die geval van direk na Elastic gestuur word, word 'n inprop gebruik elasticsearch, wat die bedieneradres en indeksnaamsjabloon spesifiseer vir die stuur van die gegenereerde dokument:

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

Finale konfigurasie

Die finale konfigurasie sal soos volg lyk:

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

Nuttige skakels:

Bron: will.com

Ons is vriende met ELK en Exchange. Deel 2

installasie

Installeer en konfigureer die Logstash-pakket

fout verdraagsaamheid

Hoe werk dit?

aanpassing

aanpassing

insette

Filters

Omskakeling van numeriese velde

Verdeel ontvanger_adres in individuele ontvangers

Verander die tydstempel

Uitgawe

Finale konfigurasie

Voeg 'n opmerking Kanselleer antwoord