influx

README

Influx Line Protocol Parser Plugin

Parses metrics using the Influx Line Protocol.

Configuration

[[inputs.file]]
  files = ["example"]

  ## Data format to consume.
  ## Each data format has its own unique set of configuration options, read
  ## more about them here:
  ##   https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
  data_format = "influx"

  ## Influx line protocol parser
  ## 'internal' is the default. 'upstream' is a newer parser that is faster
  ## and more memory efficient.
  ## influx_parser_type = "internal"

Documentation

Index

• Constants
• Variables
• func NewMachine(handler Handler) *machine
• func NewSeriesMachine(handler Handler) *machine
• func NewStreamMachine(r io.Reader, handler Handler) *streamMachine
• type Handler
• type MetricHandler
  • func NewMetricHandler() *MetricHandler
  • func (h *MetricHandler) AddBool(key []byte, value []byte) error
  • func (h *MetricHandler) AddFloat(key []byte, value []byte) error
  • func (h *MetricHandler) AddInt(key []byte, value []byte) error
  • func (h *MetricHandler) AddString(key []byte, value []byte) error
  • func (h *MetricHandler) AddTag(key []byte, value []byte) error
  • func (h *MetricHandler) AddUint(key []byte, value []byte) error
  • func (h *MetricHandler) Metric() telegraf.Metric
  • func (h *MetricHandler) SetMeasurement(name []byte) error
  • func (h *MetricHandler) SetTimeFunc(f TimeFunc)
  • func (h *MetricHandler) SetTimePrecision(p time.Duration)
  • func (h *MetricHandler) SetTimestamp(tm []byte) error
• type ParseError
  • func (e *ParseError) Error() string
• type Parser
  • func (m Parser) Column() int
  • func (p *Parser) Init() error
  • func (p *Parser) InitFromConfig(config *parsers.Config) error
  • func (m Parser) LineNumber() int
  • func (m Parser) LineOffset() int
  • func (m Parser) Next() error
  • func (p *Parser) Parse(input []byte) ([]telegraf.Metric, error)
  • func (p *Parser) ParseLine(line string) (telegraf.Metric, error)
  • func (m Parser) Position() int
  • func (m Parser) SetData(data []byte)
  • func (p *Parser) SetDefaultTags(tags map[string]string)
  • func (p *Parser) SetTimeFunc(f TimeFunc)
  • func (p *Parser) SetTimePrecision(u time.Duration)
• type StreamParser
  • func NewStreamParser(r io.Reader) *StreamParser
  • func (sp *StreamParser) Column() int
  • func (sp *StreamParser) LineNumber() int
  • func (sp *StreamParser) LineOffset() int
  • func (sp *StreamParser) LineText() string
  • func (sp *StreamParser) Next() (telegraf.Metric, error)
  • func (sp *StreamParser) Position() int
  • func (sp *StreamParser) SetTimeFunc(f TimeFunc)
  • func (sp *StreamParser) SetTimePrecision(u time.Duration)
• type TimeFunc

Constants

const LineProtocol_en_align int = 85

const LineProtocol_en_discard_line int = 34

const LineProtocol_en_main int = 46

const LineProtocol_en_series int = 37

const LineProtocol_error int = 0

const LineProtocol_first_final int = 46

const LineProtocol_start int = 46

Variables

var (
	ErrNameParse      = errors.New("expected measurement name")
	ErrFieldParse     = errors.New("expected field")
	ErrTagParse       = errors.New("expected tag")
	ErrTimestampParse = errors.New("expected timestamp")
	ErrParse          = errors.New("parse error")
	EOF               = errors.New("EOF")
)

var (
	ErrNoMetric = errors.New("no metric in line")
)

Functions

func NewMachine

func NewMachine(handler Handler) *machine

func NewSeriesMachine

func NewSeriesMachine(handler Handler) *machine

func NewStreamMachine

func NewStreamMachine(r io.Reader, handler Handler) *streamMachine

Types

type Handler

type Handler interface {
	SetMeasurement(name []byte) error
	AddTag(key []byte, value []byte) error
	AddInt(key []byte, value []byte) error
	AddUint(key []byte, value []byte) error
	AddFloat(key []byte, value []byte) error
	AddString(key []byte, value []byte) error
	AddBool(key []byte, value []byte) error
	SetTimestamp(tm []byte) error
}

type MetricHandler

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

MetricHandler implements the Handler interface and produces telegraf.Metric.

func NewMetricHandler

func NewMetricHandler() *MetricHandler

func (*MetricHandler) AddBool

func (h *MetricHandler) AddBool(key []byte, value []byte) error

func (*MetricHandler) AddFloat

