🥇Vi integrerer ELK og Exchange. Del 2

Jeg fortsætter min fortælling om, hvordan man forbinder Exchange og ELK (starten her). Jeg minder om, at denne kombination uden problemer kan håndtere et meget stort antal logs. Denne gang vil vi tale om, hvordan man får Exchange til at fungere sammen med komponenterne Logstash og Kibana.

Logstash i ELK-stacken bruges til intelligent behandling af logs og forberedelse til placering i Elastic i form af dokumenter, som det er nemt at lave forskellige visualiseringer af i Kibana.

Installation

Består af to faser:

Installation og konfiguration af OpenJDK-pakken.
Installation og konfiguration af Logstash-pakken.

Installation og konfiguration af OpenJDK-pakken

OpenJDK-pakken skal downloades og udpakkes i en bestemt mappe. Derefter skal stien til denne mappe tilføjes til variablerne $env:Path og $env:JAVA_HOME i Windows-operativsystemet:

Lad os tjekke Java-versionen:

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)

Installation og konfiguration af Logstash-pakken

Download arkivfilen med Logstash-distributionen from here. Arkivet skal udpakkes i rodmappen på drevet. Det anbefales ikke at udpakke det i en mappe, C:Program Files da Logstash ikke vil kunne starte korrekt. Derefter skal filen jvm.options ændres for at tildele hukommelse til Java-processen. Jeg anbefaler at angive halvdelen af serverens RAM. Hvis den har 16 GB RAM, vil standardnøglerne være:

-Xms1g
-Xmx1g

skal ændres til:

-Xms8g
-Xmx8g

Derudover er det fornuftigt at kommentere linjen -XX:+UseConcMarkSweepGC. Læs mere om dette her. Næste skridt er at oprette standardkonfigurationen i filen logstash.conf:

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

Ved brug af denne konfiguration læser Logstash data fra konsollen, sender dem gennem et tomt filter og sender dem tilbage til konsollen. Denne konfiguration vil give mulighed for at teste Logstash's funktionalitet. Lad os starte det i interaktiv tilstand:

PS C:...bin> .logstash.bat -f .logstash.conf
...
[2019-12-19T11:15:27,769][INFO ][logstash.javapipeline    ][main] Pipeline started {"pipeline.id"=>"main"}
Stdin-pluginet venter nu på 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 blev startet korrekt på port 9600.

Det sidste trin i installationen: start Logstash som en Windows-tjeneste. Dette kan gøres, for eksempel ved hjælp af pakken NSSM:

PS C:...bin> .nssm.exe install logstash
Tjenesten "logstash" er installeret med succes!

Fejltolerance

Dataens integritet under overførsel fra den oprindelige server sikres af mekanismen Persistent Queues.

Hvordan det fungerer

Diagrammet for placering af køer under behandlingen af logs: input → queue → filter + output.

Input-plugin'et modtager data fra logkilden, skriver dem til køen og sender en bekræftelse til kilden om, at dataene er modtaget.

Beskeder fra køen behandles af Logstash, går gennem filteret og output-plugin'et. Når output modtager bekræftelse på logens sendelse, sletter Logstash den behandlede log fra køen. Hvis Logstash stopper, forbliver alle ubehandlede beskeder og beskeder, der ikke har modtaget bekræftelse på afsendelse, i køen, og Logstash vil fortsætte med at behandle dem ved næste opstart.

Indstillinger

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

queue.type: (mulige værdier — persisted og memory (standard)).
path.queue: (stien til mappen med køfilerne, som som standard gemmes i C:Logstashqueue).
queue.page_capacity: (den maksimale størrelse af en køside, standardværdien er — 64mb).
queue.drain: (true/false — aktiverer/deaktiverer stoppet af købehandlingen før Logstash lukkes. Jeg anbefaler ikke at aktivere dette, da det direkte vil påvirke serverens nedlukningshastighed).
queue.max_events: (det maksimale antal hændelser i køen, standardværdien er — 0 (ubegrænset)).
queue.max_bytes: (den maksimale størrelse af køen i bytes, standardværdien er — 1024mb (1gb)).

Hvis der er konfigureret queue.max_events og queue.max_bytes, stopper beskeder med at blive accepteret i køen ved at nå værdien af nogen af disse indstillinger. Der er flere oplysninger om Persistent Queues her.

Eksempel på en del af logstash.yml, der vedrører køkonfigurationen:

queue.type: persisted
queue.max_bytes: 10gb

Indstillinger

Konfigurationen af Logstash består normalt af tre dele, der svarer til de forskellige faser i behandlingen af indgående logs: modtagelse (input-sektionen), parsing (filter-sektionen) og afsendelse til Elastic (output-sektionen). Nedenfor vil vi se nærmere på hver af dem.

Input

Den indgående strøm af rå logs modtages fra filbeat-agenter. Det er netop denne plugin, vi angiver i input-sektionen:

input {
  beats {
    port => 5044
  }
}

Efter sådan en konfiguration begynder Logstash at lytte til port 5044, og ved modtagelse af logs behandler den dem i overensstemmelse med indstillingerne i filter-sektionen. Om nødvendigt kan logkanalen fra filebeat indkapsles i SSL. Der er flere oplysninger om indstillingerne for beats-plugin'et. her.

Filter

Alle interessante tekstlogfiler, der genereres af Exchange, er i csv-format med felter, der er beskrevet i selve logfilen. Til at parse csv-poster tilbyder Logstash os tre plugins: dissect, csv og grok. Den første er den hurtigste, men den kan kun håndtere de enkleste logs.
For eksempel vil den følgende post blive opdelt i to (på grund af tilstedeværelsen af et komma inden for feltet), hvilket vil føre til, at loggen bliver parsed forkert:

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

Den kan bruges til at parse logs, for eksempel fra IIS. I dette tilfælde kan filtersektionen se således ud:

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-konfigurationen giver mulighed for at bruge betingede operatorer, så vi kan sende logs til dissect-pluginet, som er blevet mærket med filebeat-tagget IIS. Inden i pluginet matcher vi værdierne af felterne med deres navne, fjerner det oprindelige felt, messagesom indeholdt posten fra loggen, og vi kan tilføje et vilkårligt felt, som for eksempel kan indeholde navnet på den applikation, hvorfra vi samler logs.

I tilfælde af tracking-logs er det bedre at bruge csv-pluginet, da det kan håndtere komplekse felter korrekt:

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

Inden i pluginet matcher vi værdierne af felterne med deres navne, fjerner det oprindelige felt, message (samt felterne tenant-id og schema-version), som indeholdt posten fra loggen, og vi kan tilføje et vilkårligt felt, som for eksempel kan indeholde navnet på den applikation, hvorfra vi samler logs.

I outputtet fra filtreringsstadiet vil vi få dokumenter, der først og fremmest er klar til visualisering i Kibana. Vi vil mangle følgende:

Numeriske felter vil blive genkendt som tekst, hvilket forhindrer operationer på dem. Det vil sige felterne time-taken fra IIS-loggen samt felterne recipient-count og total-bites fra Tracking-loggen.
Den standard tidsstempel for dokumentet vil indeholde behandlingstiden for loggen og ikke tidspunktet for, hvornår den blev skrevet på serversiden.
Felt recipient-address vil fremstå som én enkelt enhed, hvilket forhindrer analyse med optælling af modtagere af e-mails.

Det er tid til at tilføje lidt magi til logbehandlingsprocessen.

Konvertering af numeriske felter

Plugin'en dissect har en mulighed convert_datatype, som kan bruges til at konvertere et tekstfelt til digitalt format. For eksempel:

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

Det er værd at huske, at denne metode kun er anvendelig, hvis feltet helt klart vil indeholde en streng. Null-værdier i felter behandles ikke af denne option og vil give en fejl.

For tracking-logs er det bedre ikke at bruge den samme konverteringsmetode, da felterne recipient-count og total-bites kan være tomme. For at konvertere disse felter er det bedst at bruge plugin'en mutate:

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

Opdeling af recipient_address i separate modtagere

Denne opgave kan også løses med plugin'en mutate:

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

Ændring af timestamp

I tilfælde af tracking-logs løses opgaven meget enkelt med plugin'en date, som hjælper med at skrive dato og tid i det ønskede format fra feltet timestamp date-time date { match => [ "date-time", "ISO8601" ] timezone => "Europe/Moscow" remove_field => [ "date-time" ] }:

I tilfælde af IIS-logs skal vi kombinere data fra felterne

ved hjælp af plugin'en mutate, angive den ønskede tidszone og placere dette tidsstempel i date og time ved hjælp af plugin'en date: timestamp 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-sektionen bruges til at sende de behandlede logs til logmodtageren. I tilfælde af direkte sending til Elastic anvendes plugin'en

Output

, hvor serveradressen og skabelonnavnet for indekset til den genererede dokumentation angives: elasticsearchoutput { elasticsearch { hosts => ["127.0.0.1:9200", "127.0.0.2:9200"] manage_template => false index => "Exchange-%{+YYYY.MM.dd}" } }

Den endelige konfiguration vil se således ud:

Final configuration

Итоговая конфигурация будет выглядеть следующим образом:

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 links: