Add liveness reporting for the yamux observer (#101)

* Add liveness reporting for the yamux observer

* Add missing vectors to registry

* IF the health server doesn't start immediately, wait up to 50ms

Author: temporal-nick

metrics/metrics.go

@@ -62,7 +62,7 @@ func DefaultGauge(name string, help string) prometheus.Gauge {
 }
 
 // DefaultGaugeVec provides a prometheus GaugeVec for the requested name. The name will be sanitized, and the recommended
-// namespace and subsystem will be set. Vector metrics allow the use of labels, so if you need a label, then use this.
+// namespace and subsystem will be set. Vector metrics allow the use of labels, so if you need labels on your metrics, then use this.
 func DefaultGaugeVec(name string, help string, labels ...string) *prometheus.GaugeVec {
 	return prometheus.NewGaugeVec(prometheus.GaugeOpts{
 		Namespace: "temporal",
@@ -83,6 +83,17 @@ func DefaultCounter(name string, help string) prometheus.Counter {
 	})
 }
 
+// DefaultCounterVec provides a prometheus CounterVec for the requested name. The name will be sanitized, and the recommended
+// namespace and subsystem will be set. Vector metrics allow the use of labels, so if you need labels on your metrics, then use this.
+func DefaultCounterVec(name string, help string, labels ...string) *prometheus.CounterVec {
+	return prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "temporal",
+		Subsystem: "s2s_proxy",
+		Name:      SanitizeForPrometheus(name),
+		Help:      help,
+	}, labels)
+}
+
 // wrapLoggerForPrometheus is necessary to adapt our temporal Logger to Prometheus's Println interface
 type wrapLoggerForPrometheus struct {
 	log.Logger

metrics/prometheus_defs.go

@@ -28,6 +28,8 @@ var (
 		muxSessionLabels...)
 	MuxStreamsActive = DefaultGaugeVec("mux_streams_active", "Immediate count of the current streams open",
 		muxSessionLabels...)
+	MuxObserverReportCount = DefaultCounterVec("mux_observer_report_count", "Number of observer executions",
+		muxSessionLabels...)
 )
 
 func init() {
@@ -36,4 +38,7 @@ func init() {
 	prometheus.MustRegister(HealthCheckIsHealthy)
 	prometheus.MustRegister(HealthCheckHealthyCount)
 	prometheus.MustRegister(AdminServiceStreamsActive)
+	prometheus.MustRegister(MuxSessionOpen)
+	prometheus.MustRegister(MuxStreamsActive)
+	prometheus.MustRegister(MuxObserverReportCount)
 }

proxy/test/wiring_test.go

@@ -5,6 +5,7 @@ import (
 	"io"
 	"net/http"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"go.temporal.io/api/workflowservice/v1"
@@ -61,8 +62,14 @@ func TestWiringWithEchoService(t *testing.T) {
 		echoServer.stop()
 	}()
 	// Test s2s-proxy health check
-	_, err := http.Get(fmt.Sprintf("http://%s/health", echoServerInfo.s2sProxyConfig.HealthCheck.ListenAddress))
-	assert.NoError(t, err)
+
+	// The server may take a few 10s of ms to start
+	var healthErr = fmt.Errorf("Not started")
+	for attempts := 0; healthErr != nil && attempts < 5; attempts++ {
+		_, healthErr = http.Get(fmt.Sprintf("http://%s/health", echoServerInfo.s2sProxyConfig.HealthCheck.ListenAddress))
+		time.Sleep(10 * time.Millisecond)
+	}
+	assert.NoError(t, healthErr)
 
 	// Confirm that Prometheus initialized and is reporting. We should see proxy_start_count
 	serverMetrics := scrapePrometheus(t, echoServerInfo.s2sProxyConfig.Metrics.Prometheus.ListenAddress)

transport/mux_connection_manager.go

@@ -156,7 +156,7 @@ func (m *muxConnectMananger) serverLoop(setting config.TCPServerSetting) error {
 				m.logger.Info("Accept new connection", tag.NewStringTag("remoteAddr", conn.RemoteAddr().String()))
 
 				session, err := yamux.Server(conn, nil)
-				go observeYamuxSession(session, m.config)
+				go observeYamuxSession(session, m.config, m.logger)
 				if err != nil {
 					m.logger.Fatal("yamux.Server failed", tag.Error(err))
 				}
@@ -218,7 +218,7 @@ func (m *muxConnectMananger) clientLoop(setting config.TCPClientSetting) error {
 				}
 
 				session, err := yamux.Client(conn, nil)
-				go observeYamuxSession(session, m.config)
+				go observeYamuxSession(session, m.config, m.logger)
 				if err != nil {
 					m.logger.Fatal("yamux.Client failed", tag.Error(err))
 				}
@@ -234,14 +234,15 @@ func (m *muxConnectMananger) clientLoop(setting config.TCPClientSetting) error {
 
 // observeYamuxSession creates a goroutine that pings the provided yamux session repeatedly and gathers its two
 // metrics: Whether the server is alive and how many streams it has open.
-func observeYamuxSession(session *yamux.Session, config config.MuxTransportConfig) {
+func observeYamuxSession(session *yamux.Session, config config.MuxTransportConfig, logger log.Logger) {
 	if session == nil {
 		// If we got a null session, we can't even generate tags to report
 		return
 	}
 	defer func() {
 		// This is an async monitor. Don't let it crash the rest of the program if there's a problem
-		recover()
+		err := recover()
+		logger.Warn("Yamux observer died!", tag.NewStringTag("muxConfigName", config.Name), tag.NewAnyTag("err", err))
 	}()
 	labels := []string{session.LocalAddr().String(),
 		session.RemoteAddr().String(),
@@ -261,6 +262,7 @@ func observeYamuxSession(session *yamux.Session, config config.MuxTransportConfi
 		}
 		metrics.MuxSessionOpen.WithLabelValues(labels...).Set(float64(sessionActive))
 		metrics.MuxStreamsActive.WithLabelValues(labels...).Set(float64(session.NumStreams()))
+		metrics.MuxObserverReportCount.WithLabelValues(labels...).Inc()
 	}
 }