TUN-6016: Push local managed tunnels configuration to the edge

Author: joliveirinha

cmd/cloudflared/tunnel/cmd.go

@@ -343,13 +343,13 @@ func StartServer(
 		observer.SendURL(quickTunnelURL)
 	}
 
-	tunnelConfig, dynamicConfig, err := prepareTunnelConfig(c, info, log, logTransport, observer, namedTunnel)
+	tunnelConfig, orchestratorConfig, err := prepareTunnelConfig(c, info, log, logTransport, observer, namedTunnel)
 	if err != nil {
 		log.Err(err).Msg("Couldn't start tunnel")
 		return err
 	}
 
-	orchestrator, err := orchestration.NewOrchestrator(ctx, dynamicConfig, tunnelConfig.Tags, tunnelConfig.Log)
+	orchestrator, err := orchestration.NewOrchestrator(ctx, orchestratorConfig, tunnelConfig.Tags, tunnelConfig.Log)
 	if err != nil {
 		return err
 	}
@@ -388,7 +388,7 @@ func StartServer(
 			info.Version(),
 			hostname,
 			metricsListener.Addr().String(),
-			dynamicConfig.Ingress,
+			orchestratorConfig.Ingress,
 			tunnelConfig.HAConnections,
 		)
 		app := tunnelUI.Launch(ctx, log, logTransport)

cmd/cloudflared/tunnel/configuration.go

@@ -43,6 +43,8 @@ var (
 
 	secretFlags     = [2]*altsrc.StringFlag{credentialsContentsFlag, tunnelTokenFlag}
 	defaultFeatures = []string{supervisor.FeatureAllowRemoteConfig, supervisor.FeatureSerializedHeaders}
+
+	configFlags = []string{"autoupdate-freq", "no-autoupdate", "retries", "protocol", "loglevel", "transport-loglevel", "origincert", "metrics", "metrics-update-freq"}
 )
 
 // returns the first path that contains a cert.pem file. If none of the DefaultConfigSearchDirectories
@@ -348,11 +350,24 @@ func prepareTunnelConfig(
 		ProtocolSelector: protocolSelector,
 		EdgeTLSConfigs:   edgeTLSConfigs,
 	}
