processor

Documentation

Index

• Constants
• type JSONConfig
• type JSONProcessor
  • func NewJSONProcessor(in LineProcessor, cfg *JSONConfig) *JSONProcessor
  • func (p *JSONProcessor) Next() (Line, error)
• type LimitProcessor
  • func NewLimitProcessor(in LineProcessor, maxBytes int) *LimitProcessor
  • func (p *LimitProcessor) Next() (Line, error)
• type Line
• type LineEncoder
  • func NewLineEncoder(in io.Reader, codec encoding.Encoding, bufferSize int) (LineEncoder, error)
  • func (p LineEncoder) Next() (Line, error)
• type LineProcessor
• type MultiLine
  • func NewMultiline(r LineProcessor, separator string, maxBytes int, config *MultilineConfig) (*MultiLine, error)
  • func (mlr *MultiLine) Next() (Line, error)
• type MultilineConfig
  • func (c *MultilineConfig) Validate() error
• type StripNewline
  • func NewStripNewline(r LineProcessor) *StripNewline
  • func (p *StripNewline) Next() (Line, error)

Constants

const (
	JsonErrorKey = "json_error"
)

Types

type JSONConfig

type JSONConfig struct {
	MessageKey    string `config:"message_key"`
	KeysUnderRoot bool   `config:"keys_under_root"`
	OverwriteKeys bool   `config:"overwrite_keys"`
	AddErrorKey   bool   `config:"add_error_key"`
}

type JSONProcessor

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

func NewJSONProcessor

func NewJSONProcessor(in LineProcessor, cfg *JSONConfig) *JSONProcessor

NewJSONProcessor creates a new processor that can decode JSON.

func (*JSONProcessor) Next

func (p *JSONProcessor) Next() (Line, error)

Next decodes JSON and returns the filled Line object.

type LimitProcessor

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

LimitProcessor sets an upper limited on line length. Lines longer then the max configured line length will be snapped short.

func NewLimitProcessor

func NewLimitProcessor(in LineProcessor, maxBytes int) *LimitProcessor

NewLimitProcessor creates a new processor limiting the line length.

func (*LimitProcessor) Next

func (p *LimitProcessor) Next() (Line, error)

Next returns the next line.

type Line

type Line struct {
	Ts      time.Time     // timestamp the line was read
	Content []byte        // actual line read
	Bytes   int           // total number of bytes read to generate the line
	Fields  common.MapStr // optional fields that can be added by processors
}

Line represents a line event with timestamp, content and actual number of bytes read from input before decoding.

type LineEncoder

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

LineEncoder produces lines by reading lines from an io.Reader through a decoder converting the reader it's encoding to utf-8.

func NewLineEncoder

func NewLineEncoder(
	in io.Reader,
	codec encoding.Encoding,
	bufferSize int,
) (LineEncoder, error)

NewLineEncoder creates a new LineEncoder from input reader by applying the given codec.

func (LineEncoder) Next

func (p LineEncoder) Next() (Line, error)

Next reads the next line from it's initial io.Reader

type LineProcessor

type LineProcessor interface {
	Next() (Line, error)
}

LineProcessor is the interface that wraps the basic Next method for getting a new line. Next returns the line being read or and error. EOF is returned if processor will not return any new lines on subsequent calls.

type MultiLine

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

MultiLine processor combining multiple line events into one multi-line event.

Lines to be combined are matched by some configurable predicate using regular expression.

The maximum number of bytes and lines to be returned is fully configurable. Even if limits are reached subsequent lines are matched, until event is fully finished.

Errors will force the multiline processor to return the currently active multiline event first and finally return the actual error on next call to Next.

func NewMultiline

func NewMultiline(
	r LineProcessor,
	separator string,
	maxBytes int,
	config *MultilineConfig,
) (*MultiLine, error)

NewMultiline creates a new multi-line processor combining stream of line events into stream of multi-line events.

func (*MultiLine) Next

func (mlr *MultiLine) Next() (Line, error)

Next returns next multi-line event.

type MultilineConfig

type MultilineConfig struct {
	Negate   bool           `config:"negate"`
	Match    string         `config:"match"       validate:"required"`
	MaxLines *int           `config:"max_lines"`
	Pattern  *regexp.Regexp `config:"pattern"`
	Timeout  *time.Duration `config:"timeout"     validate:"positive"`
}

func (*MultilineConfig) Validate

func (c *MultilineConfig) Validate() error

type StripNewline

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

StripNewline processor removes the last trailing newline characters from read lines.

func NewStripNewline

func NewStripNewline(r LineProcessor) *StripNewline

NewStripNewline creates a new line reader stripping the last tailing newline.

func (*StripNewline) Next

func (p *StripNewline) Next() (Line, error)

Next returns the next line.