🥇Amicizia tra ELK e Exchange. Parte 2

Continuo il mio racconto su come integrare Exchange ed ELK (inizio qui). Ricordo che questa combinazione è in grado di gestire senza esitazioni un numero molto elevato di log. Questa volta parleremo di come configurare Exchange con i componenti Logstash e Kibana.

Logstash nel stack ELK viene utilizzato per l'elaborazione intelligente dei log e la loro preparazione per la memorizzazione in Elastic sotto forma di documenti, sulla base dei quali è comodo costruire diverse visualizzazioni in Kibana.

Installazione

Si compone di due fasi:

Installazione e configurazione del pacchetto OpenJDK.
Installazione e configurazione del pacchetto Logstash.

Installazione e configurazione del pacchetto OpenJDK

È necessario scaricare il pacchetto OpenJDK ed estrarlo in una directory specifica. Quindi, il percorso di questa directory deve essere aggiunto alle variabili $env:Path e $env:JAVA_HOME del sistema operativo Windows:

Verifichiamo la versione di Java:

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)

Installazione e configurazione del pacchetto Logstash

Scaricate il file-archivio con la distribuzione di Logstash da qui. L'archivio deve essere estratto nella radice del disco. Non è consigliabile estrarre in una cartella C:Program Files , perché Logstash potrebbe non avviarsi normalmente. Successivamente, è necessario modificare il file jvm.options per quanto riguarda l'assegnazione della memoria RAM per il processo Java. Si consiglia di specificare la metà della memoria RAM del server. Se ha 16 GB di RAM, le chiavi predefinite sono:

-Xms1g
-Xmx1g

dovrebbero essere sostituite con:

-Xms8g
-Xmx8g

Inoltre, è utile commentare la riga -XX:+UseConcMarkSweepGC. Per ulteriori informazioni su questo qui. Il passaggio successivo è la creazione di una configurazione predefinita nel file logstash.conf:

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

Utilizzando questa configurazione, Logstash legge i dati dalla console, li passa attraverso un filtro vuoto e li restituisce alla console. L'uso di questa configurazione consentirà di verificare il funzionamento di Logstash. Per farlo, avviamolo in modalità interattiva:

PS C:...bin .logstash.bat -f .logstash.conf
...
[2019-12-19T11:15:27,769][INFO ][logstash.javapipeline    ][main] Pipeline avviato {"pipeline.id"=>"main"}
Il plugin stdin è ora in attesa di input:
[2019-12-19T11:15:27,847][INFO ][logstash.agent           ] Pipeline in esecuzione {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2019-12-19T11:15:28,113][INFO ][logstash.agent           ] Endpoint API Logstash avviato con successo {:port=>9600}

Logstash avviato con successo sulla porta 9600.

L'ultimo passaggio dell'installazione: avviare Logstash come servizio Windows. Questo può essere fatto, ad esempio, attraverso il pacchetto NSSM:

PS C:...bin> .nssm.exe install logstash
Servizio "logstash" installato con successo!

Resilienza

La conservazione dei log durante la trasmissione dal server sorgente è garantita dal meccanismo delle Coda Persistenti.

Come funziona

Schema di emittente delle code durante l'elaborazione dei log: input → queue → filter + output.

Il plugin input riceve i dati dalla sorgente dei log, li scrive nella coda e invia una conferma alla sorgente per la ricezione dei dati.

I messaggi dalla coda vengono elaborati da Logstash, passano attraverso il filtro e il plugin di output. Al ricevimento di una conferma di invio dal plugin di output, Logstash rimuove il log elaborato dalla coda. Se Logstash viene fermato, tutti i messaggi non elaborati e quelli per cui non è stata ricevuta la conferma di invio rimangono nella coda e Logstash continuerà la loro elaborazione al prossimo avvio.

Configurazione

Regolato dalle chiavi nel file C:Logstashconfiglogstash.yml:

queue.type: (valori possibili — persistito e memoria (predefinito)).
path.queue: (percorso della cartella con i file delle code, che di default sono memorizzati in C:Logstashqueue).
queue.page_capacity: (dimensione massima della pagina della coda, valore predefinito — 64mb).
queue.drain: (true/false — attiva/disattiva l'interruzione dell'elaborazione della coda prima di spegnere Logstash. Non consiglio di attivarla, poiché influisce direttamente sulla velocità di spegnimento del server).
queue.max_events: (numero massimo di eventi nella coda, di default — 0 (illimitato)).
queue.max_bytes: (dimensione massima della coda in byte, di default — 1024mb (1gb)).

Se configurato queue.max_events e queue.max_bytes, i messaggi smettono di essere accettati nella coda all'arrivo del valore di una di queste impostazioni. Maggiori dettagli sulle Coda Persistenti sono forniti qui.

