Release 2022.5.2

Allow for cloudflared to be built with debug mode to be used with dlv exec

.teamcity/update-homebrew-core.sh

@@ -19,8 +19,8 @@ eval "$(tmp/homebrew/bin/brew shellenv)"
 brew update --force --quiet
 chmod -R go-w "$(brew --prefix)/share/zsh"
 
-git config user.name "cloudflare-warp-bot"
-git config user.email "warp-bot@cloudflare.com"
+git config --global user.name "cloudflare-warp-bot"
+git config --global user.email "warp-bot@cloudflare.com"
 
 # bump formula pr
-brew bump-formula-pr cloudflared --version="$VERSION"
+brew bump-formula-pr cloudflared --version="$VERSION" --no-browse

Makefile

@@ -40,6 +40,10 @@ ifneq ($(GO_BUILD_TAGS),)
 	GO_BUILD_TAGS := -tags "$(GO_BUILD_TAGS)"
 endif
 
+ifeq ($(debug), 1)
+	GO_BUILD_TAGS += -gcflags="all=-N -l"
+endif
+
 IMPORT_PATH    := github.com/cloudflare/cloudflared
 PACKAGE_DIR    := $(CURDIR)/packaging
 PREFIX         := /usr
@@ -261,6 +265,10 @@ github-release: cloudflared
 github-release-built-pkgs:
 	python3 github_release.py --path $(PWD)/built_artifacts --release-version $(VERSION)
 
+.PHONY: release-pkgs-linux
+release-pkgs-linux:
+	python3 ./release_pkgs.py
+
 .PHONY: github-message
 github-message:
 	python3 github_message.py --release-version $(VERSION)

RELEASE_NOTES

@@ -1,3 +1,26 @@
+2022.5.2
+- 2022-05-23 TUN-6270: Import gpg keys from environment variables
+- 2022-05-24 TUN-6209: Improve feedback process if release_pkgs to deb and rpm fail
+- 2022-05-24 TUN-6280: Don't wrap qlog connection tracer for gatethering QUIC metrics since we're not writing qlog files.
+- 2022-05-25 TUN-6209: Sign RPM packages
+- 2022-05-25 TUN-6285: Upload pkg assets to repos when cloudflared is released.
+- 2022-05-24 TUN-6282: Upgrade golang to 1.17.10, go-boring to 1.17.9
+- 2022-05-26 TUN-6292: Debug builds for cloudflared
+- 2022-05-28 TUN-6304: Fixed some file permission issues
+- 2022-05-11 TUN-6197: Publish to brew core should not try to open the browser
+- 2022-05-12 TUN-5943: Add RPM support
+- 2022-05-18 TUN-6248: Fix panic in cloudflared during tracing when origin doesn't provide header map
+- 2022-05-18 TUN-6250: Add upstream response status code to tracing span attributes
+
+2022.5.1
+- 2022-05-06 TUN-6146: Release_pkgs is now a generic command line script
+- 2022-05-06 TUN-6185: Fix tcpOverWSOriginService not using original scheme for String representation
+- 2022-05-05 TUN-6175: Simply debian packaging by structural upload
+- 2022-05-05 TUN-5945: Added support for Ubuntu releases
+- 2022-05-04 TUN-6054: Create and upload deb packages to R2
+- 2022-05-03 TUN-6161: Set git user/email for brew core release
+- 2022-05-03 TUN-6166: Fix mocked QUIC transport for UDP proxy manager to return expected error
+- 2022-04-27 TUN-6016: Push local managed tunnels configuration to the edge
 2022.5.0
 - 2022-05-02 TUN-6158: Update golang.org/x/crypto
 - 2022-04-20 VULN-8383 Bump yaml.v2 to yaml.v3

cfsetup.yaml

@@ -1,5 +1,5 @@
-pinned_go: &pinned_go go=1.17.5-1
-pinned_go_fips: &pinned_go_fips go-boring=1.17.5-1
+pinned_go: &pinned_go go=1.17.10-1
+pinned_go_fips: &pinned_go_fips go-boring=1.17.9-1
 
 build_dir: &build_dir /cfsetup_build
 default-flavor: buster
@@ -40,16 +40,22 @@ stretch: &stretch
       - libffi-dev
       - python3-setuptools
       - python3-pip
+      - reprepro
+      - createrepo
     pre-cache: &github_release_pkgs_pre_cache
       - wget https://github.com/sudarshan-reddy/msitools/releases/download/v0.101b/wixl -P /usr/local/bin
       - chmod a+x /usr/local/bin/wixl
       - pip3 install pynacl==1.4.0
       - pip3 install pygithub==1.55
+      - pip3 install boto3==1.22.9
+      - pip3 install python-gnupg==0.4.9
     post-cache:
       # build all packages (except macos and FIPS) and move them to /cfsetup/built_artifacts
       - ./build-packages.sh
       # release the packages built and moved to /cfsetup/built_artifacts
       - make github-release-built-pkgs
+      # publish packages to linux repos
+      - make release-pkgs-linux
   # handle FIPS separately so that we built with gofips compiler
   github-fips-release-pkgs:
     build_dir: *build_dir
@@ -249,10 +255,10 @@ centos-7:
     pre-cache:
       - yum install -y fakeroot
       - yum upgrade -y binutils-2.27-44.base.el7.x86_64
-      - wget https://go.dev/dl/go1.17.5.linux-amd64.tar.gz -P /tmp/
-      - tar -C /usr/local -xzf /tmp/go1.17.5.linux-amd64.tar.gz
+      - wget https://go.dev/dl/go1.17.10.linux-amd64.tar.gz -P /tmp/
+      - tar -C /usr/local -xzf /tmp/go1.17.10.linux-amd64.tar.gz
     post-cache:
       - export PATH=$PATH:/usr/local/go/bin
       - export GOOS=linux
       - export GOARCH=amd64
-      - make publish-rpm
+      - make publish-rpm

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

connection/quic_test.go

@@ -163,7 +163,7 @@ type fakeControlStream struct {
 	ControlStreamHandler
 }
 
-func (fakeControlStream) ServeControlStream(ctx context.Context, rw io.ReadWriteCloser, connOptions *tunnelpogs.ConnectionOptions) error {
+func (fakeControlStream) ServeControlStream(ctx context.Context, rw io.ReadWriteCloser, connOptions *tunnelpogs.ConnectionOptions, tunnelConfigGetter TunnelConfigJSONGetter) error {
 	<-ctx.Done()
 	return nil
 }

connection/rpc.go

@@ -58,6 +58,11 @@ type NamedTunnelRPCClient interface {
 		options *tunnelpogs.ConnectionOptions,
 		connIndex uint8,
 		observer *Observer,
+	) (*tunnelpogs.ConnectionDetails, error)
+	SendLocalConfiguration(
+		c context.Context,
+		config []byte,
+		observer *Observer,
 	) error
 	GracefulShutdown(ctx context.Context, gracePeriod time.Duration)
 	Close()
@@ -90,7 +95,7 @@ func (rsc *registrationServerClient) RegisterConnection(
 	options *tunnelpogs.ConnectionOptions,
 	connIndex uint8,
 	observer *Observer,
-) error {
+) (*tunnelpogs.ConnectionDetails, error) {
 	conn, err := rsc.client.RegisterConnection(
 		ctx,
 		properties.Credentials.Auth(),
@@ -101,10 +106,10 @@ func (rsc *registrationServerClient) RegisterConnection(
 	if err != nil {
 		if err.Error() == DuplicateConnectionError {
 			observer.metrics.regFail.WithLabelValues("dup_edge_conn", "registerConnection").Inc()
-			return errDuplicationConnection
+			return nil, errDuplicationConnection
 		}
 		observer.metrics.regFail.WithLabelValues("server_error", "registerConnection").Inc()
-		return serverRegistrationErrorFromRPC(err)
+		return nil, serverRegistrationErrorFromRPC(err)
 	}
 
 	observer.metrics.regSuccess.WithLabelValues("registerConnection").Inc()
@@ -112,7 +117,18 @@ func (rsc *registrationServerClient) RegisterConnection(
 	observer.logServerInfo(connIndex, conn.Location, fmt.Sprintf("Connection %s registered", conn.UUID))
 	observer.sendConnectedEvent(connIndex, conn.Location)
 
-	return nil
+	return conn, nil
+}
+
+func (rsc *registrationServerClient) SendLocalConfiguration(ctx context.Context, config []byte, observer *Observer) (err error) {
+	observer.metrics.localConfigMetrics.pushes.Inc()
+	defer func() {
+		if err != nil {
+			observer.metrics.localConfigMetrics.pushesErrors.Inc()
+		}
+	}()
+
+	return rsc.client.SendLocalConfiguration(ctx, config)
 }
 
 func (rsc *registrationServerClient) GracefulShutdown(ctx context.Context, gracePeriod time.Duration) {
@@ -274,7 +290,7 @@ func (h *h2muxConnection) registerNamedTunnel(
 	rpcClient := h.newRPCClientFunc(ctx, stream, h.observer.log)
 	defer rpcClient.Close()
 
-	if err = rpcClient.RegisterConnection(ctx, namedTunnel, connOptions, h.connIndex, h.observer); err != nil {
+	if _, err = rpcClient.RegisterConnection(ctx, namedTunnel, connOptions, h.connIndex, h.observer); err != nil {
 		return err
 	}
 	return nil

datagramsession/manager_test.go

@@ -255,7 +255,10 @@ func (rc *datagramChannel) Send(ctx context.Context, sessionID uuid.UUID, payloa
 	case <-ctx.Done():
 		return ctx.Err()
 	case <-rc.closedChan:
-		return fmt.Errorf("datagram channel closed")
+		return &errClosedSession{
+			message:  fmt.Errorf("datagram channel closed").Error(),
+			byRemote: true,
+		}
 	case rc.datagramChan <- &newDatagram{sessionID: sessionID, payload: payload}:
 		return nil
 	}
@@ -266,7 +269,11 @@ func (rc *datagramChannel) Receive(ctx context.Context) (uuid.UUID, []byte, erro
 	case <-ctx.Done():
 		return uuid.Nil, nil, ctx.Err()
 	case <-rc.closedChan:
-		return uuid.Nil, nil, fmt.Errorf("datagram channel closed")
+		err := &errClosedSession{
+			message:  fmt.Errorf("datagram channel closed").Error(),
+			byRemote: true,
+		}
+		return uuid.Nil, nil, err
 	case msg := <-rc.datagramChan:
 		return msg.sessionID, msg.payload, nil
 	}

dev.Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.17.5 as builder
+FROM golang:1.17.10 as builder
 ENV GO111MODULE=on \
     CGO_ENABLED=0
 WORKDIR /go/src/github.com/cloudflare/cloudflared/
@@ -6,4 +6,4 @@ RUN apt-get update
 COPY . .
 # compile cloudflared
 RUN make cloudflared
-RUN cp /go/src/github.com/cloudflare/cloudflared/cloudflared /usr/local/bin/
+RUN cp /go/src/github.com/cloudflare/cloudflared/cloudflared /usr/local/bin/

ingress/config.go

@@ -47,18 +47,26 @@ type RemoteConfig struct {
 	WarpRouting config.WarpRoutingConfig
 }
 
-type remoteConfigJSON struct {
-	GlobalOriginRequest config.OriginRequestConfig      `json:"originRequest"`
+type RemoteConfigJSON struct {
+	GlobalOriginRequest *config.OriginRequestConfig     `json:"originRequest,omitempty"`
 	IngressRules        []config.UnvalidatedIngressRule `json:"ingress"`
 	WarpRouting         config.WarpRoutingConfig        `json:"warp-routing"`
 }
 
 func (rc *RemoteConfig) UnmarshalJSON(b []byte) error {
-	var rawConfig remoteConfigJSON
+	var rawConfig RemoteConfigJSON
+
 	if err := json.Unmarshal(b, &rawConfig); err != nil {
 		return err
 	}
-	ingress, err := validateIngress(rawConfig.IngressRules, originRequestFromConfig(rawConfig.GlobalOriginRequest))
+
+	// if nil, just assume the default values.
+	globalOriginRequestConfig := rawConfig.GlobalOriginRequest
+	if globalOriginRequestConfig == nil {
+		globalOriginRequestConfig = &config.OriginRequestConfig{}
+	}
+
+	ingress, err := validateIngress(rawConfig.IngressRules, originRequestFromConfig(*globalOriginRequestConfig))
 	if err != nil {
 		return err
 	}
@@ -387,3 +395,91 @@ func setConfig(defaults OriginRequestConfig, overrides config.OriginRequestConfi
 	cfg.setIPRules(overrides)
 	return cfg
 }
+
+func ConvertToRawOriginConfig(c OriginRequestConfig) config.OriginRequestConfig {
+	var connectTimeout *config.CustomDuration
+	var tlsTimeout *config.CustomDuration
+	var tcpKeepAlive *config.CustomDuration
+	var keepAliveConnections *int
+	var keepAliveTimeout *config.CustomDuration
+	var proxyAddress *string
+
+	if c.ConnectTimeout != defaultConnectTimeout {
+		connectTimeout = &c.ConnectTimeout
+	}
+	if c.TLSTimeout != defaultTLSTimeout {
+		tlsTimeout = &c.TLSTimeout
+	}
+	if c.TCPKeepAlive != defaultTCPKeepAlive {
+		tcpKeepAlive = &c.TCPKeepAlive
+	}
+	if c.KeepAliveConnections != defaultKeepAliveConnections {
+		keepAliveConnections = &c.KeepAliveConnections
+	}
+	if c.KeepAliveTimeout != defaultKeepAliveTimeout {
+		keepAliveTimeout = &c.KeepAliveTimeout
+	}
+	if c.ProxyAddress != defaultProxyAddress {
+		proxyAddress = &c.ProxyAddress
+	}
+
+	return config.OriginRequestConfig{
+		ConnectTimeout:         connectTimeout,
+		TLSTimeout:             tlsTimeout,
+		TCPKeepAlive:           tcpKeepAlive,
+		NoHappyEyeballs:        defaultBoolToNil(c.NoHappyEyeballs),
+		KeepAliveConnections:   keepAliveConnections,
+		KeepAliveTimeout:       keepAliveTimeout,
+		HTTPHostHeader:         emptyStringToNil(c.HTTPHostHeader),
+		OriginServerName:       emptyStringToNil(c.OriginServerName),
+		CAPool:                 emptyStringToNil(c.CAPool),
+		NoTLSVerify:            defaultBoolToNil(c.NoTLSVerify),
+		DisableChunkedEncoding: defaultBoolToNil(c.DisableChunkedEncoding),
+		BastionMode:            defaultBoolToNil(c.BastionMode),
+		ProxyAddress:           proxyAddress,
+		ProxyPort:              zeroUIntToNil(c.ProxyPort),
+		ProxyType:              emptyStringToNil(c.ProxyType),
+		IPRules:                convertToRawIPRules(c.IPRules),
+	}
+}
+
+func convertToRawIPRules(ipRules []ipaccess.Rule) []config.IngressIPRule {
+	result := make([]config.IngressIPRule, 0)
+	for _, r := range ipRules {
+		cidr := r.StringCIDR()
+
+		newRule := config.IngressIPRule{
+			Prefix: &cidr,
+			Ports:  r.Ports(),
+			Allow:  r.RulePolicy(),
+		}
+
+		result = append(result, newRule)
+	}
+
+	return result
+}
+
+func defaultBoolToNil(b bool) *bool {
+	if b == false {
+		return nil
+	}
+
+	return &b
+}
+
+func emptyStringToNil(s string) *string {
+	if s == "" {
+		return nil
+	}
+
+	return &s
+}
+
+func zeroUIntToNil(v uint) *uint {
+	if v == 0 {
+		return nil
+	}
+
+	return &v
+}

ingress/origin_proxy.go

@@ -1,6 +1,7 @@
 package ingress
 
 import (
+	"fmt"
 	"net"
 	"net/http"
 
@@ -49,7 +50,13 @@ func (o *httpService) RoundTrip(req *http.Request) (*http.Response, error) {
 }
 
 func (o *statusCode) RoundTrip(_ *http.Request) (*http.Response, error) {
-	return o.resp, nil
+	resp := &http.Response{
+		StatusCode: o.code,
+		Status:     fmt.Sprintf("%d %s", o.code, http.StatusText(o.code)),
+		Body:       new(NopReadCloser),
+	}
+
+	return resp, nil
 }
 
 func (o *rawTCPService) EstablishConnection(dest string) (OriginConnection, error) {

ingress/origin_service.go

@@ -109,6 +109,7 @@ func (o rawTCPService) MarshalJSON() ([]byte, error) {
 // tcpOverWSService models TCP origins serving eyeballs connecting over websocket, such as
 // cloudflared access commands.
 type tcpOverWSService struct {
+	scheme        string
 	dest          string
 	isBastion     bool
 	streamHandler streamHandlerFunc
@@ -130,7 +131,8 @@ func newTCPOverWSService(url *url.URL) *tcpOverWSService {
 		addPortIfMissing(url, 7864) // just a random port since there isn't a default in this case
 	}
 	return &tcpOverWSService{
-		dest: url.Host,
+		scheme: url.Scheme,
+		dest:   url.Host,
 	}
 }
 
@@ -160,7 +162,12 @@ func (o *tcpOverWSService) String() string {
 	if o.isBastion {
 		return ServiceBastion
 	}
-	return o.dest
+
+	if o.scheme != "" {
+		return fmt.Sprintf("%s://%s", o.scheme, o.dest)
+	} else {
+		return o.dest
+	}
 }
 
 func (o *tcpOverWSService) start(log *zerolog.Logger, _ <-chan struct{}, cfg OriginRequestConfig) error {
@@ -231,20 +238,15 @@ func (o helloWorld) MarshalJSON() ([]byte, error) {
 // statusCode is an OriginService that just responds with a given HTTP status.
 // Typical use-case is "user wants the catch-all rule to just respond 404".
 type statusCode struct {
-	resp *http.Response
+	code int
 }
 
 func newStatusCode(status int) statusCode {
-	resp := &http.Response{
-		StatusCode: status,
-		Status:     fmt.Sprintf("%d %s", status, http.StatusText(status)),
-		Body:       new(NopReadCloser),
-	}
-	return statusCode{resp: resp}
+	return statusCode{code: status}
 }
 
 func (o *statusCode) String() string {
-	return fmt.Sprintf("http_status:%d", o.resp.StatusCode)
+	return fmt.Sprintf("http_status:%d", o.code)
 }
 
 func (o *statusCode) start(

ipaccess/access.go

@@ -99,3 +99,15 @@ func (ipr *Rule) PortsString() string {
 	}
 	return "all"
 }
+
+func (ipr *Rule) Ports() []int {
+	return ipr.ports
+}
+
+func (ipr *Rule) RulePolicy() bool {
+	return ipr.allow
+}
+
+func (ipr *Rule) StringCIDR() string {
+	return ipr.ipNet.String()
+}

metrics/metrics.go

@@ -23,7 +23,7 @@ const (
 )
 
 type orchestrator interface {
-	GetConfigJSON() ([]byte, error)
+	GetVersionedConfigJSON() ([]byte, error)
 }
 
 func newMetricsHandler(
@@ -47,7 +47,7 @@ func newMetricsHandler(
 	})
 	if orchestrator != nil {
 		router.HandleFunc("/config", func(w http.ResponseWriter, r *http.Request) {
-			json, err := orchestrator.GetConfigJSON()
+			json, err := orchestrator.GetVersionedConfigJSON()
 			if err != nil {
 				w.WriteHeader(500)
 				_, _ = fmt.Fprintf(w, "ERR: %v", err)

orchestration/config.go

@@ -1,15 +1,66 @@
 package orchestration
 
 import (
+	"encoding/json"
+
+	"github.com/cloudflare/cloudflared/config"
 	"github.com/cloudflare/cloudflared/ingress"
 )
 
-type newConfig struct {
+type newRemoteConfig struct {
 	ingress.RemoteConfig
 	// Add more fields when we support other settings in tunnel orchestration
 }
 
+type newLocalConfig struct {
+	RemoteConfig       ingress.RemoteConfig
+	ConfigurationFlags map[string]string `json:"__configuration_flags,omitempty"`
+}
+
+// Config is the original config as read and parsed by cloudflared.
 type Config struct {
 	Ingress            *ingress.Ingress
 	WarpRoutingEnabled bool
+
+	// Extra settings used to configure this instance but that are not eligible for remotely management
+	// ie. (--protocol, --loglevel, ...)
+	ConfigurationFlags map[string]string
+}
+
+func (rc *newLocalConfig) MarshalJSON() ([]byte, error) {
+	var r = struct {
+		ConfigurationFlags map[string]string `json:"__configuration_flags,omitempty"`
+		ingress.RemoteConfigJSON
+	}{
+		ConfigurationFlags: rc.ConfigurationFlags,
+		RemoteConfigJSON: ingress.RemoteConfigJSON{
+			// UI doesn't support top level configs, so we reconcile to individual ingress configs.
+			GlobalOriginRequest: nil,
+			IngressRules:        convertToUnvalidatedIngressRules(rc.RemoteConfig.Ingress),
+			WarpRouting:         rc.RemoteConfig.WarpRouting,
+		},
+	}
+
+	return json.Marshal(r)
+}
+
+func convertToUnvalidatedIngressRules(i ingress.Ingress) []config.UnvalidatedIngressRule {
+	result := make([]config.UnvalidatedIngressRule, 0)
+	for _, rule := range i.Rules {
+		var path string
+		if rule.Path != nil {
+			path = rule.Path.String()
+		}
+
+		newRule := config.UnvalidatedIngressRule{
+			Hostname:      rule.Hostname,
+			Path:          path,
+			Service:       rule.Service.String(),
+			OriginRequest: ingress.ConvertToRawOriginConfig(rule.Config),
+		}
+
+		result = append(result, newRule)
+	}
+
+	return result
 }

orchestration/config_test.go

@@ -0,0 +1,82 @@
+package orchestration
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/cloudflare/cloudflared/ingress"
+)
+
+// TestNewLocalConfig_MarshalJSON tests that we are able to converte a compiled and validated config back
+// into an "unvalidated" format which is compatible with Remote Managed configurations.
+func TestNewLocalConfig_MarshalJSON(t *testing.T) {
+
+	rawConfig := []byte(`
+	{
+		"originRequest": {
+					"connectTimeout": 160,
+					"httpHostHeader": "default"
+		},
+		"ingress": [
+			{
+				"hostname": "tun.example.com",
+				"service": "https://localhost:8000"
+			},
+			{
+				"hostname": "*",
+				"service": "https://localhost:8001",
+				"originRequest": {
+					"connectTimeout": 121,
+					"tlsTimeout": 2,
+					"noHappyEyeballs": false,
+					"tcpKeepAlive": 2,
+					"keepAliveConnections": 2,
+					"keepAliveTimeout": 2,
+					"httpHostHeader": "def",
+					"originServerName": "b2",
+					"caPool": "/tmp/path1",
+					"noTLSVerify": false,
+					"disableChunkedEncoding": false,
+					"bastionMode": false,
+					"proxyAddress": "interface",
+					"proxyPort": 200,
+					"proxyType": "",
+					"ipRules": [
+						{
+							"prefix": "10.0.0.0/16",
+							"ports": [3000, 3030],
+							"allow": false
+						},
+						{
+							"prefix": "192.16.0.0/24",
+							"ports": [5000, 5050],
+							"allow": true
+						}
+					]
+				}
+			}
+		]
+	}
+	`)
+
+	var expectedConfig ingress.RemoteConfig
+	err := json.Unmarshal(rawConfig, &expectedConfig)
+	require.NoError(t, err)
+
+	c := &newLocalConfig{
+		RemoteConfig:       expectedConfig,
+		ConfigurationFlags: nil,
+	}
+
+	jsonSerde, err := json.Marshal(c)
+	require.NoError(t, err)
+
+	var config ingress.RemoteConfig
+	err = json.Unmarshal(jsonSerde, &config)
+	require.NoError(t, err)
+
+	require.Equal(t, config.WarpRouting.Enabled, false)
+	require.Equal(t, config.Ingress.Rules, expectedConfig.Ingress.Rules)
+}

orchestration/orchestrator.go

@@ -54,7 +54,7 @@ func NewOrchestrator(ctx context.Context, config *Config, tags []tunnelpogs.Tag,
 	return o, nil
 }
 
-// Update creates a new proxy with the new ingress rules
+// UpdateConfig creates a new proxy with the new ingress rules
 func (o *Orchestrator) UpdateConfig(version int32, config []byte) *tunnelpogs.UpdateConfigurationResponse {
 	o.lock.Lock()
 	defer o.lock.Unlock()
@@ -63,12 +63,12 @@ func (o *Orchestrator) UpdateConfig(version int32, config []byte) *tunnelpogs.Up
 		o.log.Debug().
 			Int32("current_version", o.currentVersion).
 			Int32("received_version", version).
-			Msg("Current version is equal or newer than receivied version")
+			Msg("Current version is equal or newer than received version")
 		return &tunnelpogs.UpdateConfigurationResponse{
 			LastAppliedVersion: o.currentVersion,
 		}
 	}
-	var newConf newConfig
+	var newConf newRemoteConfig
 	if err := json.Unmarshal(config, &newConf); err != nil {
 		o.log.Err(err).
 			Int32("version", version).
@@ -131,10 +131,26 @@ func (o *Orchestrator) updateIngress(ingressRules ingress.Ingress, warpRoutingEn
 	return nil
 }
 
-// GetConfigJSON returns the current version and configuration as JSON
+// GetConfigJSON returns the current json serialization of the config as the edge understands it
 func (o *Orchestrator) GetConfigJSON() ([]byte, error) {
 	o.lock.RLock()
 	defer o.lock.RUnlock()
+
+	c := &newLocalConfig{
+		RemoteConfig: ingress.RemoteConfig{
+			Ingress:     *o.config.Ingress,
+			WarpRouting: config.WarpRoutingConfig{Enabled: o.config.WarpRoutingEnabled},
+		},
+		ConfigurationFlags: o.config.ConfigurationFlags,
+	}
+
+	return json.Marshal(c)
+}
+
+// GetVersionedConfigJSON returns the current version and configuration as JSON
+func (o *Orchestrator) GetVersionedConfigJSON() ([]byte, error) {
+	o.lock.RLock()
+	defer o.lock.RUnlock()
 	var currentConfiguration = struct {
 		Version int32 `json:"version"`
 		Config  struct {

orchestration/orchestrator_test.go

@@ -2,6 +2,7 @@ package orchestration
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -641,6 +642,19 @@ func TestPersistentConnection(t *testing.T) {
 	wg.Wait()
 }
 
+func TestSerializeLocalConfig(t *testing.T) {
+	c := &newLocalConfig{
+		RemoteConfig: ingress.RemoteConfig{
+			Ingress:     ingress.Ingress{},
+			WarpRouting: config.WarpRoutingConfig{},
+		},
+		ConfigurationFlags: map[string]string{"a": "b"},
+	}
+
+	result, _ := json.Marshal(c)
+	fmt.Println(string(result))
+}
+
 func wsEcho(w http.ResponseWriter, r *http.Request) {
 	upgrader := gows.Upgrader{}
 

proxy/proxy.go

@@ -11,7 +11,6 @@ import (
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/codes"
 	"go.opentelemetry.io/otel/trace"
 
 	"github.com/cloudflare/cloudflared/carrier"
@@ -197,12 +196,18 @@ func (p *Proxy) proxyHTTPRequest(
 	_, ttfbSpan := tr.Tracer().Start(tr.Context(), "ttfb_origin")
 	resp, err := httpService.RoundTrip(roundTripReq)
 	if err != nil {
-		tracing.EndWithStatus(ttfbSpan, codes.Error, "")
+		tracing.EndWithErrorStatus(ttfbSpan, err)
 		return errors.Wrap(err, "Unable to reach the origin service. The service may be down or it may not be responding to traffic from cloudflared")
 	}
-	tracing.EndWithStatus(ttfbSpan, codes.Ok, resp.Status)
+
+	tracing.EndWithStatusCode(ttfbSpan, resp.StatusCode)
 	defer resp.Body.Close()
 
+	// resp headers can be nil
+	if resp.Header == nil {
+		resp.Header = make(http.Header)
+	}
+
 	// Add spans to response header (if available)
 	tr.AddSpans(resp.Header, p.log)
 

quic/tracing.go

@@ -3,9 +3,9 @@ package quic
 import (
 	"context"
 	"net"
+	"time"
 
 	"github.com/lucas-clemente/quic-go/logging"
-	"github.com/lucas-clemente/quic-go/qlog"
 	"github.com/rs/zerolog"
 )
 
@@ -40,75 +40,116 @@ func NewServerTracer(logger *zerolog.Logger) logging.Tracer {
 	}
 }
 
-func (t *tracer) TracerForConnection(_ context.Context, p logging.Perspective, odcid logging.ConnectionID) logging.ConnectionTracer {
-	connID := logging.ConnectionID(odcid).String()
-	ql := &quicLogger{
-		logger:       t.logger,
-		connectionID: connID,
-	}
+func (t *tracer) TracerForConnection(_ctx context.Context, _p logging.Perspective, _odcid logging.ConnectionID) logging.ConnectionTracer {
 	if t.config.isClient {
-		return newConnTracer(ql, p, odcid, newClientCollector(t.config.index))
+		return newConnTracer(newClientCollector(t.config.index))
 	}
-	return newConnTracer(ql, p, odcid, newServiceCollector())
+	return newConnTracer(newServiceCollector())
 }
 
 func (*tracer) SentPacket(net.Addr, *logging.Header, logging.ByteCount, []logging.Frame) {}
 func (*tracer) DroppedPacket(net.Addr, logging.PacketType, logging.ByteCount, logging.PacketDropReason) {
 }
 
-// connTracer is a wrapper around https://pkg.go.dev/github.com/lucas-clemente/quic-go@v0.23.0/qlog#NewConnectionTracer to collect metrics
+var _ logging.Tracer = (*tracer)(nil)
+
+// connTracer collects connection level metrics
 type connTracer struct {
-	logging.ConnectionTracer
 	metricsCollector MetricsCollector
-	connectionID     string
 }
 
-func newConnTracer(ql *quicLogger, p logging.Perspective, odcid logging.ConnectionID, metricsCollector MetricsCollector) logging.ConnectionTracer {
+var _ logging.ConnectionTracer = (*connTracer)(nil)
+
+func newConnTracer(metricsCollector MetricsCollector) logging.ConnectionTracer {
 	return &connTracer{
-		qlog.NewConnectionTracer(ql, p, odcid),
-		metricsCollector,
-		logging.ConnectionID(odcid).String(),
+		metricsCollector: metricsCollector,
 	}
 }
 
 func (ct *connTracer) StartedConnection(local, remote net.Addr, srcConnID, destConnID logging.ConnectionID) {
 	ct.metricsCollector.startedConnection()
-	ct.ConnectionTracer.StartedConnection(local, remote, srcConnID, destConnID)
 }
 
 func (ct *connTracer) ClosedConnection(err error) {
 	ct.metricsCollector.closedConnection(err)
-	ct.ConnectionTracer.ClosedConnection(err)
 }
 
 func (ct *connTracer) SentPacket(hdr *logging.ExtendedHeader, packetSize logging.ByteCount, ack *logging.AckFrame, frames []logging.Frame) {
 	ct.metricsCollector.sentPackets(packetSize)
-	ct.ConnectionTracer.SentPacket(hdr, packetSize, ack, frames)
 }
 
 func (ct *connTracer) ReceivedPacket(hdr *logging.ExtendedHeader, size logging.ByteCount, frames []logging.Frame) {
 	ct.metricsCollector.receivedPackets(size)
-	ct.ConnectionTracer.ReceivedPacket(hdr, size, frames)
 }
 
 func (ct *connTracer) BufferedPacket(pt logging.PacketType) {
 	ct.metricsCollector.bufferedPackets(pt)
-	ct.ConnectionTracer.BufferedPacket(pt)
 }
 
 func (ct *connTracer) DroppedPacket(pt logging.PacketType, size logging.ByteCount, reason logging.PacketDropReason) {
 	ct.metricsCollector.droppedPackets(pt, size, reason)
-	ct.ConnectionTracer.DroppedPacket(pt, size, reason)
 }
 
 func (ct *connTracer) LostPacket(level logging.EncryptionLevel, number logging.PacketNumber, reason logging.PacketLossReason) {
 	ct.metricsCollector.lostPackets(reason)
-	ct.ConnectionTracer.LostPacket(level, number, reason)
 }
 
 func (ct *connTracer) UpdatedMetrics(rttStats *logging.RTTStats, cwnd, bytesInFlight logging.ByteCount, packetsInFlight int) {
 	ct.metricsCollector.updatedRTT(rttStats)
-	ct.ConnectionTracer.UpdatedMetrics(rttStats, cwnd, bytesInFlight, packetsInFlight)
+}
+
+func (ct *connTracer) NegotiatedVersion(chosen logging.VersionNumber, clientVersions, serverVersions []logging.VersionNumber) {
+}
+
+func (ct *connTracer) SentTransportParameters(parameters *logging.TransportParameters) {
+}
+
+func (ct *connTracer) ReceivedTransportParameters(parameters *logging.TransportParameters) {
+}
+
+func (ct *connTracer) RestoredTransportParameters(parameters *logging.TransportParameters) {
+}
+
+func (ct *connTracer) ReceivedVersionNegotiationPacket(header *logging.Header, numbers []logging.VersionNumber) {
+}
+
+func (ct *connTracer) ReceivedRetry(header *logging.Header) {
+}
+
+func (ct *connTracer) AcknowledgedPacket(level logging.EncryptionLevel, number logging.PacketNumber) {
+}
+
+func (ct *connTracer) UpdatedCongestionState(state logging.CongestionState) {
+}
+
+func (ct *connTracer) UpdatedPTOCount(value uint32) {
+}
+
+func (ct *connTracer) UpdatedKeyFromTLS(level logging.EncryptionLevel, perspective logging.Perspective) {
+}
+
+func (ct *connTracer) UpdatedKey(generation logging.KeyPhase, remote bool) {
+}
+
+func (ct *connTracer) DroppedEncryptionLevel(level logging.EncryptionLevel) {
+}
+
+func (ct *connTracer) DroppedKey(generation logging.KeyPhase) {
+}
+
+func (ct *connTracer) SetLossTimer(timerType logging.TimerType, level logging.EncryptionLevel, time time.Time) {
+}
+
+func (ct *connTracer) LossTimerExpired(timerType logging.TimerType, level logging.EncryptionLevel) {
+}
+
+func (ct *connTracer) LossTimerCanceled() {
+}
+
+func (ct *connTracer) Close() {
+}
+
+func (ct *connTracer) Debug(name, msg string) {
 }
 
 type quicLogger struct {

release_pkgs.py

@@ -0,0 +1,327 @@
+"""
+    This is a utility for creating deb and rpm packages, signing them 
+    and uploading them to a storage and adding metadata to workers KV.
+
+    It has two over-arching responsiblities:
+    1. Create deb and yum repositories from .deb and .rpm files. 
+       This is also responsible for signing the packages and generally preparing 
+       them to be in an uploadable state.
+    2. Upload these packages to a storage in a format that apt and yum expect.
+"""
+from subprocess import Popen, PIPE
+import os
+import argparse
+import base64
+from pathlib import Path
+import logging
+import shutil
+from hashlib import sha256
+
+import gnupg
+import boto3
+from botocore.client import Config
+from botocore.exceptions import ClientError
+
+# The front facing R2 URL to access assets from.
+R2_ASSET_URL = 'https://demo-r2-worker.cloudflare-tunnel.workers.dev/'
+
+class PkgUploader:
+    def __init__(self, account_id, bucket_name, client_id, client_secret):
+        self.account_id = account_id
+        self.bucket_name = bucket_name
+        self.client_id = client_id
+        self.client_secret = client_secret
+
+    def upload_pkg_to_r2(self, filename, upload_file_path):
+        endpoint_url = f"https://{self.account_id}.r2.cloudflarestorage.com"
+        token_secret_hash = sha256(self.client_secret.encode()).hexdigest()
+         
+        config = Config(
+            region_name = 'auto',
+            s3={
+                "addressing_style": "path",
+            }
+        )
+         
+        r2 = boto3.client(
+            "s3",
+            endpoint_url=endpoint_url,
+            aws_access_key_id=self.client_id,
+            aws_secret_access_key=token_secret_hash,
+            config=config,
+        )
+
+        print(f"uploading asset: {filename} to {upload_file_path} in bucket{self.bucket_name}...")
+        try:
+            r2.upload_file(filename, self.bucket_name, upload_file_path)
+        except ClientError as e:
+            raise e
+
+                  
+class PkgCreator:
+    """
+        The distribution conf is what dictates to reprepro, the debian packaging building
+        and signing tool we use, what distros to support, what GPG key to use for signing
+        and what to call the debian binary etc. This function creates it "./conf/distributions".
+
+        origin - name of your package (String)
+        label - label of your package (could be same as the name) (String)
+        release - release you want this to be distributed for (List of Strings)
+        components - could be a channel like main/stable/beta
+        archs - Architecture (List of Strings)
+        description - (String)
+        gpg_key_id - gpg key id of what you want to use to sign the packages.(String) 
+    """
+    def create_distribution_conf(self, 
+            file_path,
+            origin,
+            label,
+            releases,
+            archs,
+            components, 
+            description,
+            gpg_key_id ):
+        with open(file_path, "w+") as distributions_file:
+            for release in releases:
+                distributions_file.write(f"Origin: {origin}\n")
+                distributions_file.write(f"Label: {label}\n")
+                distributions_file.write(f"Codename: {release}\n")
+                archs_list = " ".join(archs)
+                distributions_file.write(f"Architectures: {archs_list}\n")
+                distributions_file.write(f"Components: {components}\n")
+                distributions_file.write(f"Description: {description} - {release}\n")
+                distributions_file.write(f"SignWith: {gpg_key_id}\n")
+                distributions_file.write("\n")
+        return distributions_file
+
+    """
+        Uses the reprepro tool to generate packages, sign them and create the InRelease as specified
+        by the distribution_conf file. 
+
+        This function creates three folders db, pool and dist. 
+        db and pool contain information and metadata about builds. We can ignore these.
+        dist: contains all the pkgs and signed releases that are necessary for an apt download.
+    """
+    def create_deb_pkgs(self, release, deb_file):
+        print(f"creating deb pkgs: {release} : {deb_file}")
+        p = Popen(["reprepro", "includedeb", release, deb_file], stdout=PIPE, stderr=PIPE)
+        out, err = p.communicate()
+        if p.returncode != 0:
+            print(f"create deb_pkgs result => {out}, {err}")
+            raise
+
+    def create_rpm_pkgs(self, artifacts_path, gpg_key_name):
+        self._setup_rpm_pkg_directories(artifacts_path, gpg_key_name)
+        p = Popen(["createrepo", "./rpm"], stdout=PIPE, stderr=PIPE)
+        out, err = p.communicate()
+        if p.returncode != 0:
+            print(f"create rpm_pkgs result => {out}, {err}")
+            raise
+
+        self._sign_repomd()
+
+    def _sign_rpms(self, file_path):
+        p = Popen(["rpm" , "--define", f"_gpg_name {gpg_key_name}", "--addsign", file_path], stdout=PIPE, stderr=PIPE)
+        out, err = p.communicate()
+        if p.returncode != 0:
+            print(f"rpm sign result result => {out}, {err}")
+            raise
+
+    def _sign_repomd(self):
+        p = Popen(["gpg", "--batch", "--detach-sign", "--armor", "./rpm/repodata/repomd.xml"], stdout=PIPE, stderr=PIPE) 
+        out, err = p.communicate()
+        if p.returncode != 0:
+            print(f"sign repomd result => {out}, {err}")
+            raise
+        
+    """
+        sets up and signs the RPM directories in the following format:
+        - rpm 
+           - aarch64
+           - x86_64
+           - 386
+
+        this assumes the assets are in the format <prefix>-<aarch64/x86_64/386>.rpm
+    """
+    def _setup_rpm_pkg_directories(self, artifacts_path, gpg_key_name, archs=["aarch64", "x86_64", "386"]):
+        for arch in archs:
+            for root, _ , files in os.walk(artifacts_path):
+                for file in files:
+                    if file.endswith(f"{arch}.rpm"):
+                        new_dir = f"./rpm/{arch}"
+                        os.makedirs(new_dir, exist_ok=True)
+                        old_path = os.path.join(root, file)
+                        new_path = os.path.join(new_dir, file)
+                        shutil.copyfile(old_path, new_path)
+                        self._sign_rpms(new_path)
+    
+    """
+        imports gpg keys into the system so reprepro and createrepo can use it to sign packages.
+        it returns the GPG ID after a successful import
+    """
+    def import_gpg_keys(self, private_key, public_key):
+        gpg = gnupg.GPG()
+        private_key = base64.b64decode(private_key)
+        gpg.import_keys(private_key)
+        public_key = base64.b64decode(public_key)
+        gpg.import_keys(public_key)
+        data = gpg.list_keys(secret=True)
+        return (data[0]["fingerprint"], data[0]["uids"][0])
+
+    """
+        basically rpm --import <key_file>
+        This enables us to sign rpms.
+    """
+    def import_rpm_key(self, public_key):
+        file_name = "pb.key"
+        with open(file_name, "wb") as f:
+            public_key = base64.b64decode(public_key)
+            f.write(public_key)
+        
+        p = Popen(["rpm", "--import", file_name], stdout=PIPE, stderr=PIPE)
+        out, err = p.communicate()
+        if p.returncode != 0:
+            print(f"create rpm import result => {out}, {err}")
+            raise
+
+
+"""
+    Walks through a directory and uploads it's assets to R2.
+    directory : root directory to walk through (String).
+    release: release string. If this value is none, a specific release path will not be created 
+              and the release will be uploaded to the default path. 
+    binary: name of the binary to upload
+"""
+def upload_from_directories(pkg_uploader, directory, release, binary):
+     for root, _ , files in os.walk(directory):
+        for file in files:
+            upload_file_name = os.path.join(binary, root, file)
+            if release:
+                upload_file_name = os.path.join(release, upload_file_name)
+            filename = os.path.join(root,file)
+            try: 
+                pkg_uploader.upload_pkg_to_r2(filename, upload_file_name)
+            except ClientError as e:
+                logging.error(e)
+                return 
+
+""" 
+    1. looks into a built_artifacts folder for cloudflared debs
+    2. creates Packages.gz, InRelease (signed) files
+    3. uploads them to Cloudflare R2 
+
+    pkg_creator, pkg_uploader: are instantiations of the two classes above.
+
+    gpg_key_id: is an id indicating the key the package should be signed with. The public key of this id will be 
+    uploaded to R2 so it can be presented to apt downloaders.
+
+    release_version: is the cloudflared release version. Only provide this if you want a permanent backup.
+"""
+def create_deb_packaging(pkg_creator, pkg_uploader, releases, gpg_key_id, binary_name, archs, package_component, release_version):
+    # set configuration for package creation.
+    print(f"initialising configuration for {binary_name} , {archs}")
+    Path("./conf").mkdir(parents=True, exist_ok=True)
+    pkg_creator.create_distribution_conf(
+    "./conf/distributions",
+    binary_name,
+    binary_name,
+    releases,
+    archs,
+    package_component,
+    f"apt repository for {binary_name}",
+    gpg_key_id)
+
+    # create deb pkgs
+    for release in releases:
+        for arch in archs:
+            print(f"creating deb pkgs for {release} and {arch}...")
+            pkg_creator.create_deb_pkgs(release, f"./built_artifacts/cloudflared-linux-{arch}.deb")
+
+    print("uploading latest to r2...")
+    upload_from_directories(pkg_uploader, "dists", None, binary_name)
+    upload_from_directories(pkg_uploader, "pool", None, binary_name)
+
+    if release_version:
+        print(f"uploading versioned release {release_version} to r2...")
+        upload_from_directories(pkg_uploader, "dists", release_version, binary_name)
+        upload_from_directories(pkg_uploader, "pool", release_version, binary_name)
+
+def create_rpm_packaging(pkg_creator, pkg_uploader, artifacts_path, release_version, binary_name, gpg_key_name):
+    print(f"creating rpm pkgs...")
+    pkg_creator.create_rpm_pkgs(artifacts_path, gpg_key_name)
+
+    print("uploading latest to r2...")
+    upload_from_directories(pkg_uploader, "rpm", None, binary_name)
+
+    if release_version:
+        print(f"uploading versioned release {release_version} to r2...")
+        upload_from_directories(pkg_uploader, "rpm", release_version, binary_name)
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(
+        description="Creates linux releases and uploads them in a packaged format"
+    )
+
+    parser.add_argument(
+            "--bucket", default=os.environ.get("R2_BUCKET"), help="R2 Bucket name"
+    )
+    parser.add_argument(
+            "--id", default=os.environ.get("R2_CLIENT_ID"), help="R2 Client ID"
+    )
+    parser.add_argument(
+            "--secret", default=os.environ.get("R2_CLIENT_SECRET"), help="R2 Client Secret"
+    )
+    parser.add_argument(
+            "--account", default=os.environ.get("R2_ACCOUNT_ID"), help="R2 Account Tag"
+    )
+    parser.add_argument(
+            "--release-tag", default=os.environ.get("RELEASE_VERSION"), help="Release version you want your pkgs to be\
+            prefixed with. Leave empty if you don't want tagged release versions backed up to R2."
+    )
+
+    parser.add_argument(
+           "--binary", default=os.environ.get("BINARY_NAME"), help="The name of the binary the packages are for"
+    )
+
+    parser.add_argument(
+            "--gpg-private-key", default=os.environ.get("LINUX_SIGNING_PRIVATE_KEY"), help="GPG private key to sign the\
+            packages"
+    )
+
+    parser.add_argument(
+            "--gpg-public-key", default=os.environ.get("LINUX_SIGNING_PUBLIC_KEY"), help="GPG public key used for\
+            signing packages"
+    )
+
+    parser.add_argument(
+            "--deb-based-releases", default=["bookworm", "bullseye", "buster", "jammy", "impish", "focal", "bionic"],
+            help="list of debian based releases that need to be packaged for"
+    )
+
+    parser.add_argument(
+            "--archs", default=["amd64", "386", "arm64"], help="list of architectures we want to package for. Note that\
+            it is the caller's responsiblity to ensure that these debs are already present in a directory. This script\
+            will not build binaries or create their debs."
+            )
+    args = parser.parse_args()
+
+    return args
+
+if __name__ == "__main__":
+    try:
+        args = parse_args()
+    except Exception as e:
+        logging.exception(e)
+        exit(1)
+
+    pkg_creator = PkgCreator()
+    (gpg_key_id, gpg_key_name) = pkg_creator.import_gpg_keys(args.gpg_private_key, args.gpg_public_key)
+    pkg_creator.import_rpm_key(args.gpg_public_key)
+
+    pkg_uploader = PkgUploader(args.account, args.bucket, args.id, args.secret)
+    print(f"signing with gpg_key: {gpg_key_id}")
+    create_deb_packaging(pkg_creator, pkg_uploader, args.deb_based_releases, gpg_key_id, args.binary, args.archs,
+            "main", args.release_tag)
+    
+    create_rpm_packaging(pkg_creator, pkg_uploader, "./built_artifacts", args.release_tag, args.binary, gpg_key_name)

tracing/tracing.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math"
 	"net/http"
 	"os"
 	"runtime"
@@ -29,6 +30,9 @@ const (
 	TracerContextNameOverride = "uber-trace-id"
 
 	IntCloudflaredTracingHeader = "cf-int-cloudflared-tracing"
+
+	MaxErrorDescriptionLen = 100
+	traceHttpStatusCodeKey = "upstreamStatusCode"
 )
 
 var (
@@ -103,6 +107,11 @@ func (cft *TracedRequest) Tracer() trace.Tracer {
 
 // Spans returns the spans as base64 encoded protobuf otlp traces.
 func (cft *TracedRequest) AddSpans(headers http.Header, log *zerolog.Logger) {
+	if headers == nil {
+		log.Error().Msgf("provided headers map is nil")
+		return
+	}
+
 	enc, err := cft.exporter.Spans()
 	switch err {
 	case nil:
@@ -121,15 +130,39 @@ func (cft *TracedRequest) AddSpans(headers http.Header, log *zerolog.Logger) {
 		log.Error().Msgf("no traces provided and no error from exporter")
 		return
 	}
+
 	headers[CanonicalCloudflaredTracingHeader] = []string{enc}
 }
 
-// EndWithStatus will set a status for the span and then end it.
-func EndWithStatus(span trace.Span, code codes.Code, status string) {
+// EndWithErrorStatus will set a status for the span and then end it.
+func EndWithErrorStatus(span trace.Span, err error) {
+	endSpan(span, -1, codes.Error, err)
+}
+
+// EndWithStatusCode will set a status for the span and then end it.
+func EndWithStatusCode(span trace.Span, statusCode int) {
+	endSpan(span, statusCode, codes.Ok, nil)
+}
+
+// EndWithErrorStatus will set a status for the span and then end it.
+func endSpan(span trace.Span, upstreamStatusCode int, spanStatusCode codes.Code, err error) {
 	if span == nil {
 		return
 	}
-	span.SetStatus(code, status)
+
+	if upstreamStatusCode > 0 {
+		span.SetAttributes(attribute.Int(traceHttpStatusCodeKey, upstreamStatusCode))
+	}
+
+	// add error to status buf cap description
+	errDescription := ""
+	if err != nil {
+		errDescription = err.Error()
+		l := int(math.Min(float64(len(errDescription)), MaxErrorDescriptionLen))
+		errDescription = errDescription[:l]
+	}
+
+	span.SetStatus(spanStatusCode, errDescription)
 	span.End()
 }
 

tracing/tracing_test.go

@@ -1,13 +1,16 @@
 package tracing
 
 import (
+	"context"
 	"net/http"
 	"net/http/httptest"
 	"testing"
 
+	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/assert"
 	tracesdk "go.opentelemetry.io/otel/sdk/trace"
 	"go.opentelemetry.io/otel/trace"
+	tracepb "go.opentelemetry.io/proto/otlp/trace/v1"
 )
 
 func TestNewCfTracer(t *testing.T) {
@@ -48,3 +51,19 @@ func TestNewCfTracerInvalidHeaders(t *testing.T) {
 		assert.IsType(t, &NoopOtlpClient{}, tr.exporter)
 	}
 }
+
+func TestAddingSpansWithNilMap(t *testing.T) {
+	req := httptest.NewRequest("GET", "http://localhost", nil)
+	req.Header.Add(TracerContextName, "14cb070dde8e51fc5ae8514e69ba42ca:b38f1bf5eae406f3:0:1")
+	tr := NewTracedRequest(req)
+
+	exporter := tr.exporter.(*InMemoryOtlpClient)
+
+	// add fake spans
+	spans := createResourceSpans([]*tracepb.Span{createOtlpSpan(traceId)})
+	err := exporter.UploadTraces(context.Background(), spans)
+	assert.NoError(t, err)
+
+	// a panic shouldn't occur
+	tr.AddSpans(nil, &zerolog.Logger{})
+}

tunnelrpc/pogs/connectionrpc.go

@@ -175,6 +175,24 @@ func (c RegistrationServer_PogsClient) RegisterConnection(ctx context.Context, a
 	return nil, newRPCError("unknown result which %d", result.Which())
 }
 
+func (c RegistrationServer_PogsClient) SendLocalConfiguration(ctx context.Context, config []byte) error {
+	client := tunnelrpc.TunnelServer{Client: c.Client}
+	promise := client.UpdateLocalConfiguration(ctx, func(p tunnelrpc.RegistrationServer_updateLocalConfiguration_Params) error {
+		if err := p.SetConfig(config); err != nil {
+			return err
+		}
+
+		return nil
+	})
+
+	_, err := promise.Struct()
+	if err != nil {
+		return wrapRPCError(err)
+	}
+
+	return nil
+}
+
 func (c RegistrationServer_PogsClient) UnregisterConnection(ctx context.Context) error {
 	client := tunnelrpc.TunnelServer{Client: c.Client}
 	promise := client.UnregisterConnection(ctx, func(p tunnelrpc.RegistrationServer_unregisterConnection_Params) error {