Continue clientLoop even client failed to dial with retry and add more logs (#103)

* Add more logging and enable debug

* disable debug log

* dial fail log

* Continue clientLoop even client failed to dial with retry

Author: hehaifengcn

proxy/adminservice.go

@@ -280,6 +280,7 @@ func (s *adminServiceProxyServer) StreamWorkflowReplicationMessages(
 		for !shutdownChan.IsShutdown() {
 			req, err := targetStreamServer.Recv()
 			if err == io.EOF {
+				logger.Info("targetStreamServer.Recv encountered EOF", tag.Error(err))
 				return
 			}
 
@@ -292,7 +293,11 @@ func (s *adminServiceProxyServer) StreamWorkflowReplicationMessages(
 			case *adminservice.StreamWorkflowReplicationMessagesRequest_SyncReplicationState:
 				logger.Debug(fmt.Sprintf("forwarding SyncReplicationState: inclusive %v", attr.SyncReplicationState.InclusiveLowWatermark))
 				if err = sourceStreamClient.Send(req); err != nil {
-					logger.Error("sourceStreamClient.Send encountered error", tag.Error(err))
+					if err != io.EOF {
+						logger.Error("sourceStreamClient.Send encountered error", tag.Error(err))
+					} else {
+						logger.Info("sourceStreamClient.Send encountered EOF", tag.Error(err))
+					}
 					return
 				}
 			default:
@@ -323,6 +328,7 @@ func (s *adminServiceProxyServer) StreamWorkflowReplicationMessages(
 		for !shutdownChan.IsShutdown() {
 			resp, err := sourceStreamClient.Recv()
 			if err == io.EOF {
+				logger.Info("sourceStreamClient.Recv encountered EOF", tag.Error(err))
 				return
 			}
 
@@ -336,7 +342,8 @@ func (s *adminServiceProxyServer) StreamWorkflowReplicationMessages(
 				if err = targetStreamServer.Send(resp); err != nil {
 					if err != io.EOF {
 						logger.Error("targetStreamServer.Send encountered error", tag.Error(err))
-
+					} else {
+						logger.Info("targetStreamServer.Send encountered EOF", tag.Error(err))
 					}
 					return
 				}

transport/mux_connection_manager.go

@@ -186,6 +186,7 @@ func (m *muxConnectMananger) clientLoop(setting config.TCPClientSetting) error {
 	m.wg.Add(1)
 	go func() {
 		defer m.wg.Done()
+	connect:
 		for {
 			select {
 			case <-m.shutdownCh:
@@ -201,10 +202,11 @@ func (m *muxConnectMananger) clientLoop(setting config.TCPClientSetting) error {
 				}
 
 				if err := backoff.ThrottleRetry(op, retryPolicy, func(err error) bool {
+					m.logger.Error("mux client failed to dial", tag.Error(err))
 					return !m.isShuttingDown()
 				}); err != nil {
-					m.logger.Error("mux client failed to dial", tag.Error(err))
-					return
+					m.logger.Error("mux client failed to dial with retry", tag.Error(err))
+					continue connect
 				}
 
 				var conn net.Conn