search

uken / fluent-plugin-elasticsearch

Apache License 2.0
890 stars 310 forks source link

data stream output overriding @timestamp field when it already exists in the record #967

Closed dschaaff closed 2 years ago

dschaaff commented 2 years ago

(check apply)

Problem

Some, but not all, of our logs include the @timestamp field. When this field is present I'd expect the plugin not to overwrite it with the current time at injestion.

Steps to replicate

@type elasticsearch_data_stream
  @id es_logs_output
  @log_level warn
  # other settings are omitted.
  log_es_400_reason true
  include_timestamp true # https://github.com/uken/fluent-plugin-elasticsearch#include_timestamp
  hosts "#{ENV['ES_LOGS_HOSTS']}"
  port 9200
  user "#{ENV['ES_LOGS_USER']}"
  password "#{ENV['ES_LOGS_PASSWORD']}"
  data_stream_name logs-stream
  data_stream_template_name cordial-logs # specifies the index template name
  data_stream_ilm_name cordial-logs # specifies the ilm policy
  bulk_message_request_threshold "#{ENV['BULK_MESSAGE_REQUEST_THRESHOLD']}"
  request_timeout "#{ENV['ES_REQUEST_TIMEOUT']}"
  retry_tag 'retry_es'
  <buffer tag>
    @type file
    chunk_limit_size  "#{ENV['ES_CHUNK_LIMIT_SIZE']}"
    flush_mode interval
    flush_interval 5s
    flush_thread_count  "#{ENV['ES_FLUSH_THREAD_COUNT']}"
    retry_timeout 4h
  </buffer>
  </store>

Send a record with the @timestamp field. Compare the @timestamp field in elasticsearch and ensure it matches the original log record.

Expected Behavior or What you need to ask

If the record already includes the field @timestamp, do no overwrite it. Only add it if @timestamp does not already exist.

Using Fluentd and ES plugin versions

*** LOCAL GEMS ***

async (1.30.1)
async-http (0.54.0)
async-io (1.33.0)
async-pool (0.3.9)
bigdecimal (default: 1.4.1)
bundler (default: 1.17.2)
cmath (default: 1.0.0)
concurrent-ruby (1.1.10)
console (1.15.0)
cool.io (1.7.1)
csv (default: 3.0.9)
date (default: 2.0.2)
dbm (default: 1.0.0)
did_you_mean (1.3.0)
e2mmap (default: 0.1.0)
elasticsearch (7.17.1)
elasticsearch-api (7.17.1)
elasticsearch-transport (7.17.1)
elasticsearch-xpack (7.17.1)
etc (default: 1.0.1)
excon (0.92.2)
ext_monitor (0.1.2)
faraday (1.8.0)
faraday-em_http (1.0.0)
faraday-em_synchrony (1.0.0)
faraday-excon (1.1.0)
faraday-httpclient (1.0.1)
faraday-net_http (1.0.1)
faraday-net_http_persistent (1.2.0)
faraday-patron (1.0.0)
faraday-rack (1.0.0)
fcntl (default: 1.0.0)
fiber-local (1.0.0)
fiddle (default: 1.0.0)
fileutils (default: 1.1.0)
fluent-config-regexp-type (1.0.0)
fluent-plugin-concat (2.5.0)
fluent-plugin-elasticsearch (5.2.1)
fluent-plugin-grok-parser (2.6.2)
fluent-plugin-multi-format-parser (1.0.0)
fluent-plugin-prometheus (2.0.2)
fluent-plugin-record-modifier (2.1.0)
fluent-plugin-rewrite-tag-filter (2.4.0)
fluentd (1.14.6, 1.14.0)
forwardable (default: 1.2.0)
gdbm (default: 2.0.0)
http_parser.rb (0.8.0, 0.7.0)
io-console (default: 0.4.7)
ipaddr (default: 1.2.2)
irb (default: 1.0.0)
json (2.4.1, default: 2.1.0)
logger (default: 1.3.0)
matrix (default: 0.1.0)
minitest (5.11.3)
msgpack (1.5.1, 1.4.5)
multi_json (1.15.0)
multipart-post (2.1.1)
mutex_m (default: 0.1.0)
net-telnet (0.2.0)
nio4r (2.5.8)
oj (3.10.18)
openssl (default: 2.1.2)
ostruct (default: 0.1.0)
power_assert (1.1.3)
prime (default: 0.1.0)
prometheus-client (4.0.0)
protocol-hpack (1.4.2)
protocol-http (0.21.0)
protocol-http1 (0.13.2)
protocol-http2 (0.14.2)
psych (default: 3.1.0)
rake (12.3.3)
rdoc (default: 6.1.2.1)
rexml (default: 3.1.9.1)
rss (default: 0.2.7)
ruby2_keywords (0.0.5)
scanf (default: 1.0.0)
sdbm (default: 1.0.0)
serverengine (2.2.5)
shell (default: 0.7)
sigdump (0.2.4)
stringio (default: 0.0.2)
strptime (0.2.5)
strscan (default: 1.0.0)
sync (default: 0.5.0)
test-unit (3.2.9)
thwait (default: 0.1.0)
timers (4.3.3)
tracer (default: 0.1.0)
tzinfo (2.0.4)
tzinfo-data (1.2022.1)
webrick (default: 1.4.4)
xmlrpc (0.3.0)
yajl-ruby (1.4.2, 1.4.1)
zlib (default: 1.0.0)