kafka

Documentation

Overview

Package kafka sends real-time testing metrics to an Apache Kafka message broker

Index

• func New(params output.Params) (output.Output, error)
• 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
• type FieldKind
• type Output
  • func (o *Output) Description() string
  • func (o *Output) Start() error
  • func (o *Output) Stop() error

Functions

func New

func New(params output.Params) (output.Output, error)

New creates a new instance of the output.

Types

type Config

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

	// Samples.
	Topic                 null.String        `json:"topic" envconfig:"K6_KAFKA_TOPIC"`
	User                  null.String        `json:"user" envconfig:"K6_KAFKA_SASL_USER"`
	Password              null.String        `json:"password" envconfig:"K6_KAFKA_SASL_PASSWORD"`
	AuthMechanism         null.String        `json:"authMechanism" envconfig:"K6_KAFKA_AUTH_MECHANISM"`
	Format                null.String        `json:"format" envconfig:"K6_KAFKA_FORMAT"`
	PushInterval          types.NullDuration `json:"pushInterval" envconfig:"K6_KAFKA_PUSH_INTERVAL"`
	Version               null.String        `json:"version" envconfig:"K6_KAFKA_VERSION"`
	SSL                   null.Bool          `json:"ssl" envconfig:"K6_KAFKA_SSL"`
	InsecureSkipTLSVerify null.Bool          `json:"insecureSkipTLSVerify" envconfig:"K6_KAFKA_INSECURE_SKIP_TLS_VERIFY"`
	LogError              null.Bool          `json:"logError" envconfig:"K6_KAFKA_LOG_ERROR"`

	InfluxDBConfig influxdbConfig `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

Apply merges applied Config into base.

type FieldKind

type FieldKind int

FieldKind defines Enum for tag-to-field type conversion

const (
	// String field (default)
	String FieldKind = iota
	// Int field
	Int
	// Float field
	Float
	// Bool field
	Bool
)

type Output

type Output struct {
	output.SampleBuffer

	Config  Config
	CloseFn func() error

	Producer sarama.AsyncProducer
	// contains filtered or unexported fields
}

Output is a k6 output that sends metrics to a Kafka broker.

func (*Output) Description

func (o *Output) Description() string

Description returns a short human-readable description of the output.

func (*Output) Start

func (o *Output) Start() error

Start initializes the output.

func (*Output) Stop

func (o *Output) Stop() error

Stop stops the output.