kafka

Documentation

Index

• type Collector
  • func New(logger logrus.FieldLogger, conf Config) (*Collector, error)
  • func (c *Collector) Collect(scs []stats.SampleContainer)
  • func (c *Collector) Init() error
  • func (c *Collector) Link() string
  • func (c *Collector) Run(ctx context.Context)
• type Config
  • func GetConsolidatedConfig(jsonRawConf json.RawMessage, env map[string]string, arg string) (Config, error)
  • func NewConfig() Config
  • func ParseArg(arg string) (Config, error)
  • func (c Config) Apply(cfg Config) Config

Types

type Collector

type Collector struct {
	Producer sarama.SyncProducer
	Config   Config

	Samples []stats.Sample
	// contains filtered or unexported fields
}

Collector implements the lib.Collector interface and should be used only for testing

func New

func New(logger logrus.FieldLogger, conf Config) (*Collector, error)

New creates an instance of the collector

func (*Collector) Collect

func (c *Collector) Collect(scs []stats.SampleContainer)

Collect just appends all of the samples passed to it to the internal sample slice. According to the the lib.Collector interface, it should never be called concurrently, so there's no locking on purpose - that way Go's race condition detector can actually detect incorrect usage. Also, theoretically the collector doesn't have to actually Run() before samples start being collected, it only has to be initialized.

func (*Collector) Init

func (c *Collector) Init() error

Init does nothing, it's only included to satisfy the lib.Collector interface

func (*Collector) Link

func (c *Collector) Link() string

Link returns a dummy string, it's only included to satisfy the lib.Collector interface

func (*Collector) Run

func (c *Collector) Run(ctx context.Context)

Run just blocks until the context is done

type Config

type Config struct {
	// Connection.
	Brokers []string `json:"brokers" envconfig:"K6_KAFKA_BROKERS"`

	// Samples.
	Topic        null.String        `json:"topic" envconfig:"K6_KAFKA_TOPIC"`
	Format       null.String        `json:"format" envconfig:"K6_KAFKA_FORMAT"`
	PushInterval types.NullDuration `json:"push_interval" envconfig:"K6_KAFKA_PUSH_INTERVAL"`

	InfluxDBConfig influxdb.Config `json:"influxdb"`
}

Config is the config for the kafka collector

func GetConsolidatedConfig

func GetConsolidatedConfig(jsonRawConf json.RawMessage, env map[string]string, arg string) (Config, error)

GetConsolidatedConfig combines {default config values + JSON config + environment vars + arg config values}, and returns the final result.

func NewConfig

func NewConfig() Config

NewConfig creates a new Config instance with default values for some fields.

func ParseArg

func ParseArg(arg string) (Config, error)

ParseArg takes an arg string and converts it to a config

func (Config) Apply

func (c Config) Apply(cfg Config) Config