telegraf/plugins/parsers/influx/machine.go.rl

package influx

import (
	"errors"
	"io"
)

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")
)

%%{
machine LineProtocol;

action begin {
	m.pb = m.p
}

action name_error {
	err = ErrNameParse
	fhold;
	fnext discard_line;
	fbreak;
}

action field_error {
	err = ErrFieldParse
	fhold;
	fnext discard_line;
	fbreak;
}

action tagset_error {
	err = ErrTagParse
	fhold;
	fnext discard_line;
	fbreak;
}

action timestamp_error {
	err = ErrTimestampParse
	fhold;
	fnext discard_line;
	fbreak;
}

action parse_error {
	err = ErrParse
	fhold;
	fnext discard_line;
	fbreak;
}

action align_error {
	err = ErrParse
	fnext discard_line;
	fbreak;
}

action hold_recover {
	fhold;
	fgoto main;
}

action goto_align {
	fgoto align;
}

action begin_metric {
	m.beginMetric = true
}

action name {
	err = m.handler.SetMeasurement(m.text())
	if err != nil {
		fhold;
		fnext discard_line;
		fbreak;
	}
}

action tagkey {
	m.key = m.text()
}

action tagvalue {
	err = m.handler.AddTag(m.key, m.text())
	if err != nil {
		fhold;
		fnext discard_line;
		fbreak;
	}
}

action fieldkey {
	m.key = m.text()
}

action integer {
	err = m.handler.AddInt(m.key, m.text())
	if err != nil {
		fhold;
		fnext discard_line;
		fbreak;
	}
}

action unsigned {
	err = m.handler.AddUint(m.key, m.text())
	if err != nil {
		fhold;
		fnext discard_line;
		fbreak;
	}
}

action float {
	err = m.handler.AddFloat(m.key, m.text())
	if err != nil {
		fhold;
		fnext discard_line;
		fbreak;
	}
}

action bool {
	err = m.handler.AddBool(m.key, m.text())
	if err != nil {
		fhold;
		fnext discard_line;
		fbreak;
	}
}

action string {
	err = m.handler.AddString(m.key, m.text())
	if err != nil {
		fhold;
		fnext discard_line;
		fbreak;
	}
}

action timestamp {
	err = m.handler.SetTimestamp(m.text())
	if err != nil {
		fhold;
		fnext discard_line;
		fbreak;
	}
}

action incr_newline {
	m.lineno++
	m.sol = m.p
	m.sol++ // next char will be the first column in the line
}

action eol {
	m.finishMetric = true
	fnext align;
	fbreak;
}

action finish_metric {
	m.finishMetric = true
}

ws =
	[\t\v\f ];

newline =
	'\r'? '\n' >incr_newline;

non_zero_digit =
	[1-9];

integer =
	'-'? ( digit | ( non_zero_digit digit* ) );

unsigned =
	( digit | ( non_zero_digit digit* ) );

number =
	'-'? (digit+ ('.' digit*)? | '.' digit+);

scientific =
	number 'e'i ["\-+"]? digit+;

timestamp =
	('-'? digit{1,19}) >begin %timestamp;

fieldkeychar =
	[^\t\n\f\r ,=\\] | ( '\\' [^\t\n\f\r] );

fieldkey =
	fieldkeychar+ >begin %fieldkey;

fieldfloat =
	(scientific | number) >begin %float;

fieldinteger =
	(integer 'i') >begin %integer;

fieldunsigned =
	(unsigned 'u') >begin %unsigned;

false =
	"false" | "FALSE" | "False" | "F" | "f";

true =
	"true" | "TRUE" | "True" | "T" | "t";

fieldbool =
	(true | false) >begin %bool;

fieldstringchar =
	[^\f\r\n\\"] | '\\' [\\"] | newline;

fieldstring =
	fieldstringchar* >begin %string;

fieldstringquoted =
	'"' fieldstring '"';

fieldvalue = fieldinteger | fieldunsigned | fieldfloat | fieldstringquoted | fieldbool;

field =
	fieldkey '=' fieldvalue;

fieldset =
	field ( ',' field )*;

tagchar =
	[^\t\n\f\r ,=\\] | ( '\\' [^\t\n\f\r\\] ) | '\\\\' %to{ fhold; };

tagkey =
	tagchar+ >begin %tagkey;

tagvalue =
	tagchar+ >begin %eof(tagvalue) %tagvalue;

tagset =
	((',' tagkey '=' tagvalue) $err(tagset_error))*;

measurement_chars =
	[^\t\n\f\r ,\\] | ( '\\' [^\t\n\f\r] );

measurement_start =
	measurement_chars - '#';

measurement =
	(measurement_start measurement_chars*) >begin %eof(name) %name;

eol_break =
	newline %to(eol)
	;

metric =
	measurement >err(name_error)
	tagset
	ws+ fieldset $err(field_error)
	(ws+ timestamp)? $err(timestamp_error)
	;

line_with_term =
	ws* metric ws* eol_break
	;

line_without_term =
	ws* metric ws*
	;

main :=
	(line_with_term*
	(line_with_term | line_without_term?)
    ) >begin_metric %eof(finish_metric)
	;

# The discard_line machine discards the current line.  Useful for recovering
# on the next line when an error occurs.
discard_line :=
	(any -- newline)* newline @goto_align;

commentline =
	ws* '#' (any -- newline)* newline;

emptyline =
	ws* newline;

# The align machine scans forward to the start of the next line.  This machine
# is used to skip over whitespace and comments, keeping this logic out of the
# main machine.
#
# Skip valid lines that don't contain line protocol, any other data will move
# control to the main parser via the err action.
align :=
	(emptyline | commentline | ws+)* %err(hold_recover);

# Series is a machine for matching measurement+tagset
series :=
	(measurement >err(name_error) tagset eol_break?)
	>begin_metric
	;
}%%

%% write data;

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 machine struct {
	data         []byte
	cs           int
	p, pe, eof   int
	pb           int
	lineno       int
	sol          int
	handler      Handler
	initState    int
	key          []byte
	beginMetric  bool
	finishMetric bool
}

func NewMachine(handler Handler) *machine {
	m := &machine{
		handler: handler,
		initState: LineProtocol_en_align,
	}

	%% access m.;
	%% variable p m.p;
	%% variable cs m.cs;
	%% variable pe m.pe;
	%% variable eof m.eof;
	%% variable data m.data;
	%% write init;

	return m
}

func NewSeriesMachine(handler Handler) *machine {
	m := &machine{
		handler: handler,
		initState: LineProtocol_en_series,
	}

	%% access m.;
	%% variable p m.p;
	%% variable pe m.pe;
	%% variable eof m.eof;
	%% variable data m.data;
	%% write init;

	return m
}

func (m *machine) SetData(data []byte) {
	m.data = data
	m.p = 0
	m.pb = 0
	m.lineno = 1
	m.sol = 0
	m.pe = len(data)
	m.eof = len(data)
	m.key = nil
	m.beginMetric = false
	m.finishMetric = false

	%% write init;
	m.cs = m.initState
}

// 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 (m *machine) Next() error {
	if m.p == m.pe && m.pe == m.eof {
		return EOF
	}

	m.key = nil
	m.beginMetric = false
	m.finishMetric = false

	return m.exec()
}

func (m *machine) exec() error {
	var err error
	%% write exec;

	if err != nil {
		return err
	}

	// This would indicate an error in the machine that was reported with a
	// more specific error.  We return a generic error but this should
	// possibly be a panic.
	if m.cs == %%{ write error; }%% {
		m.cs = LineProtocol_en_discard_line
		return ErrParse
	}

	// If we haven't found a metric line yet and we reached the EOF, report it
	// now.  This happens when the data ends with a comment or whitespace.
	//
	// Otherwise we have successfully parsed a metric line, so if we are at
	// the EOF we will report it the next call.
	if !m.beginMetric && m.p == m.pe && m.pe == m.eof {
		return EOF
	}

	return nil
}

// Position returns the current byte offset into the data.
func (m *machine) Position() int {
	return m.p
}

// LineOffset returns the byte offset of the current line.
func (m *machine) LineOffset() int {
	return m.sol
}

// LineNumber returns the current line number.  Lines are counted based on the
// regular expression `\r?\n`.
func (m *machine) LineNumber() int {
	return m.lineno
}

// Column returns the current column.
func (m *machine) Column() int {
	lineOffset := m.p - m.sol
	return lineOffset + 1
}

func (m *machine) text() []byte {
	return m.data[m.pb:m.p]
}

type streamMachine struct {
	machine *machine
	reader  io.Reader
}

func NewStreamMachine(r io.Reader, handler Handler) *streamMachine {
	m := &streamMachine{
		machine: NewMachine(handler),
		reader: r,
	}

	m.machine.SetData(make([]byte, 1024))
	m.machine.pe = 0
	m.machine.eof = -1
	return m
}

func (m *streamMachine) Next() error {
	// Check if we are already at EOF, this should only happen if called again
	// after already returning EOF.
	if m.machine.p == m.machine.pe && m.machine.pe == m.machine.eof {
		return EOF
	}

	copy(m.machine.data, m.machine.data[m.machine.p:])
	m.machine.pe = m.machine.pe - m.machine.p
	m.machine.sol = m.machine.sol - m.machine.p
	m.machine.pb = 0
	m.machine.p = 0
	m.machine.eof = -1

	m.machine.key = nil
	m.machine.beginMetric = false
	m.machine.finishMetric = false

	for {
		// Expand the buffer if it is full
		if m.machine.pe == len(m.machine.data) {
			expanded := make([]byte, 2 * len(m.machine.data))
			copy(expanded, m.machine.data)
			m.machine.data = expanded
		}

		n, err := m.reader.Read(m.machine.data[m.machine.pe:])
		if n == 0 && err == io.EOF {
			m.machine.eof = m.machine.pe
		} else if err != nil && err != io.EOF {
			return err
		}

		m.machine.pe += n

		err = m.machine.exec()
		if err != nil {
			return err
		}

		// If we have successfully parsed a full metric line break out
		if m.machine.finishMetric {
			break
		}

	}

	return nil
}

// Position returns the current byte offset into the data.
func (m *streamMachine) Position() int {
	return m.machine.Position()
}

// LineOffset returns the byte offset of the current line.
func (m *streamMachine) LineOffset() int {
	return m.machine.LineOffset()
}

// LineNumber returns the current line number.  Lines are counted based on the
// regular expression `\r?\n`.
func (m *streamMachine) LineNumber() int {
	return m.machine.LineNumber()
}

// Column returns the current column.
func (m *streamMachine) Column() int {
	return m.machine.Column()
}

// LineText returns the text of the current line that has been parsed so far.
func (m *streamMachine) LineText() string {
	return string(m.machine.data[0:m.machine.p])
}