diff --git a/cmd/config_logging.go b/cmd/config_logging.go index cb06888..9579778 100644 --- a/cmd/config_logging.go +++ b/cmd/config_logging.go @@ -3,32 +3,18 @@ package cmd import ( "github.com/mitchellh/mapstructure" "github.com/pkg/errors" - "github.com/sirupsen/logrus" + "github.com/zrepl/zrepl/logger" "os" "time" ) type LoggingConfig struct { - Stdout struct { - Level logrus.Level - Format LogFormat - } - TCP *TCPLoggingConfig -} - -type TCPLoggingConfig struct { - Level logrus.Level - Format LogFormat - Net string - Address string - RetryInterval time.Duration + Outlets logger.Outlets } func parseLogging(i interface{}) (c *LoggingConfig, err error) { c = &LoggingConfig{} - c.Stdout.Level = logrus.WarnLevel - c.Stdout.Format = LogFormatHuman if i == nil { return c, nil } @@ -50,89 +36,75 @@ func parseLogging(i interface{}) (c *LoggingConfig, err error) { return nil, errors.Wrap(err, "mapstructure error") } + c.Outlets = logger.NewOutlets() + if asMap.Stdout.Level != "" { - lvl, err := logrus.ParseLevel(asMap.Stdout.Level) + + out := WriterOutlet{ + HumanFormatter{}, + os.Stdout, + } + + level, err := logger.ParseLevel(asMap.Stdout.Level) if err != nil { return nil, errors.Wrap(err, "cannot parse stdout log level") } - c.Stdout.Level = lvl - } - if asMap.Stdout.Format != "" { - format, err := parseLogFormat(asMap.Stdout.Format) - if err != nil { - return nil, errors.Wrap(err, "cannot parse log format") + + if asMap.Stdout.Format != "" { + out.Formatter, err = parseLogFormat(asMap.Stdout.Format) + if err != nil { + return nil, errors.Wrap(err, "cannot parse log format") + } } - c.Stdout.Format = format + + c.Outlets.Add(out, level) + } if asMap.TCP.Address != "" { - c.TCP = &TCPLoggingConfig{} - c.TCP.Format, err = parseLogFormat(asMap.TCP.Format) + + out := &TCPOutlet{} + + out.Formatter, err = parseLogFormat(asMap.TCP.Format) if err != nil { - return nil, errors.Wrap(err, "cannot parse log format") + return nil, errors.Wrap(err, "cannot parse log 'format'") } - c.TCP.Level, err = logrus.ParseLevel(asMap.TCP.Level) + + lvl, err := logger.ParseLevel(asMap.TCP.Level) if err != nil { - return nil, errors.Wrap(err, "cannot parse level") + return nil, errors.Wrap(err, "cannot parse 'level'") } - c.TCP.RetryInterval, err = time.ParseDuration(asMap.TCP.RetryInterval) - c.TCP.Net, c.TCP.Address = asMap.TCP.Net, asMap.TCP.Address + + out.RetryInterval, err = time.ParseDuration(asMap.TCP.RetryInterval) + if err != nil { + return nil, errors.Wrap(err, "cannot parse 'retry_interval'") + } + + out.Net, out.Address = asMap.TCP.Net, asMap.TCP.Address + + c.Outlets.Add(out, lvl) } return c, nil } -type LogFormat string - -const ( - LogFormatHuman LogFormat = "human" - LogFormatLogfmt LogFormat = "logfmt" - LogFormatJSON LogFormat = "json" -) - -func (f LogFormat) Formatter() logrus.Formatter { - switch f { - case LogFormatHuman: - return HumanFormatter{} - case LogFormatLogfmt: - return &logrus.TextFormatter{} - case LogFormatJSON: - return &logrus.JSONFormatter{} - default: - panic("incomplete implementation") - } -} - -var LogFormats []LogFormat = []LogFormat{LogFormatHuman, LogFormatLogfmt, LogFormatJSON} - -func parseLogFormat(i interface{}) (f LogFormat, err error) { +func parseLogFormat(i interface{}) (f EntryFormatter, err error) { var is string switch j := i.(type) { case string: is = j default: - return "", errors.Errorf("invalid log format: wrong type: %T", i) + return nil, errors.Errorf("invalid log format: wrong type: %T", i) } - for _, f := range LogFormats { - if string(f) == is { - return f, nil - } + switch is { + case "human": + return HumanFormatter{}, nil + case "json": + return &JSONFormatter{}, nil + default: + return nil, errors.Errorf("invalid log format: '%s'", is) } - return "", errors.Errorf("invalid log format: '%s'", is) -} - -func (c *LoggingConfig) MakeLogrus() (l logrus.FieldLogger) { - - log := logrus.New() - log.Out = os.Stdout - log.Level = logrus.DebugLevel // FIXTHIS IN LOGRUS - log.Formatter = c.Stdout.Format.Formatter() - - th := &TCPHook{Formatter: JSONFormatter{}, MinLevel: c.TCP.Level, Net: c.TCP.Net, Address: c.TCP.Address, RetryInterval: c.TCP.RetryInterval} - log.Hooks.Add(th) - - return log } diff --git a/cmd/daemon.go b/cmd/daemon.go index de17adf..21f5e51 100644 --- a/cmd/daemon.go +++ b/cmd/daemon.go @@ -4,9 +4,11 @@ import ( "context" "fmt" "github.com/spf13/cobra" + "github.com/zrepl/zrepl/logger" "os" "os/signal" "syscall" + "time" ) // daemonCmd represents the daemon command @@ -33,7 +35,8 @@ func doDaemon(cmd *cobra.Command, args []string) { os.Exit(1) } - log := conf.Global.logging.MakeLogrus() + log := logger.NewLogger(conf.Global.logging.Outlets, 1*time.Second) + log.Debug("starting daemon") ctx := context.WithValue(context.Background(), contextKeyLog, log) ctx = context.WithValue(ctx, contextKeyLog, log) diff --git a/cmd/logging_formatters.go b/cmd/logging_formatters.go new file mode 100644 index 0000000..23f29be --- /dev/null +++ b/cmd/logging_formatters.go @@ -0,0 +1,114 @@ +package cmd + +import ( + "bytes" + "encoding/json" + "fmt" + "github.com/pkg/errors" + "github.com/zrepl/zrepl/logger" + "strings" + "time" +) + +type EntryFormatter interface { + Format(e *logger.Entry) ([]byte, error) +} + +const ( + FieldLevel = "level" + FieldMessage = "msg" + FieldTime = "time" +) + +const ( + logJobField string = "job" + logTaskField string = "task" + logFSField string = "filesystem" + logMapFromField string = "map_from" + logMapToField string = "map_to" + logIncFromField string = "inc_from" + logIncToField string = "inc_to" +) + +type NoFormatter struct{} + +func (f NoFormatter) Format(e *logger.Entry) ([]byte, error) { + return []byte(e.Message), nil +} + +type HumanFormatter struct{} + +func (f HumanFormatter) Format(e *logger.Entry) (out []byte, err error) { + + var line bytes.Buffer + + fmt.Fprintf(&line, "[%s]", e.Level.Short()) + + prefixFields := []string{logJobField, logTaskField, logFSField} + prefixed := make(map[string]bool, len(prefixFields)+2) + for _, field := range prefixFields { + val, ok := e.Fields[field].(string) + if ok { + fmt.Fprintf(&line, "[%s]", val) + prefixed[field] = true + } else { + break + } + } + // even more prefix fields + mapFrom, mapFromOk := e.Fields[logMapFromField].(string) + mapTo, mapToOk := e.Fields[logMapToField].(string) + if mapFromOk && mapToOk { + fmt.Fprintf(&line, "[%s => %s]", mapFrom, mapTo) + prefixed[logMapFromField], prefixed[logMapToField] = true, true + } + incFrom, incFromOk := e.Fields[logIncFromField].(string) + incTo, incToOk := e.Fields[logIncToField].(string) + if incFromOk && incToOk { + fmt.Fprintf(&line, "[%s => %s]", incFrom, incTo) + prefixed[logIncFromField], prefixed[logIncToField] = true, true + } + + fmt.Fprintf(&line, ": %s", e.Message) + + for field, value := range e.Fields { + + if prefixed[field] { + continue + } + + if strings.ContainsAny(field, " \t") { + return nil, errors.Errorf("field must not contain whitespace: '%s'", field) + } + fmt.Fprintf(&line, " %s=\"%s\"", field, value) + } + + return line.Bytes(), nil +} + +type JSONFormatter struct{} + +func (f JSONFormatter) Format(e *logger.Entry) ([]byte, error) { + data := make(logger.Fields, len(e.Fields)+3) + for k, v := range e.Fields { + switch v := v.(type) { + case error: + // Otherwise errors are ignored by `encoding/json` + // https://github.com/sirupsen/logrus/issues/137 + data[k] = v.Error() + default: + _, err := json.Marshal(v) + if err != nil { + return nil, errors.Errorf("field is not JSON encodable: %s", k) + } + data[k] = v + } + } + + data[FieldMessage] = e.Message + data[FieldTime] = e.Time.Format(time.RFC3339) + data[FieldLevel] = e.Level + + return json.Marshal(data) + +} diff --git a/cmd/logging_outlets.go b/cmd/logging_outlets.go new file mode 100644 index 0000000..e0d0d26 --- /dev/null +++ b/cmd/logging_outlets.go @@ -0,0 +1,63 @@ +package cmd + +import ( + "context" + "github.com/pkg/errors" + "github.com/zrepl/zrepl/logger" + "io" + "net" + "time" +) + +type WriterOutlet struct { + Formatter EntryFormatter + Writer io.Writer +} + +func (h WriterOutlet) WriteEntry(ctx context.Context, entry logger.Entry) error { + bytes, err := h.Formatter.Format(&entry) + if err != nil { + return err + } + _, err = h.Writer.Write(bytes) + h.Writer.Write([]byte("\n")) + return err +} + +type TCPOutlet struct { + Formatter EntryFormatter + Net, Address string + Dialer net.Dialer + RetryInterval time.Duration + conn net.Conn + retry time.Time +} + +func (h *TCPOutlet) WriteEntry(ctx context.Context, e logger.Entry) error { + + b, err := h.Formatter.Format(&e) + if err != nil { + return err + } + + if h.conn == nil { + if time.Now().Sub(h.retry) < h.RetryInterval { + return nil // this is not an error toward the logger + //return errors.New("TCP hook reconnect prohibited by retry interval") + } + h.conn, err = h.Dialer.DialContext(ctx, h.Net, h.Address) + if err != nil { + h.retry = time.Now() + return errors.Wrap(err, "cannot dial") + } + } + + _, err = h.conn.Write(b) + if err != nil { + h.conn.Close() + h.conn = nil + return err + } + + return nil +} diff --git a/cmd/logrus.go b/cmd/logrus.go deleted file mode 100644 index 9519fda..0000000 --- a/cmd/logrus.go +++ /dev/null @@ -1,171 +0,0 @@ -package cmd - -import ( - "bytes" - "encoding/json" - "fmt" - "github.com/pkg/errors" - "github.com/sirupsen/logrus" - "net" - "strings" - "time" -) - -const ( - logJobField string = "job" - logTaskField string = "task" - logFSField string = "filesystem" - logMapFromField string = "map_from" - logMapToField string = "map_to" - logIncFromField string = "inc_from" - logIncToField string = "inc_to" -) - -type NoFormatter struct{} - -func (f NoFormatter) Format(e *logrus.Entry) ([]byte, error) { - return []byte(e.Message), nil -} - -type HumanFormatter struct{} - -func (f HumanFormatter) shortLevel(l logrus.Level) string { - switch l { - case logrus.DebugLevel: - return "DBG" - case logrus.InfoLevel: - return "INF" - case logrus.WarnLevel: - return "WRN" - case logrus.ErrorLevel: - return "ERR" - case logrus.PanicLevel: - return "PNC" - } - panic("incomplete implementation") -} - -func (f HumanFormatter) Format(e *logrus.Entry) (out []byte, err error) { - - var line bytes.Buffer - - fmt.Fprintf(&line, "[%s]", f.shortLevel(e.Level)) - - prefixFields := []string{logJobField, logTaskField, logFSField} - prefixed := make(map[string]bool, len(prefixFields)+2) - for _, field := range prefixFields { - val, ok := e.Data[field].(string) - if ok { - fmt.Fprintf(&line, "[%s]", val) - prefixed[field] = true - } else { - break - } - } - // even more prefix fields - mapFrom, mapFromOk := e.Data[logMapFromField].(string) - mapTo, mapToOk := e.Data[logMapToField].(string) - if mapFromOk && mapToOk { - fmt.Fprintf(&line, "[%s => %s]", mapFrom, mapTo) - prefixed[logMapFromField], prefixed[logMapToField] = true, true - } - incFrom, incFromOk := e.Data[logIncFromField].(string) - incTo, incToOk := e.Data[logIncToField].(string) - if incFromOk && incToOk { - fmt.Fprintf(&line, "[%s => %s]", incFrom, incTo) - prefixed[logIncFromField], prefixed[logIncToField] = true, true - } - - fmt.Fprintf(&line, ": %s", e.Message) - - for field, value := range e.Data { - - if prefixed[field] { - continue - } - - if strings.ContainsAny(field, " \t") { - return nil, errors.Errorf("field must not contain whitespace: '%s'", field) - } - fmt.Fprintf(&line, " %s=\"%s\"", field, value) - } - - fmt.Fprintf(&line, "\n") - - return line.Bytes(), nil -} - -type JSONFormatter struct{} - -func (f JSONFormatter) Format(e *logrus.Entry) ([]byte, error) { - data := make(logrus.Fields, len(e.Data)+3) - for k, v := range e.Data { - switch v := v.(type) { - case error: - // Otherwise errors are ignored by `encoding/json` - // https://github.com/sirupsen/logrus/issues/137 - data[k] = v.Error() - default: - _, err := json.Marshal(v) - if err != nil { - return nil, errors.Errorf("field is not JSON encodable: %s", k) - } - data[k] = v - } - } - - data["msg"] = e.Message - data["time"] = e.Time.Format(time.RFC3339) - data["level"] = e.Level - - return json.Marshal(data) - -} - -type TCPHook struct { - Formatter logrus.Formatter - MinLevel logrus.Level - Net, Address string - RetryInterval time.Duration - conn net.Conn - retry time.Time -} - -func (h *TCPHook) Levels() []logrus.Level { - for i := range logrus.AllLevels { // assume it's ordered - if logrus.AllLevels[i] == h.MinLevel { - return logrus.AllLevels[:i] - } - } - return logrus.AllLevels -} - -func (h *TCPHook) Fire(e *logrus.Entry) error { - b, err := h.Formatter.Format(e) - if err != nil { - return err - } - - if h.conn == nil { - if time.Now().Sub(h.retry) < h.RetryInterval { - return errors.New("TCP hook reconnect prohibited by retry interval") - } - h.conn, err = net.Dial(h.Net, h.Address) - if err != nil { - h.retry = time.Now() - return errors.Wrap(err, "cannot dial") - } - } - - _, err = h.conn.Write(b) - if err != nil { - h.conn.Close() - h.conn = nil - } - - return nil -} - -type nopWriter int - -func (w nopWriter) Write(p []byte) (n int, err error) { return len(p), nil } diff --git a/cmd/main.go b/cmd/main.go index ca81efe..19719a0 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -11,8 +11,8 @@ package cmd import ( - "github.com/sirupsen/logrus" "github.com/spf13/cobra" + "github.com/zrepl/zrepl/logger" ) // @@ -20,7 +20,7 @@ import ( // Printf(format string, v ...interface{}) //} -type Logger logrus.FieldLogger +type Logger = *logger.Logger var RootCmd = &cobra.Command{ Use: "zrepl", diff --git a/cmd/test.go b/cmd/test.go index 91a07fb..aae9330 100644 --- a/cmd/test.go +++ b/cmd/test.go @@ -10,9 +10,10 @@ import ( "strings" "github.com/kr/pretty" - "github.com/sirupsen/logrus" "github.com/spf13/cobra" + "github.com/zrepl/zrepl/logger" "github.com/zrepl/zrepl/zfs" + "time" ) var testCmd = &cobra.Command{ @@ -64,8 +65,9 @@ func init() { func testCmdGlobalInit(cmd *cobra.Command, args []string) { - log := logrus.New() - log.Formatter = NoFormatter{} + out := logger.NewOutlets() + out.Add(WriterOutlet{NoFormatter{}, os.Stdout}, logger.Info) + log := logger.NewLogger(out, 1*time.Second) testCmdGlobal.log = log var err error diff --git a/logger/datastructures.go b/logger/datastructures.go new file mode 100644 index 0000000..fc0ff5c --- /dev/null +++ b/logger/datastructures.go @@ -0,0 +1,86 @@ +package logger + +import ( + "context" + "fmt" + "github.com/pkg/errors" + "time" +) + +type Level int + +const ( + Debug Level = iota + Info + Warn + Error +) + +func (l Level) Short() string { + switch l { + case Debug: + return "DEBG" + case Info: + return "INFO" + case Warn: + return "WARN" + case Error: + return "ERRO" + default: + return fmt.Sprintf("%s", l) + } +} + +func (l Level) String() string { + switch l { + case Debug: + return "debug" + case Info: + return "info" + case Warn: + return "warn" + case Error: + return "error" + default: + return fmt.Sprintf("%s", l) + } +} + +func ParseLevel(s string) (l Level, err error) { + for _, l := range AllLevels { + if s == l.String() { + return l, nil + } + } + return -1, errors.Errorf("unknown level '%s'", s) +} + +// Levels ordered least severe to most severe +var AllLevels []Level = []Level{Debug, Info, Warn, Error} + +type Fields map[string]interface{} + +type Entry struct { + Level Level + Message string + Time time.Time + Fields Fields +} + +type Outlet interface { + // Note: os.Stderr is also used by logger.Logger for reporting errors returned by outlets + // => you probably don't want to log there + WriteEntry(ctx context.Context, entry Entry) error +} + +type Outlets map[Level][]Outlet + +func NewOutlets() Outlets { + return make(Outlets, len(AllLevels)) +} + +func (os Outlets) Add(outlet Outlet, minLevel Level) { + for _, l := range AllLevels[minLevel:] { + os[l] = append(os[l], outlet) + } +} diff --git a/logger/logger.go b/logger/logger.go new file mode 100644 index 0000000..356dddd --- /dev/null +++ b/logger/logger.go @@ -0,0 +1,124 @@ +package logger + +import ( + "context" + "fmt" + "os" + "runtime/debug" + "sync" + "time" +) + +const ( + // The field set by WithError function + FieldError = "err" +) + +const DefaultUserFieldCapacity = 5 +const InternalErrorPrefix = "github.com/zrepl/zrepl/logger: " + +type Logger struct { + fields Fields + outlets Outlets + outletTimeout time.Duration + + mtx *sync.Mutex +} + +func NewLogger(outlets Outlets, outletTimeout time.Duration) *Logger { + return &Logger{ + make(Fields, DefaultUserFieldCapacity), + outlets, + outletTimeout, + &sync.Mutex{}, + } +} + +func (l *Logger) log(level Level, msg string) { + + l.mtx.Lock() + defer l.mtx.Unlock() + + entry := Entry{level, msg, time.Now(), l.fields} + + ctx, _ := context.WithDeadline(context.Background(), time.Now().Add(l.outletTimeout)) + ech := make(chan error) + + louts := l.outlets[level] + for i := range louts { + go func(ctx context.Context, outlet Outlet, entry Entry) { + ech <- outlet.WriteEntry(ctx, entry) + }(ctx, louts[i], entry) + } + + for fin := 0; fin < len(louts); fin++ { + select { + case err := <-ech: + if err != nil { + fmt.Fprintf(os.Stderr, "%s outlet error: %s\n", InternalErrorPrefix, err) + } + case <-ctx.Done(): + if ctx.Err() == context.DeadlineExceeded { + fmt.Fprintf(os.Stderr, "%s outlets exceeded deadline, keep waiting anyways", InternalErrorPrefix) + } + } + } + +} + +func (l *Logger) WithField(field string, val interface{}) *Logger { + + l.mtx.Lock() + defer l.mtx.Unlock() + + if _, ok := l.fields[field]; ok { + fmt.Fprintf(os.Stderr, "%s caller overwrites field '%s'. Stack:\n%s\n", InternalErrorPrefix, string(debug.Stack())) + } + + child := &Logger{ + fields: make(Fields, len(l.fields)+1), + outlets: l.outlets, // cannot be changed after logger initialized + outletTimeout: l.outletTimeout, + mtx: l.mtx, + } + for k, v := range l.fields { + child.fields[k] = v + } + child.fields[field] = val + + return child + +} + +func (l *Logger) WithFields(fields Fields) (ret *Logger) { + // TODO optimize + ret = l + for field, value := range fields { + ret = l.WithField(field, value) + } + return ret +} + +func (l *Logger) WithError(err error) *Logger { + return l.WithField(FieldError, err.Error()) +} + +func (l *Logger) Debug(msg string) { + l.log(Debug, msg) +} + +func (l *Logger) Info(msg string) { + l.log(Info, msg) +} + +func (l *Logger) Warn(msg string) { + l.log(Warn, msg) +} + +func (l *Logger) Error(msg string) { + l.log(Error, msg) +} + +func (l *Logger) Printf(format string, args ...interface{}) { + l.log(Error, fmt.Sprintf(format, args...)) +} diff --git a/logger/logger_test.go b/logger/logger_test.go new file mode 100644 index 0000000..1e12e35 --- /dev/null +++ b/logger/logger_test.go @@ -0,0 +1,47 @@ +package logger_test + +import ( + "context" + "fmt" + "github.com/kr/pretty" + "github.com/zrepl/zrepl/logger" + "testing" + "time" +) + +type TestOutlet struct { + Record []logger.Entry +} + +func (o *TestOutlet) WriteEntry(ctx context.Context, entry logger.Entry) error { + o.Record = append(o.Record, entry) + return nil +} + +func NewTestOutlet() *TestOutlet { + return &TestOutlet{make([]logger.Entry, 0)} +} + +func TestLogger_Basic(t *testing.T) { + + outlet_arr := []logger.Outlet{ + NewTestOutlet(), + NewTestOutlet(), + } + + outlets := logger.NewOutlets() + for _, o := range outlet_arr { + outlets.Add(o, logger.Debug) + } + + l := logger.NewLogger(outlets, 1*time.Second) + + l.Info("foobar") + + l.WithField("fieldname", "fieldval").Info("log with field") + + l.WithError(fmt.Errorf("fooerror")).Error("error") + + t.Log(pretty.Sprint(outlet_arr)) + +}