kafka.conf

# The following label is to ignore Fluentd warning events.
<label @FLUENT_LOG>
    <match fluent.**>
        @type null
        @id ignore_fluent_logs
    </match>
</label>

<source>
    @type tail
    @id in_tail_kafka
    path /var/log/kafka.log
    pos_file /var/log/fluentd-kafka.log.pos
    read_from_head true
    path_key tailed_path
    tag oci.kafka
    <parse>
        @type json
    </parse>
</source>

# Add below prometheus config block only when you need output plugin metrics.
<source>
    @type prometheus
    bind 0.0.0.0
    port 24231
    metrics_path /metrics
</source>

<filter oci.**>
    @type record_transformer
    enable_ruby true
    <record>
        oci_la_global_metadata ${{<key1>: <value1>, <key2>:<value2>}}
        oci_la_entity_id <LOGGING_ANALYTICS_ENTITY_OCID>              # If same across sources. Else keep this in individual filters
        oci_la_entity_type <LOGGING_ANALYTICS_ENTITY_TYPE>            # If same across sources. Else keep this in individual filters
    </record>
</filter>

<filter oci.kafka>
    @type record_transformer
    enable_ruby true
    <record>
        oci_la_metadata ${{<key1>: <value1>, <key2>:<value2>}}
        oci_la_log_source_name <LOGGING_ANALYTICS_SOURCENAME>
        oci_la_log_group_id <LOGGING_ANALYTICS_LOGGROUP_OCID>
        oci_la_log_path "${record['tailed_path']}"
        message ${record["log"]}                            # Will assign the 'log' key value from json wrapped message to 'message' field
        tag ${tag}
    </record>
</filter>

<match oci.**>
    @type oci-logging-analytics
    namespace                   <YOUR_OCI_TENANCY_NAMESPACE>
    # Auth config file details
    config_file_location        ~/.oci/config
    profile_name                DEFAULT
    # Buffer Configuration
    <buffer>
          @type	file
          path	/var/log
          retry_forever	true
          disable_chunk_backup	true
    </buffer>
</match>