Release 2020.2.0

This reverts commit 053b2c17f1.

RELEASE_NOTES

@@ -1,3 +1,14 @@
+2020.2.0
+- 2020-01-30 TUN-2651: Fix panic in h2mux reader when a stream error is encountered
+- 2020-01-27 TUN-2645: Revert "TUN-2645: Turn on reconnect tokens"
+- 2020-01-28 TUN-2693: Metrics for ReconnectTunnel
+- 2020-01-28 TUN-2696: Add unknown registerRPCName
+- 2020-01-28 TUN-2699: Metrics for Authenticate RPCs
+- 2020-01-28 TUN-2690: cloudflared reconnect uses wrong context
+- 2020-01-29 TUN-2707: Inconsistent cardinality in tunnel error metrics
+- 2020-01-13 TUN-2645: Turn on reconnect tokens
+- 2019-12-23 TUN-2646: Make --edge flag work again for local development
+
 2019.12.0
 - 2019-12-11 TUN-2631: only notify that activeStreamMap is closed if ignoreNewStreams=true
 - 2019-12-17 bug(cloudflared): Set the MaxIdleConnsPerHost of http.Transport to proxy-keepalive-connections (#155)

connection/dial.go

@@ -29,7 +29,7 @@ func DialEdge(
 	tlsEdgeConn.SetDeadline(time.Now().Add(timeout))
 
 	if err = tlsEdgeConn.Handshake(); err != nil {
-		return nil, newDialError(err, "Handshake with edge error")
+		return nil, newDialError(err, "TLS handshake with edge error")
 	}
 	// clear the deadline on the conn; h2mux has its own timeouts
 	tlsEdgeConn.SetDeadline(time.Time{})

h2mux/muxreader.go

@@ -97,7 +97,12 @@ func (r *MuxReader) run(parentLogger *log.Entry) error {
 			switch e := err.(type) {
 			case http2.StreamError:
 				errLogger.Warn("stream error")
-				r.streamError(e.StreamID, e.Code)
+				// Ideally we wouldn't return here, since that aborts the muxer.
+				// We should communicate the error to the relevant MuxedStream
+				// data structure, so that callers of MuxedStream.Read() and
+				// MuxedStream.Write() would see it. Then we could `continue`
+				// and keep the muxer going.
+				return r.streamError(e.StreamID, e.Code)
 			case http2.ConnectionError:
 				errLogger.Warn("connection error")
 				return r.connectionError(err)

origin/metrics.go

@@ -58,9 +58,11 @@ type TunnelMetrics struct {
 	// oldServerLocations stores the last server the tunnel was connected to
 	oldServerLocations map[string]string
 
-	regSuccess prometheus.Counter
-	regFail    *prometheus.CounterVec
-	rpcFail    *prometheus.CounterVec
+	regSuccess  *prometheus.CounterVec
+	regFail     *prometheus.CounterVec
+	authSuccess prometheus.Counter
+	authFail    *prometheus.CounterVec
+	rpcFail     *prometheus.CounterVec
 
 	muxerMetrics        *muxerMetrics
 	tunnelsHA           tunnelsForHA
@@ -417,7 +419,7 @@ func NewTunnelMetrics() *TunnelMetrics {
 			Name:      "tunnel_rpc_fail",
 			Help:      "Count of RPC connection errors by type",
 		},
-		[]string{"error"},
+		[]string{"error", "rpcName"},
 	)
 	prometheus.MustRegister(rpcFail)
 
@@ -428,7 +430,7 @@ func NewTunnelMetrics() *TunnelMetrics {
 			Name:      "tunnel_register_fail",
 			Help:      "Count of tunnel registration errors by type",
 		},
-		[]string{"error"},
+		[]string{"error", "rpcName"},
 	)
 	prometheus.MustRegister(registerFail)
 
@@ -443,15 +445,38 @@ func NewTunnelMetrics() *TunnelMetrics {
 	)
 	prometheus.MustRegister(userHostnamesCounts)
 
-	registerSuccess := prometheus.NewCounter(
+	registerSuccess := prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: metricsNamespace,
 			Subsystem: tunnelSubsystem,
 			Name:      "tunnel_register_success",
 			Help:      "Count of successful tunnel registrations",
-		})
+		},
+		[]string{"rpcName"},
+	)
 	prometheus.MustRegister(registerSuccess)
 
