stages

type MetricsCollector ¶

type MetricsCollector interface {
	IncrementHTTPRequestCount(podName, method, service, path string, status int)
	IncrementHTTPRequestsTotalCount(service string)
	ObserveHTTPRequestTime(podName, method, service, path string, value float64)
	ObserveHTTPUpstreamResponseTimeTotal(podName, method, service, path string, value float64)
}

MetricsCollector is a metrics counter object interface for parsing sli stage

type ParserFunction ¶

type ParserFunction func([]byte) (common.EntryMap, error)

type ParserFunctionDefault ¶

type ParserFunctionDefault func([]byte) common.EntryMap

type ParserSLI ¶

type ParserSLI interface {
	Parse(entryMap common.EntryMap)
}

ParserSLI may inject additional operations on parsed entry

type Stage ¶

type Stage interface {
	InitWorker()
	Close()
}

Stage interface

type StageFiltering ¶

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

StageFiltering filters messages from input

func NewStageFiltering(input <-chan common.EntryMap, userLogField string, logger logging.Logger) *StageFiltering

func (s *StageFiltering) Close()

func (s *StageFiltering) InitWorker()

func (s *StageFiltering) Out() <-chan common.EntryMap

type StageJSONMarshalling ¶

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

StageJSONMarshalling marshals messages from input to JSON and sends them to output

func NewStageJSONMarshalling(input <-chan common.EntryMap, logger logging.Logger) *StageJSONMarshalling

func (s *StageJSONMarshalling) Close()

func (s *StageJSONMarshalling) InitWorker()

func (s *StageJSONMarshalling) Out() <-chan string

type StageParsingEntry ¶

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

StageParsingEntry parses lines using given parser, extends resulting map with metadata and sends it downstream

func NewStageParsingEntry(input <-chan *common.Entry, parseDocker, parseContainerD ParserFunction,
	parserDefault ParserFunctionDefault, extendsField string, logger logging.Logger) *StageParsingEntry

func (s *StageParsingEntry) Close()

func (s *StageParsingEntry) InitWorker()

func (s *StageParsingEntry) Out() <-chan common.EntryMap

type StageParsingSLI ¶

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

StageParsingSLI checks if the message in its input is a SLI message, modifies metrics, then sends message downstream

func NewStageParsingSLI(input <-chan common.EntryMap, userLogField string, parser ParserSLI, logger logging.Logger) *StageParsingSLI

func (s *StageParsingSLI) Close()

func (s *StageParsingSLI) InitWorker()

func (s *StageParsingSLI) Out() <-chan common.EntryMap

type StageTransport ¶

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

StageTransport stacks messages from input into batches and tries to send them to transport

func NewStageTransport(input <-chan string, transportClient TransportClient,
	bufferSizeMax int, flushInterval time.Duration, logger logging.Logger) *StageTransport

func (s *StageTransport) Close()

func (s *StageTransport) InitWorker()

type TransportClient ¶

type TransportClient interface {
	DeliverMessages(messages []string) error
}

TransportClient is a transport interface for transport stage