https://github.com/fluent/fluent-plugin-s3
With fluentd v0.14 and fluent-plugin-s3 v1.0.0, use new buffer configuration to dynamic parameters.
<match pattern>
@type s3
aws_key_id YOUR_AWS_KEY_ID
aws_sec_key YOUR_AWS_SECRET_KEY
s3_bucket YOUR_S3_BUCKET_NAME
s3_region ap-northeast-1
path logs/${tag}/%Y/%m/%d/
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 tag,time>
@type file
path /var/log/fluent/s3
timekey 3600 # 1 hour partition
timekey_wait 10m
timekey_use_utc true # use utc
</buffer>
<format>
@type json
</format>
</match>
https://devtron.ai/blog/pushing-k8s-cluster-logs-s3-bucket-fluentd/
general.conf: |
# Prevent fluentd from handling records containing its own logs. Otherwise
# it can lead to an infinite loop, when error in sending one message generates
# another message which also fails to be sent and so on.
<match fluentd.**>
@type null
</match>
# Used for health checking
<source>
@type http
port 9880
bind 0.0.0.0
</source>
# Emits internal metrics to every minute, and also exposes them on port
# 24220. Useful for determining if an output plugin is retryring/erroring,
# or determining the buffer queue length.
<source>
@type monitor_agent
bind 0.0.0.0
port 24220
tag fluentd.monitor.metrics
</source>
system.conf: |-
<system>
root_dir /tmp/fluentd-buffers/
</system>
forward-input.conf: |
<source>
@type forward
port 24224
bind 0.0.0.0
</source>
output.conf: |
<filter kube.**>
@type record_transformer
enable_ruby
<record>
kubernetes_tag ${"%s" % [record["kubernetes"]["labels"]["app"] || record["kubernetes"]["labels"]["k8s-app"] || record["kubernetes"]["labels"]["name"] || "unspecified-app-label"]}
</record>
</filter>
<match kube.**>
@type rewrite_tag_filter
<rule>
key kubernetes_tag
pattern ^(.+)$
tag $1
</rule>
</match>
<match **>
@type s3
s3_bucket <k8s-logs-bucket>
s3_region <ap-southeast-1>
s3_object_key_format "${tag}/%{time_slice}-events_%{index}.%{file_extension}"
time_slice_format %Y/%m/%d/%H
time_slice_wait 10m
path test-logs
# 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 tag,time>
@type file
flush_mode interval
flush_interval 30s
path /var/log/fluent/s3
timekey 300 # 1 hour partition
timekey_wait 1m
timekey_use_utc true # use utc
chunk_limit_size 100m
</buffer>
<format>
@type json
</format>
</match>