Siamo amici di ELK ed Exchange. Parte 2

Continuo la mia storia su come fare amicizia con Exchange ed ELK (inizio qui). Permettetemi di ricordarvi che questa combinazione è in grado di elaborare un numero molto elevato di log senza esitazione. Questa volta parleremo di come far funzionare Exchange con i componenti Logstash e Kibana.

Logstash nello stack ELK viene utilizzato per elaborare in modo intelligente i log e prepararli per il posizionamento in Elastic sotto forma di documenti, sulla base dei quali è conveniente creare varie 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 decompresso in una directory specifica. Successivamente è necessario inserire il percorso di questa directory nelle variabili $env:Path e $env:JAVA_HOME del sistema operativo Windows:

Controlliamo la versione 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 di archivio con la distribuzione Logstash quindi. L'archivio deve essere decompresso nella radice del disco. Disimballare nella cartella C:Program Files Non ne vale la pena, Logstash si rifiuterà di avviarsi normalmente. Quindi è necessario accedere al file jvm.options correzioni responsabili dell'allocazione della RAM per il processo Java. Consiglio di specificare metà della RAM del server. Se ha 16 GB di RAM a bordo, i tasti predefiniti sono:

-Xms1g
-Xmx1g

deve essere sostituito con:

-Xms8g
-Xmx8g

Inoltre, si consiglia di commentare la riga -XX:+UseConcMarkSweepGC. Maggiori informazioni qui. Il passaggio successivo consiste nel creare una configurazione predefinita nel file logstash.conf:

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

Con questa configurazione, Logstash legge i dati dalla console, li passa attraverso un filtro vuoto e li restituisce alla console. L'utilizzo di questa configurazione metterà alla prova la funzionalità di Logstash. Per fare ciò, eseguiamolo in modalità interattiva:

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 è stato avviato correttamente sulla porta 9600.

Il passaggio finale dell'installazione: avvia Logstash come servizio Windows. Questo può essere fatto, ad esempio, utilizzando il pacchetto NSSM:

PS C:...bin> .nssm.exe install logstash
Service "logstash" installed successfully!

tolleranza ai guasti

La sicurezza dei log quando vengono trasferiti dal server di origine è garantita dal meccanismo delle code persistenti.

Come funziona

Il layout delle code durante l'elaborazione del registro è: input → coda → filtro + output.

Il plugin di input riceve i dati da un'origine log, li scrive in una coda e invia la conferma che i dati sono stati ricevuti all'origine.

I messaggi dalla coda vengono elaborati da Logstash, passati attraverso il filtro e il plugin di output. Quando si riceve la conferma dall'output che il log è stato inviato, Logstash rimuove il log elaborato dalla coda. Se Logstash si interrompe, tutti i messaggi non elaborati e i messaggi per i quali non è stata ricevuta alcuna conferma rimangono nella coda e Logstash continuerà a elaborarli al successivo avvio.

registrazione

Regolabile tramite tasti nella cartella C:Logstashconfiglogstash.yml:

• queue.type: (valori possibili - persisted и memory (default)).
• path.queue: (percorso della cartella con i file in coda, che sono archiviati in C:Logstashqueue per impostazione predefinita).
• queue.page_capacity: (dimensione massima della pagina in coda, il valore predefinito è 64 MB).
• queue.drain: (vero/falso: abilita/disabilita l'interruzione dell'elaborazione della coda prima di chiudere Logstash. Non consiglio di abilitarlo, perché ciò influirà direttamente sulla velocità di arresto del server).
• queue.max_events: (numero massimo di eventi in coda, il valore predefinito è 0 (illimitato)).
• queue.max_bytes: (dimensione massima della coda in byte, impostazione predefinita: 1024 MB (1 GB)).

Se configurato queue.max_events и queue.max_bytes, i messaggi non verranno più accettati nella coda quando viene raggiunto il valore di una di queste impostazioni. Ulteriori informazioni sulle code persistenti qui.

Un esempio della parte di logstash.yml responsabile dell'impostazione della coda:

queue.type: persisted
queue.max_bytes: 10gb

registrazione

La configurazione di Logstash è solitamente composta da tre parti, responsabili di diverse fasi di elaborazione dei log in ingresso: ricezione (sezione input), parsing (sezione filtro) e invio ad Elastic (sezione output). Di seguito daremo uno sguardo più da vicino a ciascuno di essi.

Ingresso

Riceviamo il flusso in entrata con i log grezzi dagli agenti filebeat. È questo plugin che indichiamo nella sezione input:

