Fluentd Hochverfügbarkeits-Setup und Duplikate

Question

Ich habe folgendes Problem:

Wir fluentd in einem Hochverfügbarkeits-Setup verwenden: ein paar K von Spediteuren -> Aggregatoren für geo Region und ES/S3 am Ende mit Plugin kopieren Wir haben einen Fehler festgestellt (Protokolle wurden einige Tage nicht verarbeitet) und seit der Wiederherstellung erhalten wir Tonnen von doppelten Datensätzen flüssig in unseren ES-Cluster (einschließlich duplizierter Daten nach der Wiederherstellung). Gibt es bekannte Probleme mit der @type copy plugin, die diese Art von Verhalten verursachen könnten?

Unsere Speditions config:

# TCP input 
<source> 
    @type forward 
    port X 
</source> 

# Logs Forwarding 
<match XXX> 
    @type forward 

    # forward to logs-aggregators 
     <server> 
#... 
     </server> 

    # use tcp for heartbeat 
    heartbeat_type tcp 

    # use longer flush_interval to reduce CPU usage. 
    # note that this is a trade-off against latency. 
    flush_interval 10s 

    # use file buffer to buffer events on disks. 
    # max 4096 8MB chunks = 32GB of buffer space 
    buffer_type file 
    buffer_path /var/log/td-agent/buffer/forward 
    buffer_chunk_limit 4m 
    buffer_queue_limit 4096 

    # use multi-threading to send buffered data in parallel 
    num_threads 8 

    # expire DNS cache (required for cloud environment such as EC2) 
    expire_dns_cache 600 
</match> 

Unsere Aggregatoren config:

# TCP input 
<source> 
    @type forward 
    port X 
</source> 

# rsyslog 
<source> 
    @type syslog 
    port X 
    tag rsyslog 
</source> 

# Logs storage 
<match rsyslog.**> 
    @type copy 
    <store> 
    @type elasticsearch 
    hosts X 
    logstash_format true 
    logstash_prefix rsyslog 
    logstash_dateformat %Y-%m-%d 


    num_threads 8 

    utc_index true 
    reload_on_failure true 
    </store> 
</match> 

# Bids storage 
<match X> 
    @type copy 

    # push data to elasticsearch cluster 
    <store> 
    @type elasticsearch 
    hosts X 

    # save like logstash would 
    logstash_format true 
    logstash_prefix jita 
    logstash_dateformat %Y-%m-%d 

    # 64G of buffer data 
    buffer_chunk_limit 16m 
    buffer_queue_limit 4096 
    flush_interval 5m 

    num_threads 8 

    # everything in UTC 
    utc_index true 

    # quickly remove a dead node from the list of addresses 
    reload_on_failure true 
    </store> 

    # additionally store data in s3 bucket 
    <store> 
    @type s3 
    aws_key_id X 
    aws_sec_key X 
    s3_bucket X 
    s3_region X 
    s3_object_key_format %{path}/%{time_slice}_%{index}.%{file_extension} 
    store_as gzip 
    num_threads 8 
    path logs 
    buffer_path /var/log/td-agent/buffer/s3 
    time_slice_format %Y-%m-%d/%H 
    time_slice_wait 10m 
    utc 
    </store> 
</match> 

Answer 1

Die Lösung für uns war ein Identitätsfeld hinzufügen und definieren Sie es in der fließend Konfiguration (elasticsearch plugin) mit id_key. Seitdem haben wir keine Probleme mehr, egal wie schwer td-agent erneut versucht :)

Answer 2

Ich weiß, dass dies ein alter Thread, aber bin Entsendung diese Antwort nur für den Fall jemand hier erreicht für die Suche die Lösung.

In allen gepufferten Plugins gibt es einen Konfigurationseintrag namens "retry_wait". Der Standardwert für diese Konfiguration ist 1s (1 Sekunde).

Dies bedeutet, dass fluentd eine Anfrage an Elasticsearch sendet und wenn es innerhalb von 1 Sekunde keine Antwort erhält, wird die Anfrage erneut gesendet.

Von ElasticSearch-Seite, da es kein Konzept von Transaktionen gibt, können sie nicht identifizieren, wenn die gleiche Anfrage erneut auf nicht gesendet wird. Zusätzlich zu dem, was bereits bei der ersten Anfrage indiziert wurde, versuchen alle Werte erneut indiziert zu werden, was zu Duplikaten führt.

In unserem Fall haben wir rund 40 bis 50 Duplikate für die meisten Datensätze gelandet.

Eine mögliche Lösung könnte sein, die Zeitüberschreitung zu erhöhen. 30 Sekunden funktionieren normalerweise für uns, Sie können mit diesem Wert spielen und eine Zahl herausfinden, die für Sie arbeitet.