kafka_output

Documentation

Index

• Variables
• type Config
• type KafkaOutput
  • func (k *KafkaOutput) Close() error
  • func (k *KafkaOutput) Init(ctx context.Context, name string, cfg map[string]interface{}, ...) error
  • func (k *KafkaOutput) RegisterMetrics(reg *prometheus.Registry)
  • func (k *KafkaOutput) SetClusterName(name string)
  • func (k *KafkaOutput) SetEventProcessors(ps map[string]map[string]interface{}, logger *log.Logger, ...)
  • func (k *KafkaOutput) SetLogger(logger *log.Logger)
  • func (k *KafkaOutput) SetName(name string)
  • func (k *KafkaOutput) String() string
  • func (k *KafkaOutput) Write(ctx context.Context, rsp proto.Message, meta outputs.Meta)
  • func (k *KafkaOutput) WriteEvent(ctx context.Context, ev *formatters.EventMsg)
• 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 KafkaNumberOfFailSendMsgs = prometheus.NewCounterVec(prometheus.CounterOpts{
	Namespace: "gnmic",
	Subsystem: "kafka_output",
	Name:      "number_of_kafka_msgs_sent_fail_total",
	Help:      "Number of failed msgs sent by gnmic kafka output",
}, []string{"producer_id", "reason"})

var KafkaNumberOfSentBytes = prometheus.NewCounterVec(prometheus.CounterOpts{
	Namespace: "gnmic",
	Subsystem: "kafka_output",
	Name:      "number_of_written_kafka_bytes_total",
	Help:      "Number of bytes written by gnmic kafka output",
}, []string{"producer_id"})

var KafkaNumberOfSentMsgs = prometheus.NewCounterVec(prometheus.CounterOpts{
	Namespace: "gnmic",
	Subsystem: "kafka_output",
	Name:      "number_of_kafka_msgs_sent_success_total",
	Help:      "Number of msgs successfully sent by gnmic kafka output",
}, []string{"producer_id"})

var KafkaSendDuration = prometheus.NewGaugeVec(prometheus.GaugeOpts{
	Namespace: "gnmic",
	Subsystem: "kafka_output",
	Name:      "msg_send_duration_ns",
	Help:      "gnmic kafka output send duration in ns",
}, []string{"producer_id"})

var SHA256 scram.HashGeneratorFcn = func() hash.Hash { return sha256.New() }

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

Types

type Config

type Config struct {
	Address          string        `mapstructure:"address,omitempty"`
	Topic            string        `mapstructure:"topic,omitempty"`
	Name             string        `mapstructure:"name,omitempty"`
	SASL             *sasl         `mapstructure:"sasl,omitempty"`
	MaxRetry         int           `mapstructure:"max-retry,omitempty"`
	Timeout          time.Duration `mapstructure:"timeout,omitempty"`
	RecoveryWaitTime time.Duration `mapstructure:"recovery-wait-time,omitempty"`
	Format           string        `mapstructure:"format,omitempty"`
	NumWorkers       int           `mapstructure:"num-workers,omitempty"`
	Debug            bool          `mapstructure:"debug,omitempty"`
	BufferSize       int           `mapstructure:"buffer-size,omitempty"`
	EnableMetrics    bool          `mapstructure:"enable-metrics,omitempty"`
	EventProcessors  []string      `mapstructure:"event-processors,omitempty"`
}

Config //

type KafkaOutput

type KafkaOutput struct {
	Cfg *Config
	// contains filtered or unexported fields
}

KafkaOutput //

func (*KafkaOutput) Close

func (k *KafkaOutput) Close() error

Close //

func (*KafkaOutput) Init

func (k *KafkaOutput) Init(ctx context.Context, name string, cfg map[string]interface{}, opts ...outputs.Option) error

Init /

func (*KafkaOutput) RegisterMetrics

func (k *KafkaOutput) RegisterMetrics(reg *prometheus.Registry)

Metrics //

func (*KafkaOutput) SetClusterName

func (k *KafkaOutput) SetClusterName(name string)

func (*KafkaOutput) SetEventProcessors

func (k *KafkaOutput) SetEventProcessors(ps map[string]map[string]interface{}, logger *log.Logger, tcs map[string]interface{})

func (*KafkaOutput) SetLogger

func (k *KafkaOutput) SetLogger(logger *log.Logger)

func (*KafkaOutput) SetName

func (k *KafkaOutput) SetName(name string)

func (*KafkaOutput) String

func (k *KafkaOutput) String() string

func (*KafkaOutput) Write

func (k *KafkaOutput) Write(ctx context.Context, rsp proto.Message, meta outputs.Meta)

Write //

func (*KafkaOutput) WriteEvent

func (k *KafkaOutput) WriteEvent(ctx context.Context, ev *formatters.EventMsg)

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)