+	authSuccess := prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: metricsNamespace,
+			Subsystem: tunnelSubsystem,
+			Name:      "tunnel_authenticate_success",
+			Help:      "Count of successful tunnel authenticate",
+		},
+	)
+	prometheus.MustRegister(authSuccess)
+
+	authFail := prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: metricsNamespace,
+			Subsystem: tunnelSubsystem,
+			Name:      "tunnel_authenticate_fail",
+			Help:      "Count of tunnel authenticate errors by type",
+		},
+		[]string{"error"},
+	)
+	prometheus.MustRegister(authFail)
+
 	return &TunnelMetrics{
 		haConnections:                  haConnections,
 		activeStreams:                  activeStreams,
@@ -472,6 +497,8 @@ func NewTunnelMetrics() *TunnelMetrics {
 		regFail:                        registerFail,
 		rpcFail:                        rpcFail,
 		userHostnamesCounts:            userHostnamesCounts,
+		authSuccess:                    authSuccess,
+		authFail:                       authFail,
 	}
 }
 

origin/supervisor.go

@@ -72,7 +72,15 @@ type tunnelError struct {
 }
 
 func NewSupervisor(config *TunnelConfig, u uuid.UUID) (*Supervisor, error) {
-	edgeIPs, err := connection.NewEdgeAddrResolver(config.Logger)
+	var (
+		edgeIPs connection.EdgeServiceDiscoverer
+		err     error
+	)
+	if len(config.EdgeAddrs) > 0 {
+		edgeIPs, err = connection.NewEdgeHostnameResolver(config.EdgeAddrs)
+	} else {
+		edgeIPs, err = connection.NewEdgeAddrResolver(config.Logger)
+	}
 	if err != nil {
 		return nil, err
 	}
@@ -346,6 +354,7 @@ func (s *Supervisor) refreshAuth(
 	logger := s.config.Logger.WithField("subsystem", subsystemRefreshAuth)
 	authOutcome, err := authenticate(ctx, backoff.Retries())
 	if err != nil {
+		s.config.Metrics.authFail.WithLabelValues(err.Error()).Inc()
 		if duration, ok := backoff.GetBackoffDuration(ctx); ok {
 			logger.WithError(err).Warnf("Retrying in %v", duration)
 			return backoff.BackoffTimer(), nil
@@ -358,15 +367,20 @@ func (s *Supervisor) refreshAuth(
 	switch outcome := authOutcome.(type) {
 	case tunnelpogs.AuthSuccess:
 		s.SetReconnectToken(outcome.JWT())
+		s.config.Metrics.authSuccess.Inc()
 		return timeAfter(outcome.RefreshAfter()), nil
 	case tunnelpogs.AuthUnknown:
 		duration := outcome.RefreshAfter()
+		s.config.Metrics.authFail.WithLabelValues(outcome.Error()).Inc()
 		logger.WithError(outcome).Warnf("Retrying in %v", duration)
 		return timeAfter(duration), nil
 	case tunnelpogs.AuthFail:
+		s.config.Metrics.authFail.WithLabelValues(outcome.Error()).Inc()
 		return nil, outcome
 	default:
-		return nil, fmt.Errorf("Unexpected outcome type %T", authOutcome)
+		err := fmt.Errorf("Unexpected outcome type %T", authOutcome)
+		s.config.Metrics.authFail.WithLabelValues(err.Error()).Inc()
+		return nil, err
 	}
 }
 

origin/supervisor_test.go

@@ -8,12 +8,37 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 
 	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
 )
 
+func testConfig(logger *logrus.Logger) *TunnelConfig {
+	metrics := TunnelMetrics{}
+
+	metrics.authSuccess = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: metricsNamespace,
+			Subsystem: tunnelSubsystem,
+			Name:      "tunnel_authenticate_success",
+			Help:      "Count of successful tunnel authenticate",
+		},
+	)
+
+	metrics.authFail = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: metricsNamespace,
+			Subsystem: tunnelSubsystem,
+			Name:      "tunnel_authenticate_fail",
+			Help:      "Count of tunnel authenticate errors by type",
+		},
+		[]string{"error"},
+	)
+	return &TunnelConfig{Logger: logger, Metrics: &metrics}
+}
+
 func TestRefreshAuthBackoff(t *testing.T) {
 	logger := logrus.New()
 	logger.Level = logrus.ErrorLevel
@@ -24,7 +49,7 @@ func TestRefreshAuthBackoff(t *testing.T) {
 		return time.After(d)
 	}
 
-	s, err := NewSupervisor(&TunnelConfig{Logger: logger}, uuid.New())
+	s, err := NewSupervisor(testConfig(logger), uuid.New())
 	if !assert.NoError(t, err) {
 		t.FailNow()
 	}
@@ -69,7 +94,7 @@ func TestRefreshAuthSuccess(t *testing.T) {
 		return time.After(d)
 	}
 
-	s, err := NewSupervisor(&TunnelConfig{Logger: logger}, uuid.New())
+	s, err := NewSupervisor(testConfig(logger), uuid.New())
 	if !assert.NoError(t, err) {
 		t.FailNow()
 	}
@@ -98,7 +123,7 @@ func TestRefreshAuthUnknown(t *testing.T) {
 		return time.After(d)
 	}
 
-	s, err := NewSupervisor(&TunnelConfig{Logger: logger}, uuid.New())
+	s, err := NewSupervisor(testConfig(logger), uuid.New())
 	if !assert.NoError(t, err) {
 		t.FailNow()
 	}
@@ -121,7 +146,7 @@ func TestRefreshAuthFail(t *testing.T) {
 	logger := logrus.New()
 	logger.Level = logrus.ErrorLevel
 
-	s, err := NewSupervisor(&TunnelConfig{Logger: logger}, uuid.New())
+	s, err := NewSupervisor(testConfig(logger), uuid.New())
 	if !assert.NoError(t, err) {
 		t.FailNow()
 	}

origin/tunnel.go

@@ -40,6 +40,14 @@ const (
 	DuplicateConnectionError = "EDUPCONN"
 )
 
+type registerRPCName string
+
+const (
+	register  registerRPCName = "register"
+	reconnect registerRPCName = "reconnect"
+	unknown   registerRPCName = "unknown"
+)
+
 type TunnelConfig struct {
 	BuildInfo            *buildinfo.BuildInfo
 	ClientID             string
@@ -113,8 +121,8 @@ type clientRegisterTunnelError struct {
 	cause error
 }
 
-func newClientRegisterTunnelError(cause error, counter *prometheus.CounterVec) clientRegisterTunnelError {
-	counter.WithLabelValues(cause.Error()).Inc()
+func newClientRegisterTunnelError(cause error, counter *prometheus.CounterVec, name registerRPCName) clientRegisterTunnelError {
+	counter.WithLabelValues(cause.Error(), string(name)).Inc()
 	return clientRegisterTunnelError{cause: cause}
 }
 
@@ -265,7 +273,7 @@ func ServeTunnel(
 			eventDigest, eventDigestErr := credentialManager.EventDigest()
 			// if we have both credentials, we can reconnect
 			if tokenErr == nil && eventDigestErr == nil {
-				return ReconnectTunnel(ctx, token, eventDigest, handler.muxer, config, logger, connectionID, originLocalIP, u)
+				return ReconnectTunnel(serveCtx, token, eventDigest, handler.muxer, config, logger, connectionID, originLocalIP, u)
 			}
 			// log errors and proceed to RegisterTunnel
 			if tokenErr != nil {
@@ -306,7 +314,7 @@ func ServeTunnel(
 
 	err = errGroup.Wait()
 	if err != nil {
-		_ = newClientRegisterTunnelError(err, config.Metrics.regFail)
+		_ = newClientRegisterTunnelError(err, config.Metrics.regFail, unknown)
 
 		switch castedErr := err.(type) {
 		case dupConnRegisterTunnelError:
@@ -348,7 +356,7 @@ func RegisterTunnel(
 	tunnelServer, err := connection.NewRPCClient(ctx, muxer, config.TransportLogger.WithField("subsystem", "rpc-register"), openStreamTimeout)
 	if err != nil {
 		// RPC stream open error
-		return newClientRegisterTunnelError(err, config.Metrics.rpcFail)
+		return newClientRegisterTunnelError(err, config.Metrics.rpcFail, register)
 	}
 	defer tunnelServer.Close()
 	// Request server info without blocking tunnel registration; must use capnp library directly.
@@ -364,10 +372,10 @@ func RegisterTunnel(
 	)
 	if registrationErr := registration.DeserializeError(); registrationErr != nil {
 		// RegisterTunnel RPC failure
-		return processRegisterTunnelError(registrationErr, config.Metrics)
+		return processRegisterTunnelError(registrationErr, config.Metrics, register)
 	}
 	credentialManager.SetEventDigest(registration.EventDigest)
-	return processRegistrationSuccess(config, logger, connectionID, registration)
+	return processRegistrationSuccess(config, logger, connectionID, registration, register)
 }
 
 func ReconnectTunnel(
@@ -385,7 +393,7 @@ func ReconnectTunnel(
 	tunnelServer, err := connection.NewRPCClient(ctx, muxer, config.TransportLogger.WithField("subsystem", "rpc-reconnect"), openStreamTimeout)
 	if err != nil {
 		// RPC stream open error
-		return newClientRegisterTunnelError(err, config.Metrics.rpcFail)
+		return newClientRegisterTunnelError(err, config.Metrics.rpcFail, reconnect)
 	}
 	defer tunnelServer.Close()
 	// Request server info without blocking tunnel registration; must use capnp library directly.
@@ -402,12 +410,12 @@ func ReconnectTunnel(
 	)
 	if registrationErr := registration.DeserializeError(); registrationErr != nil {
 		// ReconnectTunnel RPC failure
-		return processRegisterTunnelError(registrationErr, config.Metrics)
+		return processRegisterTunnelError(registrationErr, config.Metrics, reconnect)
 	}
-	return processRegistrationSuccess(config, logger, connectionID, registration)
+	return processRegistrationSuccess(config, logger, connectionID, registration, reconnect)
 }
 
-func processRegistrationSuccess(config *TunnelConfig, logger *log.Entry, connectionID uint8, registration *tunnelpogs.TunnelRegistration) error {
+func processRegistrationSuccess(config *TunnelConfig, logger *log.Entry, connectionID uint8, registration *tunnelpogs.TunnelRegistration, name registerRPCName) error {
 	for _, logLine := range registration.LogLines {
 		logger.Info(logLine)
 	}
@@ -432,16 +440,16 @@ func processRegistrationSuccess(config *TunnelConfig, logger *log.Entry, connect
 	config.Metrics.userHostnamesCounts.WithLabelValues(registration.Url).Inc()
 
 	logger.Infof("Route propagating, it may take up to 1 minute for your new route to become functional")
-	config.Metrics.regSuccess.Inc()
+	config.Metrics.regSuccess.WithLabelValues(string(name)).Inc()
 	return nil
 }
 
-func processRegisterTunnelError(err tunnelpogs.TunnelRegistrationError, metrics *TunnelMetrics) error {
+func processRegisterTunnelError(err tunnelpogs.TunnelRegistrationError, metrics *TunnelMetrics, name registerRPCName) error {
 	if err.Error() == DuplicateConnectionError {
-		metrics.regFail.WithLabelValues("dup_edge_conn").Inc()
+		metrics.regFail.WithLabelValues("dup_edge_conn", string(name)).Inc()
 		return dupConnRegisterTunnelError{}
 	}
-	metrics.regFail.WithLabelValues("server_error").Inc()
+	metrics.regFail.WithLabelValues("server_error", string(name)).Inc()
 	return serverRegisterTunnelError{
 		cause:     fmt.Errorf("Server error: %s", err.Error()),
 		permanent: err.IsPermanent(),
@@ -539,7 +547,7 @@ func NewTunnelHandler(ctx context.Context,
 	// Client mux handshake with agent server
 	h.muxer, err = h2mux.Handshake(edgeConn, edgeConn, config.muxerConfig(h), h.metrics.activeStreams)
 	if err != nil {
-		return nil, "", errors.Wrap(err, "Handshake with edge error")
+		return nil, "", errors.Wrap(err, "h2mux handshake with edge error")
 	}
 	return h, edgeConn.LocalAddr().String(), nil
 }