logbus

README

Logbus

logbus is a log processing pipeline that from one input datasource like message queue, to multiple output channel like stdout, files, common store db (MongoDB, MySQL, ES).

You can implement your own input/output interface.

Note: logbus is not support distributed deploy.

Feature

• Input source
  • Redis List
• Output channel
  • Stdout
  • MongoDB
• Your can custom a transformer callback function processing log decide finally output log

Intro

logbus define a standard log format StdLog, I use logrusredis-hook output log to redis LIST by StdLogWash, but all of this is not necessary.

logbus a pipeline framework, the core task is Read and Write, checkout serve.go code. Only you need do is implement your owne input, output, transformer (of course, you can use default, or PR a new).

For data processing, it provide:

• DefaultTransformer do nothing, all log passed.
• StatLogTransformer only pass which log has a valid StateId

Usage

go get github.com/zhangjie2012/logbus

For example code example/main.go, log from redis LIST and to stdout/MongoDB, It's a real scenes for me (already running in prod env). You can implement your input, output, transformer and call Serve built up.

TODO

• merge code from loki-redis-reporter, support grafana/loki.
• output channel support local files and auto rotate.

Documentation

Index

• Constants
• Variables
• func Serve(ctx context.Context, in Input, outputs []Output)
• func StdLogWash(appName string, t time.Time, metadata logrus.Fields, caller *runtime.Frame, ...) []byte
• type Input
  • func NewRedisListInput(addr string, password string, db int, key string) (Input, error)
• type MongoOutput
  • func (out *MongoOutput) Close() error
  • func (out *MongoOutput) Write(l *StdLog) error
• type Output
  • func NewMongoOutput(host string, port int, username string, password string, dbName string, ...) (Output, error)
  • func NewStdoutOutput(transformer TransformerFunc) (Output, error)
• type RedisListInput
  • func (in *RedisListInput) Close() error
  • func (in *RedisListInput) Read(ctx context.Context) (*StdLog, error)
• type StdLog
  • func DefaultTransformer(l *StdLog) (*StdLog, bool)
  • func StatLogTransformer(l *StdLog) (*StdLog, bool)
• type StdoutOutput
  • func (out *StdoutOutput) Close() error
  • func (out *StdoutOutput) Write(l *StdLog) error
• type TransformerFunc

Constants

const StateIdInvalid string = "_invalid"

only state log has one valid id

Variables

var NoInputData = fmt.Errorf("no input data")

Functions

func Serve

func Serve(ctx context.Context, in Input, outputs []Output)

func StdLogWash

func StdLogWash(appName string, t time.Time, metadata logrus.Fields, caller *runtime.Frame, level logrus.Level, message string) []byte

StdLogWash redefined "LogWashFunc" in logrusredis-hook use "StandardLog" replace logrusredis-hook's "DefaultLogS"

Types

type Input

type Input interface {
	Read(ctx context.Context) (*StdLog, error)
	Close() error
}

func NewRedisListInput

func NewRedisListInput(addr string, password string, db int, key string) (Input, error)

type MongoOutput

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

func (*MongoOutput) Close

func (out *MongoOutput) Close() error

func (*MongoOutput) Write

func (out *MongoOutput) Write(l *StdLog) error

type Output

type Output interface {
	Write(l *StdLog) error
	Close() error
}

func NewMongoOutput

func NewMongoOutput(host string, port int, username string, password string, dbName string,
	transformer TransformerFunc) (Output, error)

func NewStdoutOutput

func NewStdoutOutput(transformer TransformerFunc) (Output, error)

type RedisListInput

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

RedisListInput data from redis LIST

func (*RedisListInput) Close

func (in *RedisListInput) Close() error

func (*RedisListInput) Read

func (in *RedisListInput) Read(ctx context.Context) (*StdLog, error)

type StdLog

type StdLog struct {
	AppName     string                 `msgpack:"appname" bson:"appname"`
	Timestamp   int64                  `msgpack:"timestamp" bson:"timestamp"`
	Level       string                 `msgpack:"level" bson:"level"`
	StateId     string                 `msgpack:"stateid" bson:"stateid"`
	Caller      string                 `msgpack:"caller" bson:"caller"`
	Message     string                 `msgpack:"message" bson:"message"`
	Annotations map[string]interface{} `msgpack:"annotations" bson:"annotations"`
}

StandardLog standard log format

• AppName define where log from
• Annotations for struct log or message expand

func DefaultTransformer

func DefaultTransformer(l *StdLog) (*StdLog, bool)

func StatLogTransformer

func StatLogTransformer(l *StdLog) (*StdLog, bool)

type StdoutOutput

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

func (*StdoutOutput) Close

func (out *StdoutOutput) Close() error

func (*StdoutOutput) Write

func (out *StdoutOutput) Write(l *StdLog) error

type TransformerFunc

type TransformerFunc func(l *StdLog) (*StdLog, bool)