Я все еще новичок в fluentd, я пробовал различные конфигурации, но я застрял.
Предположим, я отправил эту запись в fluend, в которой есть _epoch, чтобы указать эпоху, когда была создана запись.
{"data":"dummy", "_epoch": <epochtime_in_second>}
Вместо использования атрибута time, обрабатываемого fluentd, я хочу переопределить время с помощью этого поля _epoch. Как произвести вывод fluentd с переопределением time?
Я пробовал это
# TCP input to receive logs from the forwarders
<source>
@type forward
bind 0.0.0.0
port 24224
</source>
# HTTP input for the liveness and readiness probes
<source>
@type http
bind 0.0.0.0
port 9880
</source>
# rds2fluentd_test
<filter rds2fluentd_test.*>
@type parser
key_name _epoch
reserve_data true
<parse>
@type regexp
expression /^(?<time>.*)$/
time_type unixtime
utc true
</parse>
</filter>
<filter rds2fluentd_test.*>
@type stdout
</filter>
<match rds2fluentd_test.*>
@type s3
@log_level debug
aws_key_id "#{ENV['AWS_ACCESS_KEY']}"
aws_sec_key "#{ENV['AWS_SECRET_KEY']}"
s3_bucket foo-bucket
s3_region ap-southeast-1
path ingestion-test-01/${_db}/${_table}/%Y-%m-%d-%H-%M/
#s3_object_key_format %{path}%{time_slice}_%{index}.%{file_extension}
# if you want to use ${tag} or %Y/%m/%d/ like syntax in path / s3_object_key_format,
# need to specify tag for ${tag} and time for %Y/%m/%d in <buffer> argument.
<buffer time,_db,_table>
@type file
path /var/log/fluent/s3
timekey 1m # 5 minutes partition
timekey_wait 10s
timekey_use_utc true # use utc
chunk_limit_size 256m
</buffer>
time_slice_format %Y%m%d%H
store_as json
</match>
Но после получения данных, как указано выше, появляется предупреждение об ошибке, например:
#0 dump an error event: error_class=Fluent::Plugin::Parser::ParserError error="parse failed no implicit conversion of Integer into Hash" location="/usr/local/bundle/gems/fluentd-1.10.4/lib/fluent/plugin/filter_parser.rb:110:in `rescue in filter_with_time'" tag="rds2fluentd_test." time=1590578507 record={....