fluentd collect k8s containerd log do not use the timestamp from realy time

[arthas3014]

What is a problem?

I use fluentd-kubernetes-daemonset:v1.11.5-debian-elasticsearch7-1.1 to collect k8s containerd log to es,and I find that the log from es timestamp is not consistent with the time of the log file on the k8s node, which is more than ten seconds later than the real time, so I suspect that fluentd failed to parse time as timestamp to elasticsearch

Describe the configuration of Fluentd

<match kubernetes.**>
@type elasticsearch_dynamic
@id kubernetes_elasticsearch
@log_level info
include_tag_key true
host "#{ENV['FLUENT_ELASTICSEARCH_HOST']}"
port "#{ENV['FLUENT_ELASTICSEARCH_PORT']}"
path "#{ENV['FLUENT_ELASTICSEARCH_PATH']}"
scheme "#{ENV['FLUENT_ELASTICSEARCH_SCHEME'] || 'http'}"
ssl_verify "#{ENV['FLUENT_ELASTICSEARCH_SSL_VERIFY'] || 'true'}"
ssl_version "#{ENV['FLUENT_ELASTICSEARCH_SSL_VERSION'] || 'TLSv1_2'}"
user "#{ENV['FLUENT_ELASTICSEARCH_USER'] || use_default}"
password "#{ENV['FLUENT_ELASTICSEARCH_PASSWORD'] || use_default}"
reload_connections "#{ENV['FLUENT_ELASTICSEARCH_RELOAD_CONNECTIONS'] || 'false'}"
reconnect_on_error "#{ENV['FLUENT_ELASTICSEARCH_RECONNECT_ON_ERROR'] || 'true'}"
reload_on_failure "#{ENV['FLUENT_ELASTICSEARCH_RELOAD_ON_FAILURE'] || 'true'}"
log_es_400_reason "#{ENV['FLUENT_ELASTICSEARCH_LOG_ES_400_REASON'] || 'false'}"
logstash_prefix "#{ENV['FLUENT_ELASTICSEARCH_INDEX_PREFIX'] || 'logstash'}-${record['kubernetes']['namespace_name']}"
logstash_dateformat "#{ENV['FLUENT_ELASTICSEARCH_LOGSTASH_DATEFORMAT'] || '%Y-%m-%d'}"
logstash_format "#{ENV['FLUENT_ELASTICSEARCH_LOGSTASH_FORMAT'] || 'true'}"
index_name "#{ENV['FLUENT_ELASTICSEARCH_LOGSTASH_INDEX_NAME'] || 'logstash'}"
target_index_key "#{ENV['FLUENT_ELASTICSEARCH_TARGET_INDEX_KEY'] || use_nil}"
type_name "#{ENV['FLUENT_ELASTICSEARCH_LOGSTASH_TYPE_NAME'] || 'fluentd'}"
include_timestamp "#{ENV['FLUENT_ELASTICSEARCH_INCLUDE_TIMESTAMP'] || 'false'}"
#time_key logtime
template_name "#{ENV['FLUENT_ELASTICSEARCH_TEMPLATE_NAME'] || use_nil}"
template_file "#{ENV['FLUENT_ELASTICSEARCH_TEMPLATE_FILE'] || use_nil}"
template_overwrite "#{ENV['FLUENT_ELASTICSEARCH_TEMPLATE_OVERWRITE'] || use_default}"
sniffer_class_name "#{ENV['FLUENT_SNIFFER_CLASS_NAME'] || 'Fluent::Plugin::ElasticsearchSimpleSniffer'}"
request_timeout "#{ENV['FLUENT_ELASTICSEARCH_REQUEST_TIMEOUT'] || '5s'}"
suppress_type_name "#{ENV['FLUENT_ELASTICSEARCH_SUPPRESS_TYPE_NAME'] || 'true'}"
enable_ilm "#{ENV['FLUENT_ELASTICSEARCH_ENABLE_ILM'] || 'false'}"
ilm_policy_id "#{ENV['FLUENT_ELASTICSEARCH_ILM_POLICY_ID'] || use_default}"
ilm_policy "#{ENV['FLUENT_ELASTICSEARCH_ILM_POLICY'] || use_default}"
ilm_policy_overwrite "#{ENV['FLUENT_ELASTICSEARCH_ILM_POLICY_OVERWRITE'] || 'false'}"

flush_thread_count "#{ENV['FLUENT_ELASTICSEARCH_BUFFER_FLUSH_THREAD_COUNT'] || '8'}"
flush_interval "#{ENV['FLUENT_ELASTICSEARCH_BUFFER_FLUSH_INTERVAL'] || '5s'}"
chunk_limit_size "#{ENV['FLUENT_ELASTICSEARCH_BUFFER_CHUNK_LIMIT_SIZE'] || '10M'}"
queue_limit_length "#{ENV['FLUENT_ELASTICSEARCH_BUFFER_QUEUE_LIMIT_LENGTH'] || '32'}"
retry_max_interval "#{ENV['FLUENT_ELASTICSEARCH_BUFFER_RETRY_MAX_INTERVAL'] || '30'}"
retry_forever true


kubernetes.conf:

<label @FLUENT_LOG>
<match fluent.**>
@type null
@id ignore_fluent_logs

