Skip to content

feat: add raw format and rawLogKey support for Fluent Bit Kafka #2000

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Vaibhav-C-S wants to merge 6 commits into
base: master
Choose a base branch
from
Open

feat: add raw format and rawLogKey support for Fluent Bit Kafka #2000

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
727d7aa
fluentbit(kafka): add raw format and rawLogKey support
Vaibhav-C-S Jun 20, 2026
2937c43
fluentbit(kafka): add raw format and rawLogKey support
Vaibhav-C-S Jun 20, 2026
82217a4
Merge branch 'master' into fluentbit-kafka-raw-log-key
Vaibhav-C-S Jun 22, 2026
279964f
Potential fix for pull request finding
Vaibhav-C-S Jun 22, 2026
9ffafba
fluentbit(kafka): add raw format support
Vaibhav-C-S Jun 22, 2026
f128444
fluentbit(kafka): add raw format support
Vaibhav-C-S Jun 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion apis/fluentbit/v1alpha2/plugins/output/kafka_types.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,8 +12,10 @@ import (
// Kafka output plugin allows to ingest your records into an Apache Kafka service. <br />
// **For full documentation, refer to https://docs.fluentbit.io/manual/pipeline/outputs/kafka**
type Kafka struct {
// Specify data format, options available: json, msgpack.
// Specify data format, options available: json, msgpack, raw.
Format string `json:"format,omitempty"`
// When using the raw format, the value of the record field specified by RawLogKey is sent to Kafka as the payload.
RawLogKey string `json:"rawLogKey,omitempty"`
// Optional key to store the message
MessageKey string `json:"messageKey,omitempty"`
// If set, the value of Message_Key_Field in the record will indicate the message key.
Expand Down Expand Up @@ -60,6 +62,9 @@ func (k *Kafka) Params(_ plugins.SecretLoader) (*params.KVs, error) {
if k.Format != "" {
kvs.Insert("Format", k.Format)
}
if k.RawLogKey != "" {
kvs.Insert("Raw_Log_Key", k.RawLogKey)
}
Comment on lines +65 to +67
Comment on lines +65 to +67
if k.MessageKey != "" {
kvs.Insert("Message_Key", k.MessageKey)
}
Expand Down
7 changes: 6 additions & 1 deletion charts/fluent-operator-fluent-bit-crds/templates/fluentbit.fluent.io_clusteroutputs.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2258,7 +2258,8 @@ spec:
So in Topics only a default topic needs to be configured
type: boolean
format:
description: 'Specify data format, options available: json, msgpack.'
description: 'Specify data format, options available: json, msgpack,
raw.'
type: string
messageKey:
description: Optional key to store the message
Expand All @@ -2277,6 +2278,10 @@ spec:
Setting the queue_full_retries value to 0 set's an unlimited number of retries.
format: int64
type: integer
rawLogKey:
description: When using the raw format, the value of the record
field specified by RawLogKey is sent to Kafka as the payload.
type: string
rdkafka:
additionalProperties:
type: string
Expand Down
7 changes: 6 additions & 1 deletion charts/fluent-operator-fluent-bit-crds/templates/fluentbit.fluent.io_outputs.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2258,7 +2258,8 @@ spec:
So in Topics only a default topic needs to be configured
type: boolean
format:
description: 'Specify data format, options available: json, msgpack.'
description: 'Specify data format, options available: json, msgpack,
raw.'
type: string
messageKey:
description: Optional key to store the message
Expand All @@ -2277,6 +2278,10 @@ spec:
Setting the queue_full_retries value to 0 set's an unlimited number of retries.
format: int64
type: integer
rawLogKey:
description: When using the raw format, the value of the record
field specified by RawLogKey is sent to Kafka as the payload.
type: string
rdkafka:
additionalProperties:
type: string
Expand Down
7 changes: 6 additions & 1 deletion charts/fluent-operator/crds/fluentbit.fluent.io_clusteroutputs.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2256,7 +2256,8 @@ spec:
So in Topics only a default topic needs to be configured
type: boolean
format:
description: 'Specify data format, options available: json, msgpack.'
description: 'Specify data format, options available: json, msgpack,
raw.'
type: string
messageKey:
description: Optional key to store the message
Expand All @@ -2275,6 +2276,10 @@ spec:
Setting the queue_full_retries value to 0 set's an unlimited number of retries.
format: int64
type: integer
rawLogKey:
description: When using the raw format, the value of the record
field specified by RawLogKey is sent to Kafka as the payload.
type: string
rdkafka:
additionalProperties:
type: string
Expand Down
7 changes: 6 additions & 1 deletion charts/fluent-operator/crds/fluentbit.fluent.io_outputs.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2256,7 +2256,8 @@ spec:
So in Topics only a default topic needs to be configured
type: boolean
format:
description: 'Specify data format, options available: json, msgpack.'
description: 'Specify data format, options available: json, msgpack,
raw.'
type: string
messageKey:
description: Optional key to store the message
Expand All @@ -2275,6 +2276,10 @@ spec:
Setting the queue_full_retries value to 0 set's an unlimited number of retries.
format: int64
type: integer
rawLogKey:
description: When using the raw format, the value of the record
field specified by RawLogKey is sent to Kafka as the payload.
type: string
rdkafka:
additionalProperties:
type: string
Expand Down
7 changes: 6 additions & 1 deletion config/crd/bases/fluentbit.fluent.io_clusteroutputs.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2257,7 +2257,8 @@ spec:
So in Topics only a default topic needs to be configured
type: boolean
format:
description: 'Specify data format, options available: json, msgpack.'
description: 'Specify data format, options available: json, msgpack,
raw.'
type: string
messageKey:
description: Optional key to store the message
Expand All @@ -2276,6 +2277,10 @@ spec:
Setting the queue_full_retries value to 0 set's an unlimited number of retries.
format: int64
type: integer
rawLogKey:
description: When using the raw format, the value of the record
field specified by RawLogKey is sent to Kafka as the payload.
type: string
rdkafka:
additionalProperties:
type: string
Expand Down
7 changes: 6 additions & 1 deletion config/crd/bases/fluentbit.fluent.io_outputs.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2257,7 +2257,8 @@ spec:
So in Topics only a default topic needs to be configured
type: boolean
format:
description: 'Specify data format, options available: json, msgpack.'
description: 'Specify data format, options available: json, msgpack,
raw.'
type: string
messageKey:
description: Optional key to store the message
Expand All @@ -2276,6 +2277,10 @@ spec:
Setting the queue_full_retries value to 0 set's an unlimited number of retries.
format: int64
type: integer
rawLogKey:
description: When using the raw format, the value of the record
field specified by RawLogKey is sent to Kafka as the payload.
type: string
rdkafka:
additionalProperties:
type: string
Expand Down
3 changes: 2 additions & 1 deletion docs/plugins/fluentbit/output/kafka.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,8 @@ Kafka output plugin allows to ingest your records into an Apache Kafka service.

| Field | Description | Scheme |
| ----- | ----------- | ------ |
| format | Specify data format, options available: json, msgpack. | string |
| format | Specify data format, options available: json, msgpack, raw. | string |
| rawLogKey | When using the raw format, the value of the record field specified by RawLogKey is sent to Kafka as the payload. | string |
| messageKey | Optional key to store the message | string |
| messageKeyField | If set, the value of Message_Key_Field in the record will indicate the message key. If not set nor found in the record, Message_Key will be used (if set). | string |
| timestampKey | Set the key to store the record timestamp | string |
Expand Down
14 changes: 12 additions & 2 deletions manifests/setup/setup.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6457,7 +6457,8 @@ spec:
So in Topics only a default topic needs to be configured
type: boolean
format:
description: 'Specify data format, options available: json, msgpack.'
description: 'Specify data format, options available: json, msgpack,
raw.'
type: string
messageKey:
description: Optional key to store the message
Expand All @@ -6476,6 +6477,10 @@ spec:
Setting the queue_full_retries value to 0 set's an unlimited number of retries.
format: int64
type: integer
rawLogKey:
description: When using the raw format, the value of the record
field specified by RawLogKey is sent to Kafka as the payload.
type: string
rdkafka:
additionalProperties:
type: string
Expand Down Expand Up @@ -37123,7 +37128,8 @@ spec:
So in Topics only a default topic needs to be configured
type: boolean
format:
description: 'Specify data format, options available: json, msgpack.'
description: 'Specify data format, options available: json, msgpack,
raw.'
type: string
messageKey:
description: Optional key to store the message
Expand All @@ -37142,6 +37148,10 @@ spec:
Setting the queue_full_retries value to 0 set's an unlimited number of retries.
format: int64
type: integer
rawLogKey:
description: When using the raw format, the value of the record
field specified by RawLogKey is sent to Kafka as the payload.
type: string
rdkafka:
additionalProperties:
type: string
Expand Down