Come posso creare un istogramma dei delta timestamp?

Question

Archiviamo piccoli documenti in ES che rappresentano una sequenza di eventi per un oggetto. Ogni evento ha un timbro data/ora. Abbiamo bisogno di analizzare il tempo tra gli eventi per tutti gli oggetti in un periodo di tempo.

Per esempio, immaginate questi documenti evento JSON:

{ "oggetto": "uno", "evento": "start", "datetime": "2016/02/09 11:23:01" }

{ "oggetto": "uno", "evento": "stop", "datetime": "2016/02/09 11:25:01"}

{ "oggetto": "due "," evento ":" start "," datetime ":" 2016-01-02 11:23:01 "}

{" oggetto ":" due "," evento ":" stop "," datetime ":" 2016-01-02 11:24:01 "}

Quello che vorremmo ottenere da questo è un istogramma che traccia i due delta timestamp risultanti (dall'inizio alla fine): 2 minuti/120 secondi per l'oggetto uno e 1 minuto/60 secondi per l'oggetto due.

In definitiva, vogliamo monitorare il tempo tra gli eventi di inizio e fine, ma è necessario calcolare il tempo tra questi eventi e quindi aggregarli o fornirli all'interfaccia utente di Kibana per essere aggregati/tracciati. Idealmente vorremmo fornire i risultati direttamente a Kibana in modo da evitare la creazione di un'interfaccia utente personalizzata.

Grazie in anticipo per eventuali idee o suggerimenti.

Answer 1

Dal momento che siete aperti a utilizzare Logstash, c'è un modo per farlo utilizzando il aggregate filter

Si noti che questo è un plugin comunità che ha bisogno di essere installato prima. (per impostazione predefinita non viene fornito con Logstash)

L'idea principale del filtro aggregate consiste nell'unire due righe di registro "correlate". È possibile configurare il plug-in in modo che sappia cosa significa "correlato". Nel tuo caso, "relativo" significa che entrambi gli eventi devono condividere lo stesso object nome (cioè one o two) e poi che il primo evento ha il suo event campo con il valore start e il secondo evento ha il suo event campo con il valore stop.

Quando il filtro incontra l'evento start, memorizza il campo datetime di quell'evento in una mappa interna. Quando incontra l'evento stop, calcola la differenza di tempo tra i due datetime e memorizza la durata in secondi nel nuovo campo duration.

input { 
    ... 
} 
filter { 
    ...other filters 

    if [event] == "start" { 
     aggregate { 
      task_id => "%{object}" 
      code => "map['start'] = event['datetime']" 
      map_action => "create" 
     } 
    } else if [event] == "stop" { 
     aggregate { 
      task_id => "%{object}" 
      code => "map['duration'] = event['datetime'] - map['start']" 
      end_of_task => true 
      timeout => 120 
     } 
    } 
} 
output { 
    elasticsearch { 
     ... 
    } 
} 

Si noti che è possibile regolare il valore timeout (qui 120 secondi) per soddisfare meglio le vostre esigenze. Quando è scaduto il timeout e non si è ancora verificato un evento di arresto, l'evento di inizio esistente verrà abbandonato.