gonx

README

gonx

gonx is Nginx access log reader library for Go. In fact you can use it for any format.

Usage

The library provides Reader type and two constructors for it.

Common constructor NewReader gets opened file (any io.Reader in fact) and log format of type string as argumets. Format is in form os nginx log_format string.

reader := gonx.NewReader(file, format)

NewNginxReader provides more magic. It gets log file io.Reader, nginx config file io.Reader and log_format name string as a third. The actual format for Parser will be extracted from given nginx config.

reader := gonx.NewNginxReader(file, nginxConfig, format_name)

Reader implements io.Reader. Here is example usage

for {
	rec, err := reader.Read()
	if err == io.EOF {
		break
	}
	// Process the record... e.g.
}

See more examples in example/*.go sources.

Performance

NOTE All benchmarks was made on my old 11" MacBook Air 2011, so you should get the better results for your brand new hardware ;-)

I have a few benchmarks for parsing string log record into Entry using gonx.Parser

BenchmarkParseSimpleLogRecord      100000            19457 ns/op
BenchmarkParseLogRecord             20000            84425 ns/op

And here is some real wold stats. I got ~300Mb log file with ~700K records and process with simple scripts.

• Reading whole file line by line with bufio.Scanner without any other processing takes a one second.
• Read in the same manner plus parsing with gonx.Parser takes about 80 seconds
• But for reading this file with gonx.Reader which parses records using separate goroutines it takes about 45 seconds (but I want to make it faster)

Format

As I said above this library is primary for nginx access log parsing, but it can be configured to parse any other format. NewReader accepts format argument, it will be transformed to regular expression and used for log line by line parsing. Format is nginx-like, here is example

`$remote_addr [$time_local] "$request"`

It should contain variables in form $name. The regular expression will be created using this string format representation

`^(?P<remote_addr>[^ ]+) \[(?P<time_local>[^]]+)\] "(?P<request>[^"]+)"$`

Reader.Read returns a record of type Entry (which is customized map[string][string]). For this example the returned record map will contain remote_addr, time_local and request keys filled with parsed values.

Stability

This library API and internal representation can be changed at any moment, but I guarantee that backward capability will be supported for the following public interfaces.

• func NewReader(logFile io.Reader, format string) *Reader
• func NewNginxReader(logFile io.Reader, nginxConf io.Reader, formatName string) (reader *Reader, err error)
• func (r *Reader) Read() (record Entry, err error)

Changelog

All major changes will be noticed in release notes.

Roadmap

I have no roadmap for this project at the moment for a few reasons. At the first, it is a simple library and I want to keep it. At the second, there is no feature requests, and for me, this library do its job. A few things may happen: the default binary, to use this not only as a library, and performance improvements if they will be needed.

Contributing

Fork the repo, create a feature branch then send me pull request. Feel free to create new issues or contact me using email.

Documentation

Index

• func MapReduce(file io.Reader, parser StringParser, reducer Reducer) chan *Entry
• type Avg
  • func (r *Avg) Reduce(input chan *Entry, output chan *Entry)
• type Chain
  • func NewChain(reducers ...Reducer) *Chain
  • func (r *Chain) Reduce(input chan *Entry, output chan *Entry)
• type Count
  • func (r *Count) Reduce(input chan *Entry, output chan *Entry)
• type Datetime
  • func (i *Datetime) Filter(entry *Entry) (validEntry *Entry)
  • func (i *Datetime) Reduce(input chan *Entry, output chan *Entry)
• type Entry
  • func NewEmptyEntry() *Entry
  • func NewEntry(fields Fields) *Entry
  • func (entry *Entry) Field(name string) (value string, err error)
  • func (entry *Entry) FieldsHash(fields []string) string
  • func (entry *Entry) FloatField(name string) (value float64, err error)
  • func (master *Entry) Merge(entry *Entry)
  • func (entry *Entry) Partial(fields []string) *Entry
  • func (entry *Entry) SetField(name string, value string)
  • func (entry *Entry) SetFloatField(name string, value float64)
  • func (entry *Entry) SetUintField(name string, value uint64)
• type Fields
• type Filter
• type GroupBy
  • func NewGroupBy(fields []string, reducers ...Reducer) *GroupBy
  • func (r *GroupBy) Reduce(input chan *Entry, output chan *Entry)
• type Parser
  • func NewNginxParser(conf io.Reader, name string) (parser *Parser, err error)
  • func NewParser(format string) *Parser
  • func (parser *Parser) ParseString(line string) (entry *Entry, err error)
• type ReadAll
  • func (r *ReadAll) Reduce(input chan *Entry, output chan *Entry)
• type Reader
  • func NewNginxReader(logFile io.Reader, nginxConf io.Reader, formatName string) (reader *Reader, err error)
  • func NewReader(logFile io.Reader, format string) *Reader
  • func (r *Reader) Read() (entry *Entry, err error)
• type Reducer
• type StringParser
• type Sum
  • func (r *Sum) Reduce(input chan *Entry, output chan *Entry)

Functions

func MapReduce

func MapReduce(file io.Reader, parser StringParser, reducer Reducer) chan *Entry

Iterate over given file and map each it's line into Entry record using parser and apply reducer to the Entries channel. Execution terminates when result will be readed from reducer's output channel, but the mapper works and fills input Entries channel until all lines will be read from the fiven file.

Types

type Avg

type Avg struct {
	Fields []string
}

Implements Reducer interface for average entries values calculation

func (*Avg) Reduce

func (r *Avg) Reduce(input chan *Entry, output chan *Entry)

Calculate average value for input channel Entries, using configured Fields of the struct. Write result to the output channel as map[string]float64

type Chain

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

Implements Reducer interface for chaining other reducers

func NewChain

func NewChain(reducers ...Reducer) *Chain

func (*Chain) Reduce

func (r *Chain) Reduce(input chan *Entry, output chan *Entry)

Apply chain of reducers to the input channel of entries and merge results

type Count

type Count struct {
}

Implements Reducer interface to count entries

func (*Count) Reduce

func (r *Count) Reduce(input chan *Entry, output chan *Entry)

Simply count entrries and write a sum to the output channel

type Datetime

type Datetime struct {
	Field  string
	Format string
	Start  time.Time
	End    time.Time
}

Implements Filter interface to filter Entries with timestamp fields within the specified datetime interval.

func (*Datetime) Filter

func (i *Datetime) Filter(entry *Entry) (validEntry *Entry)

Check field value to be in desired datetime range.

func (*Datetime) Reduce

func (i *Datetime) Reduce(input chan *Entry, output chan *Entry)

Reducer interface too. Go throught input and apply Filter.

type Entry

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

Parsed log record. Use Get method to retrieve a value by name instead of threating this as a map, because inner representation is in design.

func NewEmptyEntry

func NewEmptyEntry() *Entry

Creates an empty Entry to be filled later

func NewEntry

func NewEntry(fields Fields) *Entry

Creates an Entry with fiven fields

func (*Entry) Field

func (entry *Entry) Field(name string) (value string, err error)

Return entry field value by name or empty string and error if it does not exist.

func (*Entry) FieldsHash

func (entry *Entry) FieldsHash(fields []string) string

func (*Entry) FloatField

func (entry *Entry) FloatField(name string) (value float64, err error)

Return entry field value as float64. Rutuen nil if field does not exists and convertion error if cannot cast a type.

func (*Entry) Merge

func (master *Entry) Merge(entry *Entry)

Merge two entries by updating values for master entry with given.

func (*Entry) Partial

func (entry *Entry) Partial(fields []string) *Entry

func (*Entry) SetField

func (entry *Entry) SetField(name string, value string)

Field value setter

func (*Entry) SetFloatField

func (entry *Entry) SetFloatField(name string, value float64)

Float field value setter. It accepts float64, but still store it as a string in the same fields map. The precision is 2, its enough for log parsing task

func (*Entry) SetUintField

func (entry *Entry) SetUintField(name string, value uint64)

Integer field value setter. It accepts float64, but still store it as a string in the same fields map.

type Fields

type Fields map[string]string

Shortcut for the map of strings

type Filter

type Filter interface {
	Reducer
	Filter(*Entry) *Entry
}

Filter interface for Entries channel limiting.

Filter method should accept *Entry and return *Entry if it meets filter condition, otherwise it returns nil.

type GroupBy

type GroupBy struct {
	Fields []string
	// contains filtered or unexported fields
}

Implements Reducer interface to apply other reducers and get data grouped by given fields.

func NewGroupBy

func NewGroupBy(fields []string, reducers ...Reducer) *GroupBy

func (*GroupBy) Reduce

func (r *GroupBy) Reduce(input chan *Entry, output chan *Entry)

Apply related reducers and group data by Fields.

type Parser

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

Log record parser. Use specific constructors to initialize it.

func NewNginxParser

func NewNginxParser(conf io.Reader, name string) (parser *Parser, err error)

NewNginxParser parse nginx conf file to find log_format with given name and returns parser for this format. It returns an error if cannot find the needle.

func NewParser

func NewParser(format string) *Parser

Returns a new Parser, use given log format to create its internal strings parsing regexp.

func (*Parser) ParseString

func (parser *Parser) ParseString(line string) (entry *Entry, err error)

Parse log file line using internal format regexp. If line do not match given format an error will be returned.

type ReadAll

type ReadAll struct {
}

Implements Reducer interface for simple input entries redirection to the output channel.

func (*ReadAll) Reduce

func (r *ReadAll) Reduce(input chan *Entry, output chan *Entry)

Redirect input Entries channel directly to the output without any modifications. It is useful when you want jast to read file fast using asynchronous with mapper routines.

type Reader

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

Log file reader. Use specific constructors to create it.

func NewNginxReader

func NewNginxReader(logFile io.Reader, nginxConf io.Reader, formatName string) (reader *Reader, err error)

Creates reader for nginx log format. Nginx config parser will be used to get particular format from the conf file.

func NewReader

func NewReader(logFile io.Reader, format string) *Reader

Creates reader for custom log format.

func (*Reader) Read

func (r *Reader) Read() (entry *Entry, err error)

Get next parsed Entry from the log file. Return EOF if there is no Entries to read.

type Reducer

type Reducer interface {
	Reduce(input chan *Entry, output chan *Entry)
}

Reducer interface for Entries channel redure.

Each Reduce method should accept input channel of Entries, do it's job and the result should be written to the output channel.

It does not return values because usually it runs in a separate goroutine and it is handy to use channel for reduced data retrieval.

type StringParser

type StringParser interface {
	ParseString(line string) (entry *Entry, err error)
}

StringParser is the interface that wraps the ParseString method.

type Sum

type Sum struct {
	Fields []string
}

Implements Reducer interface for summarize Entry values for the given fields

func (*Sum) Reduce

func (r *Sum) Reduce(input chan *Entry, output chan *Entry)

Summarize given Entry fields and return a map with result for each field.