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

package influx

import (
	"errors"
)

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

%%{
machine LineProtocol;

action begin {
	m.pb = m.p
}

action yield {
	yield = true
	fnext align;
	fbreak;
}

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

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

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

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

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

action hold_recover {
	fhold;
	fgoto main;
}

action discard {
	fgoto align;
}

action name {
	m.handler.SetMeasurement(m.text())
}

action tagkey {
	key = m.text()
}

action tagvalue {
	m.handler.AddTag(key, m.text())
}

action fieldkey {
	key = m.text()
}

action integer {
	m.handler.AddInt(key, m.text())
}

action unsigned {
	m.handler.AddUint(key, m.text())
}

action float {
	m.handler.AddFloat(key, m.text())
}

action bool {
	m.handler.AddBool(key, m.text())
}

action string {
	m.handler.AddString(key, m.text())
}

action timestamp {
	m.handler.SetTimestamp(m.text())
}

ws =
	[\t\v\f ];

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 =
	[^\n\f\r\\"] | '\\' [\\"];

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] );

tagkey =
	tagchar+ >begin %tagkey;

tagvalue =
	tagchar+ >begin %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 %name;

newline =
	[\r\n];

comment =
	'#' (any -- newline)* newline;

eol =
	ws* newline? >yield %eof(yield);

line =
	measurement
	tagset
	(ws+ fieldset) $err(field_error)
	(ws+ timestamp)? $err(timestamp_error)
	eol;

# The main machine parses a single line of line protocol.
main := line $err(parse_error);

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

# 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.
align :=
	(space* comment)* space* measurement_start @hold_recover %eof(yield);

series := measurement tagset $err(parse_error) eol;
}%%

%% write data;

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

type machine struct {
	data       []byte
	cs         int
	p, pe, eof int
	pb         int
	handler    Handler
	initState  int
	err        error
}

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

	%% access m.;
	%% variable p m.p;
	%% 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.pe = len(data)
	m.eof = len(data)
	m.err = nil

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

// ParseLine parses a line of input and returns true if more data can be
// parsed.
func (m *machine) ParseLine() bool {
	if m.data == nil || m.p >= m.pe {
		m.err = nil
		return false
	}

	m.err = nil
	var key []byte
	var yield bool

	%% write exec;

	// Even if there was an error, return true. On the next call to this
	// function we will attempt to scan to the next line of input and recover.
	if m.err != nil {
		return true
	}

	// Don't check the error state in the case that we just yielded, because
	// the yield indicates we just completed parsing a line.
	if !yield && m.cs == LineProtocol_error {
		m.err = ErrParse
		return true
	}

	return true
}

// Err returns the error that occurred on the last call to ParseLine.  If the
// result is nil, then the line was parsed successfully.
func (m *machine) Err() error {
	return m.err
}

// Position returns the current position into the input.
func (m *machine) Position() int {
	return m.p
}

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