kafka

Documentation

Index

• Variables
• func ApplySaramaSASLFromParsed(pConf *service.ParsedConfig, mgr *service.Resources, conf *sarama.Config) error
• func NewKafkaWriterFromParsed(conf *service.ParsedConfig, mgr *service.Resources) (service.BatchOutput, error)
• func OSKConfigSpec() *service.ConfigSpec
• func SaramaSASLField() *service.ConfigField
• func TopicLoggerFields() []*service.ConfigField
• type TopicLogger
  • func NewTopicLogger() *TopicLogger
  • func (l *TopicLogger) Enabled(ctx context.Context, atLevel slog.Level) bool
  • func (l *TopicLogger) Handle(ctx context.Context, r slog.Record) error
  • func (l *TopicLogger) InitOutputFromParsed(pConf *service.ParsedConfig) error
  • func (l *TopicLogger) SetFallbackLogger(fLogger *service.Logger)
  • func (l *TopicLogger) WithAttrs(attrs []slog.Attr) slog.Handler
  • func (l *TopicLogger) WithGroup(name string) slog.Handler
• type XDGSCRAMClient
  • func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)
  • func (x *XDGSCRAMClient) Done() bool
  • func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)

Variables

var AWSSASLFromConfigFn = notImportedAWSFn

AWSSASLFromConfigFn is populated with the child `aws` package when imported.

var (
	ErrUnsupportedSASLMechanism = errors.New("unsupported SASL mechanism")
)

SASL specific error types.

var SHA256 scram.HashGeneratorFcn = sha256.New

SHA256 generates the SHA256 hash.

var SHA512 scram.HashGeneratorFcn = sha512.New

SHA512 generates the SHA512 hash.

Functions

func ApplySaramaSASLFromParsed

func ApplySaramaSASLFromParsed(pConf *service.ParsedConfig, mgr *service.Resources, conf *sarama.Config) error

ApplySaramaSASLFromParsed applies a parsed config containing a SASL field to a sarama.Config.

func NewKafkaWriterFromParsed

func NewKafkaWriterFromParsed(conf *service.ParsedConfig, mgr *service.Resources) (service.BatchOutput, error)

NewKafkaWriterFromParsed returns a kafka output from a parsed config.

func OSKConfigSpec

func OSKConfigSpec() *service.ConfigSpec

OSKConfigSpec creates a new config spec for a kafka output.

func SaramaSASLField

func SaramaSASLField() *service.ConfigField

SaramaSASLField returns a field spec definition for SASL within the sarama components.

func TopicLoggerFields

func TopicLoggerFields() []*service.ConfigField

Types

type TopicLogger

type TopicLogger struct {
	// contains filtered or unexported fields
}

TopicLogger provides a mechanism for sending service-wide logs into a kafka topic. The writing is done by a regular output, but this type is necessary in order to allow hot swapping of log components during start up.

func NewTopicLogger

func NewTopicLogger() *TopicLogger

func (*TopicLogger) Enabled

func (l *TopicLogger) Enabled(ctx context.Context, atLevel slog.Level) bool

func (*TopicLogger) Handle

func (l *TopicLogger) Handle(ctx context.Context, r slog.Record) error

func (*TopicLogger) InitOutputFromParsed

func (l *TopicLogger) InitOutputFromParsed(pConf *service.ParsedConfig) error

func (*TopicLogger) SetFallbackLogger

func (l *TopicLogger) SetFallbackLogger(fLogger *service.Logger)

func (*TopicLogger) WithAttrs

func (l *TopicLogger) WithAttrs(attrs []slog.Attr) slog.Handler

func (*TopicLogger) WithGroup

func (l *TopicLogger) WithGroup(name string) slog.Handler

type XDGSCRAMClient

type XDGSCRAMClient struct {
	*scram.Client
	*scram.ClientConversation
	scram.HashGeneratorFcn
}

XDGSCRAMClient represents struct to XDG Scram client to initialize conversation.

func (*XDGSCRAMClient) Begin

func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)

Begin initializes new client and conversation to securely transmit the provided credentials to Kafka.

func (*XDGSCRAMClient) Done

func (x *XDGSCRAMClient) Done() bool

Done returns true if the conversation is completed or has errored.

func (*XDGSCRAMClient) Step

func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)

Step takes a string provided from a server (or just an empty string for the very first conversation step) and attempts to move the authentication conversation forward.