funnel

README

funnel

A new approach to logging

The 12 factor rule for logging says that an app "should not attempt to write to or manage logfiles. Instead, each running process writes its event stream, unbuffered, to stdout." The execution environment should take care of capturing the logs and perform further processing with it. Funnel is this "execution environment".

All you have to do from your app is to print your log line to stdout, and pipe it to funnel. You can still use any logging library inside your app to handle other stuff like log level, structured logging etc. But don't bother about the log destination. Let funnel take care whether you want to just write to files or stream your output to Kafka. Think of it as a fluentd/logstash replacement(with minimal features!) but having only stdin as an input.

Features quick tour

• Basic use case of logging to local files:
  • Rolling over to a new file
  • Deleting old files
  • Gzipping files
  • File rename policies
• Prepend each log line with a custom string
• Supports other target outputs like Kafka, ElasticSearch. More info below.
• Live reloading of config on file save. No more messing around with SIGHUP or SIGUSR1.

Quickstart

Grab the binary for your platform and the config file from here.

To run, just pipe the output of your app to the funnel binary. Note that, funnel only consumes from stdin, so you might need to redirect stderr to stdout.

$/etc/myapp/bin 2>&1 | funnel

P.S. You also need to drop the funnel binary to your $PATH.

Use in a systemd service

In the [service] section of your file, add these lines -

[Service]
EnvironmentFile=/path/to/env/file (Can contain funnel environment flags)
ExecStart=/bin/sh -c '/path/to/binary 2>&1 | funnel'
ExecStop=/usr/bin/pkill -TERM -f "/path/to/binary"

Target outputs and Use cases

Output	Description	Log format
File	Writes to local files	No format needed.
Kafka	Send your log stream to a Kafka topic	No format needed.
Redis pub-sub	Send your log stream to a Redis pub-sub channel	No format needed.
ElasticSearch	Index, Search and Analyze structured JSON logs	Logs have to be in JSON format
Amazon S3	Upload your logs to S3	No format needed.
InfluxDB	Use InfluxDB if your app emits timeseries data which needs to be queried and graphed	Logs have to be in JSON format with tags and fields as the keys
NATS	Send your log stream to a NATS subject	No format needed.

Further details on input log format along with examples can be found in the sample config file.

Configuration

The config can be specified in a .toml file. The file is part of the repo, which you can see here. All the settings are documented and are populated with the default values. The same defaults are embedded in the app itself, so the app can even run without a config file.

To read the config, the app looks for a file named funnel.toml in these locations one by one -

• /etc/funnel/funnel.toml
• $HOME/.config/funnel/funnel.toml
• ./funnel.toml (i.e. in the current directory of your target app)

You can place a global file in /etc/funnel/ and have separate files in each app directory to have config values overriding the global ones.

Environment variables are also supported and takes the highest precedence. To get the env variable name, just capitalize the config variable. For eg-

• logging.directory becomes LOGGING_DIRECTORY
• rollup.file_rename_policy becomes ROLLUP_FILE_RENAME_POLICY

Disabling outputs

In the case that you don't intend to use the Elasticsearch, InfluxDB, Kafka, Redis or S3 features, e.g. you just want to use the log rotation features, you can reduce the size of the binary by using build tags.

The build tags are:

• disableelasticsearch
• disableinfluxdb
• disablekafka
• disableredis
• disables3
• disablenats

e.g., to build without any of the above outputs:

go build -tags "disableelasticsearch disableinfluxdb disablekafka disableredis disables3 disablenats" ./cmd/funnel

Windows Support:

Funnel logs its internal errors to syslog. As syslog is not supported on windows, funnel doesn't work on windows - https://golang.org/pkg/log/syslog/#pkg-note-BUG

To make it work on windows, just replace the syslog writer with another logging library which atleast has an Err() method implemented.

Footnote - This project was heavily inspired from the logsend project.

Documentation

Index

• Constants
• Variables
• func GetConfig(v *viper.Viper, logger *syslog.Writer) (*Config, chan *Config, OutputWriter, error)
• func RegisterNewWriter(name string, factory OutputFactory)
• type ByModTime
  • func (a ByModTime) Len() int
  • func (a ByModTime) Less(i, j int) bool
  • func (a ByModTime) Swap(i, j int)
• type Config
• type ConfigValueError
  • func (e *ConfigValueError) Error() string
• type Consumer
  • func (c *Consumer) Start(inputStream io.Reader)
• type FileOutput
  • func (f *FileOutput) Close() error
• type LineProcessor
  • func GetLineProcessor(cfg *Config) LineProcessor
• type NoProcessor
  • func (*NoProcessor) Write(w io.Writer, line string) error
• type OutputFactory
• type OutputWriter
  • func GetOutputWriter(v *viper.Viper, logger *syslog.Writer) (OutputWriter, error)
• type SimpleLineProcessor
  • func (lp *SimpleLineProcessor) Write(w io.Writer, line string) error
• type TemplateLineProcessor
  • func (lp *TemplateLineProcessor) Write(w io.Writer, line string) error
• type UnregisteredOutputError
  • func (e *UnregisteredOutputError) Error() string