-	dynamicConfig := &orchestration.Config{
+	orchestratorConfig := &orchestration.Config{
 		Ingress:            &ingressRules,
 		WarpRoutingEnabled: warpRoutingEnabled,
+		ConfigurationFlags: parseConfigFlags(c),
 	}
-	return tunnelConfig, dynamicConfig, nil
+	return tunnelConfig, orchestratorConfig, nil
+}
+
+func parseConfigFlags(c *cli.Context) map[string]string {
+	result := make(map[string]string)
+
+	for _, flag := range configFlags {
+		if v := c.String(flag); c.IsSet(flag) && v != "" {
+			result[flag] = v
+		}
+	}
+
+	return result
 }
 
 func gracePeriod(c *cli.Context) (time.Duration, error) {

config/configuration.go

@@ -177,9 +177,9 @@ func ValidateUrl(c *cli.Context, allowURLFromArgs bool) (*url.URL, error) {
 }
 
 type UnvalidatedIngressRule struct {
-	Hostname      string              `json:"hostname"`
-	Path          string              `json:"path"`
-	Service       string              `json:"service"`
+	Hostname      string              `json:"hostname,omitempty"`
+	Path          string              `json:"path,omitempty"`
+	Service       string              `json:"service,omitempty"`
 	OriginRequest OriginRequestConfig `yaml:"originRequest" json:"originRequest"`
 }
 
@@ -192,41 +192,41 @@ type UnvalidatedIngressRule struct {
 // - To specify a time.Duration in json, use int64 of the nanoseconds
 type OriginRequestConfig struct {
 	// HTTP proxy timeout for establishing a new connection
-	ConnectTimeout *CustomDuration `yaml:"connectTimeout" json:"connectTimeout"`
+	ConnectTimeout *CustomDuration `yaml:"connectTimeout" json:"connectTimeout,omitempty"`
 	// HTTP proxy timeout for completing a TLS handshake
-	TLSTimeout *CustomDuration `yaml:"tlsTimeout" json:"tlsTimeout"`
+	TLSTimeout *CustomDuration `yaml:"tlsTimeout" json:"tlsTimeout,omitempty"`
 	// HTTP proxy TCP keepalive duration
-	TCPKeepAlive *CustomDuration `yaml:"tcpKeepAlive" json:"tcpKeepAlive"`
+	TCPKeepAlive *CustomDuration `yaml:"tcpKeepAlive" json:"tcpKeepAlive,omitempty"`
 	// HTTP proxy should disable "happy eyeballs" for IPv4/v6 fallback
-	NoHappyEyeballs *bool `yaml:"noHappyEyeballs" json:"noHappyEyeballs"`
+	NoHappyEyeballs *bool `yaml:"noHappyEyeballs" json:"noHappyEyeballs,omitempty"`
 	// HTTP proxy maximum keepalive connection pool size
-	KeepAliveConnections *int `yaml:"keepAliveConnections" json:"keepAliveConnections"`
+	KeepAliveConnections *int `yaml:"keepAliveConnections" json:"keepAliveConnections,omitempty"`
 	// HTTP proxy timeout for closing an idle connection
-	KeepAliveTimeout *CustomDuration `yaml:"keepAliveTimeout" json:"keepAliveTimeout"`
+	KeepAliveTimeout *CustomDuration `yaml:"keepAliveTimeout" json:"keepAliveTimeout,omitempty"`
 	// Sets the HTTP Host header for the local webserver.
-	HTTPHostHeader *string `yaml:"httpHostHeader" json:"httpHostHeader"`
+	HTTPHostHeader *string `yaml:"httpHostHeader" json:"httpHostHeader,omitempty"`
 	// Hostname on the origin server certificate.
-	OriginServerName *string `yaml:"originServerName" json:"originServerName"`
+	OriginServerName *string `yaml:"originServerName" json:"originServerName,omitempty"`
 	// Path to the CA for the certificate of your origin.
 	// This option should be used only if your certificate is not signed by Cloudflare.
-	CAPool *string `yaml:"caPool" json:"caPool"`
+	CAPool *string `yaml:"caPool" json:"caPool,omitempty"`
 	// Disables TLS verification of the certificate presented by your origin.
 	// Will allow any certificate from the origin to be accepted.
 	// Note: The connection from your machine to Cloudflare's Edge is still encrypted.
-	NoTLSVerify *bool `yaml:"noTLSVerify" json:"noTLSVerify"`
+	NoTLSVerify *bool `yaml:"noTLSVerify" json:"noTLSVerify,omitempty"`
 	// Disables chunked transfer encoding.
 	// Useful if you are running a WSGI server.
-	DisableChunkedEncoding *bool `yaml:"disableChunkedEncoding" json:"disableChunkedEncoding"`
+	DisableChunkedEncoding *bool `yaml:"disableChunkedEncoding" json:"disableChunkedEncoding,omitempty"`
 	// Runs as jump host
-	BastionMode *bool `yaml:"bastionMode" json:"bastionMode"`
+	BastionMode *bool `yaml:"bastionMode" json:"bastionMode,omitempty"`
 	// Listen address for the proxy.
-	ProxyAddress *string `yaml:"proxyAddress" json:"proxyAddress"`
+	ProxyAddress *string `yaml:"proxyAddress" json:"proxyAddress,omitempty"`
 	// Listen port for the proxy.
-	ProxyPort *uint `yaml:"proxyPort" json:"proxyPort"`
+	ProxyPort *uint `yaml:"proxyPort" json:"proxyPort,omitempty"`
 	// Valid options are 'socks' or empty.
-	ProxyType *string `yaml:"proxyType" json:"proxyType"`
+	ProxyType *string `yaml:"proxyType" json:"proxyType,omitempty"`
 	// IP rules for the proxy service
-	IPRules []IngressIPRule `yaml:"ipRules" json:"ipRules"`
+	IPRules []IngressIPRule `yaml:"ipRules" json:"ipRules,omitempty"`
 }
 
 type IngressIPRule struct {

connection/connection.go

@@ -30,6 +30,7 @@ var switchingProtocolText = fmt.Sprintf("%d %s", http.StatusSwitchingProtocols,
 
 type Orchestrator interface {
 	UpdateConfig(version int32, config []byte) *pogs.UpdateConfigurationResponse
+	GetConfigJSON() ([]byte, error)
 	GetOriginProxy() (OriginProxy, error)
 }
 

connection/connection_test.go

@@ -42,6 +42,10 @@ type mockOrchestrator struct {
 	originProxy OriginProxy
 }
 
+func (mcr *mockOrchestrator) GetConfigJSON() ([]byte, error) {
+	return nil, fmt.Errorf("not implemented")
+}
+
 func (*mockOrchestrator) UpdateConfig(version int32, config []byte) *tunnelpogs.UpdateConfigurationResponse {
 	return &tunnelpogs.UpdateConfigurationResponse{
 		LastAppliedVersion: version,

connection/control.go

@@ -30,11 +30,15 @@ type controlStream struct {
 // ControlStreamHandler registers connections with origintunneld and initiates graceful shutdown.
 type ControlStreamHandler interface {
 	// ServeControlStream handles the control plane of the transport in the current goroutine calling this
-	ServeControlStream(ctx context.Context, rw io.ReadWriteCloser, connOptions *tunnelpogs.ConnectionOptions) error
+	ServeControlStream(ctx context.Context, rw io.ReadWriteCloser, connOptions *tunnelpogs.ConnectionOptions, tunnelConfigGetter TunnelConfigJSONGetter) error
 	// IsStopped tells whether the method above has finished
 	IsStopped() bool
 }
 
+type TunnelConfigJSONGetter interface {
+	GetConfigJSON() ([]byte, error)
+}
+
 // NewControlStream returns a new instance of ControlStreamHandler
 func NewControlStream(
 	observer *Observer,
@@ -63,15 +67,28 @@ func (c *controlStream) ServeControlStream(
 	ctx context.Context,
 	rw io.ReadWriteCloser,
 	connOptions *tunnelpogs.ConnectionOptions,
+	tunnelConfigGetter TunnelConfigJSONGetter,
 ) error {
 	rpcClient := c.newRPCClientFunc(ctx, rw, c.observer.log)
 
-	if err := rpcClient.RegisterConnection(ctx, c.namedTunnelProperties, connOptions, c.connIndex, c.observer); err != nil {
+	registrationDetails, err := rpcClient.RegisterConnection(ctx, c.namedTunnelProperties, connOptions, c.connIndex, c.observer)
+	if err != nil {
 		rpcClient.Close()
 		return err
 	}
 	c.connectedFuse.Connected()
 
+	// if conn index is 0 and tunnel is not remotely managed, then send local ingress rules configuration
+	if c.connIndex == 0 && !registrationDetails.TunnelIsRemotelyManaged {
+		if tunnelConfig, err := tunnelConfigGetter.GetConfigJSON(); err == nil {
+			if err := rpcClient.SendLocalConfiguration(ctx, tunnelConfig, c.observer); err != nil {
+				c.observer.log.Err(err).Msg("unable to send local configuration")
+			}
+		} else {
+			c.observer.log.Err(err).Msg("failed to obtain current configuration")
+		}
+	}
+
 	c.waitForUnregister(ctx, rpcClient)
 	return nil
 }

connection/http2.go

@@ -117,7 +117,7 @@ func (c *HTTP2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 	switch connType {
 	case TypeControlStream:
-		if err := c.controlStreamHandler.ServeControlStream(r.Context(), respWriter, c.connOptions); err != nil {
+		if err := c.controlStreamHandler.ServeControlStream(r.Context(), respWriter, c.connOptions, c.orchestrator); err != nil {
 			c.controlStreamErr = err
 			c.log.Error().Err(err)
 			respWriter.WriteErrorResponse()

connection/http2_test.go

@@ -15,6 +15,7 @@ import (
 	"time"
 
 	"github.com/gobwas/ws/wsutil"
+	"github.com/google/uuid"
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -166,18 +167,26 @@ type mockNamedTunnelRPCClient struct {
 	unregistered chan struct{}
 }
 
+func (mc mockNamedTunnelRPCClient) SendLocalConfiguration(c context.Context, config []byte, observer *Observer) error {
+	return nil
+}
+
 func (mc mockNamedTunnelRPCClient) RegisterConnection(
 	c context.Context,
 	properties *NamedTunnelProperties,
 	options *tunnelpogs.ConnectionOptions,
 	connIndex uint8,
 	observer *Observer,
-) error {
+) (*tunnelpogs.ConnectionDetails, error) {
 	if mc.shouldFail != nil {
-		return mc.shouldFail
+		return nil, mc.shouldFail
 	}
 	close(mc.registered)
-	return nil
+	return &tunnelpogs.ConnectionDetails{
+		Location:                "LIS",
+		UUID:                    uuid.New(),
+		TunnelIsRemotelyManaged: false,
+	}, nil
 }
 
 func (mc mockNamedTunnelRPCClient) GracefulShutdown(ctx context.Context, gracePeriod time.Duration) {
@@ -477,7 +486,7 @@ func TestGracefulShutdownHTTP2(t *testing.T) {
 
 	select {
 	case <-rpcClientFactory.registered:
-		break //ok
+		break // ok
 	case <-time.Tick(time.Second):
 		t.Fatal("timeout out waiting for registration")
 	}
@@ -487,7 +496,7 @@ func TestGracefulShutdownHTTP2(t *testing.T) {
 
 	select {
 	case <-rpcClientFactory.unregistered:
-		break //ok
+		break // ok
 	case <-time.Tick(time.Second):
 		t.Fatal("timeout out waiting for unregistered signal")
 	}

connection/metrics.go

@@ -13,6 +13,7 @@ const (
 	MetricsNamespace = "cloudflared"
 	TunnelSubsystem  = "tunnel"
 	muxerSubsystem   = "muxer"
+	configSubsystem  = "config"
 )
 
 type muxerMetrics struct {
@@ -36,6 +37,11 @@ type muxerMetrics struct {
 	compRateAve      *prometheus.GaugeVec
 }
 
+type localConfigMetrics struct {
+	pushes       prometheus.Counter
+	pushesErrors prometheus.Counter
+}
+
 type tunnelMetrics struct {
 	timerRetries    prometheus.Gauge
 	serverLocations *prometheus.GaugeVec
@@ -51,6 +57,39 @@ type tunnelMetrics struct {
 	muxerMetrics        *muxerMetrics
 	tunnelsHA           tunnelsForHA
 	userHostnamesCounts *prometheus.CounterVec
+
+	localConfigMetrics *localConfigMetrics
+}
+
+func newLocalConfigMetrics() *localConfigMetrics {
+
+	pushesMetric := prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: MetricsNamespace,
+			Subsystem: configSubsystem,
+			Name:      "local_config_pushes",
+			Help:      "Number of local configuration pushes to the edge",
+		},
+	)
+
+	pushesErrorsMetric := prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: MetricsNamespace,
+			Subsystem: configSubsystem,
+			Name:      "local_config_pushes_errors",
+			Help:      "Number of errors occurred during local configuration pushes",
+		},
+	)
+
+	prometheus.MustRegister(
+		pushesMetric,
+		pushesErrorsMetric,
+	)
+
+	return &localConfigMetrics{
+		pushes:       pushesMetric,
+		pushesErrors: pushesErrorsMetric,
+	}
 }
 
 func newMuxerMetrics() *muxerMetrics {
@@ -386,6 +425,7 @@ func initTunnelMetrics() *tunnelMetrics {
 		regFail:             registerFail,
 		rpcFail:             rpcFail,
 		userHostnamesCounts: userHostnamesCounts,
+		localConfigMetrics:  newLocalConfigMetrics(),
 	}
 }
 

connection/quic.go

@@ -111,7 +111,7 @@ func (q *QUICConnection) Serve(ctx context.Context) error {
 
 func (q *QUICConnection) serveControlStream(ctx context.Context, controlStream quic.Stream) error {
 	// This blocks until the control plane is done.
-	err := q.controlStreamHandler.ServeControlStream(ctx, controlStream, q.connOptions)
+	err := q.controlStreamHandler.ServeControlStream(ctx, controlStream, q.connOptions, q.orchestrator)
 	if err != nil {
 		// Not wrapping error here to be consistent with the http2 message.
 		return err