Improve metrics and logs. Have MuxManager.Start wait 1m so that muxes are connected before we start outbound server. (#164)

Author: temporal-nick

metrics/prometheus_defs.go

@@ -59,7 +59,7 @@ var (
 	// Mux Manager
 
 	muxManagerLabels       = []string{"addr", "mode", "config_name"}
-	MuxErrors              = DefaultCounterVec("mux_errors", "Number of errors observed from mux", muxManagerLabels...)
+	MuxErrors              = DefaultCounterVec("mux_errors", "Number of errors observed from mux", append(muxManagerLabels, "error")...)
 	MuxConnectionEstablish = DefaultCounterVec("mux_connection_establish", "Number of times mux has established", muxManagerLabels...)
 	MuxDialFailed          = DefaultCounterVec("mux_dial_failed", "Mux failed when dialing", muxManagerLabels...)
 	MuxDialSuccess         = DefaultCounterVec("mux_dial_success", "Mux succeeded on dial", muxManagerLabels...)

proxy/adminservice.go

@@ -43,7 +43,10 @@ func NewAdminServiceProxyServer(
 	reportStreamValue func(idx int32, value int32),
 	logger log.Logger,
 ) adminservice.AdminServiceServer {
-	logger = log.With(logger, common.ServiceTag(serviceName))
+	// The AdminServiceStreams will duplicate the same output for an underlying connection issue hundreds of times.
+	// Limit their output to three times per minute
+	logger = log.NewThrottledLogger(log.With(logger, common.ServiceTag(serviceName)),
+		func() float64 { return 3.0 / 60.0 })
 	return &adminServiceProxyServer{
 		adminClient:             adminClient,
 		logger:                  logger,

proxy/cluster_connection_test.go

@@ -18,10 +18,12 @@ import (
 	"github.com/temporalio/s2s-proxy/endtoendtest/testservices"
 	"github.com/temporalio/s2s-proxy/metrics"
 	"github.com/temporalio/s2s-proxy/transport/grpcutil"
+	"github.com/temporalio/s2s-proxy/transport/mux"
 )
 
 func init() {
 	_ = os.Setenv("TEMPORAL_TEST_LOG_LEVEL", "error")
+	mux.MuxManagerStartDelay = 0
 }
 
 func getDynamicPorts(t *testing.T, num int) []string {

proxy/test/echo_proxy_test.go

@@ -17,11 +17,13 @@ import (
 	"github.com/temporalio/s2s-proxy/config"
 	"github.com/temporalio/s2s-proxy/encryption"
 	"github.com/temporalio/s2s-proxy/endtoendtest"
+	"github.com/temporalio/s2s-proxy/transport/mux"
 )
 
 func init() {
 	// silence info log spam
 	_ = os.Setenv("TEMPORAL_TEST_LOG_LEVEL", "error")
+	mux.MuxManagerStartDelay = 0
 }
 
 const (

transport/mux/establisher.go

@@ -110,7 +110,6 @@ func (p *establishingConnProvider) NewConnection() (net.Conn, error) {
 		return nil, err
 	}
 	metrics.MuxDialSuccess.WithLabelValues(p.metricLabels...).Inc()
-	metrics.MuxConnectionEstablish.WithLabelValues(p.metricLabels...).Inc()
 	return client, nil
 }
 

transport/mux/grpc_mux_test.go

@@ -19,6 +19,10 @@ import (
 	"github.com/temporalio/s2s-proxy/transport/grpcutil"
 )
 
+func init() {
+	MuxManagerStartDelay = 0
+}
+
 func TestGRPCMux(t *testing.T) {
 	logger := log.NewTestLogger()
 

transport/mux/multi_mux_manager.go

@@ -7,6 +7,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/hashicorp/yamux"
 	"go.temporal.io/server/common/channel"
@@ -17,6 +18,8 @@ import (
 	"github.com/temporalio/s2s-proxy/transport/mux/session"
 )
 
+var MuxManagerStartDelay = time.Minute
+
 type (
 	multiMuxManager struct {
 		lifetime            context.Context
@@ -150,6 +153,25 @@ func (m *multiMuxManager) Start() {
 	m.init.Do(func() {
 		// Start the mux provider
 		m.muxProvider.Start()
+		go func() {
+			ticker := time.NewTicker(time.Minute)
+			for m.lifetime.Err() == nil {
+				select {
+				case <-ticker.C:
+				case <-m.lifetime.Done():
+				}
+				sb := strings.Builder{}
+				m.muxesLock.RLock()
+				for _, v := range m.muxes {
+					sb.WriteString(v.Describe())
+				}
+				m.muxesLock.RUnlock()
+				m.logger.Info("MuxManager status", tag.NewBoolTag("shutdown", m.hasShutDown.IsShutdown()),
+					tag.Name(m.name), tag.NewStringTag("sessions", sb.String()))
+			}
+		}()
+		// Allow the mux provider some time to provide connections
+		<-time.After(MuxManagerStartDelay)
 	})
 }
 func (m *multiMuxManager) Describe() string {

transport/mux/multi_mux_manager_test.go

@@ -17,6 +17,7 @@ import (
 
 func init() {
 	_ = os.Setenv("TEMPORAL_TEST_LOG_LEVEL", "error")
+	MuxManagerStartDelay = 0
 }
 
 func TestMultiMuxManager(t *testing.T) {

transport/mux/multimux_test.go

@@ -44,6 +44,30 @@ func (s *muxServer) Run(t *testing.T) {
 	}
 }
 
+func TestListenerHandoff(t *testing.T) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	serverConnCh := make(chan net.Conn)
+	listenerClosedCh := make(chan struct{})
+	go func() {
+		serverSide, err := listener.Accept()
+		require.NoError(t, err)
+		serverConnCh <- serverSide
+		err = listener.Close()
+		require.NoError(t, err)
+		close(listenerClosedCh)
+	}()
+	clientConn, err := net.Dial("tcp", listener.Addr().String())
+	require.NoError(t, err)
+	serverConn := <-serverConnCh
+	<-listenerClosedCh
+	_, _ = clientConn.Write([]byte("Hi"))
+	buf := make([]byte, 1024)
+	n, _ := serverConn.Read(buf)
+	require.Equal(t, []byte("Hi"), buf[:n])
+	close(serverConnCh)
+}
+
 func TestMultiMux(t *testing.T) {
 	wg := &sync.WaitGroup{}
 	serverCh, shutDownMux, muxListenerAddrCh := make(chan *muxServer), make(chan struct{}), make(chan string)

transport/mux/observer.go

@@ -59,6 +59,7 @@ func emitYamuxMetrics(lifetime context.Context, muxCategory string, id string, s
 		} else {
 			// Clean up the label so we don't report it forever
 			metrics.MuxStreamsActive.DeleteLabelValues(metricLabels...)
+			metrics.MuxSessionOpen.DeleteLabelValues(metricLabels...)
 		}
 		metrics.MuxObserverReportCount.WithLabelValues(metricLabels...).Inc()
 	}