Esempio di parte di logstash.yml relativa alla configurazione della coda:

queue.type: persistito
queue.max_bytes: 10gb

Configurazione

La configurazione di Logstash è solitamente composta da tre parti che si occupano di diverse fasi dell'elaborazione dei log in ingresso: ricezione (sezione input), parsing (sezione filter) e invio a Elastic (sezione output). Di seguito, analizzeremo in dettaglio ciascuna di esse.

Input

Il flusso in ingresso con i log grezzi è ricevuto dagli agenti filebeat. Questo plugin è quello che indichiamo nella sezione input:

input {
  beats {
    port => 5044
  }
}

Dopo questa configurazione, Logstash inizia ad ascoltare la porta 5044 e, al ricevimento dei log, li elabora secondo le impostazioni della sezione filter. Se necessario, è possibile crittografare il canale di ricezione dei log tramite SSL. Maggiori dettagli sulle impostazioni del plugin beats sono indicati qui.

Filter

Tutti i log di testo interessanti generati da Exchange, pronti per essere elaborati, sono in formato csv con i campi descritti all'interno del file di log. Per analizzare i record csv, Logstash ci offre tre plugin: dissect, csv e grok. Il primo è il più veloce, ma gestisce solo l'analisi dei log più semplici.
Ad esempio, il seguente record sarà suddiviso in due (a causa della presenza di una virgola all'interno del campo), il che porterà a un'analisi errata del log:

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

Può essere utilizzato per analizzare i log, ad esempio, IIS. In questo caso, la sezione filter potrebbe apparire come segue:

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

La configurazione di Logstash consente di utilizzare operatori condizionali, quindi possiamo indirizzare al plugin dissect solo i log contrassegnati con il tag filebeat IIS. All'interno del plugin, mappiamo i valori dei campi ai loro nomi, rimuoviamo il campo originale message, che conteneva la registrazione del log, e possiamo aggiungere un campo arbitrario che conterrà, ad esempio, il nome dell'applicazione da cui stiamo raccogliendo i log.

Nel caso dei log di tracciamento, è meglio utilizzare il plugin csv, che è in grado di gestire correttamente campi complessi:

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

All'interno del plugin, mappiamo i valori dei campi ai loro nomi, rimuoviamo il campo originale message (e anche i campi tenant-id e schema-version), che conteneva la registrazione del log, e possiamo aggiungere un campo arbitrario che conterrà, ad esempio, il nome dell'applicazione da cui stiamo raccogliendo i log.

All'uscita dalla fase di filtro, otterremo documenti che sono, a prima vista, pronti per la visualizzazione in Kibana. Ci mancherà quanto segue:

I campi numerici saranno riconosciuti come testo, il che non consente di eseguire operazioni su di essi. In particolare, i campi time-taken del log IIS, così come i campi recipient-count e total-bites del log Tracking.
Il timestamp standard del documento conterrà il tempo di elaborazione del log, non il tempo di registrazione sul lato server.
Campo recipient-address apparirà come un'unica riga, il che non consente di effettuare analisi con il conteggio dei destinatari delle email.

È tempo di aggiungere un po' di magia al processo di elaborazione dei log.

La conversione dei campi numerici

Il plugin dissect ha un'opzione convert_datatype, che può essere utilizzata per convertire un campo di testo in formato numerico. Ad esempio, così:

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

Vale la pena ricordare che questo metodo è adatto solo se il campo conterrà esattamente una stringa. I valori nulli nei campi non vengono gestiti dall'opzione e si verifica un'eccezione.

Per i log di tracking, è meglio non utilizzare lo stesso metodo di conversione, poiché i campi recipient-count e total-bites potrebbero essere vuoti. Per convertire questi campi è meglio utilizzare il plugin mutate:

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

Suddividere il recipient_address in destinatari separati

Questo compito può essere risolto anche con il plugin mutate:

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

Modifichiamo il timestamp

Nel caso dei log di tracking, la questione può essere facilmente risolta con il plugin date, che aiuterà a scrivere nel campo timestamp la data e l'ora nel formato necessario dal campo date-time:

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

Nel caso dei log IIS, sarà necessario unire i dati dei campi date e time utilizzando il plugin mutate, specificare il fuso orario necessario e inserire questo timestamp in timestamp utilizzando il plugin 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

La sezione output è utilizzata per inviare i log elaborati al ricevitore di log. Nel caso di invio diretto a Elastic, si utilizza il plugin elasticsearch, in cui viene specificato l'indirizzo del server e il modello del nome dell'indice per inviare il documento formato:

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

Configurazione finale

La configurazione finale apparirà come segue:

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

Link utili: