kafkaacquisition

Documentation

Index

• type KafkaConfiguration
  • func (kc *KafkaConfiguration) NewDialer() (*kafka.Dialer, error)
  • func (kc *KafkaConfiguration) NewReader(dialer *kafka.Dialer, logger *log.Entry) (*kafka.Reader, error)
  • func (kc *KafkaConfiguration) NewTLSConfig() (*tls.Config, error)
• type KafkaSource
  • func (k *KafkaSource) CanRun() error
  • func (k *KafkaSource) Configure(yamlConfig []byte, logger *log.Entry, MetricsLevel int) error
  • func (k *KafkaSource) ConfigureByDSN(string, map[string]string, *log.Entry, string) error
  • func (k *KafkaSource) Dump() interface{}
  • func (k *KafkaSource) GetAggregMetrics() []prometheus.Collector
  • func (k *KafkaSource) GetMetrics() []prometheus.Collector
  • func (k *KafkaSource) GetMode() string
  • func (k *KafkaSource) GetName() string
  • func (k *KafkaSource) GetUuid() string
  • func (k *KafkaSource) OneShotAcquisition(out chan types.Event, t *tomb.Tomb) error
  • func (k *KafkaSource) ReadMessage(out chan types.Event) error
  • func (k *KafkaSource) RunReader(out chan types.Event, t *tomb.Tomb) error
  • func (k *KafkaSource) StreamingAcquisition(out chan types.Event, t *tomb.Tomb) error
  • func (k *KafkaSource) UnmarshalConfig(yamlConfig []byte) error
• type TLSConfig

Types

type KafkaConfiguration

type KafkaConfiguration struct {
	Brokers                           []string   `yaml:"brokers"`
	Topic                             string     `yaml:"topic"`
	GroupID                           string     `yaml:"group_id"`
	Partition                         int        `yaml:"partition"`
	Timeout                           string     `yaml:"timeout"`
	TLS                               *TLSConfig `yaml:"tls"`
	configuration.DataSourceCommonCfg `yaml:",inline"`
}

func (*KafkaConfiguration) NewDialer

func (kc *KafkaConfiguration) NewDialer() (*kafka.Dialer, error)

func (*KafkaConfiguration) NewReader

func (kc *KafkaConfiguration) NewReader(dialer *kafka.Dialer, logger *log.Entry) (*kafka.Reader, error)

func (*KafkaConfiguration) NewTLSConfig

func (kc *KafkaConfiguration) NewTLSConfig() (*tls.Config, error)

type KafkaSource

type KafkaSource struct {
	Config KafkaConfiguration

	Reader *kafka.Reader
	// contains filtered or unexported fields
}

func (*KafkaSource) CanRun

func (k *KafkaSource) CanRun() error

func (*KafkaSource) Configure

func (k *KafkaSource) Configure(yamlConfig []byte, logger *log.Entry, MetricsLevel int) error

func (*KafkaSource) ConfigureByDSN

func (k *KafkaSource) ConfigureByDSN(string, map[string]string, *log.Entry, string) error

func (*KafkaSource) Dump

func (k *KafkaSource) Dump() interface{}

func (*KafkaSource) GetAggregMetrics

func (k *KafkaSource) GetAggregMetrics() []prometheus.Collector

func (*KafkaSource) GetMetrics

func (k *KafkaSource) GetMetrics() []prometheus.Collector

func (*KafkaSource) GetMode

func (k *KafkaSource) GetMode() string

func (*KafkaSource) GetName

func (k *KafkaSource) GetName() string

func (*KafkaSource) GetUuid

func (k *KafkaSource) GetUuid() string

func (*KafkaSource) OneShotAcquisition

func (k *KafkaSource) OneShotAcquisition(out chan types.Event, t *tomb.Tomb) error

func (*KafkaSource) ReadMessage

func (k *KafkaSource) ReadMessage(out chan types.Event) error

func (*KafkaSource) RunReader

func (k *KafkaSource) RunReader(out chan types.Event, t *tomb.Tomb) error

func (*KafkaSource) StreamingAcquisition

func (k *KafkaSource) StreamingAcquisition(out chan types.Event, t *tomb.Tomb) error

func (*KafkaSource) UnmarshalConfig

func (k *KafkaSource) UnmarshalConfig(yamlConfig []byte) error

type TLSConfig

type TLSConfig struct {
	InsecureSkipVerify bool   `yaml:"insecure_skip_verify"`
	ClientCert         string `yaml:"client_cert"`
	ClientKey          string `yaml:"client_key"`
	CaCert             string `yaml:"ca_cert"`
}