🥇Kopplar ihop ELK och Exchange. Del 2

Jag fortsätter min berättelse om hur man förenar Exchange och ELK (början ). Ett typiskt exempel på användning:). Jag påminner om att denna kombination kan utan tvekan hantera en mycket stor mängd loggar. Denna gång kommer vi att diskutera hur man ställer in Exchange med komponenterna Logstash och Kibana.

Logstash i ELK-stacken används för intelligent logghantering och förberedelse av loggar för lagring i Elastic i form av dokument, vilket gör det lättare att skapa olika visualiseringar i Kibana.

Installation

Består av två steg:

Installation och konfiguration av paketet OpenJDK.
Installation och konfiguration av paketet Logstash.

Installation och konfiguration av paketet OpenJDK

Paketet OpenJDK måste laddas ner och packas upp i en viss katalog. Sedan behöver sökvägen till denna katalog läggas till i $env:Path och $env:JAVA_HOME variablerna i Windows-operativsystemet:

Låt oss kontrollera 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 och konfiguration av paketet Logstash

Ladda ner arkivfilen med Logstash-distributionen härifrån. Arkivet bör packas upp i roten av disken. Det är inte bra att packa upp i en mapp C:Program Files , Logstash vägrar att starta ordentligt. Sedan måste du justera filen jvm.options , som ansvarar för minnesallokering för Java-processen. Jag rekommenderar att ange hälften av serverns RAM. Om den har 16 GB RAM, bör standardnycklarna vara:

-Xms1g
-Xmx1g

måste bytas ut mot:

-Xms8g
-Xmx8g

Dessutom är det rimligt att kommentera raden -XX:+UseConcMarkSweepGC. Mer om detta ). Ett typiskt exempel på användning:. Nästa steg är att skapa en standardkonfiguration i filen logstash.conf:

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

Med denna konfiguration läser Logstash data från konsolen, passerar genom en tom filter och skriver ut tillbaka till konsolen. Användning av denna konfiguration gör det möjligt att kontrollera Logstash-funktionaliteten. För att göra detta, kör vi den i interaktivt läge:

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 framgångsrikt startat på port 9600.

Det sista steget i installationen: starta Logstash som en Windows-tjänst. Detta kan göras till exempel med hjälp av paketet NSSM:

PS C:...bin> .nssm.exe install logstash
Tjänsten "logstash" harinstallerats framgångsrikt!

Fel-tolerans

Skydd av loggar vid överföring från den ursprungliga servern säkerställs genom mekanismen för Persistent Queues.

Hur det fungerar

Schemat för placering av köer i processen för logghantering: input → queue → filter + output.

Input-plugin tar emot data från loggkällan, skriver dem till kön och skickar en bekräftelse på mottagandet av data till källan.

Meddelanden från kön bearbetas av Logstash, går igenom en filter och output-plugin. När bekräftelsen på att loggen har skickats tas emot från output tar Logstash bort den bearbetade loggen från kön. Om Logstash stoppas så förblir alla obehandlade meddelanden och meddelanden utan bekräftelse om skickande kvar i kön, och Logstash kommer att fortsätta bearbeta dem vid nästa start.

Inställning

Regleras med nycklar i filen C:Logstashconfiglogstash.yml:

queue.type: (möjliga värden — persisted och memory (standard)).
path.queue: (sökväg till mappen med köfiler, som som standard lagras i C:Logstashqueue).
queue.page_capacity: (maximal storlek på kö-sidan, standardvärdet är 64mb).
queue.drain: (true/false — aktiverar/deaktiverar stopp av köbehandling innan Logstash stängs. Jag rekommenderar inte att aktivera detta, eftersom det direkt påverkar hastigheten på serveravstängning).
queue.max_events: (maximalt antal händelser i kön, standardvärdet är 0 (obegränsat)).
queue.max_bytes: (maximal storlek på kön i byte, standardvärdet är 1024mb (1gb)).

Om de är inställda queue.max_events och queue.max_bytes, kommer meddelanden att sluta tas emot i kön när något av dessa värden nås. Mer information om Persistent Queues finns i ). Ett typiskt exempel på användning:.

Exempel på en del av logstash.yml som ansvarar för köinställningen:

queue.type: persisted
queue.max_bytes: 10gb

Inställning

Logstash-konfigurationen består vanligtvis av tre delar som ansvarar för olika faser av bearbetningen av inkommande loggar: mottagande (input-sektion), parsning (filter-sektion) och utskick till Elastic (output-sektion). Nedan kommer vi att titta närmare på varje del.

Input

Vi tar emot den inkommande strömmen med råa loggar från Filebeat-agenter. Det är just denna plugin vi anger i input-sektionen:

input {
  beats {
    port => 5044
  }
}

Med denna inställning börjar Logstash att lyssna på port 5044, och vid mottagande av loggar bearbetas de enligt inställningarna i filter-sektionen. Vid behov kan loggkanalen från Filebeat säkras med SSL. Mer information om inställningarna för Beats-plugin finns beskrivet. ). Ett typiskt exempel på användning:.

Filter

Alla intressanta textloggar som genereras av Exchange har csv-format med de fält som beskrivs i själva loggfilerna. För att parsa csv-poster erbjuder Logstash oss tre plugins: dissect, csv och grok. Den första är den mest snabba, men klarar bara av att parsa de enklaste loggarna.
Till exempel kommer följande post att delas upp i två (på grund av närvaron av ett kommatecken inuti fältet), vilket kommer att göra att loggen parsas fel:

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

Den kan användas vid parning av loggar, till exempel från IIS. I detta fall kan filtersektionen se ut på följande sätt:

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

Konfigurationen av Logstash gör att vi kan använda villkorssatser, därför kan vi i dissect-pluginet rikta oss enbart till loggar som har märkts med filebeat-taggen IIS. Inuti pluginet mappar vi fältvärden med deras namn, tar bort det ursprungliga fältet meddelande, som innehöll posten från loggen, och vi kan lägga till ett valfritt fält som till exempel kan innehålla namnet på applikationen från vilken vi samlar loggar.

När det gäller spårningsloggar är det bättre att använda csv-pluginet, den kan korrekt hantera komplexa fält:

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

Inuti pluginet mappar vi fältvärden med deras namn, tar bort det ursprungliga fältet meddelande (och även fältet tenant-id och schema-version), som innehöll posten från loggen, och vi kan lägga till ett valfritt fält som kan innehålla namnet på applikationen från vilken vi samlar loggar.

Vid utgången av filtersteget kommer vi att få dokument som i grova drag är klara för visualisering i Kibana. Det vi saknar är följande:

Numeriska fält kommer att tolkas som text, vilket hindrar operationer på dem. Nämligen fälten time-taken från IIS-loggen, samt fälten recipient-count och total-bites från Tracking-loggen.
Den standard tidsstämpeln för dokumentet kommer att innehålla tiden för behandling av loggen, inte tiden för registreringen på serversidan.
Fältet recipient-address kommer att se ut som en enda struktur, vilket gör det omöjligt att genomföra analys med räkning av e-postmottagare.

Det är dags att lägga till lite magi i processen för logghantering.

Konvertering av numeriska fält

Pluginet dissect har en option convert_datatype, som kan användas för att konvertera textfält till digitalt format. Till exempel så här:

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

Det är värt att komma ihåg att denna metod endast fungerar om fältet säkerligen innehåller en sträng. Null-värden i fälten hanteras inte av denna option och ger ett undantag.

För tracking-loggar är det bäst att inte använda den motsvarande metoden convert, eftersom fälten recipient-count och total-bites kan vara tomma. För att konvertera dessa fält är det bättre att använda pluginet mutate:

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

Uppdelning av recipient_address i individuella mottagare

Denna uppgift kan också lösas med hjälp av pluginet mutate:

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

Ändra timestamp

I fallet med tracking-loggar är uppgiften mycket enkel att lösa med pluginet date, som hjälper till att skriva in datum och tid i rätt format från fältet timestamp date-time date { match => [ "date-time", "ISO8601" ] timezone => "Europe/Moscow" remove_field => [ "date-time" ] }:

I fallet med IIS-loggar måste vi slå samman data i fälten

med hjälp av pluginet mutate, ange vår önskade tidszon och placera denna tidsstämpel i date och time med hjälp av pluginet 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 används för att skicka de bearbetade loggarna till loggmottagaren. Vid direkt sändning till Elastic används pluginet

Output

elasticsearch , där serveradressen och mönstret för indexnamnet anges för att skicka det formaterade dokumentet:output { elasticsearch { hosts => ["127.0.0.1:9200", "127.0.0.2:9200"] manage_template => false index => "Exchange-%{+YYYY.MM.dd}" } }

Slutlig konfiguration

Den slutliga konfigurationen kommer att se ut på följande sätt:

Slutlig konfiguration

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

Nyttiga länkar: