🥇Vi kobler sammen ELK og Exchange. Del 2

Jeg fortsetter min fortelling om hvordan man kan få Exchange og ELK til å samarbeide (start here). Jeg minner om at denne kombinasjonen uten problemer kan håndtere et veldig stort volum loggfiler. Denne gangen skal vi snakke om hvordan man setter opp Exchange med komponentene Logstash og Kibana.

Logstash i ELK-stakken brukes til intelligent behandling av logger og forberedelse til plassering i Elastic i form av dokumenter, som gjør det praktisk å bygge ulike visualiseringer i Kibana.

Installasjon

Består av to trinn:

Installasjon og konfigurasjon av OpenJDK-pakken.
Installasjon og konfigurasjon av Logstash-pakken.

Installasjon og konfigurasjon av OpenJDK-pakken

OpenJDK-pakken må lastes ned og pakkes ut til en bestemt katalog. Deretter må stien til denne katalogen legges til variablene $env:Path og $env:JAVA_HOME i Windows-operativsystemet:

La oss sjekke Java-versjonen:

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)

Installasjon og konfigurasjon av Logstash-pakken

Last ned arkivfilen med Logstash-distribusjonen herfrå. Arkivet må pakkes ut i rotmappen. Det er ikke lurt å pakke ut i mappen C:Program Files , da Logstash ikke vil starte normalt. Deretter må du gjøre endringer i filen jvm.options , som regulerer tildeling av RAM til Java-prosessen. Jeg anbefaler å angi halvparten av serverens RAM. Hvis den har 16 GB RAM, er de standard nøklene:

-Xms1g
-Xmx1g

som må erstattes med:

-Xms8g
-Xmx8g

I tillegg er det fornuftig å kommentere ut linjen -XX:+UseConcMarkSweepGC. Mer om dette here. Neste steg er å opprette standardkonfigurasjonen i logstash.conf-filen:

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

Når du bruker denne konfigurasjonen, leser Logstash data fra konsollen, passerer dem gjennom et tomt filter og skriver dem tilbake til konsollen. Bruk av denne konfigurasjonen vil tillate å sjekke Logstashs funksjonalitet. For dette starter vi det i interaktiv modus:

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 har blitt startet vellykket på port 9600.

Det siste trinnet i installasjonen: kjør Logstash som en Windows-tjeneste. Dette kan gjøres, for eksempel, ved hjelp av pakken NSSM:

PS C:...bin> .nssm.exe install logstash
Tjenesten "logstash" er installert med suksess!

Feiltoleranse

Sikring av logger under overføring fra den opprinnelige serveren skjer ved hjelp av mekanismen Persistent Queues.

Slik fungerer det

Oversikten over plasseringen av køene under behandlingen av logger: input → queue → filter + output.

Input-pluginet mottar data fra loggkilder, skriver dem til køen og sender bekreftelse tilbake til kilden om mottak av data.

Meldinger fra køen behandles av Logstash, går gjennom filteret og output-pluginet. Når output bekrefter sending av loggen, fjerner Logstash den behandlede loggen fra køen. Hvis Logstash stopper, forblir alle ubehandlede meldinger og meldinger som ikke har fått bekreftelse om sending i køen, og Logstash fortsetter behandlingen av dem ved neste oppstart.

Konfigurasjon

Reguleres med nøkler i filen C:Logstashconfiglogstash.yml:

queue.type: (mulige verdier — persisted og memory (standard)).
path.queue: (stien til mappen med køfilene, som som standard lagres i C:Logstashqueue).
queue.page_capacity: (maksimal størrelse på køsiden, standardverdi — 64mb).
queue.drain: (true/false — slår på/av stopp av købehandling før Logstash slås av. Anbefales ikke å aktivere, da dette vil påvirke serverens avstengningshastighet direkte).
queue.max_events: (maksimalt antall hendelser i køen, som standard — 0 (uten begrensning)).
queue.max_bytes: (maksimal størrelse på køen i byte, som standard — 1024mb (1gb)).

Hvis de er konfigurert queue.max_events og queue.max_bytes, vil ikke meldinger bli akseptert i køen når verdien av noen av disse innstillingene er nådd. Mer om Persistent Queues er beskrevet here.

Eksempel på en del av logstash.yml som omhandler køkonfigurasjonen:

queue.type: persisted
queue.max_bytes: 10gb

Konfigurasjon

Logstash-konfigurasjonen består vanligvis av tre deler, som håndterer forskjellige faser av behandlingen av innkommende logger: mottak (input-seksjonen), parsing (filter-seksjonen) og sending til Elastic (output-seksjonen). Nedenfor ser vi nærmere på hver av dem.

Input

Den innkommende strømmen med rålogger mottas fra filebeat-agenter. Dette er pluginet vi angir i input-seksjonen:

input {
  beats {
    port => 5044
  }
}

Etter slik oppsett begynner Logstash å lytte på port 5044, og ved mottak av logger behandler den dem i henhold til innstillingene i filter-seksjonen. Om nødvendig kan logginnkommende fra filebit pakkes inn i SSL. Mer om konfigurasjonen av beats-pluginet er skrevet. here.

Filter

Alle interessante textlogger som genereres av Exchange, har CSV-format med feltene beskrevet i loggfilene selv. For å analysere CSV-poster tilbyr Logstash tre plugins: dissect, csv og grok. Den første er den raskeste, men klarer bare å analysere de enkleste loggene.
For eksempel vil følgende post bli delt opp i to (på grunn av en komma i et av feltene), noe som vil gjøre at loggen blir analysert feil:

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

Den kan brukes når man analyserer logger, for eksempel IIS. I dette tilfellet kan filterseksjonen se slik ut:

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-konfigurasjonen tillater bruk av betingede operatorer, så vi kan sende bare logger merket med filebeat-taggen IIS. Inne i pluginet mapper vi verdiene av feltene med navnene deres, fjerner det opprinnelige feltet message, som inneholdt posten fra loggen, og vi kan legge til et vilkårlig felt som for eksempel vil inneholde navnet på applikasjonen vi samler logger fra.

I tilfelle av sporinglogger er det bedre å bruke csv-pluginen, som kan håndtere komplekse felt riktig:

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

Inne i pluginet mapper vi verdiene av feltene med navnene deres, fjerner det opprinnelige feltet message (og også feltene tenant-id og schema-version), som inneholdt posten fra loggen, og vi kan legge til et vilkårlig felt som for eksempel vil inneholde navnet på applikasjonen vi samler logger fra.

Ut fra filtreringsstadiet vil vi få dokumenter som i første omgang er klare for visualisering i Kibana. Det som mangler er følgende:

Numeriske felt vil bli gjenkjent som tekst, noe som gjør det umulig å utføre operasjoner med dem. Spesielt feltene time-taken fra IIS-loggen, samt feltene recipient-count og total-bites fra Tracking-loggen.
Standard tidsstempel for dokumentet vil inneholde tiden for behandlingen av loggen, ikke tiden for skrivingen på serversiden.
Felt recipient-address vil se ut som én enkelt streng, noe som forhindrer analyse med telling av e-postmottakere.

Det er på tide å legge til litt magi i prosessen med loggbehandling.

Konvertering av numeriske felt

Plugin-modulen dissect har alternativet convert_datatype, som kan brukes til å konvertere et tekstfelt til digitalt format. For eksempel:

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

Det er viktig å huske at denne metoden bare fungerer hvis feltet faktisk vil inneholde en streng. Null-verdier fra feltene blir ikke håndtert av alternativet og vil kaste en unntak.

For tracking-loggene er det best å ikke bruke den lignende convert-metoden, siden feltene recipient-count og total-bites kan være tomme. For konvertering av disse feltene er det bedre å bruke plugin-modulen mutate:

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

Splitting recipient_address til individuelle mottakere

Denne oppgaven kan også løses ved hjelp av plugin-modulen mutate:

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

Endrer timestamp

I tilfelle med tracking-loggene løses oppgaven veldig enkelt av plugin-modulen date, som vil hjelpe med å legge inn i feltet timestamp dato og tid i ønsket format fra feltet date-time:

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

I tilfelle med IIS-loggene må vi kombinere data fra feltene date og time ved hjelp av plugin-modulen mutate, angi ønsket tidssone og plassere dette tidsstempelet i timestamp ved hjelp av plugin-modulen 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

Output-seksjonen brukes til å sende de behandlede loggene til loggmottakeren. Når du sender direkte til Elastic, brukes plugin-modulen elasticsearch, der serveradressen og navneskjemaet for indeksen for sending av det genererte dokumentet angis:

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

Den endelige konfigurasjonen

Den endelige konfigurasjonen vil se slik ut:

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

Nyttige lenker: