kafka

README

Kafka output

It sends the event batches to kafka brokers using sarama lib.

Config params

brokers []string required

List of kafka brokers to write to.

default_topic string required

The default topic name if nothing will be found in the event field or should_use_topic_field isn't set.

use_topic_field bool default=false

If set, the plugin will use topic name from the event field.

topic_field string default=topic

Which event field to use as topic name. It works only if should_use_topic_field is set.

workers_count cfg.Expression default=gomaxprocs*4

How many workers will be instantiated to send batches.

batch_size cfg.Expression default=capacity/4

A maximum quantity of the events to pack into one batch.

batch_flush_timeout cfg.Duration default=200ms

After this timeout the batch will be sent even if batch isn't full.

Generated using insane-doc

Documentation

Index

• func Factory() (pipeline.AnyPlugin, pipeline.AnyConfig)
• type Config
• type Plugin
  • func (p *Plugin) Out(event *pipeline.Event)
  • func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.OutputPluginParams)
  • func (p *Plugin) Stop()

Functions

func Factory

func Factory() (pipeline.AnyPlugin, pipeline.AnyConfig)

Types

type Config

type Config struct {
	//> @3@4@5@6
	//>
	//> List of kafka brokers to write to.
	Brokers []string `json:"brokers" required:"true"` //*

	//> @3@4@5@6
	//>
	//> The default topic name if nothing will be found in the event field or `should_use_topic_field` isn't set.
	DefaultTopic string `json:"default_topic" required:"true"` //*

	//> @3@4@5@6
	//>
	//> If set, the plugin will use topic name from the event field.
	UseTopicField bool `json:"use_topic_field" default:"false"` //*

	//> @3@4@5@6
	//>
	//> Which event field to use as topic name. It works only if `should_use_topic_field` is set.
	TopicField string `json:"topic_field" default:"topic"` //*

	//> @3@4@5@6
	//>
	//> How many workers will be instantiated to send batches.
	WorkersCount  cfg.Expression `json:"workers_count" default:"gomaxprocs*4" parse:"expression"` //*
	WorkersCount_ int

	//> @3@4@5@6
	//>
	//> A maximum quantity of the events to pack into one batch.
	BatchSize  cfg.Expression `json:"batch_size" default:"capacity/4" parse:"expression"` //*
	BatchSize_ int

	//> @3@4@5@6
	//>
	//> After this timeout the batch will be sent even if batch isn't full.
	BatchFlushTimeout  cfg.Duration `json:"batch_flush_timeout" default:"200ms" parse:"duration"` //*
	BatchFlushTimeout_ time.Duration
}

! config-params ^ config-params

type Plugin

type Plugin struct {
	// contains filtered or unexported fields
}

func (*Plugin) Out

func (p *Plugin) Out(event *pipeline.Event)

func (*Plugin) Start

func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.OutputPluginParams)

func (*Plugin) Stop

func (p *Plugin) Stop()