Documentation
¶
Overview ¶
Package serde is kafui's pluggable message serialization/deserialization framework. It formalizes the previously ad-hoc Avro/Protobuf/MessagePack decode paths into a Registry of named Serdes with auto-detection and a UI-visible fallback.
Extension point (MSG-20) ¶
The Registry is the extension point. There is no plugin system (Go's `plugin` package requires identical toolchain/version and is Linux/macOS only, so it is not worth the complexity here). To add a custom serde, implement the Serde interface and register it in code — see BuildRegistry in config.go, which is the single place built-in and configured serdes are wired up. A custom serde added there is indistinguishable from a built-in one to the rest of the app.
Index ¶
- Constants
- func Decode(reg *Registry, chosen string, data []byte) (text, name string, fallback bool)
- func IsFallback(name string) bool
- func SchemaID(data []byte) (uint32, bool)
- func SelectSerde(configs []SerdeConfig, topic string, isKey bool) string
- func Validate(reg *Registry, chosen string, sample []byte) error
- type ConsumerOffsetsKeySerde
- type ConsumerOffsetsValueSerde
- type DecodeFunc
- type DescriptorProtobufSerde
- type DuplicateSerdeError
- type HexSerde
- type JSONSerde
- type MsgpackSerde
- type NullSerde
- type RawProtobufSerde
- type Registry
- type SchemaRegistrySerde
- type Serde
- type SerdeConfig
- type Serializer
- type StringSerde
- type UnknownSerdeError
Constants ¶
const ( NameString = "string" NameHex = "hex" // also serves as the bytes representation NameJSON = "json" // pretty-printed JSON NameNull = "null" // empty / absent payloads NameInt = "int" // 4-byte big-endian signed NameLong = "long" // 8-byte big-endian signed NameFloat = "float" // 4-byte big-endian IEEE-754 NameDouble = "double" // 8-byte big-endian IEEE-754 )
Built-in serde names.
const ( NameConsumerOffsetsKey = "consumer-offsets-key" NameConsumerOffsetsValue = "consumer-offsets-value" )
Internal-topic serde names. These decode the well-known binary formats Kafka uses for its internal topics. They are read-only (no Serialize).
const Auto = "auto"
Auto is the sentinel serde name meaning "auto-detect".
const NameMsgpack = "msgpack"
NameMsgpack is the MessagePack serde name.
const NameRawProtobuf = "protobuf-raw"
NameRawProtobuf is the schemaless wire-format protobuf serde name.
const NameSchemaRegistry = "schema-registry"
NameSchemaRegistry is the Confluent-wire-format schema-registry serde name. It covers Avro, JSON Schema and Protobuf payloads: all three use the same framing (magic byte 0x00 + 4-byte big-endian schema id), and the concrete decode is delegated to the injected DecodeFunc (which reuses kafds' existing schema-registry client / Avro cache).
Variables ¶
This section is empty.
Functions ¶
func Decode ¶
Decode renders data using the chosen serde (empty or "auto" = auto-detect). When the selected serde fails (or none is found) it falls back to string for valid UTF-8, otherwise hex, and marks the returned name with a " (fallback)" suffix. Decoding always yields some text, so no error is returned.
func IsFallback ¶
IsFallback reports whether a serde name (as returned by Decode) denotes a fallback rendering.
func SelectSerde ¶
func SelectSerde(configs []SerdeConfig, topic string, isKey bool) string
SelectSerde returns the name of the first configured serde bound to the given topic/target, or "" when none matches (the caller then auto-detects).
Types ¶
type ConsumerOffsetsKeySerde ¶
type ConsumerOffsetsKeySerde struct{}
ConsumerOffsetsKeySerde decodes __consumer_offsets record keys.
func (ConsumerOffsetsKeySerde) CanDeserialize ¶
func (s ConsumerOffsetsKeySerde) CanDeserialize(d []byte) bool
func (ConsumerOffsetsKeySerde) Deserialize ¶
func (s ConsumerOffsetsKeySerde) Deserialize(d []byte) (string, error)
func (ConsumerOffsetsKeySerde) Name ¶
func (ConsumerOffsetsKeySerde) Name() string
type ConsumerOffsetsValueSerde ¶
type ConsumerOffsetsValueSerde struct{}
ConsumerOffsetsValueSerde decodes __consumer_offsets offset-commit values.
func (ConsumerOffsetsValueSerde) CanDeserialize ¶
func (s ConsumerOffsetsValueSerde) CanDeserialize(d []byte) bool
func (ConsumerOffsetsValueSerde) Deserialize ¶
func (s ConsumerOffsetsValueSerde) Deserialize(d []byte) (string, error)
func (ConsumerOffsetsValueSerde) Name ¶
func (ConsumerOffsetsValueSerde) Name() string
type DecodeFunc ¶
DecodeFunc decodes a full Confluent-framed payload (including the magic byte and schema id) into human-readable bytes (typically JSON). kafds supplies the Avro-cache-backed implementation; a nil decoder makes the serde report an explanatory error so decoding falls back.
type DescriptorProtobufSerde ¶
type DescriptorProtobufSerde struct {
// contains filtered or unexported fields
}
DescriptorProtobufSerde decodes protobuf payloads against a message type loaded from a compiled FileDescriptorSet (`protoc --descriptor_set_out`).
func NewDescriptorProtobufSerde ¶
func NewDescriptorProtobufSerde(name, descPath, messageName string) (*DescriptorProtobufSerde, error)
NewDescriptorProtobufSerde loads descPath (a serialized FileDescriptorSet) and resolves messageName (fully-qualified, e.g. "pkg.MyMessage").
func (*DescriptorProtobufSerde) CanDeserialize ¶
func (s *DescriptorProtobufSerde) CanDeserialize(d []byte) bool
func (*DescriptorProtobufSerde) Deserialize ¶
func (s *DescriptorProtobufSerde) Deserialize(d []byte) (string, error)
func (*DescriptorProtobufSerde) Name ¶
func (s *DescriptorProtobufSerde) Name() string
type DuplicateSerdeError ¶
type DuplicateSerdeError struct {
Name string
}
DuplicateSerdeError is returned when registering a serde whose name is already taken.
func (DuplicateSerdeError) Error ¶
func (e DuplicateSerdeError) Error() string
type HexSerde ¶
type HexSerde struct{}
HexSerde renders bytes as a hex string. It never fails and accepts any input, so it is the binary fallback.
func (HexSerde) CanDeserialize ¶
type JSONSerde ¶
type JSONSerde struct{}
JSONSerde pretty-prints JSON payloads. CanDeserialize only claims valid JSON so auto-detection prefers it over plain string.
func (JSONSerde) CanDeserialize ¶
type MsgpackSerde ¶
type MsgpackSerde struct{}
MsgpackSerde decodes MessagePack payloads to JSON. It wraps the decode path previously hardwired in kafds.handleMessageWithConfig.
func (MsgpackSerde) CanDeserialize ¶
func (MsgpackSerde) CanDeserialize(d []byte) bool
func (MsgpackSerde) Deserialize ¶
func (MsgpackSerde) Deserialize(d []byte) (string, error)
func (MsgpackSerde) Name ¶
func (MsgpackSerde) Name() string
type NullSerde ¶
type NullSerde struct{}
NullSerde renders empty/nil payloads.
func (NullSerde) CanDeserialize ¶
type RawProtobufSerde ¶
type RawProtobufSerde struct{}
RawProtobufSerde performs a best-effort, schemaless dump of a protobuf wire-format payload: it walks the tag/field structure and renders it as JSON keyed by field number. No message type is required.
func (RawProtobufSerde) CanDeserialize ¶
func (RawProtobufSerde) CanDeserialize(d []byte) bool
func (RawProtobufSerde) Deserialize ¶
func (RawProtobufSerde) Deserialize(d []byte) (string, error)
func (RawProtobufSerde) Name ¶
func (RawProtobufSerde) Name() string
type Registry ¶
type Registry struct {
// contains filtered or unexported fields
}
Registry holds the serdes available for lookup and auto-detection. A subset of registered serdes participate in auto-detection (in registration order); the rest are selectable only by explicit name (e.g. numeric/hex serdes that would falsely match arbitrary bytes). Registry is not safe for concurrent registration, but lookups after construction are read-only and safe.
func BuildRegistry ¶
func BuildRegistry(decode DecodeFunc, configs []SerdeConfig) (*Registry, error)
BuildRegistry assembles the standard registry: the schema-registry serde (using the given decoder), then configured descriptor-protobuf serdes, then the primitive/format built-ins. Auto-detection order (MSG-15) is schema-registry → configured → JSON → string. Numeric/hex/msgpack/raw-proto and internal-topic serdes are selectable by name only (they would falsely match arbitrary bytes during auto-detection). Duplicate configured names fail (MSG-11/17).
func (*Registry) AutoDetect ¶
AutoDetect returns the first auto-detection serde that claims the bytes, or nil when none match.
func (*Registry) Register ¶
Register adds a serde selectable by name only. It rejects duplicate names.
func (*Registry) RegisterAuto ¶
RegisterAuto adds a serde that participates in auto-detection (appended to the detection order) in addition to being selectable by name.
type SchemaRegistrySerde ¶
type SchemaRegistrySerde struct {
// contains filtered or unexported fields
}
SchemaRegistrySerde decodes payloads framed with the Confluent wire format.
func NewSchemaRegistrySerde ¶
func NewSchemaRegistrySerde(decode DecodeFunc) *SchemaRegistrySerde
NewSchemaRegistrySerde builds the serde with the given decoder. A nil decoder is allowed (CanDeserialize still recognises the framing, but Deserialize reports that no registry is configured, triggering fallback).
func (*SchemaRegistrySerde) CanDeserialize ¶
func (s *SchemaRegistrySerde) CanDeserialize(data []byte) bool
CanDeserialize reports whether data carries the Confluent magic-byte framing.
func (*SchemaRegistrySerde) Deserialize ¶
func (s *SchemaRegistrySerde) Deserialize(data []byte) (string, error)
func (*SchemaRegistrySerde) Name ¶
func (s *SchemaRegistrySerde) Name() string
type Serde ¶
type Serde interface {
// Name is the unique identifier used for lookup and shown in the UI.
Name() string
// CanDeserialize reports whether this serde is a plausible decoder for the
// given bytes. It is used for auto-detection and must not panic on
// arbitrary input.
CanDeserialize(data []byte) bool
// Deserialize renders the bytes as text, or returns an error when the bytes
// are not valid for this serde (which triggers fallback).
Deserialize(data []byte) (string, error)
}
Serde decodes (and optionally encodes) the raw bytes of a Kafka message key or value into a human-readable string. Implementations must be safe for concurrent use.
func DoubleSerde ¶
func DoubleSerde() Serde
func FloatSerde ¶
func FloatSerde() Serde
type SerdeConfig ¶
type SerdeConfig struct {
Name string `yaml:"name"` // registered serde name to apply / define
TopicPattern string `yaml:"topicPattern"` // regex; empty = all topics
Target string `yaml:"target"` // "key" | "value" | "both" (default both)
DescriptorPath string `yaml:"descriptorPath"` // FileDescriptorSet path (descriptor protobuf)
MessageType string `yaml:"messageType"` // fully-qualified message name
}
SerdeConfig is a per-cluster serde binding. For topics whose name matches TopicPattern (a regex; empty = all topics), the named serde is applied to the key and/or value. When DescriptorPath is set the binding also *defines* a descriptor-file Protobuf serde (registered under Name) rather than merely referencing a built-in.
type Serializer ¶
type Serializer interface {
// Serialize encodes the given text back into wire bytes.
Serialize(text string) ([]byte, error)
}
Serializer is optionally implemented by serdes that support producing. It is separate from Serde so that read-only serdes (e.g. schema-registry decode, internal-topic decoders) need not implement it.
type StringSerde ¶
type StringSerde struct{}
StringSerde renders bytes as UTF-8 text. It is the universal fallback.
func (StringSerde) CanDeserialize ¶
func (StringSerde) CanDeserialize(_ []byte) bool
func (StringSerde) Deserialize ¶
func (StringSerde) Deserialize(d []byte) (string, error)
func (StringSerde) Name ¶
func (StringSerde) Name() string
type UnknownSerdeError ¶
type UnknownSerdeError struct {
Name string
}
UnknownSerdeError is returned when an explicitly chosen serde name is not registered.
func (UnknownSerdeError) Error ¶
func (e UnknownSerdeError) Error() string
Source Files