fluent.conf.erb

<% target = (dockerfile.split("/").last.split("-").last) %>

# AUTOMATICALLY GENERATED
# DO NOT EDIT THIS FILE DIRECTLY, USE /templates/conf/fluent.conf.erb

@include "#{ENV['FLUENTD_SYSTEMD_CONF'] || 'systemd'}.conf"
@include "#{ENV['FLUENTD_PROMETHEUS_CONF'] || 'prometheus'}.conf"
@include kubernetes.conf
@include conf.d/*.conf

<% case target when "elasticsearch6","elasticsearch7" %>
<match **>
   @type elasticsearch
   @id out_es
   @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'}"
<% if target == "elasticsearch7" %>
   ssl_version "#{ENV['FLUENT_ELASTICSEARCH_SSL_VERSION'] || 'TLSv1_2'}"
<% else %>
   ssl_version "#{ENV['FLUENT_ELASTICSEARCH_SSL_VERSION'] || 'TLSv1'}"
<% end %>
   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_LOGSTASH_PREFIX'] || 'logstash'}"
   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'}"
   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'}"
   application_name "#{ENV['FLUENT_ELASTICSEARCH_APPLICATION_NAME'] || use_default}"
<% if target == "elasticsearch7" %>
   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'}"
<% end %>
   <buffer>
     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'] || '2M'}"
     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
   </buffer>
</match>
<%  when "opensearch"%>
<match **>
  @type opensearch
  @id out_os
  @log_level info
  include_tag_key true
  host "#{ENV['FLUENT_OPENSEARCH_HOST']}"
  port "#{ENV['FLUENT_OPENSEARCH_PORT']}"
  path "#{ENV['FLUENT_OPENSEARCH_PATH']}"
  scheme "#{ENV['FLUENT_OPENSEARCH_SCHEME'] || 'http'}"
  ssl_verify "#{ENV['FLUENT_OPENSEARCH_SSL_VERIFY'] || 'true'}"
  ssl_version "#{ENV['FLUENT_OPENSEARCH_SSL_VERSION'] || 'TLSv1_2'}"
  ca_file "#{ENV['FLUENT_OPENSEARCH_CA_PATH']}"
  user "#{ENV['FLUENT_OPENSEARCH_USER']}"
  password "#{ENV['FLUENT_OPENSEARCH_PASSWORD']}"
  client_cert "#{ENV['FLUENT_OPENSEARCH_CLIENT_CERT']}"
  client_key "#{ENV['FLUENT_OPENSEARCH_CLIENT_KEY']}"
  client_key_pass "#{ENV['FLUENT_OPENSEARCH_CLIENT_KEY_PASS']}"
  index_name "#{ENV['FLUENT_OPENSEARCH_INDEX_NAME'] || 'fluentd'}"
</match>
<%  when "logentries"%>
<match **>
   @type logentries
   @id out_logentries
   use_json true
   tag_access_log stdout
   tag_error_log stderr
   config_path /etc/logentries/tokens.yaml
</match>
<%  when "loggly"%>
<match **>
   @type loggly
   @id out_loggly
   @log_level info
   loggly_url "https://logs-01.loggly.com/bulk/#{ENV['LOGGLY_TOKEN']}/tag/#{ENV['LOGGLY_TAGS'] || 'fluentd'}/bulk"
</match>
<%  when "cloudwatch"%>
<match **>
  @type cloudwatch_logs
  @id out_cloudwatch_logs
  log_group_name "#{ENV['LOG_GROUP_NAME']}"
  auto_create_stream true
  use_tag_as_stream true
  retention_in_days "#{ENV['RETENTION_IN_DAYS'] || 'nil'}"
  json_handler yajl # To avoid UndefinedConversionError
  log_rejected_request "#{ENV['LOG_REJECTED_REQUEST']}" # Log rejected request for missing parts
</match>
<%  when "s3"%>
<match **>
  # docs: https://docs.fluentd.org/v0.12/articles/out_s3
  # note: this configuration relies on the nodes have an IAM instance profile with access to your S3 bucket
  @type s3
  @id out_s3
  @log_level info
  s3_bucket "#{ENV['S3_BUCKET_NAME']}"
  s3_region "#{ENV['S3_BUCKET_REGION']}"
  s3_object_key_format "#{ENV['S3_OBJECT_KEY_FORMAT'] || '%{path}%Y/%m/%d/cluster-log-%{index}.%{file_extension}'}"
  <inject>
    time_key time
    tag_key tag
    localtime false
  </inject>
  <buffer>
    @type file
    path /var/log/fluentd-buffers/s3.buffer
    timekey "#{ENV['S3_TIMEKEY'] || '3600'}"
    timekey_use_utc true
    chunk_limit_size "#{ENV['S3_CHUNK_LIMIT_SIZE'] || '256m'}"
  </buffer>
</match>
<%  when "gcs"%>
<match **>
  # docs: https://github.com/daichirata/fluent-plugin-gcs
  # this configuration relies on the nodes having permission to write on your gs bucket
  @type gcs
  @id out_gcs
  project "#{ENV['GCS_BUCKET_PROJECT']}"
  bucket "#{ENV['GCS_BUCKET_NAME']}"
  object_key_format %{path}%{time_slice}/%{hostname}/%{index}.%{file_extension}
  path logs/
  buffer_path /var/log/fluentd-buffers/gcs.buffer
  time_slice_format %Y/%m/%d
  time_slice_wait 10m
  utc
</match>
<% when "stackdriver"%>
<match **>
  @type google_cloud
  @id out_google_cloud

  # Set the buffer type to file to improve the reliability and reduce the memory consumption
  buffer_type file
  buffer_path /var/log/fluentd-buffers/stackdriver.buffer
  # Set queue_full action to block because we want to pause gracefully
  # in case of the off-the-limits load instead of throwing an exception
  buffer_queue_full_action block
  # Set the chunk limit conservatively to avoid exceeding the GCL limit
  # of 10MiB per write request.
  buffer_chunk_limit 2M
  # Cap the combined memory usage of this buffer and the one below to
  # 2MiB/chunk * (6 + 2) chunks = 16 MiB
  buffer_queue_limit 6
  # Never wait more than 5 seconds before flushing logs in the non-error case.
  flush_interval 5s
  # Never wait longer than 30 seconds between retries.
  max_retry_wait 30
  # Disable the limit on the number of retries (retry forever).
  # disable_retry_limit
  # Use multiple threads for processing.
  num_threads 2
</match>
<% when "azureblob"%>
<match **>
  @type azure-storage-append-blob
  @id out_azure_storage_append_blob

  azure_cloud                     "#{ENV['AZUREBLOB_AZURE_CLOUD']}"
  azure_storage_account           "#{ENV['AZUREBLOB_ACCOUNT_NAME']}"
  azure_storage_access_key        "#{ENV['AZUREBLOB_ACCOUNT_KEY']}"
  azure_storage_connection_string "#{ENV['AZUREBLOB_CONNECTION_STRING']}"
  azure_storage_sas_token         "#{ENV['AZUREBLOB_SAS_TOKEN']}"
  azure_container                 "#{ENV['AZUREBLOB_CONTAINER']}"
  azure_imds_api_version          "#{ENV['AZUREBLOB_IMDS_API_VERSION']}"
  azure_token_refresh_interval    "#{ENV['AZUREBLOB_TOKEN_REFRESH_INTERVAL']}"
  auto_create_container           true
  path                            "#{ENV['AZUREBLOB_LOG_PATH']}"
  azure_object_key_format         %{path}%{time_slice}_%{index}.log
  time_slice_format               %Y%m%d-%H
  # if you want to use %{tag} or %Y/%m/%d/ like syntax in path / azure_blob_name_format,
  # need to specify tag for %{tag} and time for %Y/%m/%d in <buffer> argument.
  <buffer>
    @type file
    path /var/log/fluent/azurestorageappendblob
    timekey 60 # 1 minute
    timekey_wait 60
    timekey_use_utc true # use utc
    chunk_limit_size 256m
  </buffer>
</match>
<% when "graylog"%>
# The gelf plugin assumes input in utf-8
<filter **>
  @type record_modifier
  @id graylog_encode_utf8
  char_encoding utf-8
</filter>

<match **>
  @type gelf
  @id out_graylog
  @log_level info
  include_tag_key true
  host "#{ENV['FLUENT_GRAYLOG_HOST']}"
  port "#{ENV['FLUENT_GRAYLOG_PORT']}"
  protocol "#{ENV['FLUENT_GRAYLOG_PROTOCOL'] || 'udp'}"
  <buffer>
    flush_thread_count 8
    flush_interval 5s
    chunk_limit_size 8M
    queue_limit_length 512
    retry_max_interval 30
    retry_forever true
  </buffer>
</match>
<% when "logzio"%>
<match **>
  @type logzio_buffered
  @id out_logzio
  endpoint_url "https://listener.logz.io:8071?token=#{ENV['LOGZIO_TOKEN']}&type=#{ENV['LOGZIO_LOGTYPE']}"
  output_include_time true
  output_include_tags true
  <buffer>
    # Set the buffer type to file to improve the reliability and reduce the memory consumption
    @type file
    path /var/log/fluentd-buffers/stackdriver.buffer
    # Set queue_full action to block because we want to pause gracefully
    # in case of the off-the-limits load instead of throwing an exception
    overflow_action block
    # Set the chunk limit conservatively to avoid exceeding the GCL limit
    # of 10MiB per write request.
    chunk_limit_size 2M
    # Cap the combined memory usage of this buffer and the one below to
    # 2MiB/chunk * (6 + 2) chunks = 16 MiB
    queue_limit_length 6
    # Never wait more than 5 seconds before flushing logs in the non-error case.
    flush_interval 5s
    # Never wait longer than 30 seconds between retries.
    retry_max_interval 30
    # Disable the limit on the number of retries (retry forever).
    retry_forever true
    # Use multiple threads for processing.
    flush_thread_count 2
  </buffer>
</match>

<% when "papertrail"%>

## Capture audit logs
#<match kube-apiserver-audit>
#  @type papertrail
#
#  papertrail_host "#{ENV['FLUENT_PAPERTRAIL_AUDIT_HOST']}"
#  papertrail_port "#{ENV['FLUENT_PAPERTRAIL_AUDIT_PORT']}"
#</match>

<match **>
  @type papertrail
  @id out_papertrail

  papertrail_host "#{ENV['FLUENT_PAPERTRAIL_HOST']}"
  papertrail_port "#{ENV['FLUENT_PAPERTRAIL_PORT']}"

</match>
<% when "kafka" %>
<match **>
  @type kafka_buffered
  @id out_kafka

  brokers "#{ENV['FLUENT_KAFKA_BROKERS']}"

  default_topic "#{ENV['FLUENT_KAFKA_DEFAULT_TOPIC'] || nil}"
  default_partition_key "#{ENV['FLUENT_KAFKA_DEFAULT_PARTITION_KEY'] || nil}"
  default_message_key "#{ENV['FLUENT_KAFKA_DEFAULT_MESSAGE_KEY'] || nil}"
  output_data_type "#{ENV['FLUENT_KAFKA_OUTPUT_DATA_TYPE'] || 'json'}"
  output_include_tag "#{ENV['FLUENT_KAFKA_OUTPUT_INCLUDE_TAG'] || false}"
  output_include_time "#{ENV['FLUENT_KAFKA_OUTPUT_INCLUDE_TIME'] || false}"
  exclude_topic_key "#{ENV['FLUENT_KAFKA_EXCLUDE_TOPIC_KEY'] || false}"
  exclude_partition_key "#{ENV['FLUENT_KAFKA_EXCLUDE_PARTITION_KEY'] || false}"
  get_kafka_client_log "#{ENV['FLUENT_KAFKA_GET_KAFKA_CLIENT_LOG'] || false}"

  # ruby-kafka producer options
  max_send_retries "#{ENV['FLUENT_KAFKA_MAX_SEND_RETRIES'] || 1}"
  required_acks "#{ENV['FLUENT_KAFKA_REQUIRED_ACKS'] || -1}"
  ack_timeout "#{ENV['FLUENT_KAFKA_ACK_TIMEOUT'] || nil}"
  compression_codec "#{ENV['FLUENT_KAFKA_COMPRESSION_CODEC'] || nil}"
  max_send_limit_bytes "#{ENV['FLUENT_KAFKA_MAX_SEND_LIMIT_BYTES'] || nil}"
  discard_kafka_delivery_failed "#{ENV['FLUENT_KAFKA_DISCARD_KAFKA_DELIVERY_FAILED'] || false}"
</match>
<% when "kafka2" %>
<match **>
  @type kafka2
  @id out_kafka2

  brokers "#{ENV['FLUENT_KAFKA2_BROKERS']}"

  default_topic "#{ENV['FLUENT_KAFKA2_DEFAULT_TOPIC'] || nil}"
  default_partition_key "#{ENV['FLUENT_KAFKA2_DEFAULT_PARTITION_KEY'] || nil}"
  default_message_key "#{ENV['FLUENT_KAFKA2_DEFAULT_MESSAGE_KEY'] || nil}"
  exclude_topic_key "#{ENV['FLUENT_KAFKA2_EXCLUDE_TOPIC_KEY'] || false}"
  exclude_partition_key "#{ENV['FLUENT_KAFKA2_EXCLUDE_PARTITION_KEY'] || false}"
  get_kafka_client_log "#{ENV['FLUENT_KAFKA2_GET_KAFKA_CLIENT_LOG'] || false}"
  <format>
    @type "#{ENV['FLUENT_KAFKA2_OUTPUT_FORMAT_TYPE'] || 'json'}"
  </format>
  <inject>
    tag_key "#{ENV['FLUENT_KAFKA2_OUTPUT_TAG_KEY'] || use_nil}"
    time_key "#{ENV['FLUENT_KAFKA2_OUTPUT_TIME_KEY'] || use_nil}"
  </inject>

  # ruby-kafka producer options
  max_send_retries "#{ENV['FLUENT_KAFKA2_MAX_SEND_RETRIES'] || 1}"
  required_acks "#{ENV['FLUENT_KAFKA2_REQUIRED_ACKS'] || -1}"
  ack_timeout "#{ENV['FLUENT_KAFKA2_ACK_TIMEOUT'] || nil}"
  compression_codec "#{ENV['FLUENT_KAFKA2_COMPRESSION_CODEC'] || nil}"
  max_send_limit_bytes "#{ENV['FLUENT_KAFKA2_MAX_SEND_LIMIT_BYTES'] || nil}"
  discard_kafka_delivery_failed "#{ENV['FLUENT_KAFKA2_DISCARD_KAFKA_DELIVERY_FAILED'] || false}"
</match>
<% when "splunkhec" %>
<match **>
  @type splunk_hec
  @id out_splunk_hec

  host "#{ENV['FLUENT_SPLUNK_HEC_HOST'] || nil}"
  port "#{ENV['FLUENT_SPLUNK_HEC_PORT'] || '8089'}"
  token "#{ENV['FLUENT_SPLUNK_HEC_TOKEN'] || nil}"

  # metadata parameter
  default_source "#{ENV['FLUENT_SPLUNK_HEC_SOURCE'] || 'fluentd-splunkhec'}"
  sourcetype "#{ENV['FLUENT_SPLUNK_HEC_SOURCE_TYPE'] || 'fluentd-splunkhec'}"

  use_fluentd_time "#{ENV['FLUENT_SPLUNK_HEC_USE_FLUENTD_TIME'] || true}"

  # ack parameter
  use_ack "#{ENV['FLUENT_SPLUNK_HEC_USE_ACK'] || false}"
  ack_interval "#{ENV['FLUENT_SPLUNK_HEC_ACK_INTERVAL'] || '1'}"
  ack_retry_limit "#{ENV['FLUENT_SPLUNK_HEC_ACK_RETRY_LIMIT'] || '3'}"

  # ssl parameter
  use_ssl "#{ENV['FLUENT_SPLUNK_HEC_USE_SSL'] || true}"
  ssl_verify "#{ENV['FLUENT_SPLUNK_HEC_SSL_VERIFY'] || true}"
  ca_file "#{ENV['FLUENT_SPLUNK_HEC_CA_FILE'] || '/etc/ssl/certs/ca-bundle.crt'}"

  # buffered output parameter
  flush_interval "#{ENV['FLUENT_SPLUNK_HEC_FLUSH_INTERVAL'] || '10s'}"

</match>

<% when "syslog" %>
<match **>
  @type remote_syslog
  @id out_kube_remote_syslog
  host "#{ENV['SYSLOG_HOST']}"
  port "#{ENV['SYSLOG_PORT']}"
  severity debug
  program fluentd
  hostname ${kubernetes_host}

  protocol "#{ENV['SYSLOG_PROTOCOL'] || 'tcp'}"
  tls "#{ENV['SYSLOG_TLS'] || 'false'}"
  ca_file "#{ENV['SYSLOG_CA_FILE'] || ''}"
  verify_mode "#{ENV['SYSLOG_VERIFY_MODE'] || ''}"
  packet_size 65535

  <buffer kubernetes_host>
  </buffer>

  <format>
    @type ltsv
  </format>
</match>

<% when "forward" %>
<match **>
  @type forward
  @id out_fwd
  @log_level info
  <server>
    host "#{ENV['FLUENT_FOWARD_HOST']}"
    port "#{ENV['FLUENT_FOWARD_PORT']}"
  </server>
</match>

<% when "kinesis" %>
<match **>
  @type kinesis_streams
  @id out_kinesis_streams
  region "#{ENV['FLUENT_KINESIS_STREAMS_REGION'] || nil}"
  stream_name "#{ENV['FLUENT_KINESIS_STREAMS_STREAM_NAME']}"
  include_time_key "#{ENV['FLUENT_KINESIS_STREAMS_INCLUDE_TIME_KEY'] || false}"
  <buffer>
    flush_interval 1
    chunk_limit_size "#{ENV['FLUENT_KINESIS_STREAMS_CHUNK_LIMIT_SIZE'] || '1m'}"
    flush_thread_interval 0.1
    flush_thread_burst_interval 0.01
    flush_thread_count 15
  </buffer>
</match>

<% end %>