kafka

Documentation

Index

• type Collector
  • func New(logger logrus.FieldLogger, conf Config) (*Collector, error)
  • func (c *Collector) Collect(scs []stats.SampleContainer)
  • func (c *Collector) GetRequiredSystemTags() stats.SystemTagSet
  • func (c *Collector) Init() error
  • func (c *Collector) Link() string
  • func (c *Collector) Run(ctx context.Context)
  • func (c *Collector) SetRunStatus(status lib.RunStatus)
• type Config
  • 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) GetRequiredSystemTags

func (c *Collector) GetRequiredSystemTags() stats.SystemTagSet

GetRequiredSystemTags returns which sample tags are needed by this collector

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

func (*Collector) SetRunStatus

func (c *Collector) SetRunStatus(status lib.RunStatus)

SetRunStatus does nothing in the Kafka collector

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 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