Component(s)

exporter/elasticsearch

What happened?

Description

If the Elasticsearch exporter is configured with a persistent queue (sending_queue.storage: filestorage) and new data is written to this persistent queue while the Elasticsearch cluster is temporarily unavailable, this data should be indexed into Elasticsearch once the cluster becomes available again.

Steps to Reproduce

1. Configure the collector like so:

   receivers:
     filelog:
       include: /tmp/*.log
       storage: file_storage/filelogreceiver
   
   exporters:
     elasticsearch:
       endpoints: [ "http://localhost:9200" ]
       index: foo
       flush:
         interval: 1s
       sending_queue:
         enabled: true
         storage: file_storage/elasticsearchexporter
   
   extensions:
     file_storage/filelogreceiver:
       directory: /tmp/otelcol/file_storage/filelogreceiver
     file_storage/elasticsearchexporter:
       directory: /tmp/otelcol/file_storage/elasticsearchexporter
    
   service:
     extensions: [file_storage/filelogreceiver, file_storage/elasticsearchexporter]
     pipelines:
       logs:
         receivers: [filelog]
         exporters: [elasticsearch]
     telemetry:
       logs:
         level: debug

2. Start the collector.
3. Add a log entry to the file.

   echo "1 foo bar $RANDOM" >> /tmp/test.log

4. Query the Elasticsearch cluster to verify that the log entry made it into the index.
5. Stop the Elasticsearch cluster.
6. Add a new log entry to the file.

   echo "2 foo bar $RANDOM" >> /tmp/test.log

7. Restart the Elasticsearch cluster.
8. Query the Elasticsearch cluster to verify that the second log entry made it into the index.

Expected Result

• In step 8, there should be two log entries in the index.

Actual Result

• In step 8, only the first log entry (added in step 3) is in the index.

Collector version

v0.93.0

Environment information

Environment

OS: (e.g., "Ubuntu 20.04") MacOS
Compiler(if manually compiled): (e.g., "go 14.2") go 1.21.6

OpenTelemetry Collector configuration

receivers:
  filelog:
    include: /tmp/*.log
    storage: file_storage/filelogreceiver

exporters:
  elasticsearch:
    endpoints: [ "http://localhost:9200" ]
    index: foo
    flush:
      interval: 1s
    sending_queue:
      enabled: true
      storage: file_storage/elasticsearchexporter

extensions:
  file_storage/filelogreceiver:
    directory: /tmp/otelcol/file_storage/filelogreceiver
  file_storage/elasticsearchexporter:
    directory: /tmp/otelcol/file_storage/elasticsearchexporter
    
service:
  extensions: [file_storage/filelogreceiver, file_storage/elasticsearchexporter]
  pipelines:
    logs:
      receivers: [filelog]
      exporters: [elasticsearch]
  telemetry:
    logs:
      level: debug

Log output

No response

Additional context

No response