TUN-5749: Refactor cloudflared to pave way for reconfigurable ingress

- Split origin into supervisor and proxy packages
- Create configManager to handle dynamic config

Author: chungthuang

cmd/cloudflared/tunnel/cmd.go

@@ -31,8 +31,8 @@ import (
 	"github.com/cloudflare/cloudflared/ingress"
 	"github.com/cloudflare/cloudflared/logger"
 	"github.com/cloudflare/cloudflared/metrics"
-	"github.com/cloudflare/cloudflared/origin"
 	"github.com/cloudflare/cloudflared/signal"
+	"github.com/cloudflare/cloudflared/supervisor"
 	"github.com/cloudflare/cloudflared/tlsconfig"
 	"github.com/cloudflare/cloudflared/tunneldns"
 )
@@ -223,7 +223,7 @@ func routeFromFlag(c *cli.Context) (route cfapi.HostnameRoute, ok bool) {
 func StartServer(
 	c *cli.Context,
 	info *cliutil.BuildInfo,
-	namedTunnel *connection.NamedTunnelConfig,
+	namedTunnel *connection.NamedTunnelProperties,
 	log *zerolog.Logger,
 	isUIEnabled bool,
 ) error {
@@ -333,7 +333,7 @@ func StartServer(
 		observer.SendURL(quickTunnelURL)
 	}
 
-	tunnelConfig, ingressRules, err := prepareTunnelConfig(c, info, log, logTransport, observer, namedTunnel)
+	tunnelConfig, dynamicConfig, err := prepareTunnelConfig(c, info, log, logTransport, observer, namedTunnel)
 	if err != nil {
 		log.Err(err).Msg("Couldn't start tunnel")
 		return err
@@ -353,11 +353,11 @@ func StartServer(
 		errC <- metrics.ServeMetrics(metricsListener, ctx.Done(), readinessServer, quickTunnelURL, log)
 	}()
 
-	if err := ingressRules.StartOrigins(&wg, log, ctx.Done(), errC); err != nil {
+	if err := dynamicConfig.Ingress.StartOrigins(&wg, log, ctx.Done(), errC); err != nil {
 		return err
 	}
 
-	reconnectCh := make(chan origin.ReconnectSignal, 1)
+	reconnectCh := make(chan supervisor.ReconnectSignal, 1)
 	if c.IsSet("stdin-control") {
 		log.Info().Msg("Enabling control through stdin")
 		go stdinControl(reconnectCh, log)
@@ -369,15 +369,15 @@ func StartServer(
 			wg.Done()
 			log.Info().Msg("Tunnel server stopped")
 		}()
-		errC <- origin.StartTunnelDaemon(ctx, tunnelConfig, connectedSignal, reconnectCh, graceShutdownC)
+		errC <- supervisor.StartTunnelDaemon(ctx, tunnelConfig, dynamicConfig, connectedSignal, reconnectCh, graceShutdownC)
 	}()
 
 	if isUIEnabled {
 		tunnelUI := ui.NewUIModel(
 			info.Version(),
 			hostname,
 			metricsListener.Addr().String(),
-			&ingressRules,
+			dynamicConfig.Ingress,
 			tunnelConfig.HAConnections,
 		)
 		app := tunnelUI.Launch(ctx, log, logTransport)
@@ -998,7 +998,7 @@ func configureProxyDNSFlags(shouldHide bool) []cli.Flag {
 	}
 }
 
-func stdinControl(reconnectCh chan origin.ReconnectSignal, log *zerolog.Logger) {
+func stdinControl(reconnectCh chan supervisor.ReconnectSignal, log *zerolog.Logger) {
 	for {
 		scanner := bufio.NewScanner(os.Stdin)
 		for scanner.Scan() {
@@ -1009,7 +1009,7 @@ func stdinControl(reconnectCh chan origin.ReconnectSignal, log *zerolog.Logger)
 			case "":
 				break
 			case "reconnect":
-				var reconnect origin.ReconnectSignal
+				var reconnect supervisor.ReconnectSignal
 				if len(parts) > 1 {
 					var err error
 					if reconnect.Delay, err = time.ParseDuration(parts[1]); err != nil {

cmd/cloudflared/tunnel/configuration.go

@@ -23,7 +23,7 @@ import (
 	"github.com/cloudflare/cloudflared/edgediscovery"
 	"github.com/cloudflare/cloudflared/h2mux"
 	"github.com/cloudflare/cloudflared/ingress"
-	"github.com/cloudflare/cloudflared/origin"
+	"github.com/cloudflare/cloudflared/supervisor"
 	"github.com/cloudflare/cloudflared/tlsconfig"
 	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
 	"github.com/cloudflare/cloudflared/validation"
@@ -87,7 +87,7 @@ func logClientOptions(c *cli.Context, log *zerolog.Logger) {
 	}
 }
 
-func dnsProxyStandAlone(c *cli.Context, namedTunnel *connection.NamedTunnelConfig) bool {
+func dnsProxyStandAlone(c *cli.Context, namedTunnel *connection.NamedTunnelProperties) bool {
 	return c.IsSet("proxy-dns") && (!c.IsSet("hostname") && !c.IsSet("tag") && !c.IsSet("hello-world") && namedTunnel == nil)
 }
 
@@ -152,44 +152,44 @@ func prepareTunnelConfig(
 	info *cliutil.BuildInfo,
 	log, logTransport *zerolog.Logger,
 	observer *connection.Observer,
-	namedTunnel *connection.NamedTunnelConfig,
-) (*origin.TunnelConfig, ingress.Ingress, error) {
+	namedTunnel *connection.NamedTunnelProperties,
+) (*supervisor.TunnelConfig, *supervisor.DynamicConfig, error) {
 	isNamedTunnel := namedTunnel != nil
 
 	configHostname := c.String("hostname")
 	hostname, err := validation.ValidateHostname(configHostname)
 	if err != nil {
 		log.Err(err).Str(LogFieldHostname, configHostname).Msg("Invalid hostname")
-		return nil, ingress.Ingress{}, errors.Wrap(err, "Invalid hostname")
+		return nil, nil, errors.Wrap(err, "Invalid hostname")
 	}
 	clientID := c.String("id")
 	if !c.IsSet("id") {
 		clientID, err = generateRandomClientID(log)
 		if err != nil {
-			return nil, ingress.Ingress{}, err
+			return nil, nil, err
 		}
 	}
 
 	tags, err := NewTagSliceFromCLI(c.StringSlice("tag"))
 	if err != nil {
 		log.Err(err).Msg("Tag parse failure")
-		return nil, ingress.Ingress{}, errors.Wrap(err, "Tag parse failure")
+		return nil, nil, errors.Wrap(err, "Tag parse failure")
 	}
 
 	tags = append(tags, tunnelpogs.Tag{Name: "ID", Value: clientID})
 
 	var (
 		ingressRules  ingress.Ingress
-		classicTunnel *connection.ClassicTunnelConfig
+		classicTunnel *connection.ClassicTunnelProperties
 	)
 	cfg := config.GetConfiguration()
 	if isNamedTunnel {
 		clientUUID, err := uuid.NewRandom()
 		if err != nil {
-			return nil, ingress.Ingress{}, errors.Wrap(err, "can't generate connector UUID")
+			return nil, nil, errors.Wrap(err, "can't generate connector UUID")
 		}
 		log.Info().Msgf("Generated Connector ID: %s", clientUUID)
-		features := append(c.StringSlice("features"), origin.FeatureSerializedHeaders)
+		features := append(c.StringSlice("features"), supervisor.FeatureSerializedHeaders)
 		namedTunnel.Client = tunnelpogs.ClientInfo{
 			ClientID: clientUUID[:],
 			Features: dedup(features),
@@ -198,10 +198,10 @@ func prepareTunnelConfig(
 		}
 		ingressRules, err = ingress.ParseIngress(cfg)
 		if err != nil && err != ingress.ErrNoIngressRules {
-			return nil, ingress.Ingress{}, err
+			return nil, nil, err
 		}
 		if !ingressRules.IsEmpty() && c.IsSet("url") {
-			return nil, ingress.Ingress{}, ingress.ErrURLIncompatibleWithIngress
+			return nil, nil, ingress.ErrURLIncompatibleWithIngress
 		}
 	} else {
 
@@ -212,10 +212,10 @@ func prepareTunnelConfig(
 
 		originCert, err := getOriginCert(originCertPath, &originCertLog)
 		if err != nil {
-			return nil, ingress.Ingress{}, errors.Wrap(err, "Error getting origin cert")
+			return nil, nil, errors.Wrap(err, "Error getting origin cert")
 		}
 
-		classicTunnel = &connection.ClassicTunnelConfig{
+		classicTunnel = &connection.ClassicTunnelProperties{
 			Hostname:   hostname,
 			OriginCert: originCert,
 			// turn off use of reconnect token and auth refresh when using named tunnels
@@ -227,48 +227,36 @@ func prepareTunnelConfig(
 	if ingressRules.IsEmpty() {
 		ingressRules, err = ingress.NewSingleOrigin(c, !isNamedTunnel)
 		if err != nil {
-			return nil, ingress.Ingress{}, err
+			return nil, nil, err
 		}
 	}
 
-	var warpRoutingService *ingress.WarpRoutingService
 	warpRoutingEnabled := isWarpRoutingEnabled(cfg.WarpRouting, isNamedTunnel)
-	if warpRoutingEnabled {
-		warpRoutingService = ingress.NewWarpRoutingService()
-		log.Info().Msgf("Warp-routing is enabled")
-	}
-
-	protocolSelector, err := connection.NewProtocolSelector(c.String("protocol"), warpRoutingEnabled, namedTunnel, edgediscovery.ProtocolPercentage, origin.ResolveTTL, log)
+	protocolSelector, err := connection.NewProtocolSelector(c.String("protocol"), warpRoutingEnabled, namedTunnel, edgediscovery.ProtocolPercentage, supervisor.ResolveTTL, log)
 	if err != nil {
-		return nil, ingress.Ingress{}, err
+		return nil, nil, err
 	}
 	log.Info().Msgf("Initial protocol %s", protocolSelector.Current())
 
 	edgeTLSConfigs := make(map[connection.Protocol]*tls.Config, len(connection.ProtocolList))
 	for _, p := range connection.ProtocolList {
 		tlsSettings := p.TLSSettings()
 		if tlsSettings == nil {
-			return nil, ingress.Ingress{}, fmt.Errorf("%s has unknown TLS settings", p)
+			return nil, nil, fmt.Errorf("%s has unknown TLS settings", p)
 		}
 		edgeTLSConfig, err := tlsconfig.CreateTunnelConfig(c, tlsSettings.ServerName)
 		if err != nil {
-			return nil, ingress.Ingress{}, errors.Wrap(err, "unable to create TLS config to connect with edge")
+			return nil, nil, errors.Wrap(err, "unable to create TLS config to connect with edge")
 		}
 		if len(tlsSettings.NextProtos) > 0 {
 			edgeTLSConfig.NextProtos = tlsSettings.NextProtos
 		}
 		edgeTLSConfigs[p] = edgeTLSConfig
 	}
 
-	originProxy := origin.NewOriginProxy(ingressRules, warpRoutingService, tags, log)
 	gracePeriod, err := gracePeriod(c)
 	if err != nil {
-		return nil, ingress.Ingress{}, err
-	}
-	connectionConfig := &connection.Config{
-		OriginProxy:     originProxy,
-		GracePeriod:     gracePeriod,
-		ReplaceExisting: c.Bool("force"),
+		return nil, nil, err
 	}
 	muxerConfig := &connection.MuxerConfig{
 		HeartbeatInterval: c.Duration("heartbeat-interval"),
@@ -279,21 +267,22 @@ func prepareTunnelConfig(
 		MetricsUpdateFreq:  c.Duration("metrics-update-freq"),
 	}
 
-	return &origin.TunnelConfig{
-		ConnectionConfig: connectionConfig,
-		OSArch:           info.OSArch(),
-		ClientID:         clientID,
-		EdgeAddrs:        c.StringSlice("edge"),
-		Region:           c.String("region"),
-		HAConnections:    c.Int("ha-connections"),
-		IncidentLookup:   origin.NewIncidentLookup(),
-		IsAutoupdated:    c.Bool("is-autoupdated"),
-		LBPool:           c.String("lb-pool"),
-		Tags:             tags,
-		Log:              log,
-		LogTransport:     logTransport,
-		Observer:         observer,
-		ReportedVersion:  info.Version(),
+	tunnelConfig := &supervisor.TunnelConfig{
+		GracePeriod:     gracePeriod,
+		ReplaceExisting: c.Bool("force"),
+		OSArch:          info.OSArch(),
+		ClientID:        clientID,
+		EdgeAddrs:       c.StringSlice("edge"),
+		Region:          c.String("region"),
+		HAConnections:   c.Int("ha-connections"),
+		IncidentLookup:  supervisor.NewIncidentLookup(),
+		IsAutoupdated:   c.Bool("is-autoupdated"),
+		LBPool:          c.String("lb-pool"),
+		Tags:            tags,
+		Log:             log,
+		LogTransport:    logTransport,
+		Observer:        observer,
+		ReportedVersion: info.Version(),
 		// Note TUN-3758 , we use Int because UInt is not supported with altsrc
 		Retries:          uint(c.Int("retries")),
 		RunFromTerminal:  isRunningFromTerminal(),
@@ -302,7 +291,12 @@ func prepareTunnelConfig(
 		MuxerConfig:      muxerConfig,
 		ProtocolSelector: protocolSelector,
 		EdgeTLSConfigs:   edgeTLSConfigs,
-	}, ingressRules, nil
+	}
+	dynamicConfig := &supervisor.DynamicConfig{
+		Ingress:            &ingressRules,
+		WarpRoutingEnabled: warpRoutingEnabled,
+	}
+	return tunnelConfig, dynamicConfig, nil
 }
 
 func gracePeriod(c *cli.Context) (time.Duration, error) {

cmd/cloudflared/tunnel/quick_tunnel.go

@@ -77,7 +77,7 @@ func RunQuickTunnel(sc *subcommandContext) error {
 	return StartServer(
 		sc.c,
 		buildInfo,
-		&connection.NamedTunnelConfig{Credentials: credentials, QuickTunnelUrl: data.Result.Hostname},
+		&connection.NamedTunnelProperties{Credentials: credentials, QuickTunnelUrl: data.Result.Hostname},
 		sc.log,
 		sc.isUIEnabled,
 	)

cmd/cloudflared/tunnel/subcommand_context.go

@@ -304,7 +304,7 @@ func (sc *subcommandContext) run(tunnelID uuid.UUID) error {
 	return StartServer(
 		sc.c,
 		buildInfo,
-		&connection.NamedTunnelConfig{Credentials: credentials},
+		&connection.NamedTunnelProperties{Credentials: credentials},
 		sc.log,
 		sc.isUIEnabled,
 	)

connection/connection.go

@@ -25,13 +25,12 @@ const (
 
 var switchingProtocolText = fmt.Sprintf("%d %s", http.StatusSwitchingProtocols, http.StatusText(http.StatusSwitchingProtocols))
 
-type Config struct {
-	OriginProxy     OriginProxy
-	GracePeriod     time.Duration
-	ReplaceExisting bool
+type ConfigManager interface {
+	Update(version int32, config []byte) *pogs.UpdateConfigurationResponse
+	GetOriginProxy() OriginProxy
 }
 
-type NamedTunnelConfig struct {
+type NamedTunnelProperties struct {
 	Credentials    Credentials
 	Client         pogs.ClientInfo
 	QuickTunnelUrl string
@@ -52,7 +51,7 @@ func (c *Credentials) Auth() pogs.TunnelAuth {
 	}
 }
 
-type ClassicTunnelConfig struct {
+type ClassicTunnelProperties struct {
 	Hostname   string
 	OriginCert []byte
 	// feature-flag to use new edge reconnect tokens

connection/connection_test.go

@@ -14,18 +14,19 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/cloudflare/cloudflared/ingress"
+	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
 	"github.com/cloudflare/cloudflared/websocket"
 )
 
 const (
-	largeFileSize = 2 * 1024 * 1024
+	largeFileSize   = 2 * 1024 * 1024
+	testGracePeriod = time.Millisecond * 100
 )
 
 var (
 	unusedWarpRoutingService = (*ingress.WarpRoutingService)(nil)
-	testConfig               = &Config{
-		OriginProxy: &mockOriginProxy{},
-		GracePeriod: time.Millisecond * 100,
+	testConfigManager        = &mockConfigManager{
+		originProxy: &mockOriginProxy{},
 	}
 	log           = zerolog.Nop()
 	testOriginURL = &url.URL{
@@ -43,6 +44,20 @@ type testRequest struct {
 	isProxyError   bool
 }
 
+type mockConfigManager struct {
+	originProxy OriginProxy
+}
+
+func (*mockConfigManager) Update(version int32, config []byte) *tunnelpogs.UpdateConfigurationResponse {
+	return &tunnelpogs.UpdateConfigurationResponse{
+		LastAppliedVersion: version,
+	}
+}
+
+func (mcr *mockConfigManager) GetOriginProxy() OriginProxy {
+	return mcr.originProxy
+}
+
 type mockOriginProxy struct{}
 
 func (moc *mockOriginProxy) ProxyHTTP(