TUN-1456: Only make one UUID

This commit is contained in:
Adam Chalmers 2019-02-01 14:11:12 -06:00
parent ca33ed9f6a
commit 77f12c7c34
2 changed files with 26 additions and 19 deletions

View File

@ -5,6 +5,8 @@ import (
"fmt" "fmt"
"net" "net"
"time" "time"
"github.com/google/uuid"
) )
const ( const (
@ -49,9 +51,9 @@ func NewSupervisor(config *TunnelConfig) *Supervisor {
} }
} }
func (s *Supervisor) Run(ctx context.Context, connectedSignal chan struct{}) error { func (s *Supervisor) Run(ctx context.Context, connectedSignal chan struct{}, u uuid.UUID) error {
logger := s.config.Logger logger := s.config.Logger
if err := s.initialize(ctx, connectedSignal); err != nil { if err := s.initialize(ctx, connectedSignal, u); err != nil {
return err return err
} }
var tunnelsWaiting []int var tunnelsWaiting []int
@ -94,7 +96,7 @@ func (s *Supervisor) Run(ctx context.Context, connectedSignal chan struct{}) err
case <-backoffTimer: case <-backoffTimer:
backoffTimer = nil backoffTimer = nil
for _, index := range tunnelsWaiting { for _, index := range tunnelsWaiting {
go s.startTunnel(ctx, index, s.newConnectedTunnelSignal(index)) go s.startTunnel(ctx, index, s.newConnectedTunnelSignal(index), u)
} }
tunnelsActive += len(tunnelsWaiting) tunnelsActive += len(tunnelsWaiting)
tunnelsWaiting = nil tunnelsWaiting = nil
@ -118,7 +120,7 @@ func (s *Supervisor) Run(ctx context.Context, connectedSignal chan struct{}) err
} }
} }
func (s *Supervisor) initialize(ctx context.Context, connectedSignal chan struct{}) error { func (s *Supervisor) initialize(ctx context.Context, connectedSignal chan struct{}, u uuid.UUID) error {
logger := s.config.Logger logger := s.config.Logger
edgeIPs, err := ResolveEdgeIPs(s.config.EdgeAddrs) edgeIPs, err := ResolveEdgeIPs(s.config.EdgeAddrs)
if err != nil { if err != nil {
@ -133,7 +135,7 @@ func (s *Supervisor) initialize(ctx context.Context, connectedSignal chan struct
s.lastResolve = time.Now() s.lastResolve = time.Now()
// check entitlement and version too old error before attempting to register more tunnels // check entitlement and version too old error before attempting to register more tunnels
s.nextUnusedEdgeIP = s.config.HAConnections s.nextUnusedEdgeIP = s.config.HAConnections
go s.startFirstTunnel(ctx, connectedSignal) go s.startFirstTunnel(ctx, connectedSignal, u)
select { select {
case <-ctx.Done(): case <-ctx.Done():
<-s.tunnelErrors <-s.tunnelErrors
@ -145,7 +147,7 @@ func (s *Supervisor) initialize(ctx context.Context, connectedSignal chan struct
} }
// At least one successful connection, so start the rest // At least one successful connection, so start the rest
for i := 1; i < s.config.HAConnections; i++ { for i := 1; i < s.config.HAConnections; i++ {
go s.startTunnel(ctx, i, make(chan struct{})) go s.startTunnel(ctx, i, make(chan struct{}), u)
time.Sleep(registrationInterval) time.Sleep(registrationInterval)
} }
return nil return nil
@ -153,8 +155,8 @@ func (s *Supervisor) initialize(ctx context.Context, connectedSignal chan struct
// startTunnel starts the first tunnel connection. The resulting error will be sent on // startTunnel starts the first tunnel connection. The resulting error will be sent on
// s.tunnelErrors. It will send a signal via connectedSignal if registration succeed // s.tunnelErrors. It will send a signal via connectedSignal if registration succeed
func (s *Supervisor) startFirstTunnel(ctx context.Context, connectedSignal chan struct{}) { func (s *Supervisor) startFirstTunnel(ctx context.Context, connectedSignal chan struct{}, u uuid.UUID) {
err := ServeTunnelLoop(ctx, s.config, s.getEdgeIP(0), 0, connectedSignal) err := ServeTunnelLoop(ctx, s.config, s.getEdgeIP(0), 0, connectedSignal, u)
defer func() { defer func() {
s.tunnelErrors <- tunnelError{index: 0, err: err} s.tunnelErrors <- tunnelError{index: 0, err: err}
}() }()
@ -175,14 +177,14 @@ func (s *Supervisor) startFirstTunnel(ctx context.Context, connectedSignal chan
default: default:
return return
} }
err = ServeTunnelLoop(ctx, s.config, s.getEdgeIP(0), 0, connectedSignal) err = ServeTunnelLoop(ctx, s.config, s.getEdgeIP(0), 0, connectedSignal, u)
} }
} }
// startTunnel starts a new tunnel connection. The resulting error will be sent on // startTunnel starts a new tunnel connection. The resulting error will be sent on
// s.tunnelErrors. // s.tunnelErrors.
func (s *Supervisor) startTunnel(ctx context.Context, index int, connectedSignal chan struct{}) { func (s *Supervisor) startTunnel(ctx context.Context, index int, connectedSignal chan struct{}, u uuid.UUID) {
err := ServeTunnelLoop(ctx, s.config, s.getEdgeIP(index), uint8(index), connectedSignal) err := ServeTunnelLoop(ctx, s.config, s.getEdgeIP(index), uint8(index), connectedSignal, u)
s.tunnelErrors <- tunnelError{index: index, err: err} s.tunnelErrors <- tunnelError{index: index, err: err}
} }

View File

@ -132,15 +132,21 @@ func StartTunnelDaemon(config *TunnelConfig, shutdownC <-chan struct{}, connecte
<-shutdownC <-shutdownC
cancel() cancel()
}() }()
u, err := uuid.NewRandom()
if err != nil {
return err
}
// If a user specified negative HAConnections, we will treat it as requesting 1 connection // If a user specified negative HAConnections, we will treat it as requesting 1 connection
if config.HAConnections > 1 { if config.HAConnections > 1 {
return NewSupervisor(config).Run(ctx, connectedSignal) return NewSupervisor(config).Run(ctx, connectedSignal, u)
} else { } else {
addrs, err := ResolveEdgeIPs(config.EdgeAddrs) addrs, err := ResolveEdgeIPs(config.EdgeAddrs)
if err != nil { if err != nil {
return err return err
} }
return ServeTunnelLoop(ctx, config, addrs[0], 0, connectedSignal) return ServeTunnelLoop(ctx, config, addrs[0], 0, connectedSignal, u)
} }
} }
@ -149,6 +155,7 @@ func ServeTunnelLoop(ctx context.Context,
addr *net.TCPAddr, addr *net.TCPAddr,
connectionID uint8, connectionID uint8,
connectedSignal chan struct{}, connectedSignal chan struct{},
u uuid.UUID,
) error { ) error {
logger := config.Logger logger := config.Logger
config.Metrics.incrementHaConnections() config.Metrics.incrementHaConnections()
@ -164,7 +171,7 @@ func ServeTunnelLoop(ctx context.Context,
// Ensure the above goroutine will terminate if we return without connecting // Ensure the above goroutine will terminate if we return without connecting
defer connectedFuse.Fuse(false) defer connectedFuse.Fuse(false)
for { for {
err, recoverable := ServeTunnel(ctx, config, addr, connectionID, connectedFuse, &backoff) err, recoverable := ServeTunnel(ctx, config, addr, connectionID, connectedFuse, &backoff, u)
if recoverable { if recoverable {
if duration, ok := backoff.GetBackoffDuration(ctx); ok { if duration, ok := backoff.GetBackoffDuration(ctx); ok {
logger.Infof("Retrying in %s seconds", duration) logger.Infof("Retrying in %s seconds", duration)
@ -183,6 +190,7 @@ func ServeTunnel(
connectionID uint8, connectionID uint8,
connectedFuse *h2mux.BooleanFuse, connectedFuse *h2mux.BooleanFuse,
backoff *BackoffHandler, backoff *BackoffHandler,
u uuid.UUID,
) (err error, recoverable bool) { ) (err error, recoverable bool) {
// Treat panics as recoverable errors // Treat panics as recoverable errors
defer func() { defer func() {
@ -222,7 +230,7 @@ func ServeTunnel(
errGroup, serveCtx := errgroup.WithContext(ctx) errGroup, serveCtx := errgroup.WithContext(ctx)
errGroup.Go(func() error { errGroup.Go(func() error {
err := RegisterTunnel(serveCtx, handler.muxer, config, connectionID, originLocalIP) err := RegisterTunnel(serveCtx, handler.muxer, config, connectionID, originLocalIP, u)
if err == nil { if err == nil {
connectedFuse.Fuse(true) connectedFuse.Fuse(true)
backoff.SetGracePeriod() backoff.SetGracePeriod()
@ -302,6 +310,7 @@ func RegisterTunnel(
config *TunnelConfig, config *TunnelConfig,
connectionID uint8, connectionID uint8,
originLocalIP string, originLocalIP string,
uuid uuid.UUID,
) error { ) error {
config.TransportLogger.Debug("initiating RPC stream to register") config.TransportLogger.Debug("initiating RPC stream to register")
stream, err := muxer.OpenStream([]h2mux.Header{ stream, err := muxer.OpenStream([]h2mux.Header{
@ -328,10 +337,6 @@ func RegisterTunnel(
serverInfoPromise := tsClient.GetServerInfo(ctx, func(tunnelrpc.TunnelServer_getServerInfo_Params) error { serverInfoPromise := tsClient.GetServerInfo(ctx, func(tunnelrpc.TunnelServer_getServerInfo_Params) error {
return nil return nil
}) })
uuid, err := uuid.NewRandom()
if err != nil {
return clientRegisterTunnelError{cause: err}
}
registration, err := ts.RegisterTunnel( registration, err := ts.RegisterTunnel(
ctx, ctx,
config.OriginCert, config.OriginCert,