kafka

Documentation

Index

• Variables
• func SASLVersion(kafkaVersion sarama.KafkaVersion, saslVersion *int) (int16, error)
• type BackoffFunc
• type Config
  • func (k *Config) SetConfig(config *sarama.Config, log telegraf.Logger) error
• type DebugLogger
  • func (l *DebugLogger) Print(v ...interface{})
  • func (l *DebugLogger) Printf(format string, v ...interface{})
  • func (l *DebugLogger) Println(v ...interface{})
• type Logger
  • func (k *Logger) SetLogger()
• type ReadConfig
  • func (k *ReadConfig) SetConfig(config *sarama.Config, log telegraf.Logger) error
• type SASLAuth
  • func (k *SASLAuth) SetSASLConfig(cfg *sarama.Config) error
  • func (k *SASLAuth) Token() (*sarama.AccessToken, error)
• type WriteConfig
  • func (k *WriteConfig) SetConfig(config *sarama.Config, log telegraf.Logger) error
• 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 SHA256 scram.HashGeneratorFcn = func() hash.Hash { return sha256.New() }

var SHA512 scram.HashGeneratorFcn = func() hash.Hash { return sha512.New() }

Functions

func SASLVersion

func SASLVersion(kafkaVersion sarama.KafkaVersion, saslVersion *int) (int16, error)

Types

type BackoffFunc

type BackoffFunc func(retries, maxRetries int) time.Duration

type Config

type Config struct {
	SASLAuth
	tls.ClientConfig

	Version          string           `toml:"version"`
	ClientID         string           `toml:"client_id"`
	CompressionCodec int              `toml:"compression_codec"`
	EnableTLS        *bool            `toml:"enable_tls"`
	KeepAlivePeriod  *tgConf.Duration `toml:"keep_alive_period"`

	MetadataRetryMax         int             `toml:"metadata_retry_max"`
	MetadataRetryType        string          `toml:"metadata_retry_type"`
	MetadataRetryBackoff     tgConf.Duration `toml:"metadata_retry_backoff"`
	MetadataRetryMaxDuration tgConf.Duration `toml:"metadata_retry_max_duration"`

	// Disable full metadata fetching
	MetadataFull *bool `toml:"metadata_full"`
}

Config common to all Kafka clients.

func (*Config) SetConfig

func (k *Config) SetConfig(config *sarama.Config, log telegraf.Logger) error

SetConfig on the sarama.Config object from the Config struct.

type DebugLogger

type DebugLogger struct {
	Log telegraf.Logger
}

DebugLogger logs messages from sarama at the debug level.

func (*DebugLogger) Print

func (l *DebugLogger) Print(v ...interface{})

func (*DebugLogger) Printf

func (l *DebugLogger) Printf(format string, v ...interface{})

func (*DebugLogger) Println

func (l *DebugLogger) Println(v ...interface{})

type Logger

type Logger struct {
}

func (*Logger) SetLogger

func (k *Logger) SetLogger()

SetLogger configures a debug logger for kafka (sarama)

type ReadConfig

type ReadConfig struct {
	Config
}

ReadConfig for kafka clients meaning to read from Kafka.

func (*ReadConfig) SetConfig

func (k *ReadConfig) SetConfig(config *sarama.Config, log telegraf.Logger) error

SetConfig on the sarama.Config object from the ReadConfig struct.

type SASLAuth

type SASLAuth struct {
	SASLUsername  config.Secret `toml:"sasl_username"`
	SASLPassword  config.Secret `toml:"sasl_password"`
	SASLMechanism string        `toml:"sasl_mechanism"`
	SASLVersion   *int          `toml:"sasl_version"`

	// GSSAPI config
	SASLGSSAPIServiceName        string `toml:"sasl_gssapi_service_name"`
	SASLGSSAPIAuthType           string `toml:"sasl_gssapi_auth_type"`
	SASLGSSAPIDisablePAFXFAST    bool   `toml:"sasl_gssapi_disable_pafxfast"`
	SASLGSSAPIKerberosConfigPath string `toml:"sasl_gssapi_kerberos_config_path"`
	SASLGSSAPIKeyTabPath         string `toml:"sasl_gssapi_key_tab_path"`
	SASLGSSAPIRealm              string `toml:"sasl_gssapi_realm"`

	// OAUTHBEARER config
	SASLAccessToken config.Secret `toml:"sasl_access_token"`
}

func (*SASLAuth) SetSASLConfig

func (k *SASLAuth) SetSASLConfig(cfg *sarama.Config) error

SetSASLConfig configures SASL for kafka (sarama)

func (*SASLAuth) Token

func (k *SASLAuth) Token() (*sarama.AccessToken, error)

Token does nothing smart, it just grabs a hard-coded token from config.

type WriteConfig

type WriteConfig struct {
	Config

	RequiredAcks     int  `toml:"required_acks"`
	MaxRetry         int  `toml:"max_retry"`
	MaxMessageBytes  int  `toml:"max_message_bytes"`
	IdempotentWrites bool `toml:"idempotent_writes"`
}

WriteConfig for kafka clients meaning to write to kafka

func (*WriteConfig) SetConfig

func (k *WriteConfig) SetConfig(config *sarama.Config, log telegraf.Logger) error

SetConfig on the sarama.Config object from the WriteConfig struct.

type XDGSCRAMClient

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

func (*XDGSCRAMClient) Begin

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

func (*XDGSCRAMClient) Done

func (x *XDGSCRAMClient) Done() bool

func (*XDGSCRAMClient) Step

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