func (h *MetricHandler) AddFloat(key []byte, value []byte) error

func (*MetricHandler) AddInt

func (h *MetricHandler) AddInt(key []byte, value []byte) error

func (*MetricHandler) AddString

func (h *MetricHandler) AddString(key []byte, value []byte) error

func (*MetricHandler) AddTag

func (h *MetricHandler) AddTag(key []byte, value []byte) error

func (*MetricHandler) AddUint

func (h *MetricHandler) AddUint(key []byte, value []byte) error

func (*MetricHandler) Metric

func (h *MetricHandler) Metric() telegraf.Metric

func (*MetricHandler) SetMeasurement

func (h *MetricHandler) SetMeasurement(name []byte) error

func (*MetricHandler) SetTimeFunc

func (h *MetricHandler) SetTimeFunc(f TimeFunc)

func (*MetricHandler) SetTimePrecision

func (h *MetricHandler) SetTimePrecision(p time.Duration)

func (*MetricHandler) SetTimestamp

func (h *MetricHandler) SetTimestamp(tm []byte) error

type ParseError

type ParseError struct {
	Offset     int
	LineOffset int
	LineNumber int
	Column     int
	// contains filtered or unexported fields
}

ParseError indicates a error in the parsing of the text.

func (*ParseError) Error

func (e *ParseError) Error() string

type Parser

type Parser struct {
	DefaultTags map[string]string `toml:"-"`
	// If set to "series" a series machine will be initialized, defaults to regular machine
	Type string `toml:"-"`

	sync.Mutex
	// contains filtered or unexported fields
}

Parser is an InfluxDB Line Protocol parser that implements the parsers.Parser interface.

func (Parser) Column

func (m Parser) Column() int

Column returns the current column.

func (*Parser) Init

func (p *Parser) Init() error

func (*Parser) InitFromConfig

func (p *Parser) InitFromConfig(config *parsers.Config) error

InitFromConfig is a compatibility function to construct the parser the old way

func (Parser) LineNumber

func (m Parser) LineNumber() int

LineNumber returns the current line number. Lines are counted based on the regular expression `\r?\n`.

func (Parser) LineOffset

func (m Parser) LineOffset() int

LineOffset returns the byte offset of the current line.

func (Parser) Next

func (m Parser) Next() error

Next parses the next metric line and returns nil if it was successfully processed. If the line contains a syntax error an error is returned, otherwise if the end of file is reached before finding a metric line then EOF is returned.

func (*Parser) Parse

func (p *Parser) Parse(input []byte) ([]telegraf.Metric, error)

func (*Parser) ParseLine

func (p *Parser) ParseLine(line string) (telegraf.Metric, error)

func (Parser) Position

func (m Parser) Position() int

Position returns the current byte offset into the data.

func (Parser) SetData

func (m Parser) SetData(data []byte)

func (*Parser) SetDefaultTags

func (p *Parser) SetDefaultTags(tags map[string]string)

func (*Parser) SetTimeFunc

func (p *Parser) SetTimeFunc(f TimeFunc)

func (*Parser) SetTimePrecision

func (p *Parser) SetTimePrecision(u time.Duration)

type StreamParser

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

StreamParser is an InfluxDB Line Protocol parser. It is not safe for concurrent use in multiple goroutines.

func NewStreamParser

func NewStreamParser(r io.Reader) *StreamParser

func (*StreamParser) Column

func (sp *StreamParser) Column() int

Column returns the current column.

func (*StreamParser) LineNumber

func (sp *StreamParser) LineNumber() int

LineNumber returns the current line number. Lines are counted based on the regular expression `\r?\n`.

func (*StreamParser) LineOffset

func (sp *StreamParser) LineOffset() int

LineOffset returns the byte offset of the current line.

func (*StreamParser) LineText

func (sp *StreamParser) LineText() string

LineText returns the text of the current line that has been parsed so far.

func (*StreamParser) Next

func (sp *StreamParser) Next() (telegraf.Metric, error)

Next parses the next item from the stream. You can repeat calls to this function if it returns ParseError to get the next metric or error.

func (*StreamParser) Position

func (sp *StreamParser) Position() int

Position returns the current byte offset into the data.

func (*StreamParser) SetTimeFunc

func (sp *StreamParser) SetTimeFunc(f TimeFunc)

SetTimeFunc changes the function used to determine the time of metrics without a timestamp. The default TimeFunc is time.Now. Useful mostly for testing, or perhaps if you want all metrics to have the same timestamp.

func (*StreamParser) SetTimePrecision

func (sp *StreamParser) SetTimePrecision(u time.Duration)

type TimeFunc

type TimeFunc func() time.Time