🥇Colleghiamo ELK e Exchange. Parte 2

Continuo il mio racconto su come integrare Exchange ed ELK (inizio qui). Ricordo che questa combinazione è capace senza esitazioni di gestire una grande quantità 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 l'archiviazione in Elasticsearch come documenti, sui quali è facile costruire diverse visualizzazioni in Kibana.

Installazione

Consiste in due fasi:

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

Installazione e configurazione del pacchetto OpenJDK

Il pacchetto OpenJDK deve essere scaricato e estratto in una directory specifica. Successivamente, 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

Scarica il file-archivio con il distributore di Logstash da qui. L'archivio deve essere estratto nella radice del disco. Non è consigliabile estrarlo nella cartella C:Program Files , poiché Logstash si rifiuterà di avviarsi correttamente. Quindi è necessario inserire nel file jvm.options modifiche che si occupano dell'allocazione della memoria RAM per il processo Java. Si consiglia di indicare metà della RAM del server. Se dispone di 16 GB di RAM, le chiavi predefinite sono:

-Xms1g
-Xmx1g

devono essere sostituite con:

-Xms8g
-Xmx8g

Inoltre, è opportuno commentare la riga -XX:+UseConcMarkSweepGC. Maggiori dettagli su questo qui. Il passo successivo è creare 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. Applicare questa configurazione permetterà di verificare il funzionamento di Logstash. Per fare questo, avviamolo in modalità interattiva:

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

Logstash è stato avviato con successo sulla porta 9600.

L'ultimo passo per l'installazione: avviare Logstash come servizio Windows. Questo può essere fatto, ad esempio, utilizzando il pacchetto NSSM:

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

Affidabilità

La conservazione dei log durante il trasferimento dal server sorgente è garantita dal meccanismo delle Code Persistenti.

Come funziona

Schema di posizionamento delle code nel processo di 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 di ricezione alla sorgente.

I messaggi dalla coda vengono elaborati da Logstash, passano attraverso il filtro e il plugin output. Alla ricezione della conferma di invio del log dall'output, Logstash rimuove il log elaborato dalla coda. Se Logstash si ferma, tutti i messaggi non elaborati e quelli per cui non è stata ricevuta conferma di invio rimangono nella coda, e Logstash continuerà a elaborarli al prossimo avvio.

Impostazione

Regolato tramite le chiavi nel file C:Logstashconfiglogstash.yml:

queue.type: (valori possibili — persisted e memory (default)).
path.queue: (percorso della cartella con i file delle code, che per impostazione predefinita 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 dello spegnimento di Logstash. Non consiglio di attivarlo, in quanto 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 impostati queue.max_events e queue.max_bytes, i messaggi smettono di essere accettati nella coda al raggiungimento del valore di una qualsiasi di queste impostazioni. Maggiori informazioni sulle Code Persistenti sono disponibili qui.

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

queue.type: persisted
queue.max_bytes: 10gb

Impostazione

La configurazione di Logstash di solito consiste in tre parti, ognuna responsabile di diverse fasi dell'elaborazione dei log in ingresso: ricezione (sezione input), parsing (sezione filter) e invio a Elastic (sezione output). Di seguito esamineremo in dettaglio ciascuna di esse.

Ingresso

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

input {
  beats {
    port => 5044
  }
}

Dopo questa configurazione, Logstash inizia ad ascoltare sulla porta 5044 e, al ricevimento dei log, li elabora secondo le impostazioni della sezione filter. Se necessario, è possibile incapsulare il canale di acquisizione dei log di filebeat in SSL. Maggiori informazioni sulle impostazioni del plugin beats sono disponibili. qui.

Filter

Tutti i log testuali interessanti generati da Exchange sono in formato csv con i campi descritti nel file di log stesso. Per analizzare le registrazioni csv, Logstash ci offre tre plugin: dissect, csv e grok. Il primo è il più veloce, ma si occupa dell'analisi solo dei log più semplici.
Ad esempio, la seguente registrazione sarà suddivisa 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, di IIS. In tal caso, la sezione filter potrebbe apparire nel seguente modo:

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 nel plugin dissect possiamo indirizzare solo i log che sono stati contrassegnati con il tag filebeat IIS. All'interno del plugin, associamo 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 raccogliamo 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, associamo i valori dei campi ai loro nomi, rimuoviamo il campo originale message (e anche i campi tenant-id e schema-version), che conteneva una registrazione dal log, e possiamo aggiungere un campo arbitrario che, ad esempio, conterrà il nome dell'applicazione da cui raccogliamo i log.

In uscita dalla fase di filtraggio, otterremo documenti che sono preliminarmente 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, e non il tempo in cui è stato registrato sul server.
Campo recipient-address sarà visualizzato in un'unica stringa, 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.

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

È importante ricordare che questo metodo è adatto solo se il campo conterrà esattamente una stringa. I valori nulli nei campi non vengono gestiti da questa opzione e causano un'eccezione.

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

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

Suddivisione di recipient_address in destinatari separati

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

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

Modifica del timestamp

Nel caso dei log di tracking, il problema si risolve facilmente con il plugin date, che aiuterà a scrivere nel campo timestamp data e ora nel formato corretto 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 ora utilizzando il plugin mutate, specificare il fuso orario necessario e inserire questo timestamp con timestamp 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" ]
}

Uscita

La sezione output viene utilizzata per inviare i log elaborati a un ricevitore di log. In caso di invio diretto a Elastic, viene utilizzato un plugin elasticsearch, in cui è 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:

Fonte: habr.com