TUN-9319: Add dynamic loading of features to connections via ConnectionOptionsSnapshot

Make sure to enforce snapshots of features and client information for each connection
so that the feature information can change in the background. This allows for new features
to only be applied to a connection if it completely disconnects and attempts a reconnect.

Updates the feature refresh time to 1 hour from previous cloudflared versions which
refreshed every 6 hours.

Closes TUN-9319

Author: DevinCarr

client/config.go

@@ -0,0 +1,74 @@
+package client
+
+import (
+	"fmt"
+	"net"
+
+	"github.com/google/uuid"
+	"github.com/rs/zerolog"
+
+	"github.com/cloudflare/cloudflared/features"
+	"github.com/cloudflare/cloudflared/tunnelrpc/pogs"
+)
+
+// Config captures the local client runtime configuration.
+type Config struct {
+	ConnectorID uuid.UUID
+	Version     string
+	Arch        string
+
+	featureSelector features.FeatureSelector
+}
+
+func NewConfig(version string, arch string, featureSelector features.FeatureSelector) (*Config, error) {
+	connectorID, err := uuid.NewRandom()
+	if err != nil {
+		return nil, fmt.Errorf("unable to generate a connector UUID: %w", err)
+	}
+	return &Config{
+		ConnectorID:     connectorID,
+		Version:         version,
+		Arch:            arch,
+		featureSelector: featureSelector,
+	}, nil
+}
+
+// ConnectionOptionsSnapshot is a snapshot of the current client information used to initialize a connection.
+//
+// The FeatureSnapshot is the features that are available for this connection. At the client level they may
+// change, but they will not change within the scope of this struct.
+type ConnectionOptionsSnapshot struct {
+	client              pogs.ClientInfo
+	originLocalIP       net.IP
+	numPreviousAttempts uint8
+	FeatureSnapshot     features.FeatureSnapshot
+}
+
+func (c *Config) ConnectionOptionsSnapshot(originIP net.IP, previousAttempts uint8) *ConnectionOptionsSnapshot {
+	snapshot := c.featureSelector.Snapshot()
+	return &ConnectionOptionsSnapshot{
+		client: pogs.ClientInfo{
+			ClientID: c.ConnectorID[:],
+			Version:  c.Version,
+			Arch:     c.Arch,
+			Features: snapshot.FeaturesList,
+		},
+		originLocalIP:       originIP,
+		numPreviousAttempts: previousAttempts,
+		FeatureSnapshot:     snapshot,
+	}
+}
+
+func (c ConnectionOptionsSnapshot) ConnectionOptions() *pogs.ConnectionOptions {
+	return &pogs.ConnectionOptions{
+		Client:              c.client,
+		OriginLocalIP:       c.originLocalIP,
+		ReplaceExisting:     false,
+		CompressionQuality:  0,
+		NumPreviousAttempts: c.numPreviousAttempts,
+	}
+}
+
+func (c ConnectionOptionsSnapshot) LogFields(event *zerolog.Event) *zerolog.Event {
+	return event.Strs("features", c.client.Features)
+}

client/config_test.go

@@ -0,0 +1,50 @@
+package client
+
+import (
+	"net"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/cloudflare/cloudflared/features"
+)
+
+func TestGenerateConnectionOptions(t *testing.T) {
+	version := "1234"
+	arch := "linux_amd64"
+	originIP := net.ParseIP("192.168.1.1")
+	var previousAttempts uint8 = 4
+
+	config, err := NewConfig(version, arch, &mockFeatureSelector{})
+	require.NoError(t, err)
+	require.Equal(t, version, config.Version)
+	require.Equal(t, arch, config.Arch)
+
+	// Validate ConnectionOptionsSnapshot fields
+	connOptions := config.ConnectionOptionsSnapshot(originIP, previousAttempts)
+	require.Equal(t, version, connOptions.client.Version)
+	require.Equal(t, arch, connOptions.client.Arch)
+	require.Equal(t, config.ConnectorID[:], connOptions.client.ClientID)
+
+	// Vaidate snapshot feature fields against the connOptions generated
+	snapshot := config.featureSelector.Snapshot()
+	require.Equal(t, features.DatagramV3, snapshot.DatagramVersion)
+	require.Equal(t, features.DatagramV3, connOptions.FeatureSnapshot.DatagramVersion)
+
+	pogsConnOptions := connOptions.ConnectionOptions()
+	require.Equal(t, connOptions.client, pogsConnOptions.Client)
+	require.Equal(t, originIP, pogsConnOptions.OriginLocalIP)
+	require.False(t, pogsConnOptions.ReplaceExisting)
+	require.Equal(t, uint8(0), pogsConnOptions.CompressionQuality)
+	require.Equal(t, previousAttempts, pogsConnOptions.NumPreviousAttempts)
+}
+
+type mockFeatureSelector struct{}
+
+func (m *mockFeatureSelector) Snapshot() features.FeatureSnapshot {
+	return features.FeatureSnapshot{
+		PostQuantum:     features.PostQuantumPrefer,
+		DatagramVersion: features.DatagramV3,
+		FeaturesList:    []string{features.FeaturePostQuantum, features.FeatureDatagramV3_1},
+	}
+}

cmd/cloudflared/tunnel/cmd.go

@@ -15,7 +15,6 @@ import (
 	"github.com/coreos/go-systemd/v22/daemon"
 	"github.com/facebookgo/grace/gracenet"
 	"github.com/getsentry/sentry-go"
-	"github.com/google/uuid"
 	"github.com/mitchellh/go-homedir"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
@@ -446,14 +445,7 @@ func StartServer(
 		log.Err(err).Msg("Couldn't start tunnel")
 		return err
 	}
-	var clientID uuid.UUID
-	if tunnelConfig.NamedTunnel != nil {
-		clientID, err = uuid.FromBytes(tunnelConfig.NamedTunnel.Client.ClientID)
-		if err != nil {
-			// set to nil for classic tunnels
-			clientID = uuid.Nil
-		}
-	}
+	connectorID := tunnelConfig.ClientConfig.ConnectorID
 
 	// Disable ICMP packet routing for quick tunnels
 	if quickTunnelURL != "" {
@@ -471,7 +463,7 @@ func StartServer(
 		c.String("management-hostname"),
 		c.Bool("management-diagnostics"),
 		serviceIP,
-		clientID,
+		connectorID,
 		c.String(cfdflags.ConnectorLabel),
 		logger.ManagementLogger.Log,
 		logger.ManagementLogger,
@@ -503,14 +495,14 @@ func StartServer(
 			sources = append(sources, ipv6.String())
 		}
 
-		readinessServer := metrics.NewReadyServer(clientID, tracker)
+		readinessServer := metrics.NewReadyServer(connectorID, tracker)
 		cliFlags := nonSecretCliFlags(log, c, nonSecretFlagsList)
 		diagnosticHandler := diagnostic.NewDiagnosticHandler(
 			log,
 			0,
 			diagnostic.NewSystemCollectorImpl(buildInfo.CloudflaredVersion),
 			tunnelConfig.NamedTunnel.Credentials.TunnelID,
-			clientID,
+			connectorID,
 			tracker,
 			cliFlags,
 			sources,

cmd/cloudflared/tunnel/configuration.go

@@ -10,13 +10,13 @@ import (
 	"strings"
 	"time"
 
-	"github.com/google/uuid"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 	"github.com/urfave/cli/v2"
 	"github.com/urfave/cli/v2/altsrc"
 	"golang.org/x/term"
 
+	"github.com/cloudflare/cloudflared/client"
 	"github.com/cloudflare/cloudflared/cmd/cloudflared/cliutil"
 	"github.com/cloudflare/cloudflared/cmd/cloudflared/flags"
 	"github.com/cloudflare/cloudflared/config"
@@ -125,27 +125,29 @@ func prepareTunnelConfig(
 	observer *connection.Observer,
 	namedTunnel *connection.TunnelProperties,
 ) (*supervisor.TunnelConfig, *orchestration.Config, error) {
-	clientID, err := uuid.NewRandom()
+	transportProtocol := c.String(flags.Protocol)
+	isPostQuantumEnforced := c.Bool(flags.PostQuantum)
+	featureSelector, err := features.NewFeatureSelector(ctx, namedTunnel.Credentials.AccountTag, c.StringSlice(flags.Features), isPostQuantumEnforced, log)
 	if err != nil {
-		return nil, nil, errors.Wrap(err, "can't generate connector UUID")
+		return nil, nil, errors.Wrap(err, "Failed to create feature selector")
 	}
-	log.Info().Msgf("Generated Connector ID: %s", clientID)
-	tags, err := NewTagSliceFromCLI(c.StringSlice(flags.Tag))
+
+	clientConfig, err := client.NewConfig(info.Version(), info.OSArch(), featureSelector)
 	if err != nil {
-		log.Err(err).Msg("Tag parse failure")
-		return nil, nil, errors.Wrap(err, "Tag parse failure")
+		return nil, nil, err
 	}
-	tags = append(tags, pogs.Tag{Name: "ID", Value: clientID.String()})
 
-	transportProtocol := c.String(flags.Protocol)
-	isPostQuantumEnforced := c.Bool(flags.PostQuantum)
+	log.Info().Msgf("Generated Connector ID: %s", clientConfig.ConnectorID)
 
-	featureSelector, err := features.NewFeatureSelector(ctx, namedTunnel.Credentials.AccountTag, c.StringSlice(flags.Features), c.Bool(flags.PostQuantum), log)
+	tags, err := NewTagSliceFromCLI(c.StringSlice(flags.Tag))
 	if err != nil {
-		return nil, nil, errors.Wrap(err, "Failed to create feature selector")
+		log.Err(err).Msg("Tag parse failure")
+		return nil, nil, errors.Wrap(err, "Tag parse failure")
 	}
-	clientFeatures := featureSelector.ClientFeatures()
-	pqMode := featureSelector.PostQuantumMode()
+	tags = append(tags, pogs.Tag{Name: "ID", Value: clientConfig.ConnectorID.String()})
+
+	clientFeatures := featureSelector.Snapshot()
+	pqMode := clientFeatures.PostQuantum
 	if pqMode == features.PostQuantumStrict {
 		// Error if the user tries to force a non-quic transport protocol
 		if transportProtocol != connection.AutoSelectFlag && transportProtocol != connection.QUIC.String() {
@@ -154,12 +156,6 @@ func prepareTunnelConfig(
 		transportProtocol = connection.QUIC.String()
 	}
 
-	namedTunnel.Client = pogs.ClientInfo{
-		ClientID: clientID[:],
-		Features: clientFeatures,
-		Version:  info.Version(),
-		Arch:     info.OSArch(),
-	}
 	cfg := config.GetConfiguration()
 	ingressRules, err := ingress.ParseIngressFromConfigAndCLI(cfg, c, log)
 	if err != nil {
@@ -224,10 +220,8 @@ func prepareTunnelConfig(
 	}
 
 	tunnelConfig := &supervisor.TunnelConfig{
+		ClientConfig:    clientConfig,
 		GracePeriod:     gracePeriod,
-		ReplaceExisting: c.Bool(flags.Force),
-		OSArch:          info.OSArch(),
-		ClientID:        clientID.String(),
 		EdgeAddrs:       c.StringSlice(flags.Edge),
 		Region:          resolvedRegion,
 		EdgeIPVersion:   edgeIPVersion,
@@ -246,7 +240,6 @@ func prepareTunnelConfig(
 		NamedTunnel:                         namedTunnel,
 		ProtocolSelector:                    protocolSelector,
 		EdgeTLSConfigs:                      edgeTLSConfigs,
-		FeatureSelector:                     featureSelector,
 		MaxEdgeAddrRetries:                  uint8(c.Int(flags.MaxEdgeAddrRetries)), // nolint: gosec
 		RPCTimeout:                          c.Duration(flags.RpcTimeout),
 		WriteStreamTimeout:                  c.Duration(flags.WriteStreamTimeout),

connection/connection.go

@@ -57,7 +57,6 @@ type Orchestrator interface {
 
 type TunnelProperties struct {
 	Credentials    Credentials
-	Client         pogs.ClientInfo
 	QuickTunnelUrl string
 }
 

connection/http2.go

@@ -16,10 +16,10 @@ import (
 	"github.com/rs/zerolog"
 	"golang.org/x/net/http2"
 
+	"github.com/cloudflare/cloudflared/client"
 	cfdflow "github.com/cloudflare/cloudflared/flow"
 
 	"github.com/cloudflare/cloudflared/tracing"
-	"github.com/cloudflare/cloudflared/tunnelrpc/pogs"
 )
 
 // note: these constants are exported so we can reuse them in the edge-side code
@@ -39,7 +39,7 @@ type HTTP2Connection struct {
 	conn         net.Conn
 	server       *http2.Server
 	orchestrator Orchestrator
-	connOptions  *pogs.ConnectionOptions
+	connOptions  *client.ConnectionOptionsSnapshot
 	observer     *Observer
 	connIndex    uint8
 
@@ -54,7 +54,7 @@ type HTTP2Connection struct {
 func NewHTTP2Connection(
 	conn net.Conn,
 	orchestrator Orchestrator,
-	connOptions *pogs.ConnectionOptions,
+	connOptions *client.ConnectionOptionsSnapshot,
 	observer *Observer,
 	connIndex uint8,
 	controlStreamHandler ControlStreamHandler,
@@ -118,7 +118,7 @@ func (c *HTTP2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	var requestErr error
 	switch connType {
 	case TypeControlStream:
-		requestErr = c.controlStreamHandler.ServeControlStream(r.Context(), respWriter, c.connOptions, c.orchestrator)
+		requestErr = c.controlStreamHandler.ServeControlStream(r.Context(), respWriter, c.connOptions.ConnectionOptions(), c.orchestrator)
 		if requestErr != nil {
 			c.controlStreamErr = requestErr
 		}

connection/http2_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"golang.org/x/net/http2"
 
+	"github.com/cloudflare/cloudflared/client"
 	"github.com/cloudflare/cloudflared/tracing"
 
 	"github.com/cloudflare/cloudflared/tunnelrpc"
@@ -51,7 +52,7 @@ func newTestHTTP2Connection() (*HTTP2Connection, net.Conn) {
 		cfdConn,
 		// OriginProxy is set in testConfigManager
 		testOrchestrator,
-		&pogs.ConnectionOptions{},
+		&client.ConnectionOptionsSnapshot{},
 		obs,
 		connIndex,
 		controlStream,
@@ -74,7 +75,7 @@ func TestHTTP2ConfigurationSet(t *testing.T) {
 	require.NoError(t, err)
 
 	reqBody := []byte(`{
-"version": 2, 
+"version": 2,
 "config": {"warp-routing": {"enabled": true},  "originRequest" : {"connectTimeout": 10}, "ingress" : [ {"hostname": "test", "service": "https://localhost:8000" } , {"service": "http_status:404"} ]}}
 `)
 	reader := bytes.NewReader(reqBody)

connection/quic_connection.go

@@ -17,6 +17,7 @@ import (
 	"github.com/rs/zerolog"
 	"golang.org/x/sync/errgroup"
 
+	"github.com/cloudflare/cloudflared/client"
 	cfdflow "github.com/cloudflare/cloudflared/flow"
 
 	cfdquic "github.com/cloudflare/cloudflared/quic"
@@ -43,7 +44,7 @@ type quicConnection struct {
 	orchestrator         Orchestrator
 	datagramHandler      DatagramSessionHandler
 	controlStreamHandler ControlStreamHandler
-	connOptions          *pogs.ConnectionOptions
+	connOptions          *client.ConnectionOptionsSnapshot
 	connIndex            uint8
 
 	rpcTimeout         time.Duration
@@ -59,7 +60,7 @@ func NewTunnelConnection(
 	orchestrator Orchestrator,
 	datagramSessionHandler DatagramSessionHandler,
 	controlStreamHandler ControlStreamHandler,
-	connOptions *pogs.ConnectionOptions,
+	connOptions *client.ConnectionOptionsSnapshot,
 	rpcTimeout time.Duration,
 	streamWriteTimeout time.Duration,
 	gracePeriod time.Duration,
@@ -130,7 +131,7 @@ func (q *quicConnection) Serve(ctx context.Context) error {
 
 // serveControlStream will serve the RPC; blocking until the control plane is done.
 func (q *quicConnection) serveControlStream(ctx context.Context, controlStream quic.Stream) error {
-	return q.controlStreamHandler.ServeControlStream(ctx, controlStream, q.connOptions, q.orchestrator)
+	return q.controlStreamHandler.ServeControlStream(ctx, controlStream, q.connOptions.ConnectionOptions(), q.orchestrator)
 }
 
 // Close the connection with no errors specified.

connection/quic_connection_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"golang.org/x/net/nettest"
 
+	"github.com/cloudflare/cloudflared/client"
 	cfdflow "github.com/cloudflare/cloudflared/flow"
 
 	"github.com/cloudflare/cloudflared/datagramsession"
@@ -843,7 +844,7 @@ func testTunnelConnection(t *testing.T, serverAddr netip.AddrPort, index uint8)
 		&mockOrchestrator{originProxy: &mockOriginProxyWithRequest{}},
 		datagramConn,
 		fakeControlStream{},
-		&pogs.ConnectionOptions{},
+		&client.ConnectionOptionsSnapshot{},
 		15*time.Second,
 		0*time.Second,
 		0*time.Second,