output

Documentation

Index

• Constants
• Variables
• func BuildOutput(outputType string, config map[interface{}]interface{}) *topology.OutputBox
• func Register(outputType string, bf BuildOutputFunc)
• type Action
  • func (action *Action) Encode() []byte
• type BuildOutputFunc
• type BulkProcessor
• type BulkRequest
• type ClickhouseOutput
  • func (c *ClickhouseOutput) Emit(event map[string]interface{})
  • func (c *ClickhouseOutput) Shutdown()
• type DotOutput
  • func (outputPlugin *DotOutput) Emit(event map[string]interface{})
  • func (outputPlugin *DotOutput) Shutdown()
• type ESBulkRequest
• type ElasticsearchOutput
  • func (p *ElasticsearchOutput) Emit(event map[string]interface{})
  • func (outputPlugin *ElasticsearchOutput) Shutdown()
• type Event
• type GetRetryEventsFunc
• type HTTPBulkProcessor
  • func NewHTTPBulkProcessor(headers map[string]string, hosts []string, requestMethod string, ...) *HTTPBulkProcessor
• type HostSelector
• type InAction
  • func (action *InAction) Encode() []byte
• type InfluxdbBulkRequest
• type InfluxdbOutput
  • func (p *InfluxdbOutput) Emit(event map[string]interface{})
  • func (outputPlugin *InfluxdbOutput) Shutdown()
• type KafkaOutput
  • func (p *KafkaOutput) Emit(event map[string]interface{})
  • func (p *KafkaOutput) Shutdown()
• type NewBulkRequestFunc
• type RRHostSelector
  • func NewRRHostSelector(hosts []interface{}, weight int) *RRHostSelector
  • func (s *RRHostSelector) AddWeight()
  • func (s *RRHostSelector) Next() interface{}
  • func (s *RRHostSelector) ReduceWeight()
  • func (s *RRHostSelector) Size() int
• type StdoutOutput
  • func (p *StdoutOutput) Emit(event map[string]interface{})
  • func (p *StdoutOutput) Shutdown()
• type TCPOutput
  • func (p *TCPOutput) Emit(event map[string]interface{})
  • func (p *TCPOutput) Shutdown()

Constants

const (
	DEFAULT_BULK_SIZE      = 15 * 1024 * 1024
	DEFAULT_BULK_ACTIONS   = 5000
	DEFAULT_FLUSH_INTERVAL = 30
	DEFAULT_CONCURRENT     = 1

	MAX_BYTE_SIZE_APPLIED_IN_ADVANCE = 1024 * 1024 * 50
)

const (
	CLICKHOUSE_DEFAULT_BULK_ACTIONS   = 1000
	CLICKHOUSE_DEFAULT_FLUSH_INTERVAL = 30
)

Variables

var (
	REMOVE_HTTP_AUTH_REGEXP = regexp.MustCompile(`^(?i)(http(s?)://)[^:]+:[^@]+@`)
)

Functions

func BuildOutput

func BuildOutput(outputType string, config map[interface{}]interface{}) *topology.OutputBox

BuildOutput builds OutputBox. it firstly tries built-in plugin, and then try 3rd party plugin

func Register

func Register(outputType string, bf BuildOutputFunc)

Register is used by output plugins to register themselves

Types

type Action

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

func (*Action) Encode

func (action *Action) Encode() []byte

type BuildOutputFunc

type BuildOutputFunc func(map[interface{}]interface{}) topology.Output

type BulkProcessor

type BulkProcessor interface {
	// contains filtered or unexported methods
}

type BulkRequest

type BulkRequest interface {
	// contains filtered or unexported methods
}

type ClickhouseOutput

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

func (*ClickhouseOutput) Emit

func (c *ClickhouseOutput) Emit(event map[string]interface{})

Emit appends event to c.events, and push to bulkChan if needed

func (*ClickhouseOutput) Shutdown

func (c *ClickhouseOutput) Shutdown()

Shutdown would stop receiving message and emiting

type DotOutput

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

func (*DotOutput) Emit

func (outputPlugin *DotOutput) Emit(event map[string]interface{})

func (*DotOutput) Shutdown

func (outputPlugin *DotOutput) Shutdown()

type ESBulkRequest

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

type ElasticsearchOutput

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

func (*ElasticsearchOutput) Emit

func (p *ElasticsearchOutput) Emit(event map[string]interface{})

Emit adds the event to bulkProcessor

func (*ElasticsearchOutput) Shutdown

func (outputPlugin *ElasticsearchOutput) Shutdown()

type Event

type Event interface {
	Encode() []byte
}

type GetRetryEventsFunc

type GetRetryEventsFunc func(*http.Response, []byte, *BulkRequest) ([]int, []int, BulkRequest)

type HTTPBulkProcessor

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

func NewHTTPBulkProcessor

func NewHTTPBulkProcessor(headers map[string]string, hosts []string, requestMethod string, retryResponseCode map[int]bool, bulk_size, bulk_actions, flush_interval, concurrent int, compress bool, newBulkRequestFunc NewBulkRequestFunc, getRetryEventsFunc GetRetryEventsFunc) *HTTPBulkProcessor

type HostSelector

type HostSelector interface {
	Next() interface{}
	ReduceWeight()
	AddWeight()
	Size() int
}

type InAction

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

func (*InAction) Encode

func (action *InAction) Encode() []byte

type InfluxdbBulkRequest

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

type InfluxdbOutput

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

func (*InfluxdbOutput) Emit

func (p *InfluxdbOutput) Emit(event map[string]interface{})

func (*InfluxdbOutput) Shutdown

func (outputPlugin *InfluxdbOutput) Shutdown()

type KafkaOutput

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

func (*KafkaOutput) Emit

func (p *KafkaOutput) Emit(event map[string]interface{})

func (*KafkaOutput) Shutdown

func (p *KafkaOutput) Shutdown()

type NewBulkRequestFunc

type NewBulkRequestFunc func() BulkRequest

type RRHostSelector

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

func NewRRHostSelector

func NewRRHostSelector(hosts []interface{}, weight int) *RRHostSelector

func (*RRHostSelector) AddWeight

func (s *RRHostSelector) AddWeight()

func (*RRHostSelector) Next

func (s *RRHostSelector) Next() interface{}

func (*RRHostSelector) ReduceWeight

func (s *RRHostSelector) ReduceWeight()

func (*RRHostSelector) Size

func (s *RRHostSelector) Size() int

type StdoutOutput

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

func (*StdoutOutput) Emit

func (p *StdoutOutput) Emit(event map[string]interface{})

func (*StdoutOutput) Shutdown

func (p *StdoutOutput) Shutdown()

type TCPOutput

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

func (*TCPOutput) Emit

func (p *TCPOutput) Emit(event map[string]interface{})

func (*TCPOutput) Shutdown

func (p *TCPOutput) Shutdown()