input {
  beats {
    port => 5044
  }
}

Dopo questa configurazione, Logstash inizia ad ascoltare la porta 5044 e, quando riceve i log, li elabora in base alle impostazioni della sezione filtro. Se necessario, puoi racchiudere il canale per ricevere i log da filebit in SSL. Ulteriori informazioni sulle impostazioni del plug-in Beats qui.

Filtro

Tutti i registri di testo generati da Exchange che sono interessanti per l'elaborazione sono in formato CSV con i campi descritti nel file di registro stesso. Per l'analisi dei record CSV, Logstash ci offre tre plugin: sezionare, csv e grok. Il primo è il massimo veloce, ma gestisce l'analisi solo dei log più semplici.
Ad esempio, dividerà in due il seguente record (a causa della presenza di una virgola all'interno del campo), motivo per cui il log verrà analizzato in modo errato:

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

Può essere utilizzato durante l'analisi dei log, ad esempio IIS. In questo caso, la sezione del filtro potrebbe assomigliare a questa:

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 ti consente di utilizzare dichiarazioni condizionali, quindi possiamo inviare solo i log contrassegnati con il tag filebeat al plugin dissect IIS. All'interno del plugin abbiniamo i valori dei campi con i loro nomi, eliminiamo il campo originale message, che conteneva una voce dal log, e possiamo aggiungere un campo personalizzato che conterrà, ad esempio, il nome dell'applicazione da cui raccogliamo i log.

Nel caso dei log di tracciamento è meglio utilizzare il plugin csv; può elaborare 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 abbiniamo i valori dei campi con i loro nomi, eliminiamo il campo originale message (e anche campi tenant-id и schema-version), che conteneva una voce dal registro, e possiamo aggiungere un campo personalizzato che conterrà, ad esempio, il nome dell'applicazione da cui raccogliamo i registri.

All'uscita dalla fase di filtraggio, riceveremo documenti in prima approssimazione, pronti per la visualizzazione a Kibana. Ci mancherà quanto segue:

• I campi numerici verranno riconosciuti come testo, il che impedisce operazioni su di essi. Cioè, i campi time-taken Registro IIS e campi recipient-count и total-bites Monitoraggio del registro.
• Il timestamp standard del documento conterrà l'ora in cui il registro è stato elaborato, non l'ora in cui è stato scritto sul lato server.
• Campo recipient-address sembrerà un cantiere, che non consente l'analisi per contare i destinatari delle lettere.

È giunto il momento di aggiungere un po' di magia al processo di elaborazione dei log.

Conversione di campi numerici

Il plugin di dissezione ha un'opzione convert_datatype, che può essere utilizzato per convertire un campo di testo in un formato digitale. Ad esempio, in questo modo:

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

Vale la pena ricordare che questo metodo è adatto solo se il campo conterrà sicuramente una stringa. L'opzione non elabora i valori Null dai campi e genera un'eccezione.

Per il tracciamento dei log, è meglio non utilizzare un metodo di conversione simile, poiché i campi recipient-count и total-bites potrebbe essere vuoto. Per convertire questi campi è meglio utilizzare un plugin mutare:

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

Suddivisione di recipient_address in singoli destinatari

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

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

Modifica del timestamp

Nel caso dei log di tracciamento il problema viene risolto molto facilmente dal plugin quando, che ti aiuterà a scrivere sul campo timestamp data e ora nel formato richiesto dal campo date-time:

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

Nel caso dei registri IIS, dovremo combinare i dati dei campi date и time utilizzando il plugin mutate, registra il fuso orario di cui abbiamo bisogno e inserisci questo timestamp timestamp utilizzando il plug-in della data:

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 di output viene utilizzata per inviare i log elaborati al destinatario dei log. In caso di invio diretto a Elastic viene utilizzato un plugin elasticsearch, che specifica l'indirizzo del server e il modello del nome dell'indice per l'invio del documento generato:

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 sarà simile a questa:

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:

• Come installare OpenJDK 11 su Windows?
• Scarica Logstash
• Elastic utilizza l'opzione deprecata UseConcMarkSweepGC #36828
• NSSM
• Code persistenti
• Beats plug in input
• Logstash Amico, dov'è la mia motosega? Devo analizzare i miei registri
• Plug-in per il filtro di dissezione
• Condizionali
• Plug-in filtro muta
• Plug-in per il filtro della data
• Plug-in di output Elasticsearch

Fonte: habr.com