Constants

const (
	// config keys
	LoggingDirectory         = "logging.directory"
	LoggingActiveFileName    = "logging.active_file_name"
	RotationMaxLines         = "rotation.max_lines"
	RotationMaxFileSizeBytes = "rotation.max_file_size_bytes"
	FlushingTimeIntervalSecs = "flushing.time_interval_secs"
	PrependValue             = "misc.prepend_value"
	FileRenamePolicy         = "rollup.file_rename_policy"
	MaxAge                   = "rollup.max_age"
	MaxCount                 = "rollup.max_count"
	Gzip                     = "rollup.gzip"
	Target                   = "target.name"
)

XXX: Move it to constants.go if needed

Variables

var (
	// ErrInvalidFileRenamePolicy is raised for invalid values to file rename policy
	ErrInvalidFileRenamePolicy = errors.New(FileRenamePolicy + " can only be timestamp or serial")
	// ErrInvalidMaxAge is raised for invalid value in max age - life bad suffixes or no integer value at all
	ErrInvalidMaxAge = errors.New(MaxAge + " must end with either d or h and start with a number")
)

Functions

func GetConfig

func GetConfig(v *viper.Viper, logger *syslog.Writer) (*Config, chan *Config, OutputWriter, error)

GetConfig returns the config struct which is then passed to the consumer

func RegisterNewWriter

func RegisterNewWriter(name string, factory OutputFactory)

RegisterNewWriter is called by the init function from every output Adds the constructor to the registry

Types

type ByModTime

type ByModTime []os.FileInfo

ByModTime implements sorting for files by mod time from recent to old

func (ByModTime) Len

func (a ByModTime) Len() int

func (ByModTime) Less

func (a ByModTime) Less(i, j int) bool

func (ByModTime) Swap

func (a ByModTime) Swap(i, j int)

type Config

type Config struct {
	DirName        string
	ActiveFileName string

	RotationMaxLines int
	RotationMaxBytes uint64

	FlushingTimeIntervalSecs int

	PrependValue string

	FileRenamePolicy string
	MaxAge           int64
	MaxCount         int
	Gzip             bool

	Target string
}

Config holds all the config settings

type ConfigValueError

type ConfigValueError struct {
	Key string
}

ConfigValueError holds the error value if a config key contains an invalid value

func (*ConfigValueError) Error

func (e *ConfigValueError) Error() string

type Consumer

type Consumer struct {
	Config        *Config
	LineProcessor LineProcessor
	Logger        *syslog.Writer
	Writer        OutputWriter

	ReloadChan chan *Config
	// contains filtered or unexported fields
}

Consumer is the main struct which holds all the stuff necessary to run the code

func (*Consumer) Start

func (c *Consumer) Start(inputStream io.Reader)

Start takes the input stream and begins reading line by line buffering the output to a file and flushing at set intervals

type FileOutput

type FileOutput struct {
	*bufio.Writer
}

FileOutput is just an embed type which adds the Close method to buffered writer to satisfy the OutputWriter interface XXX: Might need to implement this in a better way

func (*FileOutput) Close

func (f *FileOutput) Close() error

Close function is just a no-op. Its never called.

type LineProcessor

type LineProcessor interface {
	Write(io.Writer, string) error
}

LineProcessor interface is passed down to the consumer which just calls the write function

func GetLineProcessor

func GetLineProcessor(cfg *Config) LineProcessor

GetLineProcessor function returns the particular processor depending on the config.

type NoProcessor

type NoProcessor struct {
}

NoProcessor is used when there is no prepend value. It just prints the line without any other action

func (*NoProcessor) Write

func (*NoProcessor) Write(w io.Writer, line string) error

type OutputFactory

type OutputFactory func(v *viper.Viper, logger *syslog.Writer) (OutputWriter, error)

OutputFactory is a function type which holds the output registry

type OutputWriter

type OutputWriter interface {
	io.Writer
	Flush() error
	Close() error
}

OutputWriter interface is to be implemented by all remote target writers It embeds the io.Writer and has basic function calls which should be needed by all writers Can be modified later if needed

func GetOutputWriter

func GetOutputWriter(v *viper.Viper, logger *syslog.Writer) (OutputWriter, error)

GetOutputWriter gets the constructor by extracting the target. Then returns the corresponding output writer by calling the constructor

type SimpleLineProcessor

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

SimpleLineProcessor is used when the prependValue is only a simple string It just concatenates the string with the line and prints it

func (*SimpleLineProcessor) Write

func (lp *SimpleLineProcessor) Write(w io.Writer, line string) error

type TemplateLineProcessor

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

TemplateLineProcessor is used when there is a template action in the prependValue It parses the prependValue and store the template. Then for every write call, it executes the template and writes it

func (*TemplateLineProcessor) Write

func (lp *TemplateLineProcessor) Write(w io.Writer, line string) error

type UnregisteredOutputError

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

UnregisteredOutputError holds the error if some target was passed from the config which was not registered

func (*UnregisteredOutputError) Error

func (e *UnregisteredOutputError) Error() string