package logging import ( "bytes" "context" "crypto/tls" "io" "log/syslog" "net" "time" "github.com/pkg/errors" "github.com/zrepl/zrepl/logger" ) type EntryFormatter interface { SetMetadataFlags(flags MetadataFlags) Format(e *logger.Entry) ([]byte, error) } type WriterOutlet struct { formatter EntryFormatter writer io.Writer } func (h WriterOutlet) WriteEntry(entry logger.Entry) error { bytes, err := h.formatter.Format(&entry) if err != nil { return err } _, err = h.writer.Write(bytes) if err != nil { return err } _, err = h.writer.Write([]byte("\n")) return err } type TCPOutlet struct { formatter EntryFormatter // Specifies how much time must pass between a connection error and a reconnection attempt // Log entries written to the outlet during this time interval are silently dropped. connect func(ctx context.Context) (net.Conn, error) entryChan chan *bytes.Buffer } func NewTCPOutlet(formatter EntryFormatter, network, address string, tlsConfig *tls.Config, retryInterval time.Duration) *TCPOutlet { connect := func(ctx context.Context) (conn net.Conn, err error) { deadl, ok := ctx.Deadline() if !ok { deadl = time.Time{} } dialer := net.Dialer{ Deadline: deadl, } if tlsConfig != nil { conn, err = tls.DialWithDialer(&dialer, network, address, tlsConfig) } else { conn, err = dialer.DialContext(ctx, network, address) } return } entryChan := make(chan *bytes.Buffer, 1) // allow one message in flight while previous is in io.Copy() o := &TCPOutlet{ formatter: formatter, connect: connect, entryChan: entryChan, } go o.outLoop(retryInterval) return o } // FIXME: use this method func (h *TCPOutlet) Close() { close(h.entryChan) } func (h *TCPOutlet) outLoop(retryInterval time.Duration) { var retry time.Time var conn net.Conn for msg := range h.entryChan { var err error for conn == nil { time.Sleep(time.Until(retry)) ctx, cancel := context.WithDeadline(context.TODO(), time.Now().Add(retryInterval)) conn, err = h.connect(ctx) cancel() if err != nil { retry = time.Now().Add(retryInterval) conn = nil } } err = conn.SetWriteDeadline(time.Now().Add(retryInterval)) if err == nil { _, err = io.Copy(conn, msg) } if err != nil { retry = time.Now().Add(retryInterval) conn.Close() conn = nil } } } func (h *TCPOutlet) WriteEntry(e logger.Entry) error { ebytes, err := h.formatter.Format(&e) if err != nil { return err } buf := new(bytes.Buffer) buf.Write(ebytes) buf.WriteString("\n") select { case h.entryChan <- buf: return nil default: return errors.New("connection broken or not fast enough") } } type SyslogOutlet struct { Formatter EntryFormatter RetryInterval time.Duration Facility syslog.Priority writer *syslog.Writer lastConnectAttempt time.Time } func (o *SyslogOutlet) WriteEntry(entry logger.Entry) error { bytes, err := o.Formatter.Format(&entry) if err != nil { return err } s := string(bytes) if o.writer == nil { now := time.Now() if now.Sub(o.lastConnectAttempt) < o.RetryInterval { return nil // not an error toward logger } o.writer, err = syslog.New(o.Facility, "zrepl") o.lastConnectAttempt = time.Now() if err != nil { o.writer = nil return err } } switch entry.Level { case logger.Debug: return o.writer.Debug(s) case logger.Info: return o.writer.Info(s) case logger.Warn: return o.writer.Warning(s) case logger.Error: return o.writer.Err(s) default: return o.writer.Err(s) // write as error as reaching this case is in fact an error } }