MessagePack is an open-source binary serialization format designed to serialize and deserialize data compactly and quickly. In Fluent Bit, it is the internal representation for events held outside visible record processing, reducing the bytes needed to store a message. Console output can therefore show an internal event representation rather than JSON. section 2.1.3 section 1.2.1
An event still has recognizable operational parts: a tag, timestamp, and record payload. The record may be a key-value structure or text for logs; metadata can identify metrics or traces. MessagePack preserves event data efficiently while plugins route, transform, or emit it. section 1.2.1 section 2.1.3
Use MessagePack where Fluent Bit needs compact event exchange or storage:
- Hold records internally when they are not being processed as visible event data. section 1.2.1
- Transmit prescribed forward-protocol payloads between Fluent Bit and Fluentd instances or across connected Fluent Bit deployments. section 3.6.4
- Receive a plugin flush chunk through C pointers, decode its MessagePack records, and convert values into appropriate Go types before output. section 10.6.4
The forward protocol uses MessagePack without requiring a schema or generated payload code. That flexibility contrasts with gRPC use in OpenTelemetry, but the payload cannot be compressed as efficiently as gRPC. Unlike the HTTP plugin, forward keeps the tag and timestamp with the core message payload. section 3.6.4
Decode an output chunk
An output plugin's flush callback receives a data pointer and length, creates a decoder, iterates through records, interprets timestamps, and prepares each decoded record for database insertion. This is the boundary where opaque MessagePack data becomes application-usable values. section 10.6.4
Listing 10.7 Implementation of the flush callback: out.go
//export FLBPluginFlushCtx
func FLBPluginFlushCtx(ctx, data unsafe.Pointer,
length C.int,
tag *C.char) int { #1
params := NewSqlParams()
myContext := output.FLBPluginGetContext(ctx) #2
if myContext != nil {
strContext := myContext.(*string) #3
params = JSONToParams(*strContext, PluginName)
} else {
params = envToParams(PluginName) #4
if params == nil {
return output.FLB_ERROR
}
}
if params == nil {
return output.FLB_ERROR
}
dec := output.NewDecoder(data, int(length))
count := 0
for { #5
ret, ts, record := output.GetRecord(dec)
if ret != 0 {
break
}
var timestamp time.Time #6
switch t := ts.(type) {
case output.FLBTime:
timestamp = ts.(output.FLBTime).Time
case uint64:
timestamp = time.Unix(int64(t), 0)
default:
timestamp = time.Now()
}
count++
insertErr := execInsert(params, record) #7
if insertErr != nil {
return output.FLB_ERROR
}
}
return output.FLB_OK #8
}
The flush callback shows the decoder creation and record-by-record loop used to turn a MessagePack chunk into values an output plugin can send onward. section 10.6.4
