diff --git a/broadcast.go b/broadcast.go index d07d41bb6..9167e947b 100644 --- a/broadcast.go +++ b/broadcast.go @@ -57,7 +57,7 @@ func (m *Memberlist) encodeAndBroadcast(node string, msgType messageType, msg in func (m *Memberlist) encodeBroadcastNotify(node string, msgType messageType, msg interface{}, notify chan struct{}) { buf, err := encode(msgType, msg) if err != nil { - m.logger.Printf("[ERR] memberlist: Failed to encode message for broadcast: %s", err) + m.logger.Errorf("Failed to encode message for broadcast: %s", err) } else { m.queueBroadcast(node, buf.Bytes(), notify) } diff --git a/config.go b/config.go index 31099e75f..ae5cd8c39 100644 --- a/config.go +++ b/config.go @@ -3,7 +3,6 @@ package memberlist import ( "fmt" "io" - "log" "net" "os" "strings" @@ -209,7 +208,7 @@ type Config struct { // this for the internal logger. If Logger is not set, it will fall back to the // behavior for using LogOutput. You cannot specify both LogOutput and Logger // at the same time. - Logger *log.Logger + Logger Logger // Size of Memberlist's internal channel which handles UDP messages. The // size of this determines the size of the queue which Memberlist will keep diff --git a/integ_test.go b/integ_test.go index c23d04f09..bd312883c 100644 --- a/integ_test.go +++ b/integ_test.go @@ -2,7 +2,6 @@ package memberlist import ( "fmt" - "log" "os" "testing" "time" @@ -42,7 +41,7 @@ func TestMemberlist_Integ(t *testing.T) { c.GossipInterval = 20 * time.Millisecond c.PushPullInterval = 200 * time.Millisecond c.SecretKey = secret - c.Logger = log.New(os.Stderr, c.Name, log.LstdFlags) + c.Logger = newNamedLogger(os.Stderr, c.Name) if i == 0 { c.Events = &ChannelEventDelegate{eventCh} diff --git a/logging.go b/logging.go index 2ca2bab4e..c0d82622d 100644 --- a/logging.go +++ b/logging.go @@ -2,9 +2,57 @@ package memberlist import ( "fmt" + "io" + "log" "net" + "os" ) +type Logger interface { + Debugf(format string, v ...interface{}) + Infof(format string, v ...interface{}) + Warnf(format string, v ...interface{}) + Errorf(format string, v ...interface{}) +} + +type loggerImpl struct { + logger *log.Logger +} + +func newLogger(output io.Writer) Logger { + return newNamedLogger(output, "") +} + +func newNamedLogger(output io.Writer, name string) Logger { + return newNamedLoggerWithFlags(output, name, log.LstdFlags) +} + +func newNamedLoggerWithFlags(output io.Writer, name string, flags int) Logger { + var logger *log.Logger + if output != nil { + logger = log.New(output, name, flags) + } else { + logger = log.New(os.Stderr, name, flags) + } + return &loggerImpl{logger} +} + +func (l *loggerImpl) Debugf(format string, v ...interface{}) { + l.logger.Printf("[DEBUG] memberlist: "+format, v...) +} + +func (l *loggerImpl) Infof(format string, v ...interface{}) { + l.logger.Printf("[INFO] memberlist: "+format, v...) +} + +func (l *loggerImpl) Warnf(format string, v ...interface{}) { + l.logger.Printf("[WARN] memberlist: "+format, v...) +} + +func (l *loggerImpl) Errorf(format string, v ...interface{}) { + l.logger.Printf("[ERROR] memberlist: "+format, v...) +} + func LogAddress(addr net.Addr) string { if addr == nil { return "from=" diff --git a/memberlist.go b/memberlist.go index 7ee040091..f9a8016da 100644 --- a/memberlist.go +++ b/memberlist.go @@ -18,7 +18,6 @@ import ( "container/list" "errors" "fmt" - "log" "net" "os" "strconv" @@ -32,7 +31,7 @@ import ( "github.com/miekg/dns" ) -var errNodeNamesAreRequired = errors.New("memberlist: node names are required by configuration but one was not provided") +var errNodeNamesAreRequired = errors.New("node names are required by configuration but one was not provided") type Memberlist struct { sequenceNum uint32 // Local sequence number @@ -76,7 +75,7 @@ type Memberlist struct { broadcasts *TransmitLimitedQueue - logger *log.Logger + logger Logger } // BuildVsnArray creates the array of Vsn @@ -127,7 +126,7 @@ func newMemberlist(conf *Config) (*Memberlist, error) { logger := conf.Logger if logger == nil { - logger = log.New(logDest, "", log.LstdFlags) + logger = newLogger(logDest) } // Set up a network transport by default if a custom one wasn't given @@ -149,7 +148,7 @@ func newMemberlist(conf *Config) (*Memberlist, error) { return nt, nil } if strings.Contains(err.Error(), "address already in use") { - logger.Printf("[DEBUG] memberlist: Got bind error: %v", err) + logger.Debugf("Got bind error: %v", err) continue } } @@ -176,14 +175,14 @@ func newMemberlist(conf *Config) (*Memberlist, error) { port := nt.GetAutoBindPort() conf.BindPort = port conf.AdvertisePort = port - logger.Printf("[DEBUG] memberlist: Using dynamic bind port %d", port) + logger.Debugf("Using dynamic bind port %d", port) } transport = nt } nodeAwareTransport, ok := transport.(NodeAwareTransport) if !ok { - logger.Printf("[DEBUG] memberlist: configured Transport is not a NodeAwareTransport and some features may not work as desired") + logger.Debugf("configured Transport is not a NodeAwareTransport and some features may not work as desired") nodeAwareTransport = &shimNodeAwareTransport{transport} } @@ -254,7 +253,7 @@ func (m *Memberlist) Join(existing []string) (int, error) { if err != nil { err = fmt.Errorf("Failed to resolve %s: %v", exist, err) errs = multierror.Append(errs, err) - m.logger.Printf("[WARN] memberlist: %v", err) + m.logger.Warnf("%v", err) continue } @@ -264,7 +263,7 @@ func (m *Memberlist) Join(existing []string) (int, error) { if err := m.pushPullNode(a, true); err != nil { err = fmt.Errorf("Failed to join %s: %v", addr.ip, err) errs = multierror.Append(errs, err) - m.logger.Printf("[DEBUG] memberlist: %v", err) + m.logger.Debugf("%v", err) continue } numSuccess++ @@ -336,7 +335,7 @@ func (m *Memberlist) tcpLookupIP(host string, defaultPort uint16, nodeName strin case (*dns.AAAA): ips = append(ips, ipPort{ip: rr.AAAA, port: defaultPort, nodeName: nodeName}) case (*dns.CNAME): - m.logger.Printf("[DEBUG] memberlist: Ignoring CNAME RR in TCP-first answer for '%s'", host) + m.logger.Debugf("Ignoring CNAME RR in TCP-first answer for '%s'", host) } } return ips, nil @@ -384,7 +383,7 @@ func (m *Memberlist) resolveAddr(hostStr string) ([]ipPort, error) { // way to query DNS, and we have a fallback below. ips, err := m.tcpLookupIP(host, port, nodeName) if err != nil { - m.logger.Printf("[DEBUG] memberlist: TCP-first lookup failed for '%s', falling back to UDP: %s", hostStr, err) + m.logger.Debugf("TCP-first lookup failed for '%s', falling back to UDP: %s", hostStr, err) } if len(ips) > 0 { return ips, nil @@ -427,7 +426,7 @@ func (m *Memberlist) setAlive() error { } _, publicIfs, err := sockaddr.IfByRFC("6890", ifAddrs) if len(publicIfs) > 0 && !m.config.EncryptionEnabled() { - m.logger.Printf("[WARN] memberlist: Binding to public address without encryption!") + m.logger.Warnf("Binding to public address without encryption!") } // Set any metadata from the delegate. @@ -639,7 +638,7 @@ func (m *Memberlist) Leave(timeout time.Duration) error { state, ok := m.nodeMap[m.config.Name] m.nodeLock.Unlock() if !ok { - m.logger.Printf("[WARN] memberlist: Leave but we're not in the node map.") + m.logger.Warnf("Leave but we're not in the node map.") return nil } @@ -718,7 +717,7 @@ func (m *Memberlist) Shutdown() error { // completely torn down. If we kill the memberlist-side handlers // those I/O handlers might get stuck. if err := m.transport.Shutdown(); err != nil { - m.logger.Printf("[ERR] Failed to shutdown transport: %v", err) + m.logger.Errorf("Failed to shutdown transport: %v", err) } // Now tear down everything else. diff --git a/memberlist_test.go b/memberlist_test.go index 70b5d780c..da9b0742d 100644 --- a/memberlist_test.go +++ b/memberlist_test.go @@ -4,7 +4,6 @@ import ( "bytes" "fmt" "io/ioutil" - "log" "net" "os" "reflect" @@ -47,7 +46,7 @@ func testConfigNet(tb testing.TB, network byte) *Config { config.Name = config.BindAddr config.BindPort = 0 // choose free port config.RequireNodeNames = true - config.Logger = log.New(os.Stderr, config.Name, log.LstdFlags) + config.Logger = newNamedLogger(os.Stderr, config.Name) return config } @@ -288,7 +287,7 @@ func TestCreate_keyringAndSecretKey(t *testing.T) { func TestCreate_invalidLoggerSettings(t *testing.T) { c := DefaultLANConfig() c.BindAddr = getBindAddr().String() - c.Logger = log.New(ioutil.Discard, "", log.LstdFlags) + c.Logger = newLogger(ioutil.Discard) c.LogOutput = ioutil.Discard m, err := Create(c) @@ -1488,7 +1487,7 @@ func TestMemberlist_Join_IPv6(t *testing.T) { c1.Name = "A" c1.BindAddr = "[::1]" c1.BindPort = 0 // choose free - c1.Logger = log.New(os.Stderr, c1.Name, log.LstdFlags) + c1.Logger = newNamedLogger(os.Stderr, c1.Name) m1, err := Create(c1) require.NoError(t, err) @@ -1499,7 +1498,7 @@ func TestMemberlist_Join_IPv6(t *testing.T) { c2.Name = "B" c2.BindAddr = "[::1]" c2.BindPort = 0 // choose free - c2.Logger = log.New(os.Stderr, c2.Name, log.LstdFlags) + c2.Logger = newNamedLogger(os.Stderr, c2.Name) m2, err := Create(c2) require.NoError(t, err) @@ -1787,7 +1786,7 @@ func TestMemberlist_EncryptedGossipTransition(t *testing.T) { // Set the gossip interval fast enough to get a reasonable test, // but slow enough to avoid "sendto: operation not permitted" conf.GossipInterval = 100 * time.Millisecond - conf.Logger = log.New(os.Stderr, shortName, log.LstdFlags) + conf.Logger = newNamedLogger(os.Stderr, shortName) pretty[conf.Name] = shortName return conf diff --git a/net.go b/net.go index 8d1d7271e..556da1624 100644 --- a/net.go +++ b/net.go @@ -221,7 +221,7 @@ func (m *Memberlist) streamListen() { // handleConn handles a single incoming stream connection from the transport. func (m *Memberlist) handleConn(conn net.Conn) { defer conn.Close() - m.logger.Printf("[DEBUG] memberlist: Stream connection %s", LogConn(conn)) + m.logger.Debugf("Stream connection %s", LogConn(conn)) metrics.IncrCounter([]string{"memberlist", "tcp", "accept"}, 1) @@ -229,18 +229,18 @@ func (m *Memberlist) handleConn(conn net.Conn) { msgType, bufConn, dec, err := m.readStream(conn) if err != nil { if err != io.EOF { - m.logger.Printf("[ERR] memberlist: failed to receive: %s %s", err, LogConn(conn)) + m.logger.Errorf("failed to receive: %s %s", err, LogConn(conn)) resp := errResp{err.Error()} out, err := encode(errMsg, &resp) if err != nil { - m.logger.Printf("[ERR] memberlist: Failed to encode error response: %s", err) + m.logger.Errorf("Failed to encode error response: %s", err) return } err = m.rawSendMsgStream(conn, out.Bytes()) if err != nil { - m.logger.Printf("[ERR] memberlist: Failed to send error: %s %s", err, LogConn(conn)) + m.logger.Errorf("Failed to send error: %s %s", err, LogConn(conn)) return } } @@ -250,7 +250,7 @@ func (m *Memberlist) handleConn(conn net.Conn) { switch msgType { case userMsg: if err := m.readUserMsg(bufConn, dec); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to receive user message: %s %s", err, LogConn(conn)) + m.logger.Errorf("Failed to receive user message: %s %s", err, LogConn(conn)) } case pushPullMsg: // Increment counter of pending push/pulls @@ -259,51 +259,51 @@ func (m *Memberlist) handleConn(conn net.Conn) { // Check if we have too many open push/pull requests if numConcurrent >= maxPushPullRequests { - m.logger.Printf("[ERR] memberlist: Too many pending push/pull requests") + m.logger.Errorf("Too many pending push/pull requests") return } join, remoteNodes, userState, err := m.readRemoteState(bufConn, dec) if err != nil { - m.logger.Printf("[ERR] memberlist: Failed to read remote state: %s %s", err, LogConn(conn)) + m.logger.Errorf("Failed to read remote state: %s %s", err, LogConn(conn)) return } if err := m.sendLocalState(conn, join); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to push local state: %s %s", err, LogConn(conn)) + m.logger.Errorf("Failed to push local state: %s %s", err, LogConn(conn)) return } if err := m.mergeRemoteState(join, remoteNodes, userState); err != nil { - m.logger.Printf("[ERR] memberlist: Failed push/pull merge: %s %s", err, LogConn(conn)) + m.logger.Errorf("Failed push/pull merge: %s %s", err, LogConn(conn)) return } case pingMsg: var p ping if err := dec.Decode(&p); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to decode ping: %s %s", err, LogConn(conn)) + m.logger.Errorf("Failed to decode ping: %s %s", err, LogConn(conn)) return } if p.Node != "" && p.Node != m.config.Name { - m.logger.Printf("[WARN] memberlist: Got ping for unexpected node %s %s", p.Node, LogConn(conn)) + m.logger.Warnf("Got ping for unexpected node %s %s", p.Node, LogConn(conn)) return } ack := ackResp{p.SeqNo, nil} out, err := encode(ackRespMsg, &ack) if err != nil { - m.logger.Printf("[ERR] memberlist: Failed to encode ack: %s", err) + m.logger.Errorf("Failed to encode ack: %s", err) return } err = m.rawSendMsgStream(conn, out.Bytes()) if err != nil { - m.logger.Printf("[ERR] memberlist: Failed to send ack: %s %s", err, LogConn(conn)) + m.logger.Errorf("Failed to send ack: %s %s", err, LogConn(conn)) return } default: - m.logger.Printf("[ERR] memberlist: Received invalid msgType (%d) %s", msgType, LogConn(conn)) + m.logger.Errorf("Received invalid msgType (%d) %s", msgType, LogConn(conn)) } } @@ -331,7 +331,7 @@ func (m *Memberlist) ingestPacket(buf []byte, from net.Addr, timestamp time.Time // Treat the message as plaintext plain = buf } else { - m.logger.Printf("[ERR] memberlist: Decrypt packet failed: %v %s", err, LogAddress(from)) + m.logger.Errorf("Decrypt packet failed: %v %s", err, LogAddress(from)) return } } @@ -345,7 +345,7 @@ func (m *Memberlist) ingestPacket(buf []byte, from net.Addr, timestamp time.Time crc := crc32.ChecksumIEEE(buf[5:]) expected := binary.BigEndian.Uint32(buf[1:5]) if crc != expected { - m.logger.Printf("[WARN] memberlist: Got invalid checksum for UDP packet: %x, %x", crc, expected) + m.logger.Warnf("Got invalid checksum for UDP packet: %x, %x", crc, expected) return } m.handleCommand(buf[5:], from, timestamp) @@ -391,7 +391,7 @@ func (m *Memberlist) handleCommand(buf []byte, from net.Addr, timestamp time.Tim // Check for overflow and append if not full m.msgQueueLock.Lock() if queue.Len() >= m.config.HandoffQueueDepth { - m.logger.Printf("[WARN] memberlist: handler queue full, dropping message (%d) %s", msgType, LogAddress(from)) + m.logger.Warnf("handler queue full, dropping message (%d) %s", msgType, LogAddress(from)) } else { queue.PushBack(msgHandoff{msgType, buf, from}) } @@ -404,7 +404,7 @@ func (m *Memberlist) handleCommand(buf []byte, from net.Addr, timestamp time.Tim } default: - m.logger.Printf("[ERR] memberlist: msg type (%d) not supported %s", msgType, LogAddress(from)) + m.logger.Errorf("msg type (%d) not supported %s", msgType, LogAddress(from)) } } @@ -451,7 +451,7 @@ func (m *Memberlist) packetHandler() { case userMsg: m.handleUser(buf, from) default: - m.logger.Printf("[ERR] memberlist: Message type (%d) not supported %s (packet handler)", msgType, LogAddress(from)) + m.logger.Errorf("Message type (%d) not supported %s (packet handler)", msgType, LogAddress(from)) } } @@ -465,13 +465,13 @@ func (m *Memberlist) handleCompound(buf []byte, from net.Addr, timestamp time.Ti // Decode the parts trunc, parts, err := decodeCompoundMessage(buf) if err != nil { - m.logger.Printf("[ERR] memberlist: Failed to decode compound request: %s %s", err, LogAddress(from)) + m.logger.Errorf("Failed to decode compound request: %s %s", err, LogAddress(from)) return } // Log any truncation if trunc > 0 { - m.logger.Printf("[WARN] memberlist: Compound request had %d truncated messages %s", trunc, LogAddress(from)) + m.logger.Warnf("Compound request had %d truncated messages %s", trunc, LogAddress(from)) } // Handle each message @@ -483,12 +483,12 @@ func (m *Memberlist) handleCompound(buf []byte, from net.Addr, timestamp time.Ti func (m *Memberlist) handlePing(buf []byte, from net.Addr) { var p ping if err := decode(buf, &p); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to decode ping request: %s %s", err, LogAddress(from)) + m.logger.Errorf("Failed to decode ping request: %s %s", err, LogAddress(from)) return } // If node is provided, verify that it is for us if p.Node != "" && p.Node != m.config.Name { - m.logger.Printf("[WARN] memberlist: Got ping for unexpected node '%s' %s", p.Node, LogAddress(from)) + m.logger.Warnf("Got ping for unexpected node '%s' %s", p.Node, LogAddress(from)) return } var ack ackResp @@ -509,14 +509,14 @@ func (m *Memberlist) handlePing(buf []byte, from net.Addr) { Name: p.SourceNode, } if err := m.encodeAndSendMsg(a, ackRespMsg, &ack); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to send ack: %s %s", err, LogAddress(from)) + m.logger.Errorf("Failed to send ack: %s %s", err, LogAddress(from)) } } func (m *Memberlist) handleIndirectPing(buf []byte, from net.Addr) { var ind indirectPingReq if err := decode(buf, &ind); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to decode indirect ping request: %s %s", err, LogAddress(from)) + m.logger.Errorf("Failed to decode indirect ping request: %s %s", err, LogAddress(from)) return } @@ -560,7 +560,7 @@ func (m *Memberlist) handleIndirectPing(buf []byte, from net.Addr) { Name: ind.SourceNode, } if err := m.encodeAndSendMsg(a, ackRespMsg, &ack); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to forward ack: %s %s", err, LogStringAddress(indAddr)) + m.logger.Errorf("Failed to forward ack: %s %s", err, LogStringAddress(indAddr)) } } m.setAckHandler(localSeqNo, respHandler, m.config.ProbeTimeout) @@ -572,7 +572,7 @@ func (m *Memberlist) handleIndirectPing(buf []byte, from net.Addr) { Name: ind.Node, } if err := m.encodeAndSendMsg(a, pingMsg, &ping); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to send indirect ping: %s %s", err, LogStringAddress(indAddr)) + m.logger.Errorf("Failed to send indirect ping: %s %s", err, LogStringAddress(indAddr)) } // Setup a timer to fire off a nack if no ack is seen in time. @@ -588,7 +588,7 @@ func (m *Memberlist) handleIndirectPing(buf []byte, from net.Addr) { Name: ind.SourceNode, } if err := m.encodeAndSendMsg(a, nackRespMsg, &nack); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to send nack: %s %s", err, LogStringAddress(indAddr)) + m.logger.Errorf("Failed to send nack: %s %s", err, LogStringAddress(indAddr)) } } }() @@ -598,7 +598,7 @@ func (m *Memberlist) handleIndirectPing(buf []byte, from net.Addr) { func (m *Memberlist) handleAck(buf []byte, from net.Addr, timestamp time.Time) { var ack ackResp if err := decode(buf, &ack); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to decode ack response: %s %s", err, LogAddress(from)) + m.logger.Errorf("Failed to decode ack response: %s %s", err, LogAddress(from)) return } m.invokeAckHandler(ack, timestamp) @@ -607,7 +607,7 @@ func (m *Memberlist) handleAck(buf []byte, from net.Addr, timestamp time.Time) { func (m *Memberlist) handleNack(buf []byte, from net.Addr) { var nack nackResp if err := decode(buf, &nack); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to decode nack response: %s %s", err, LogAddress(from)) + m.logger.Errorf("Failed to decode nack response: %s %s", err, LogAddress(from)) return } m.invokeNackHandler(nack) @@ -616,7 +616,7 @@ func (m *Memberlist) handleNack(buf []byte, from net.Addr) { func (m *Memberlist) handleSuspect(buf []byte, from net.Addr) { var sus suspect if err := decode(buf, &sus); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to decode suspect message: %s %s", err, LogAddress(from)) + m.logger.Errorf("Failed to decode suspect message: %s %s", err, LogAddress(from)) return } m.suspectNode(&sus) @@ -646,19 +646,19 @@ func (m *Memberlist) ensureCanConnect(from net.Addr) error { func (m *Memberlist) handleAlive(buf []byte, from net.Addr) { if err := m.ensureCanConnect(from); err != nil { - m.logger.Printf("[DEBUG] memberlist: Blocked alive message: %s %s", err, LogAddress(from)) + m.logger.Debugf("Blocked alive message: %s %s", err, LogAddress(from)) return } var live alive if err := decode(buf, &live); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to decode alive message: %s %s", err, LogAddress(from)) + m.logger.Errorf("Failed to decode alive message: %s %s", err, LogAddress(from)) return } if m.config.IPMustBeChecked() { innerIP := net.IP(live.Addr) if innerIP != nil { if err := m.config.IPAllowed(innerIP); err != nil { - m.logger.Printf("[DEBUG] memberlist: Blocked alive.Addr=%s message from: %s %s", innerIP.String(), err, LogAddress(from)) + m.logger.Debugf("Blocked alive.Addr=%s message from: %s %s", innerIP.String(), err, LogAddress(from)) return } } @@ -676,7 +676,7 @@ func (m *Memberlist) handleAlive(buf []byte, from net.Addr) { func (m *Memberlist) handleDead(buf []byte, from net.Addr) { var d dead if err := decode(buf, &d); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to decode dead message: %s %s", err, LogAddress(from)) + m.logger.Errorf("Failed to decode dead message: %s %s", err, LogAddress(from)) return } m.deadNode(&d) @@ -695,7 +695,7 @@ func (m *Memberlist) handleCompressed(buf []byte, from net.Addr, timestamp time. // Try to decode the payload payload, err := decompressPayload(buf) if err != nil { - m.logger.Printf("[ERR] memberlist: Failed to decompress payload: %v %s", err, LogAddress(from)) + m.logger.Errorf("Failed to decompress payload: %v %s", err, LogAddress(from)) return } @@ -753,7 +753,7 @@ func (m *Memberlist) rawSendMsgPacket(a Address, node *Node, msg []byte) error { if m.config.EnableCompression { buf, err := compressPayload(msg) if err != nil { - m.logger.Printf("[WARN] memberlist: Failed to compress payload: %v", err) + m.logger.Warnf("Failed to compress payload: %v", err) } else { // Only use compression if it reduced the size if buf.Len() < len(msg) { @@ -767,7 +767,7 @@ func (m *Memberlist) rawSendMsgPacket(a Address, node *Node, msg []byte) error { if node == nil { toAddr, _, err := net.SplitHostPort(a.Addr) if err != nil { - m.logger.Printf("[ERR] memberlist: Failed to parse address %q: %v", a.Addr, err) + m.logger.Errorf("Failed to parse address %q: %v", a.Addr, err) return err } m.nodeLock.RLock() @@ -795,7 +795,7 @@ func (m *Memberlist) rawSendMsgPacket(a Address, node *Node, msg []byte) error { primaryKey := m.config.Keyring.GetPrimaryKey() err := encryptPayload(m.encryptionVersion(), primaryKey, msg, nil, &buf) if err != nil { - m.logger.Printf("[ERR] memberlist: Encryption of message failed: %v", err) + m.logger.Errorf("Encryption of message failed: %v", err) return err } msg = buf.Bytes() @@ -813,7 +813,7 @@ func (m *Memberlist) rawSendMsgStream(conn net.Conn, sendBuf []byte) error { if m.config.EnableCompression { compBuf, err := compressPayload(sendBuf) if err != nil { - m.logger.Printf("[ERROR] memberlist: Failed to compress payload: %v", err) + m.logger.Errorf("Failed to compress payload: %v", err) } else { sendBuf = compBuf.Bytes() } @@ -823,7 +823,7 @@ func (m *Memberlist) rawSendMsgStream(conn net.Conn, sendBuf []byte) error { if m.config.EncryptionEnabled() && m.config.GossipVerifyOutgoing { crypt, err := m.encryptLocalState(sendBuf) if err != nil { - m.logger.Printf("[ERROR] memberlist: Failed to encrypt local state: %v", err) + m.logger.Errorf("Failed to encrypt local state: %v", err) return err } sendBuf = crypt @@ -883,7 +883,7 @@ func (m *Memberlist) sendAndReceiveState(a Address, join bool) ([]pushNodeState, return nil, nil, err } defer conn.Close() - m.logger.Printf("[DEBUG] memberlist: Initiating push/pull sync with: %s %s", a.Name, conn.RemoteAddr()) + m.logger.Debugf("Initiating push/pull sync with: %s %s", a.Name, conn.RemoteAddr()) metrics.IncrCounter([]string{"memberlist", "tcp", "connect"}, 1) // Send our state diff --git a/net_test.go b/net_test.go index 5c85316ca..23cb7f531 100644 --- a/net_test.go +++ b/net_test.go @@ -5,7 +5,6 @@ import ( "encoding/binary" "fmt" "io" - "log" "net" "reflect" "strconv" @@ -780,10 +779,10 @@ func TestIngestPacket_CRC(t *testing.T) { in[1] <<= 1 logs := &bytes.Buffer{} - logger := log.New(logs, "", 0) - m.logger = logger + m.logger = newNamedLoggerWithFlags(logs, "", 0) m.ingestPacket(in, udp.LocalAddr(), time.Now()) + fmt.Println("logs.String() == [" + logs.String() + "]") if !strings.Contains(logs.String(), "invalid checksum") { t.Fatalf("bad: %s", logs.String()) } @@ -801,8 +800,7 @@ func TestIngestPacket_ExportedFunc_EmptyMessage(t *testing.T) { emptyConn := &emptyReadNetConn{} logs := &bytes.Buffer{} - logger := log.New(logs, "", 0) - m.logger = logger + m.logger = newNamedLoggerWithFlags(logs, "", 0) type ingestionAwareTransport interface { IngestPacket(conn net.Conn, addr net.Addr, now time.Time, shouldClose bool) error diff --git a/net_transport.go b/net_transport.go index 058301172..4edbb8856 100644 --- a/net_transport.go +++ b/net_transport.go @@ -4,7 +4,6 @@ import ( "bytes" "fmt" "io" - "log" "net" "sync" "sync/atomic" @@ -34,7 +33,7 @@ type NetTransportConfig struct { BindPort int // Logger is a logger for operator messages. - Logger *log.Logger + Logger Logger } // NetTransport is a Transport implementation that uses connectionless UDP for @@ -43,7 +42,7 @@ type NetTransport struct { config *NetTransportConfig packetCh chan *Packet streamCh chan net.Conn - logger *log.Logger + logger Logger wg sync.WaitGroup tcpListeners []*net.TCPListener udpListeners []*net.UDPConn @@ -302,7 +301,7 @@ func (t *NetTransport) tcpListen(tcpLn *net.TCPListener) { loopDelay = maxDelay } - t.logger.Printf("[ERR] memberlist: Error accepting TCP connection: %v", err) + t.logger.Errorf("Error accepting TCP connection: %v", err) time.Sleep(loopDelay) continue } @@ -328,14 +327,14 @@ func (t *NetTransport) udpListen(udpLn *net.UDPConn) { break } - t.logger.Printf("[ERR] memberlist: Error reading UDP packet: %v", err) + t.logger.Errorf("Error reading UDP packet: %v", err) continue } // Check the length - it needs to have at least one byte to be a // proper message. if n < 1 { - t.logger.Printf("[ERR] memberlist: UDP packet too short (%d bytes) %s", + t.logger.Errorf("UDP packet too short (%d bytes) %s", len(buf), LogAddress(addr)) continue } diff --git a/state.go b/state.go index 5e4f7fdd7..00dec5190 100644 --- a/state.go +++ b/state.go @@ -324,7 +324,7 @@ func (m *Memberlist) probeNode(node *nodeState) { }() if node.State == StateAlive { if err := m.encodeAndSendMsg(node.FullAddress(), pingMsg, &ping); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to send ping: %s", err) + m.logger.Errorf("Failed to send ping: %s", err) if failedRemote(err) { goto HANDLE_REMOTE_FAILURE } else { @@ -334,14 +334,14 @@ func (m *Memberlist) probeNode(node *nodeState) { } else { var msgs [][]byte if buf, err := encode(pingMsg, &ping); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to encode ping message: %s", err) + m.logger.Errorf("Failed to encode ping message: %s", err) return } else { msgs = append(msgs, buf.Bytes()) } s := suspect{Incarnation: node.Incarnation, Node: node.Name, From: m.config.Name} if buf, err := encode(suspectMsg, &s); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to encode suspect message: %s", err) + m.logger.Errorf("Failed to encode suspect message: %s", err) return } else { msgs = append(msgs, buf.Bytes()) @@ -349,7 +349,7 @@ func (m *Memberlist) probeNode(node *nodeState) { compound := makeCompoundMessage(msgs) if err := m.rawSendMsgPacket(node.FullAddress(), &node.Node, compound.Bytes()); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to send compound ping and suspect message to %s: %s", addr, err) + m.logger.Errorf("Failed to send compound ping and suspect message to %s: %s", addr, err) if failedRemote(err) { goto HANDLE_REMOTE_FAILURE } else { @@ -388,7 +388,7 @@ func (m *Memberlist) probeNode(node *nodeState) { // probe interval it will give the TCP fallback more time, which // is more active in dealing with lost packets, and it gives more // time to wait for indirect acks/nacks. - m.logger.Printf("[DEBUG] memberlist: Failed ping: %s (timeout reached)", node.Name) + m.logger.Debugf("Failed ping: %s (timeout reached)", node.Name) } HANDLE_REMOTE_FAILURE: @@ -421,7 +421,7 @@ HANDLE_REMOTE_FAILURE: } if err := m.encodeAndSendMsg(peer.FullAddress(), indirectPingMsg, &ind); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to send indirect ping: %s", err) + m.logger.Errorf("Failed to send indirect ping: %s", err) } } @@ -444,7 +444,7 @@ HANDLE_REMOTE_FAILURE: defer close(fallbackCh) didContact, err := m.sendPingAndWaitForAck(node.FullAddress(), ping, deadline) if err != nil { - m.logger.Printf("[ERR] memberlist: Failed fallback ping: %s", err) + m.logger.Errorf("Failed fallback ping: %s", err) } else { fallbackCh <- didContact } @@ -469,7 +469,7 @@ HANDLE_REMOTE_FAILURE: // any additional time here. for didContact := range fallbackCh { if didContact { - m.logger.Printf("[WARN] memberlist: Was able to connect to %s but other probes failed, network may be misconfigured", node.Name) + m.logger.Warnf("Was able to connect to %s but other probes failed, network may be misconfigured", node.Name) return } } @@ -491,7 +491,7 @@ HANDLE_REMOTE_FAILURE: } // No acks received from target, suspect it as failed. - m.logger.Printf("[INFO] memberlist: Suspect %s has failed, no acks received", node.Name) + m.logger.Infof("Suspect %s has failed, no acks received", node.Name) s := suspect{Incarnation: node.Incarnation, Node: node.Name, From: m.config.Name} m.suspectNode(&s) } @@ -532,7 +532,7 @@ func (m *Memberlist) Ping(node string, addr net.Addr) (time.Duration, error) { // Timeout, return an error below. } - m.logger.Printf("[DEBUG] memberlist: Failed UDP ping: %v (timeout reached)", node) + m.logger.Debugf("Failed UDP ping: %v (timeout reached)", node) return 0, NoPingResponseError{ping.Node} } @@ -603,13 +603,13 @@ func (m *Memberlist) gossip() { if len(msgs) == 1 { // Send single message as is if err := m.rawSendMsgPacket(node.FullAddress(), &node, msgs[0]); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to send gossip to %s: %s", addr, err) + m.logger.Errorf("Failed to send gossip to %s: %s", addr, err) } } else { // Otherwise create and send a compound message compound := makeCompoundMessage(msgs) if err := m.rawSendMsgPacket(node.FullAddress(), &node, compound.Bytes()); err != nil { - m.logger.Printf("[ERR] memberlist: Failed to send gossip to %s: %s", addr, err) + m.logger.Errorf("Failed to send gossip to %s: %s", addr, err) } } } @@ -636,7 +636,7 @@ func (m *Memberlist) pushPull() { // Attempt a push pull if err := m.pushPullNode(node.FullAddress(), false); err != nil { - m.logger.Printf("[ERR] memberlist: Push/Pull with %s failed: %s", node.Name, err) + m.logger.Errorf("Push/Pull with %s failed: %s", node.Name, err) } } @@ -932,7 +932,7 @@ func (m *Memberlist) aliveNode(a *alive, notify chan struct{}, bootstrap bool) { pMax := a.Vsn[1] pCur := a.Vsn[2] if pMin == 0 || pMax == 0 || pMin > pMax { - m.logger.Printf("[WARN] memberlist: Ignoring an alive message for '%s' (%v:%d) because protocol version(s) are wrong: %d <= %d <= %d should be >0", a.Node, net.IP(a.Addr), a.Port, pMin, pCur, pMax) + m.logger.Warnf("Ignoring an alive message for '%s' (%v:%d) because protocol version(s) are wrong: %d <= %d <= %d should be >0", a.Node, net.IP(a.Addr), a.Port, pMin, pCur, pMax) return } } @@ -943,7 +943,7 @@ func (m *Memberlist) aliveNode(a *alive, notify chan struct{}, bootstrap bool) { // cluster merging to still occur. if m.config.Alive != nil { if len(a.Vsn) < 6 { - m.logger.Printf("[WARN] memberlist: ignoring alive message for '%s' (%v:%d) because Vsn is not present", + m.logger.Warnf("ignoring alive message for '%s' (%v:%d) because Vsn is not present", a.Node, net.IP(a.Addr), a.Port) return } @@ -960,7 +960,7 @@ func (m *Memberlist) aliveNode(a *alive, notify chan struct{}, bootstrap bool) { DCur: a.Vsn[5], } if err := m.config.Alive.NotifyAlive(node); err != nil { - m.logger.Printf("[WARN] memberlist: ignoring alive message for '%s': %s", + m.logger.Warnf("ignoring alive message for '%s': %s", a.Node, err) return } @@ -972,7 +972,7 @@ func (m *Memberlist) aliveNode(a *alive, notify chan struct{}, bootstrap bool) { if !ok { errCon := m.config.IPAllowed(a.Addr) if errCon != nil { - m.logger.Printf("[WARN] memberlist: Rejected node %s (%v): %s", a.Node, net.IP(a.Addr), errCon) + m.logger.Warnf("Rejected node %s (%v): %s", a.Node, net.IP(a.Addr), errCon) return } state = &nodeState{ @@ -1014,7 +1014,7 @@ func (m *Memberlist) aliveNode(a *alive, notify chan struct{}, bootstrap bool) { if !bytes.Equal([]byte(state.Addr), a.Addr) || state.Port != a.Port { errCon := m.config.IPAllowed(a.Addr) if errCon != nil { - m.logger.Printf("[WARN] memberlist: Rejected IP update from %v to %v for node %s: %s", a.Node, state.Addr, net.IP(a.Addr), errCon) + m.logger.Warnf("Rejected IP update from %v to %v for node %s: %s", a.Node, state.Addr, net.IP(a.Addr), errCon) return } // If DeadNodeReclaimTime is configured, check if enough time has elapsed since the node died. @@ -1023,11 +1023,11 @@ func (m *Memberlist) aliveNode(a *alive, notify chan struct{}, bootstrap bool) { // Allow the address to be updated if a dead node is being replaced. if state.State == StateLeft || (state.State == StateDead && canReclaim) { - m.logger.Printf("[INFO] memberlist: Updating address for left or failed node %s from %v:%d to %v:%d", + m.logger.Infof("Updating address for left or failed node %s from %v:%d to %v:%d", state.Name, state.Addr, state.Port, net.IP(a.Addr), a.Port) updatesNode = true } else { - m.logger.Printf("[ERR] memberlist: Conflicting address for %s. Mine: %v:%d Theirs: %v:%d Old state: %v", + m.logger.Errorf("Conflicting address for %s. Mine: %v:%d Theirs: %v:%d Old state: %v", state.Name, state.Addr, state.Port, net.IP(a.Addr), a.Port, state.State) // Inform the conflict delegate if provided @@ -1088,7 +1088,7 @@ func (m *Memberlist) aliveNode(a *alive, notify chan struct{}, bootstrap bool) { return } m.refute(state, a.Incarnation) - m.logger.Printf("[WARN] memberlist: Refuting an alive message for '%s' (%v:%d) meta:(%v VS %v), vsn:(%v VS %v)", a.Node, net.IP(a.Addr), a.Port, a.Meta, state.Meta, a.Vsn, versions) + m.logger.Warnf("Refuting an alive message for '%s' (%v:%d) meta:(%v VS %v), vsn:(%v VS %v)", a.Node, net.IP(a.Addr), a.Port, a.Meta, state.Meta, a.Vsn, versions) } else { m.encodeBroadcastNotify(a.Node, aliveMsg, a, notify) @@ -1165,7 +1165,7 @@ func (m *Memberlist) suspectNode(s *suspect) { // If this is us we need to refute, otherwise re-broadcast if state.Name == m.config.Name { m.refute(state, s.Incarnation) - m.logger.Printf("[WARN] memberlist: Refuting a suspect message (from: %s)", s.From) + m.logger.Warnf("Refuting a suspect message (from: %s)", s.From) return // Do not mark ourself suspect } else { m.encodeAndBroadcast(s.Node, suspectMsg, s) @@ -1213,7 +1213,7 @@ func (m *Memberlist) suspectNode(s *suspect) { metrics.IncrCounter([]string{"memberlist", "degraded", "timeout"}, 1) } - m.logger.Printf("[INFO] memberlist: Marking %s as failed, suspect timeout reached (%d peer confirmations)", + m.logger.Infof("Marking %s as failed, suspect timeout reached (%d peer confirmations)", state.Name, numConfirmations) m.deadNode(d) @@ -1252,7 +1252,7 @@ func (m *Memberlist) deadNode(d *dead) { // If we are not leaving we need to refute if !m.hasLeft() { m.refute(state, d.Incarnation) - m.logger.Printf("[WARN] memberlist: Refuting a dead message (from: %s)", d.From) + m.logger.Warnf("Refuting a dead message (from: %s)", d.From) return // Do not mark ourself dead } diff --git a/state_test.go b/state_test.go index 059918af8..1f84600a2 100644 --- a/state_test.go +++ b/state_test.go @@ -3,7 +3,6 @@ package memberlist import ( "bytes" "fmt" - "log" "net" "os" "reflect" @@ -24,7 +23,7 @@ func HostMemberlist(host string, t *testing.T, f func(*Config)) *Memberlist { c.Name = host c.BindAddr = host c.BindPort = 0 // choose a free port - c.Logger = log.New(os.Stderr, host, log.LstdFlags) + c.Logger = newNamedLogger(os.Stderr, host) if f != nil { f(c) } diff --git a/transport_test.go b/transport_test.go index 49745763f..506e24ca8 100644 --- a/transport_test.go +++ b/transport_test.go @@ -1,7 +1,6 @@ package memberlist import ( - "log" "net" "strings" "sync/atomic" @@ -141,7 +140,7 @@ type testCountingWriter struct { func (tw testCountingWriter) Write(p []byte) (n int, err error) { atomic.AddInt32(tw.numCalls, 1) - if !strings.Contains(string(p), "memberlist: Error accepting TCP connection") { + if !strings.Contains(string(p), "Error accepting TCP connection") { tw.t.Error("did not receive expected log message") } tw.t.Log("countingWriter:", string(p)) @@ -160,7 +159,7 @@ func TestTransport_TcpListenBackoff(t *testing.T) { var numCalls int32 countingWriter := testCountingWriter{t, &numCalls} - countingLogger := log.New(countingWriter, "test", log.LstdFlags) + countingLogger := newNamedLogger(countingWriter, "test") transport := NetTransport{ streamCh: make(chan net.Conn), logger: countingLogger,