TUN-5698: Make ingress rules and warp routing dynamically configurable

Author: chungthuang

cmd/cloudflared/tunnel/cmd.go

@@ -31,6 +31,7 @@ import (
 	"github.com/cloudflare/cloudflared/ingress"
 	"github.com/cloudflare/cloudflared/logger"
 	"github.com/cloudflare/cloudflared/metrics"
+	"github.com/cloudflare/cloudflared/orchestration"
 	"github.com/cloudflare/cloudflared/signal"
 	"github.com/cloudflare/cloudflared/supervisor"
 	"github.com/cloudflare/cloudflared/tlsconfig"
@@ -353,7 +354,8 @@ func StartServer(
 		errC <- metrics.ServeMetrics(metricsListener, ctx.Done(), readinessServer, quickTunnelURL, log)
 	}()
 
-	if err := dynamicConfig.Ingress.StartOrigins(&wg, log, ctx.Done(), errC); err != nil {
+	orchestrator, err := orchestration.NewOrchestrator(ctx, dynamicConfig, tunnelConfig.Tags, tunnelConfig.Log)
+	if err != nil {
 		return err
 	}
 
@@ -369,7 +371,7 @@ func StartServer(
 			wg.Done()
 			log.Info().Msg("Tunnel server stopped")
 		}()
-		errC <- supervisor.StartTunnelDaemon(ctx, tunnelConfig, dynamicConfig, connectedSignal, reconnectCh, graceShutdownC)
+		errC <- supervisor.StartTunnelDaemon(ctx, tunnelConfig, orchestrator, connectedSignal, reconnectCh, graceShutdownC)
 	}()
 
 	if isUIEnabled {

cmd/cloudflared/tunnel/configuration.go

@@ -23,6 +23,7 @@ import (
 	"github.com/cloudflare/cloudflared/edgediscovery"
 	"github.com/cloudflare/cloudflared/h2mux"
 	"github.com/cloudflare/cloudflared/ingress"
+	"github.com/cloudflare/cloudflared/orchestration"
 	"github.com/cloudflare/cloudflared/supervisor"
 	"github.com/cloudflare/cloudflared/tlsconfig"
 	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
@@ -153,7 +154,7 @@ func prepareTunnelConfig(
 	log, logTransport *zerolog.Logger,
 	observer *connection.Observer,
 	namedTunnel *connection.NamedTunnelProperties,
-) (*supervisor.TunnelConfig, *supervisor.DynamicConfig, error) {
+) (*supervisor.TunnelConfig, *orchestration.Config, error) {
 	isNamedTunnel := namedTunnel != nil
 
 	configHostname := c.String("hostname")
@@ -292,7 +293,7 @@ func prepareTunnelConfig(
 		ProtocolSelector: protocolSelector,
 		EdgeTLSConfigs:   edgeTLSConfigs,
 	}
-	dynamicConfig := &supervisor.DynamicConfig{
+	dynamicConfig := &orchestration.Config{
 		Ingress:            &ingressRules,
 		WarpRoutingEnabled: warpRoutingEnabled,
 	}

connection/connection.go

@@ -25,9 +25,9 @@ const (
 
 var switchingProtocolText = fmt.Sprintf("%d %s", http.StatusSwitchingProtocols, http.StatusText(http.StatusSwitchingProtocols))
 
-type ConfigManager interface {
-	Update(version int32, config []byte) *pogs.UpdateConfigurationResponse
-	GetOriginProxy() OriginProxy
+type Orchestrator interface {
+	UpdateConfig(version int32, config []byte) *pogs.UpdateConfigurationResponse
+	GetOriginProxy() (OriginProxy, error)
 }
 
 type NamedTunnelProperties struct {

connection/connection_test.go

@@ -6,14 +6,12 @@ import (
 	"io"
 	"math/rand"
 	"net/http"
-	"net/url"
 	"testing"
 	"time"
 
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/assert"
 
-	"github.com/cloudflare/cloudflared/ingress"
 	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
 	"github.com/cloudflare/cloudflared/websocket"
 )
@@ -24,15 +22,10 @@ const (
 )
 
 var (
-	unusedWarpRoutingService = (*ingress.WarpRoutingService)(nil)
-	testConfigManager        = &mockConfigManager{
+	testOrchestrator = &mockOrchestrator{
 		originProxy: &mockOriginProxy{},
 	}
 	log           = zerolog.Nop()
-	testOriginURL = &url.URL{
-		Scheme: "https",
-		Host:   "connectiontest.argotunnel.com",
-	}
 	testLargeResp = make([]byte, largeFileSize)
 )
 
@@ -44,18 +37,18 @@ type testRequest struct {
 	isProxyError   bool
 }
 
-type mockConfigManager struct {
+type mockOrchestrator struct {
 	originProxy OriginProxy
 }
 
-func (*mockConfigManager) Update(version int32, config []byte) *tunnelpogs.UpdateConfigurationResponse {
+func (*mockOrchestrator) UpdateConfig(version int32, config []byte) *tunnelpogs.UpdateConfigurationResponse {
 	return &tunnelpogs.UpdateConfigurationResponse{
 		LastAppliedVersion: version,
 	}
 }
 
-func (mcr *mockConfigManager) GetOriginProxy() OriginProxy {
-	return mcr.originProxy
+func (mcr *mockOrchestrator) GetOriginProxy() (OriginProxy, error) {
+	return mcr.originProxy, nil
 }
 
 type mockOriginProxy struct{}

connection/h2mux.go

@@ -22,10 +22,10 @@ const (
 )
 
 type h2muxConnection struct {
-	configManager ConfigManager
-	gracePeriod   time.Duration
-	muxerConfig   *MuxerConfig
-	muxer         *h2mux.Muxer
+	orchestrator Orchestrator
+	gracePeriod  time.Duration
+	muxerConfig  *MuxerConfig
+	muxer        *h2mux.Muxer
 	// connectionID is only used by metrics, and prometheus requires labels to be string
 	connIndexStr string
 	connIndex    uint8
@@ -61,7 +61,7 @@ func (mc *MuxerConfig) H2MuxerConfig(h h2mux.MuxedStreamHandler, log *zerolog.Lo
 
 // NewTunnelHandler returns a TunnelHandler, origin LAN IP and error
 func NewH2muxConnection(
-	configManager ConfigManager,
+	orchestrator Orchestrator,
 	gracePeriod time.Duration,
 	muxerConfig *MuxerConfig,
 	edgeConn net.Conn,
@@ -70,7 +70,7 @@ func NewH2muxConnection(
 	gracefulShutdownC <-chan struct{},
 ) (*h2muxConnection, error, bool) {
 	h := &h2muxConnection{
-		configManager:     configManager,
+		orchestrator:      orchestrator,
 		gracePeriod:       gracePeriod,
 		muxerConfig:       muxerConfig,
 		connIndexStr:      uint8ToString(connIndex),
@@ -227,7 +227,13 @@ func (h *h2muxConnection) ServeStream(stream *h2mux.MuxedStream) error {
 		sourceConnectionType = TypeWebsocket
 	}
 
-	err := h.configManager.GetOriginProxy().ProxyHTTP(respWriter, req, sourceConnectionType == TypeWebsocket)
+	originProxy, err := h.orchestrator.GetOriginProxy()
+	if err != nil {
+		respWriter.WriteErrorResponse()
+		return err
+	}
+
+	err = originProxy.ProxyHTTP(respWriter, req, sourceConnectionType == TypeWebsocket)
 	if err != nil {
 		respWriter.WriteErrorResponse()
 	}

connection/h2mux_test.go

@@ -48,7 +48,7 @@ func newH2MuxConnection(t require.TestingT) (*h2muxConnection, *h2mux.Muxer) {
 	}()
 	var connIndex = uint8(0)
 	testObserver := NewObserver(&log, &log, false)
-	h2muxConn, err, _ := NewH2muxConnection(testConfigManager, testGracePeriod, testMuxerConfig, originConn, connIndex, testObserver, nil)
+	h2muxConn, err, _ := NewH2muxConnection(testOrchestrator, testGracePeriod, testMuxerConfig, originConn, connIndex, testObserver, nil)
 	require.NoError(t, err)
 	return h2muxConn, <-edgeMuxChan
 }

connection/http2.go

@@ -30,12 +30,12 @@ var errEdgeConnectionClosed = fmt.Errorf("connection with edge closed")
 // HTTP2Connection represents a net.Conn that uses HTTP2 frames to proxy traffic from the edge to cloudflared on the
 // origin.
 type HTTP2Connection struct {
-	conn          net.Conn
-	server        *http2.Server
-	configManager ConfigManager
-	connOptions   *tunnelpogs.ConnectionOptions
-	observer      *Observer
-	connIndex     uint8
+	conn         net.Conn
+	server       *http2.Server
+	orchestrator Orchestrator
+	connOptions  *tunnelpogs.ConnectionOptions
+	observer     *Observer
+	connIndex    uint8
 	// newRPCClientFunc allows us to mock RPCs during testing
 	newRPCClientFunc func(context.Context, io.ReadWriteCloser, *zerolog.Logger) NamedTunnelRPCClient
 
@@ -49,7 +49,7 @@ type HTTP2Connection struct {
 // NewHTTP2Connection returns a new instance of HTTP2Connection.
 func NewHTTP2Connection(
 	conn net.Conn,
-	configManager ConfigManager,
+	orchestrator Orchestrator,
 	connOptions *tunnelpogs.ConnectionOptions,
 	observer *Observer,
 	connIndex uint8,
@@ -61,7 +61,7 @@ func NewHTTP2Connection(
 		server: &http2.Server{
 			MaxConcurrentStreams: MaxConcurrentStreams,
 		},
-		configManager:        configManager,
+		orchestrator:         orchestrator,
 		connOptions:          connOptions,
 		observer:             observer,
 		connIndex:            connIndex,
@@ -106,6 +106,12 @@ func (c *HTTP2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	originProxy, err := c.orchestrator.GetOriginProxy()
+	if err != nil {
+		c.observer.log.Error().Msg(err.Error())
+		return
+	}
+
 	switch connType {
 	case TypeControlStream:
 		if err := c.controlStreamHandler.ServeControlStream(r.Context(), respWriter, c.connOptions); err != nil {
@@ -116,7 +122,7 @@ func (c *HTTP2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 	case TypeWebsocket, TypeHTTP:
 		stripWebsocketUpgradeHeader(r)
-		if err := c.configManager.GetOriginProxy().ProxyHTTP(respWriter, r, connType == TypeWebsocket); err != nil {
+		if err := originProxy.ProxyHTTP(respWriter, r, connType == TypeWebsocket); err != nil {
 			err := fmt.Errorf("Failed to proxy HTTP: %w", err)
 			c.log.Error().Err(err)
 			respWriter.WriteErrorResponse()
@@ -131,7 +137,7 @@ func (c *HTTP2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		}
 
 		rws := NewHTTPResponseReadWriterAcker(respWriter, r)
-		if err := c.configManager.GetOriginProxy().ProxyTCP(r.Context(), rws, &TCPRequest{
+		if err := originProxy.ProxyTCP(r.Context(), rws, &TCPRequest{
 			Dest:    host,
 			CFRay:   FindCfRayHeader(r),
 			LBProbe: IsLBProbeRequest(r),

connection/http2_test.go

@@ -44,7 +44,7 @@ func newTestHTTP2Connection() (*HTTP2Connection, net.Conn) {
 	return NewHTTP2Connection(
 		cfdConn,
 		// OriginProxy is set in testConfigManager
-		testConfigManager,
+		testOrchestrator,
 		&pogs.ConnectionOptions{},
 		obs,
 		connIndex,

connection/quic.go

@@ -36,7 +36,7 @@ const (
 type QUICConnection struct {
 	session              quic.Session
 	logger               *zerolog.Logger
-	configManager        ConfigManager
+	orchestrator         Orchestrator
 	sessionManager       datagramsession.Manager
 	controlStreamHandler ControlStreamHandler
 	connOptions          *tunnelpogs.ConnectionOptions
@@ -47,7 +47,7 @@ func NewQUICConnection(
 	quicConfig *quic.Config,
 	edgeAddr net.Addr,
 	tlsConfig *tls.Config,
-	configManager ConfigManager,
+	orchestrator Orchestrator,
 	connOptions *tunnelpogs.ConnectionOptions,
 	controlStreamHandler ControlStreamHandler,
 	logger *zerolog.Logger,
@@ -66,7 +66,7 @@ func NewQUICConnection(
 
 	return &QUICConnection{
 		session:              session,
-		configManager:        configManager,
+		orchestrator:         orchestrator,
 		logger:               logger,
 		sessionManager:       sessionManager,
 		controlStreamHandler: controlStreamHandler,
@@ -175,6 +175,10 @@ func (q *QUICConnection) handleDataStream(stream *quicpogs.RequestServerStream)
 		return err
 	}
 
+	originProxy, err := q.orchestrator.GetOriginProxy()
+	if err != nil {
+		return err
+	}
 	switch connectRequest.Type {
 	case quicpogs.ConnectionTypeHTTP, quicpogs.ConnectionTypeWebsocket:
 		req, err := buildHTTPRequest(connectRequest, stream)
@@ -183,10 +187,10 @@ func (q *QUICConnection) handleDataStream(stream *quicpogs.RequestServerStream)
 		}
 
 		w := newHTTPResponseAdapter(stream)
-		return q.configManager.GetOriginProxy().ProxyHTTP(w, req, connectRequest.Type == quicpogs.ConnectionTypeWebsocket)
+		return originProxy.ProxyHTTP(w, req, connectRequest.Type == quicpogs.ConnectionTypeWebsocket)
 	case quicpogs.ConnectionTypeTCP:
 		rwa := &streamReadWriteAcker{stream}
-		return q.configManager.GetOriginProxy().ProxyTCP(context.Background(), rwa, &TCPRequest{Dest: connectRequest.Dest})
+		return originProxy.ProxyTCP(context.Background(), rwa, &TCPRequest{Dest: connectRequest.Dest})
 	}
 	return nil
 }

connection/quic_test.go

@@ -632,7 +632,7 @@ func testQUICConnection(udpListenerAddr net.Addr, t *testing.T) *QUICConnection
 		testQUICConfig,
 		udpListenerAddr,
 		tlsClientConfig,
-		&mockConfigManager{originProxy: &mockOriginProxyWithRequest{}},
+		&mockOrchestrator{originProxy: &mockOriginProxyWithRequest{}},
 		&tunnelpogs.ConnectionOptions{},
 		fakeControlStream{},
 		&log,