@id fluentd-containers.log @type tail path /var/log/containers/*.log pos_file /var/log/es-containers.log.pos exclude_path /var/log/containers/fluentd-es*.log tag raw.kubernetes.* read_from_head true @type multi_format format json time_key time time_format %Y-%m-%dT%H:%M:%S.%NZ format /^(?.+?) (?stdout|stderr) [^ ]* (?.*)$/ #format /^(?[^ ]+) (?stdout|stderr) [^ ]* (?.*)$/ time_key time time_format %Y-%m-%dT%H:%M:%S.%N%:z #time_format %Y-%m-%dT%H:%M:%S.%L%z @id raw.kubernetes @type detect_exceptions remove_tag_prefix raw message log stream stream multiline_flush_interval 5 max_bytes 500000 max_lines 1000 @type record_transformer logtime ${time} @id filter_concat @type concat key message multiline_end_regexp /\n$/ separator "" @id filter_kubernetes_metadata @type kubernetes_metadata @id filter_parser @type parser key_name log reserve_data true remove_key_name_field true @type multi_format format json format none @type grep key message pattern /.*\/healthz/ @type tail @id in_tail_minion path /var/log/salt/minion pos_file /var/log/fluentd-salt.pos tag salt @type regexp expression /^(?[^ ]* [^ ,]*)[^\[]*\[[^\]]*\]\[(?[^ \]]*) *\] (?.*)$/ time_format %Y-%m-%d %H:%M:%S @type tail @id in_tail_startupscript path /var/log/startupscript.log pos_file /var/log/fluentd-startupscript.log.pos tag startupscript @type syslog @type tail @id in_tail_docker path /var/log/docker.log pos_file /var/log/fluentd-docker.log.pos tag docker @type regexp expression /^time="(?[^)]*)" level=(?[^ ]*) msg="(?[^"]*)"( err="(?[^"]*)")?( statusCode=($\d+))?/ @type tail @id in_tail_etcd path /var/log/etcd.log pos_file /var/log/fluentd-etcd.log.pos tag etcd @type none @type tail @id in_tail_kubelet multiline_flush_interval 5s path /var/log/kubelet.log pos_file /var/log/fluentd-kubelet.log.pos tag kubelet @type kubernetes @type tail @id in_tail_kube_proxy multiline_flush_interval 5s path /var/log/kube-proxy.log pos_file /var/log/fluentd-kube-proxy.log.pos tag kube-proxy @type kubernetes @type tail @id in_tail_kube_apiserver multiline_flush_interval 5s path /var/log/kube-apiserver.log pos_file /var/log/fluentd-kube-apiserver.log.pos tag kube-apiserver @type kubernetes @type tail @id in_tail_kube_controller_manager multiline_flush_interval 5s path /var/log/kube-controller-manager.log pos_file /var/log/fluentd-kube-controller-manager.log.pos tag kube-controller-manager @type kubernetes @type tail @id in_tail_kube_scheduler multiline_flush_interval 5s path /var/log/kube-scheduler.log pos_file /var/log/fluentd-kube-scheduler.log.pos tag kube-scheduler @type kubernetes @type tail @id in_tail_rescheduler multiline_flush_interval 5s path /var/log/rescheduler.log pos_file /var/log/fluentd-rescheduler.log.pos tag rescheduler @type kubernetes @type tail @id in_tail_glbc multiline_flush_interval 5s path /var/log/glbc.log pos_file /var/log/fluentd-glbc.log.pos tag glbc @type kubernetes @type tail @id in_tail_cluster_autoscaler multiline_flush_interval 5s path /var/log/cluster-autoscaler.log pos_file /var/log/fluentd-cluster-autoscaler.log.pos tag cluster-autoscaler @type kubernetes @type tail @id in_tail_kube_apiserver_audit multiline_flush_interval 5s path /var/log/kubernetes/kube-apiserver-audit.log pos_file /var/log/kube-apiserver-audit.log.pos tag kube-apiserver-audit @type multiline format_firstline /^\S+\s+AUDIT:/ format1 /^(?\S+) AUDIT:(?: (?:id="(?(?:[^"\\]|\\.)*)"|ip="(?(?:[^"\\]|\\.)*)"|method="(?(?:[^"\\]|\\.)*)"|user="(?(?:[^"\\]|\\.)*)"|groups="(?(?:[^"\\]|\\.)*)"|as="(?(?:[^"\\]|\\.)*)"|asgroups="(?(?:[^"\\]|\\.)*)"|namespace="(?(?:[^"\\]|\\.)*)"|uri="(?(?:[^"\\]|\\.)*)"|response="(?(?:[^"\\]|\\.)*)"|\w+="(?:[^"\\]|\\.)*"))*/ time_format %Y-%m-%dT%T.%L%Z

Describe the logs of Fluentd

No response

Environment

- Fluentd version:
- TD Agent version:
- Fluent Package version:
- Docker image (tag):
- Operating system:
- Kernel version:

[arthas3014]

the time in node container file is:
2024-05-29T15:10:06.516919109+08:00 stderr F panic: logger sync failed: sync /dev/stderr: invalid argument
but the parse out the custom time is:
"logtime": "2024-05-29 07:10:06 +0000"
and the es timetamp is : 2024-05-29T07:10:26.326419109+00:00