MB-54959 - adjust log level for watcher on valid closes

bg: we create metadata provider object on all indexer nodes periodically
this starts watcher for each node. upon cleanup, peer pipe logs an error
that connection is terminated but this is expected behaviour. this error
log is creating alerts in customer log monitors

exp: if we are killing the peer pipe connection on a valid condition
don't log it as error

stubs: in my experiments, when we kill a server (with -9/-15 flag)
the connection is still gracefully terminated and go returns io.EOF
because of this we cannot diff between indexer crash and conn close
at gometa level. metadata provider will still be aware if conn was
expected to die or closed abruptly

assumptions: -
Change-Id: I015ac0e036a1ad07ffacb20d3ef760149047d45d

common/peerPipe.go

@@ -20,6 +20,7 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"io"
 	"net"
 	"sync"
 
@@ -212,9 +213,26 @@ func (p *PeerPipe) doReceive() {
 		// read packet len
 		lenBuf, err := p.readBytes(8, nil)
 		if err != nil {
-			// if encountering an error, kill the pipe.
-			log.Current.Errorf("PeerPipe.doRecieve() : ecounter error when received mesasage from Peer %s.  Error = %s. Kill Pipe.",
+			// err indicates conn termination. conditions of termination:
+			// 1. when other peer gracefully closes the connection, we get io.EOF (indexer restart closes network gracefully)
+			// 2. if we close the connection, we get "use of closed network connection" as read error
+			// 3. any network error (for eg: read: operation timed out)
+			// to diff between 2 & 3, we can read isClosed from PeerPipe with lock
+
+			isClosed := func () bool {
+				p.mutex.Lock()
+				defer p.mutex.Unlock()
+				return p.isClosed
+			}()
+
+			if err == io.EOF || isClosed {
+				log.Current.Infof("PeerPipe.doRecieve() : connection terminated for %s. Reason = %s. Cleaning Pipe.",
 				p.GetAddr(), err.Error())
+			} else {
+				// if encountering an error, kill the pipe.
+				log.Current.Errorf("PeerPipe.doRecieve() : ecounter error when received mesasage from Peer %s.  Error = %s. Kill Pipe.",
+				p.GetAddr(), err.Error())
+			}
 			return
 		}
 		size := binary.BigEndian.Uint64(lenBuf)
@@ -230,8 +248,20 @@ func (p *PeerPipe) doReceive() {
 		buf, err := p.readBytes(size, readahead)
 		if err != nil {
 			// if encountering an error, kill the pipe.
-			log.Current.Errorf("PeerPipe.doRecieve() : ecounter error when received mesasage from Peer %s.  Error = %s. Kill Pipe.",
+			isClosed := func () bool {
+				p.mutex.Lock()
+				defer p.mutex.Unlock()
+				return p.isClosed
+			}()
+
+			if err == io.EOF || isClosed {
+				log.Current.Infof("PeerPipe.doRecieve() : connection terminated for %s. Reason = %s. Cleaning Pipe.",
+				p.GetAddr(), err.Error())
+			} else {
+				// if encountering an error, kill the pipe.
+				log.Current.Errorf("PeerPipe.doRecieve() : ecounter error when received mesasage from Peer %s.  Error = %s. Kill Pipe.",
 				p.GetAddr(), err.Error())
+			}
 			return
 		}
 		// unmarshall the content and put it in the channel