Release 2024.2.1

This commit makes the remote diagnostics enabled by default, which is
a useful feature when debugging cloudflared issues without manual intervention from users.
Users can still opt-out by disabling the feature flag.

CHANGES.md

@@ -1,3 +1,7 @@
+## 2024.2.1
+### Notices
+- Starting from this version, tunnel diagnostics will be enabled by default. This will allow the engineering team to remotely get diagnostics from cloudflared during debug activities. Users still have the capability to opt-out of this feature by defining `--management-diagnostics=false` (or env `TUNNEL_MANAGEMENT_DIAGNOSTICS`).
+
 ## 2023.9.0
 ### Notices
 - The `warp-routing` `enabled: boolean` flag is no longer supported in the configuration file. Warp Routing traffic (eg TCP, UDP, ICMP) traffic is proxied to cloudflared if routes to the target tunnel are configured. This change does not affect remotely managed tunnels, but for locally managed tunnels, users that might be relying on this feature flag to block traffic should instead guarantee that tunnel has no Private Routes configured for the tunnel.

RELEASE_NOTES

@@ -1,3 +1,12 @@
+2024.2.1
+- 2024-02-20 TUN-8242: Update Changes.md file with new remote diagnostics behaviour
+- 2024-02-19 TUN-8238: Fix type mismatch introduced by fast-forward
+- 2024-02-16 TUN-8243: Collect metrics on the number of QUIC frames sent/received
+- 2024-02-15 TUN-8238: Refactor proxy logging
+- 2024-02-14 TUN-8242: Enable remote diagnostics by default
+- 2024-02-12 TUN-8236: Add write timeout to quic and tcp connections
+- 2024-02-09 TUN-8224: Fix safety of TCP stream logging, separate connect and ack log messages
+
 2024.2.0
 - 2024-02-07 TUN-8224: Count and collect metrics on stream connect successes/errors
 

cmd/cloudflared/tunnel/cmd.go

@@ -81,6 +81,9 @@ const (
 	// udpUnregisterSessionTimeout is how long we wait before we stop trying to unregister a UDP session from the edge
 	udpUnregisterSessionTimeoutFlag = "udp-unregister-session-timeout"
 
+	// writeStreamTimeout sets if we should have a timeout when writing data to a stream towards the destination (edge/origin).
+	writeStreamTimeout = "write-stream-timeout"
+
 	// quicDisablePathMTUDiscovery sets if QUIC should not perform PTMU discovery and use a smaller (safe) packet size.
 	// Packets will then be at most 1252 (IPv4) / 1232 (IPv6) bytes in size.
 	// Note that this may result in packet drops for UDP proxying, since we expect being able to send at least 1280 bytes of inner packets.
@@ -696,6 +699,13 @@ func tunnelFlags(shouldHide bool) []cli.Flag {
 			Value:  5 * time.Second,
 			Hidden: true,
 		}),
+		altsrc.NewDurationFlag(&cli.DurationFlag{
+			Name:    writeStreamTimeout,
+			EnvVars: []string{"TUNNEL_STREAM_WRITE_TIMEOUT"},
+			Usage:   "Use this option to add a stream write timeout for connections when writing towards the origin or edge. Default is 0 which disables the write timeout.",
+			Value:   0 * time.Second,
+			Hidden:  true,
+		}),
 		altsrc.NewBoolFlag(&cli.BoolFlag{
 			Name:    quicDisablePathMTUDiscovery,
 			EnvVars: []string{"TUNNEL_DISABLE_QUIC_PMTU"},
@@ -780,7 +790,7 @@ func tunnelFlags(shouldHide bool) []cli.Flag {
 			Name:    "management-diagnostics",
 			Usage:   "Enables the in-depth diagnostic routes to be made available over the management service (/debug/pprof, /metrics, etc.)",
 			EnvVars: []string{"TUNNEL_MANAGEMENT_DIAGNOSTICS"},
-			Value:   false,
+			Value:   true,
 		}),
 		selectProtocolFlag,
 		overwriteDNSFlag,

cmd/cloudflared/tunnel/configuration.go

@@ -247,6 +247,7 @@ func prepareTunnelConfig(
 		FeatureSelector:             featureSelector,
 		MaxEdgeAddrRetries:          uint8(c.Int("max-edge-addr-retries")),
 		UDPUnregisterSessionTimeout: c.Duration(udpUnregisterSessionTimeoutFlag),
+		WriteStreamTimeout:          c.Duration(writeStreamTimeout),
 		DisableQUICPathMTUDiscovery: c.Bool(quicDisablePathMTUDiscovery),
 	}
 	packetConfig, err := newPacketConfig(c, log)
@@ -259,6 +260,7 @@ func prepareTunnelConfig(
 		Ingress:            &ingressRules,
 		WarpRouting:        ingress.NewWarpRoutingConfig(&cfg.WarpRouting),
 		ConfigurationFlags: parseConfigFlags(c),
+		WriteTimeout:       c.Duration(writeStreamTimeout),
 	}
 	return tunnelConfig, orchestratorConfig, nil
 }

component-tests/test_management.py

@@ -55,7 +55,7 @@ class TestManagement:
         config = component_tests_config(cfd_mode=CfdModes.NAMED, run_proxy_dns=False, provide_ingress=False)
         LOGGER.debug(config)
         config_path = write_config(tmp_path, config.full_config)
-        with start_cloudflared(tmp_path, config, cfd_pre_args=["tunnel", "--ha-connections", "1", "--management-diagnostics"], new_process=True):
+        with start_cloudflared(tmp_path, config, cfd_pre_args=["tunnel", "--ha-connections", "1"], new_process=True):
             wait_tunnel_ready(require_min_connections=1)
             cfd_cli = CloudflaredCli(config, config_path, LOGGER)
             url = cfd_cli.get_management_url("metrics", config, config_path)
@@ -76,7 +76,7 @@ class TestManagement:
         config = component_tests_config(cfd_mode=CfdModes.NAMED, run_proxy_dns=False, provide_ingress=False)
         LOGGER.debug(config)
         config_path = write_config(tmp_path, config.full_config)
-        with start_cloudflared(tmp_path, config, cfd_pre_args=["tunnel", "--ha-connections", "1", "--management-diagnostics"], new_process=True):
+        with start_cloudflared(tmp_path, config, cfd_pre_args=["tunnel", "--ha-connections", "1"], new_process=True):
             wait_tunnel_ready(require_min_connections=1)
             cfd_cli = CloudflaredCli(config, config_path, LOGGER)
             url = cfd_cli.get_management_url("debug/pprof/heap", config, config_path)
@@ -97,7 +97,7 @@ class TestManagement:
         config = component_tests_config(cfd_mode=CfdModes.NAMED, run_proxy_dns=False, provide_ingress=False)
         LOGGER.debug(config)
         config_path = write_config(tmp_path, config.full_config)
-        with start_cloudflared(tmp_path, config, cfd_pre_args=["tunnel", "--ha-connections", "1"], new_process=True):
+        with start_cloudflared(tmp_path, config, cfd_pre_args=["tunnel", "--ha-connections", "1", "--management-diagnostics=false"], new_process=True):
             wait_tunnel_ready(require_min_connections=1)
             cfd_cli = CloudflaredCli(config, config_path, LOGGER)
             url = cfd_cli.get_management_url("metrics", config, config_path)

connection/quic.go

@@ -66,6 +66,7 @@ type QUICConnection struct {
 	connIndex            uint8
 
 	udpUnregisterTimeout time.Duration
+	streamWriteTimeout   time.Duration
 }
 
 // NewQUICConnection returns a new instance of QUICConnection.
@@ -82,6 +83,7 @@ func NewQUICConnection(
 	logger *zerolog.Logger,
 	packetRouterConfig *ingress.GlobalRouterConfig,
 	udpUnregisterTimeout time.Duration,
+	streamWriteTimeout time.Duration,
 ) (*QUICConnection, error) {
 	udpConn, err := createUDPConnForConnIndex(connIndex, localAddr, logger)
 	if err != nil {
@@ -117,6 +119,7 @@ func NewQUICConnection(
 		connOptions:          connOptions,
 		connIndex:            connIndex,
 		udpUnregisterTimeout: udpUnregisterTimeout,
+		streamWriteTimeout:   streamWriteTimeout,
 	}, nil
 }
 
@@ -195,7 +198,7 @@ func (q *QUICConnection) acceptStream(ctx context.Context) error {
 
 func (q *QUICConnection) runStream(quicStream quic.Stream) {
 	ctx := quicStream.Context()
-	stream := quicpogs.NewSafeStreamCloser(quicStream)
+	stream := quicpogs.NewSafeStreamCloser(quicStream, q.streamWriteTimeout, q.logger)
 	defer stream.Close()
 
 	// we are going to fuse readers/writers from stream <- cloudflared -> origin, and we want to guarantee that
@@ -373,7 +376,7 @@ func (q *QUICConnection) closeUDPSession(ctx context.Context, sessionID uuid.UUI
 		return
 	}
 
-	stream := quicpogs.NewSafeStreamCloser(quicStream)
+	stream := quicpogs.NewSafeStreamCloser(quicStream, q.streamWriteTimeout, q.logger)
 	defer stream.Close()
 	rpcClientStream, err := quicpogs.NewRPCClientStream(ctx, stream, q.udpUnregisterTimeout, q.logger)
 	if err != nil {

connection/quic_test.go

@@ -35,6 +35,7 @@ var (
 		KeepAlivePeriod: 5 * time.Second,
 		EnableDatagrams: true,
 	}
+	defaultQUICTimeout = 30 * time.Second
 )
 
 var _ ReadWriteAcker = (*streamReadWriteAcker)(nil)
@@ -197,7 +198,7 @@ func quicServer(
 
 	quicStream, err := session.OpenStreamSync(context.Background())
 	require.NoError(t, err)
-	stream := quicpogs.NewSafeStreamCloser(quicStream)
+	stream := quicpogs.NewSafeStreamCloser(quicStream, defaultQUICTimeout, &log)
 
 	reqClientStream := quicpogs.RequestClientStream{ReadWriteCloser: stream}
 	err = reqClientStream.WriteConnectRequestData(dest, connectionType, metadata...)
@@ -726,6 +727,7 @@ func testQUICConnection(udpListenerAddr net.Addr, t *testing.T, index uint8) *QU
 		&log,
 		nil,
 		5*time.Second,
+		0*time.Second,
 	)
 	require.NoError(t, err)
 	return qc

ingress/constants_test.go

@@ -0,0 +1,7 @@
+package ingress
+
+import "github.com/cloudflare/cloudflared/logger"
+
+var (
+	TestLogger = logger.Create(nil)
+)

ingress/origin_connection.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"io"
 	"net"
+	"time"
 
 	"github.com/rs/zerolog"
 
@@ -31,15 +32,32 @@ func DefaultStreamHandler(originConn io.ReadWriter, remoteConn net.Conn, log *ze
 
 // tcpConnection is an OriginConnection that directly streams to raw TCP.
 type tcpConnection struct {
-	conn net.Conn
+	net.Conn
+	writeTimeout time.Duration
+	logger       *zerolog.Logger
 }
 
-func (tc *tcpConnection) Stream(ctx context.Context, tunnelConn io.ReadWriter, log *zerolog.Logger) {
-	stream.Pipe(tunnelConn, tc.conn, log)
+func (tc *tcpConnection) Stream(_ context.Context, tunnelConn io.ReadWriter, _ *zerolog.Logger) {
+	stream.Pipe(tunnelConn, tc, tc.logger)
+}
+
+func (tc *tcpConnection) Write(b []byte) (int, error) {
+	if tc.writeTimeout > 0 {
+		if err := tc.Conn.SetWriteDeadline(time.Now().Add(tc.writeTimeout)); err != nil {
+			tc.logger.Err(err).Msg("Error setting write deadline for TCP connection")
+		}
+	}
+
+	nBytes, err := tc.Conn.Write(b)
+	if err != nil {
+		tc.logger.Err(err).Msg("Error writing to the TCP connection")
+	}
+
+	return nBytes, err
 }
 
 func (tc *tcpConnection) Close() {
-	tc.conn.Close()
+	tc.Conn.Close()
 }
 
 // tcpOverWSConnection is an OriginConnection that streams to TCP over WS.

ingress/origin_connection_test.go

@@ -19,7 +19,6 @@ import (
 	"golang.org/x/net/proxy"
 	"golang.org/x/sync/errgroup"
 
-	"github.com/cloudflare/cloudflared/logger"
 	"github.com/cloudflare/cloudflared/socks"
 	"github.com/cloudflare/cloudflared/stream"
 	"github.com/cloudflare/cloudflared/websocket"
@@ -31,7 +30,6 @@ const (
 )
 
 var (
-	testLogger   = logger.Create(nil)
 	testMessage  = []byte("TestStreamOriginConnection")
 	testResponse = []byte(fmt.Sprintf("echo-%s", testMessage))
 )
@@ -39,7 +37,8 @@ var (
 func TestStreamTCPConnection(t *testing.T) {
 	cfdConn, originConn := net.Pipe()
 	tcpConn := tcpConnection{
-		conn: cfdConn,
+		Conn:         cfdConn,
+		writeTimeout: 30 * time.Second,
 	}
 
 	eyeballConn, edgeConn := net.Pipe()
@@ -66,7 +65,7 @@ func TestStreamTCPConnection(t *testing.T) {
 		return nil
 	})
 
-	tcpConn.Stream(ctx, edgeConn, testLogger)
+	tcpConn.Stream(ctx, edgeConn, TestLogger)
 	require.NoError(t, errGroup.Wait())
 }
 
@@ -93,7 +92,7 @@ func TestDefaultStreamWSOverTCPConnection(t *testing.T) {
 		return nil
 	})
 
-	tcpOverWSConn.Stream(ctx, edgeConn, testLogger)
+	tcpOverWSConn.Stream(ctx, edgeConn, TestLogger)
 	require.NoError(t, errGroup.Wait())
 }
 
@@ -147,7 +146,7 @@ func TestSocksStreamWSOverTCPConnection(t *testing.T) {
 
 		errGroup, ctx := errgroup.WithContext(ctx)
 		errGroup.Go(func() error {
-			tcpOverWSConn.Stream(ctx, edgeConn, testLogger)
+			tcpOverWSConn.Stream(ctx, edgeConn, TestLogger)
 			return nil
 		})
 
@@ -159,7 +158,7 @@ func TestSocksStreamWSOverTCPConnection(t *testing.T) {
 			require.NoError(t, err)
 			defer wsForwarderInConn.Close()
 
-			stream.Pipe(wsForwarderInConn, &wsEyeball{wsForwarderOutConn}, testLogger)
+			stream.Pipe(wsForwarderInConn, &wsEyeball{wsForwarderOutConn}, TestLogger)
 			return nil
 		})
 
@@ -209,7 +208,7 @@ func TestWsConnReturnsBeforeStreamReturns(t *testing.T) {
 			originConn.Close()
 		}()
 		ctx := context.WithValue(r.Context(), websocket.PingPeriodContextKey, time.Microsecond)
-		tcpOverWSConn.Stream(ctx, eyeballConn, testLogger)
+		tcpOverWSConn.Stream(ctx, eyeballConn, TestLogger)
 	})
 	server := httptest.NewServer(handler)
 	defer server.Close()

ingress/origin_proxy.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+
+	"github.com/rs/zerolog"
 )
 
 // HTTPOriginProxy can be implemented by origin services that want to proxy http requests.
@@ -14,7 +16,7 @@ type HTTPOriginProxy interface {
 
 // StreamBasedOriginProxy can be implemented by origin services that want to proxy ws/TCP.
 type StreamBasedOriginProxy interface {
-	EstablishConnection(ctx context.Context, dest string) (OriginConnection, error)
+	EstablishConnection(ctx context.Context, dest string, log *zerolog.Logger) (OriginConnection, error)
 }
 
 // HTTPLocalProxy can be implemented by cloudflared services that want to handle incoming http requests.
@@ -62,19 +64,21 @@ func (o *statusCode) RoundTrip(_ *http.Request) (*http.Response, error) {
 	return resp, nil
 }
 
-func (o *rawTCPService) EstablishConnection(ctx context.Context, dest string) (OriginConnection, error) {
+func (o *rawTCPService) EstablishConnection(ctx context.Context, dest string, logger *zerolog.Logger) (OriginConnection, error) {
 	conn, err := o.dialer.DialContext(ctx, "tcp", dest)
 	if err != nil {
 		return nil, err
 	}
 
 	originConn := &tcpConnection{
-		conn: conn,
+		Conn:         conn,
+		writeTimeout: o.writeTimeout,
+		logger:       logger,
 	}
 	return originConn, nil
 }
 
-func (o *tcpOverWSService) EstablishConnection(ctx context.Context, dest string) (OriginConnection, error) {
+func (o *tcpOverWSService) EstablishConnection(ctx context.Context, dest string, _ *zerolog.Logger) (OriginConnection, error) {
 	var err error
 	if !o.isBastion {
 		dest = o.dest
@@ -92,6 +96,6 @@ func (o *tcpOverWSService) EstablishConnection(ctx context.Context, dest string)
 
 }
 
-func (o *socksProxyOverWSService) EstablishConnection(_ctx context.Context, _dest string) (OriginConnection, error) {
+func (o *socksProxyOverWSService) EstablishConnection(_ context.Context, _ string, _ *zerolog.Logger) (OriginConnection, error) {
 	return o.conn, nil
 }

ingress/origin_proxy_test.go

@@ -36,7 +36,7 @@ func TestRawTCPServiceEstablishConnection(t *testing.T) {
 	require.NoError(t, err)
 
 	// Origin not listening for new connection, should return an error
-	_, err = rawTCPService.EstablishConnection(context.Background(), req.URL.String())
+	_, err = rawTCPService.EstablishConnection(context.Background(), req.URL.String(), TestLogger)
 	require.Error(t, err)
 }
 
@@ -87,7 +87,7 @@ func TestTCPOverWSServiceEstablishConnection(t *testing.T) {
 		t.Run(test.testCase, func(t *testing.T) {
 			if test.expectErr {
 				bastionHost, _ := carrier.ResolveBastionDest(test.req)
-				_, err := test.service.EstablishConnection(context.Background(), bastionHost)
+				_, err := test.service.EstablishConnection(context.Background(), bastionHost, TestLogger)
 				assert.Error(t, err)
 			}
 		})
@@ -99,7 +99,7 @@ func TestTCPOverWSServiceEstablishConnection(t *testing.T) {
 	for _, service := range []*tcpOverWSService{newTCPOverWSService(originURL), newBastionService()} {
 		// Origin not listening for new connection, should return an error
 		bastionHost, _ := carrier.ResolveBastionDest(bastionReq)
-		_, err := service.EstablishConnection(context.Background(), bastionHost)
+		_, err := service.EstablishConnection(context.Background(), bastionHost, TestLogger)
 		assert.Error(t, err)
 	}
 }
@@ -132,7 +132,7 @@ func TestHTTPServiceHostHeaderOverride(t *testing.T) {
 		url: originURL,
 	}
 	shutdownC := make(chan struct{})
-	require.NoError(t, httpService.start(testLogger, shutdownC, cfg))
+	require.NoError(t, httpService.start(TestLogger, shutdownC, cfg))
 
 	req, err := http.NewRequest(http.MethodGet, originURL.String(), nil)
 	require.NoError(t, err)
@@ -167,7 +167,7 @@ func TestHTTPServiceUsesIngressRuleScheme(t *testing.T) {
 		url: originURL,
 	}
 	shutdownC := make(chan struct{})
-	require.NoError(t, httpService.start(testLogger, shutdownC, cfg))
+	require.NoError(t, httpService.start(TestLogger, shutdownC, cfg))
 
 	// Tunnel uses scheme defined in the service field of the ingress rule, independent of the X-Forwarded-Proto header
 	protos := []string{"https", "http", "dne"}

ingress/origin_service.go

@@ -94,15 +94,17 @@ func (o httpService) MarshalJSON() ([]byte, error) {
 // rawTCPService dials TCP to the destination specified by the client
 // It's used by warp routing
 type rawTCPService struct {
-	name   string
-	dialer net.Dialer
+	name         string
+	dialer       net.Dialer
+	writeTimeout time.Duration
+	logger       *zerolog.Logger
 }
 
 func (o *rawTCPService) String() string {
 	return o.name
 }
 
-func (o *rawTCPService) start(log *zerolog.Logger, _ <-chan struct{}, cfg OriginRequestConfig) error {
+func (o *rawTCPService) start(_ *zerolog.Logger, _ <-chan struct{}, _ OriginRequestConfig) error {
 	return nil
 }
 
@@ -285,13 +287,14 @@ type WarpRoutingService struct {
 	Proxy StreamBasedOriginProxy
 }
 
-func NewWarpRoutingService(config WarpRoutingConfig) *WarpRoutingService {
+func NewWarpRoutingService(config WarpRoutingConfig, writeTimeout time.Duration) *WarpRoutingService {
 	svc := &rawTCPService{
 		name: ServiceWarpRouting,
 		dialer: net.Dialer{
 			Timeout:   config.ConnectTimeout.Duration,
 			KeepAlive: config.TCPKeepAlive.Duration,
 		},
+		writeTimeout: writeTimeout,
 	}
 
 	return &WarpRoutingService{Proxy: svc}

orchestration/config.go

@@ -2,6 +2,7 @@ package orchestration
 
 import (
 	"encoding/json"
+	"time"
 
 	"github.com/cloudflare/cloudflared/config"
 	"github.com/cloudflare/cloudflared/ingress"
@@ -19,8 +20,9 @@ type newLocalConfig struct {
 
 // Config is the original config as read and parsed by cloudflared.
 type Config struct {
-	Ingress     *ingress.Ingress
-	WarpRouting ingress.WarpRoutingConfig
+	Ingress      *ingress.Ingress
+	WarpRouting  ingress.WarpRoutingConfig
+	WriteTimeout time.Duration
 
 	// Extra settings used to configure this instance but that are not eligible for remotely management
 	// ie. (--protocol, --loglevel, ...)

orchestration/orchestrator.go

@@ -17,10 +17,10 @@ import (
 	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
 )
 
-// Orchestrator manages configurations so they can be updatable during runtime
+// Orchestrator manages configurations, so they can be updatable during runtime
 // properties are static, so it can be read without lock
 // currentVersion and config are read/write infrequently, so their access are synchronized with RWMutex
-// access to proxy is synchronized with atmoic.Value, because it uses copy-on-write to provide scalable frequently
+// access to proxy is synchronized with atomic.Value, because it uses copy-on-write to provide scalable frequently
 // read when update is infrequent
 type Orchestrator struct {
 	currentVersion int32
@@ -30,9 +30,10 @@ type Orchestrator struct {
 	proxy atomic.Value
 	// Set of internal ingress rules defined at cloudflared startup (separate from user-defined ingress rules)
 	internalRules []ingress.Rule
-	config        *Config
-	tags          []tunnelpogs.Tag
-	log           *zerolog.Logger
+	// cloudflared Configuration
+	config *Config
+	tags   []tunnelpogs.Tag
+	log    *zerolog.Logger
 
 	// orchestrator must not handle any more updates after shutdownC is closed
 	shutdownC <-chan struct{}
@@ -40,7 +41,11 @@ type Orchestrator struct {
 	proxyShutdownC chan<- struct{}
 }
 
-func NewOrchestrator(ctx context.Context, config *Config, tags []tunnelpogs.Tag, internalRules []ingress.Rule, log *zerolog.Logger) (*Orchestrator, error) {
+func NewOrchestrator(ctx context.Context,
+	config *Config,
+	tags []tunnelpogs.Tag,
+	internalRules []ingress.Rule,
+	log *zerolog.Logger) (*Orchestrator, error) {
 	o := &Orchestrator{
 		// Lowest possible version, any remote configuration will have version higher than this
 		// Starting at -1 allows a configuration migration (local to remote) to override the current configuration as it
@@ -131,7 +136,7 @@ func (o *Orchestrator) updateIngress(ingressRules ingress.Ingress, warpRouting i
 	if err := ingressRules.StartOrigins(o.log, proxyShutdownC); err != nil {
 		return errors.Wrap(err, "failed to start origin")
 	}
-	proxy := proxy.NewOriginProxy(ingressRules, warpRouting, o.tags, o.log)
+	proxy := proxy.NewOriginProxy(ingressRules, warpRouting, o.tags, o.config.WriteTimeout, o.log)
 	o.proxy.Store(proxy)
 	o.config.Ingress = &ingressRules
 	o.config.WarpRouting = warpRouting

orchestration/orchestrator_test.go

@@ -27,7 +27,7 @@ import (
 )
 
 var (
-	testLogger = zerolog.Logger{}
+	testLogger = zerolog.Nop()
 	testTags   = []tunnelpogs.Tag{
 		{
 			Name:  "package",

proxy/logger.go

@@ -0,0 +1,78 @@
+package proxy
+
+import (
+	"net/http"
+	"strconv"
+
+	"github.com/rs/zerolog"
+
+	"github.com/cloudflare/cloudflared/connection"
+	"github.com/cloudflare/cloudflared/ingress"
+	"github.com/cloudflare/cloudflared/management"
+)
+
+const (
+	logFieldCFRay         = "cfRay"
+	logFieldLBProbe       = "lbProbe"
+	logFieldRule          = "ingressRule"
+	logFieldOriginService = "originService"
+	logFieldFlowID        = "flowID"
+	logFieldConnIndex     = "connIndex"
+	logFieldDestAddr      = "destAddr"
+)
+
+// newHTTPLogger creates a child zerolog.Logger from the provided with added context from the HTTP request, ingress
+// services, and connection index.
+func newHTTPLogger(logger *zerolog.Logger, connIndex uint8, req *http.Request, rule int, serviceName string) zerolog.Logger {
+	ctx := logger.With().
+		Int(management.EventTypeKey, int(management.HTTP)).
+		Uint8(logFieldConnIndex, connIndex)
+	cfRay := connection.FindCfRayHeader(req)
+	lbProbe := connection.IsLBProbeRequest(req)
+	if cfRay != "" {
+		ctx.Str(logFieldCFRay, cfRay)
+	}
+	if lbProbe {
+		ctx.Bool(logFieldLBProbe, lbProbe)
+	}
+	return ctx.
+		Str(logFieldOriginService, serviceName).
+		Interface(logFieldRule, rule).
+		Logger()
+}
+
+// newTCPLogger creates a child zerolog.Logger from the provided with added context from the TCPRequest.
+func newTCPLogger(logger *zerolog.Logger, req *connection.TCPRequest) zerolog.Logger {
+	return logger.With().
+		Int(management.EventTypeKey, int(management.TCP)).
+		Uint8(logFieldConnIndex, req.ConnIndex).
+		Str(logFieldOriginService, ingress.ServiceWarpRouting).
+		Str(logFieldFlowID, req.FlowID).
+		Str(logFieldDestAddr, req.Dest).
+		Uint8(logFieldConnIndex, req.ConnIndex).
+		Logger()
+}
+
+// logHTTPRequest logs a Debug message with the corresponding HTTP request details from the eyeball.
+func logHTTPRequest(logger *zerolog.Logger, r *http.Request) {
+	logger.Debug().
+		Str("host", r.Host).
+		Str("path", r.URL.Path).
+		Interface("headers", r.Header).
+		Int64("content-length", r.ContentLength).
+		Msgf("%s %s %s", r.Method, r.URL, r.Proto)
+}
+
+// logOriginHTTPResponse logs a Debug message of the origin response.
+func logOriginHTTPResponse(logger *zerolog.Logger, resp *http.Response) {
+	responseByCode.WithLabelValues(strconv.Itoa(resp.StatusCode)).Inc()
+	logger.Debug().
+		Int64("content-length", resp.ContentLength).
+		Msgf("%s", resp.Status)
+}
+
+// logRequestError logs an error for the proxied request.
+func logRequestError(logger *zerolog.Logger, err error) {
+	requestErrors.Inc()
+	logger.Error().Err(err).Send()
+}

proxy/proxy.go

@@ -17,7 +17,6 @@ import (
 	"github.com/cloudflare/cloudflared/cfio"
 	"github.com/cloudflare/cloudflared/connection"
 	"github.com/cloudflare/cloudflared/ingress"
-	"github.com/cloudflare/cloudflared/management"
 	"github.com/cloudflare/cloudflared/stream"
 	"github.com/cloudflare/cloudflared/tracing"
 	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
@@ -25,16 +24,8 @@ import (
 
 const (
 	// TagHeaderNamePrefix indicates a Cloudflared Warp Tag prefix that gets appended for warp traffic stream headers.
-	TagHeaderNamePrefix   = "Cf-Warp-Tag-"
-	LogFieldCFRay         = "cfRay"
-	LogFieldLBProbe       = "lbProbe"
-	LogFieldRule          = "ingressRule"
-	LogFieldOriginService = "originService"
-	LogFieldFlowID        = "flowID"
-	LogFieldConnIndex     = "connIndex"
-	LogFieldDestAddr      = "destAddr"
-
-	trailerHeaderName = "Trailer"
+	TagHeaderNamePrefix = "Cf-Warp-Tag-"
+	trailerHeaderName   = "Trailer"
 )
 
 // Proxy represents a means to Proxy between cloudflared and the origin services.
@@ -51,6 +42,7 @@ func NewOriginProxy(
 	ingressRules ingress.Ingress,
 	warpRouting ingress.WarpRoutingConfig,
 	tags []tunnelpogs.Tag,
+	writeTimeout time.Duration,
 	log *zerolog.Logger,
 ) *Proxy {
 	proxy := &Proxy{
@@ -59,7 +51,7 @@ func NewOriginProxy(
 		log:          log,
 	}
 
-	proxy.warpRouting = ingress.NewWarpRoutingService(warpRouting)
+	proxy.warpRouting = ingress.NewWarpRoutingService(warpRouting, writeTimeout)
 
 	return proxy
 }
@@ -90,26 +82,18 @@ func (p *Proxy) ProxyHTTP(
 	defer decrementConcurrentRequests()
 
 	req := tr.Request
-	cfRay := connection.FindCfRayHeader(req)
-	lbProbe := connection.IsLBProbeRequest(req)
 	p.appendTagHeaders(req)
 
 	_, ruleSpan := tr.Tracer().Start(req.Context(), "ingress_match",
 		trace.WithAttributes(attribute.String("req-host", req.Host)))
 	rule, ruleNum := p.ingressRules.FindMatchingRule(req.Host, req.URL.Path)
-	logFields := logFields{
-		cfRay:     cfRay,
-		lbProbe:   lbProbe,
-		rule:      ruleNum,
-		connIndex: tr.ConnIndex,
-	}
-	p.logRequest(req, logFields)
 	ruleSpan.SetAttributes(attribute.Int("rule-num", ruleNum))
 	ruleSpan.End()
+	logger := newHTTPLogger(p.log, tr.ConnIndex, req, ruleNum, rule.Service.String())
+	logHTTPRequest(&logger, req)
 	if err, applied := p.applyIngressMiddleware(rule, req, w); err != nil {
 		if applied {
-			rule, srv := ruleField(p.ingressRules, ruleNum)
-			p.logRequestError(err, cfRay, "", rule, srv)
+			logRequestError(&logger, err)
 			return nil
 		}
 		return err
@@ -123,10 +107,9 @@ func (p *Proxy) ProxyHTTP(
 			originProxy,
 			isWebsocket,
 			rule.Config.DisableChunkedEncoding,
-			logFields,
+			&logger,
 		); err != nil {
-			rule, srv := ruleField(p.ingressRules, ruleNum)
-			p.logRequestError(err, cfRay, "", rule, srv)
+			logRequestError(&logger, err)
 			return err
 		}
 		return nil
@@ -140,13 +123,9 @@ func (p *Proxy) ProxyHTTP(
 			return fmt.Errorf("response writer is not a flusher")
 		}
 		rws := connection.NewHTTPResponseReadWriterAcker(w, flusher, req)
-		connectedLogger := p.log.Debug().
-			Int(management.EventTypeKey, int(management.HTTP)).
-			Str(LogFieldDestAddr, dest).
-			Uint8(LogFieldConnIndex, tr.ConnIndex)
-		if err := p.proxyStream(tr.ToTracedContext(), rws, dest, originProxy, connectedLogger); err != nil {
-			rule, srv := ruleField(p.ingressRules, ruleNum)
-			p.logRequestError(err, cfRay, "", rule, srv)
+		logger := logger.With().Str(logFieldDestAddr, dest).Logger()
+		if err := p.proxyStream(tr.ToTracedContext(), rws, dest, originProxy, &logger); err != nil {
+			logRequestError(&logger, err)
 			return err
 		}
 		return nil
@@ -176,36 +155,20 @@ func (p *Proxy) ProxyTCP(
 	serveCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-	tracedCtx := tracing.NewTracedContext(serveCtx, req.CfTraceID, p.log)
+	logger := newTCPLogger(p.log, req)
+	tracedCtx := tracing.NewTracedContext(serveCtx, req.CfTraceID, &logger)
+	logger.Debug().Msg("tcp proxy stream started")
 
-	debugLogger := func() *zerolog.Event {
-		return p.log.Debug().
-			Int(management.EventTypeKey, int(management.TCP)).
-			Str(LogFieldFlowID, req.FlowID).
-			Str(LogFieldDestAddr, req.Dest).
-			Uint8(LogFieldConnIndex, req.ConnIndex)
-	}
-
-	debugLogger().Msg("tcp proxy stream started")
-
-	if err := p.proxyStream(tracedCtx, rwa, req.Dest, p.warpRouting.Proxy, debugLogger()); err != nil {
-		p.logRequestError(err, req.CFRay, req.FlowID, "", ingress.ServiceWarpRouting)
+	if err := p.proxyStream(tracedCtx, rwa, req.Dest, p.warpRouting.Proxy, &logger); err != nil {
+		logRequestError(&logger, err)
 		return err
 	}
 
-	debugLogger().Msg("tcp proxy stream finished successfully")
+	logger.Debug().Msg("tcp proxy stream finished successfully")
 
 	return nil
 }
 
-func ruleField(ing ingress.Ingress, ruleNum int) (ruleID string, srv string) {
-	srv = ing.Rules[ruleNum].Service.String()
-	if ing.IsSingleRule() {
-		return "", srv
-	}
-	return fmt.Sprintf("%d", ruleNum), srv
-}
-
 // ProxyHTTPRequest proxies requests of underlying type http and websocket to the origin service.
 func (p *Proxy) proxyHTTPRequest(
 	w connection.ResponseWriter,
@@ -213,7 +176,7 @@ func (p *Proxy) proxyHTTPRequest(
 	httpService ingress.HTTPOriginProxy,
 	isWebsocket bool,
 	disableChunkedEncoding bool,
-	fields logFields,
+	logger *zerolog.Logger,
 ) error {
 	roundTripReq := tr.Request
 	if isWebsocket {
@@ -280,7 +243,7 @@ func (p *Proxy) proxyHTTPRequest(
 			reader: tr.Request.Body,
 		}
 
-		stream.Pipe(eyeballStream, rwc, p.log)
+		stream.Pipe(eyeballStream, rwc, logger)
 		return nil
 	}
 
@@ -291,7 +254,7 @@ func (p *Proxy) proxyHTTPRequest(
 	// copy trailers
 	copyTrailers(w, resp)
 
-	p.logOriginResponse(resp, fields)
+	logOriginHTTPResponse(logger, resp)
 	return nil
 }
 
@@ -303,13 +266,13 @@ func (p *Proxy) proxyStream(
 	rwa connection.ReadWriteAcker,
 	dest string,
 	connectionProxy ingress.StreamBasedOriginProxy,
-	connectedLogger *zerolog.Event,
+	logger *zerolog.Logger,
 ) error {
 	ctx := tr.Context
 	_, connectSpan := tr.Tracer().Start(ctx, "stream-connect")
 
 	start := time.Now()
-	originConn, err := connectionProxy.EstablishConnection(ctx, dest)
+	originConn, err := connectionProxy.EstablishConnection(ctx, dest, logger)
 	if err != nil {
 		connectStreamErrors.Inc()
 		tracing.EndWithErrorStatus(connectSpan, err)
@@ -317,6 +280,7 @@ func (p *Proxy) proxyStream(
 	}
 	connectSpan.End()
 	defer originConn.Close()
+	logger.Debug().Msg("origin connection established")
 
 	encodedSpans := tr.GetSpans()
 
@@ -326,9 +290,9 @@ func (p *Proxy) proxyStream(
 	}
 
 	connectLatency.Observe(float64(time.Since(start).Milliseconds()))
-	connectedLogger.Msg("proxy stream established")
+	logger.Debug().Msg("proxy stream acknowledged")
 
-	originConn.Stream(ctx, rwa, p.log)
+	originConn.Stream(ctx, rwa, logger)
 	return nil
 }
 
@@ -362,14 +326,6 @@ func (p *Proxy) appendTagHeaders(r *http.Request) {
 	}
 }
 
-type logFields struct {
-	cfRay     string
-	lbProbe   bool
-	rule      int
-	flowID    string
-	connIndex uint8
-}
-
 func copyTrailers(w connection.ResponseWriter, response *http.Response) {
 	for trailerHeader, trailerValues := range response.Trailer {
 		for _, trailerValue := range trailerValues {
@@ -378,64 +334,6 @@ func copyTrailers(w connection.ResponseWriter, response *http.Response) {
 	}
 }
 
-func (p *Proxy) logRequest(r *http.Request, fields logFields) {
-	log := p.log.With().Int(management.EventTypeKey, int(management.HTTP)).Logger()
-	event := log.Debug()
-	if fields.cfRay != "" {
-		event = event.Str(LogFieldCFRay, fields.cfRay)
-	}
-	if fields.lbProbe {
-		event = event.Bool(LogFieldLBProbe, fields.lbProbe)
-	}
-	if fields.cfRay == "" && !fields.lbProbe {
-		log.Debug().Msgf("All requests should have a CF-RAY header. Please open a support ticket with Cloudflare. %s %s %s ", r.Method, r.URL, r.Proto)
-	}
-	event.
-		Uint8(LogFieldConnIndex, fields.connIndex).
-		Str("host", r.Host).
-		Str("path", r.URL.Path).
-		Interface(LogFieldRule, fields.rule).
-		Interface("headers", r.Header).
-		Int64("content-length", r.ContentLength).
-		Msgf("%s %s %s", r.Method, r.URL, r.Proto)
-}
-
-func (p *Proxy) logOriginResponse(resp *http.Response, fields logFields) {
-	responseByCode.WithLabelValues(strconv.Itoa(resp.StatusCode)).Inc()
-	event := p.log.Debug()
-	if fields.cfRay != "" {
-		event = event.Str(LogFieldCFRay, fields.cfRay)
-	}
-	if fields.lbProbe {
-		event = event.Bool(LogFieldLBProbe, fields.lbProbe)
-	}
-	event.
-		Int(management.EventTypeKey, int(management.HTTP)).
-		Uint8(LogFieldConnIndex, fields.connIndex).
-		Int64("content-length", resp.ContentLength).
-		Msgf("%s", resp.Status)
-}
-
-func (p *Proxy) logRequestError(err error, cfRay string, flowID string, rule, service string) {
-	requestErrors.Inc()
-	log := p.log.Error().Err(err)
-	if cfRay != "" {
-		log = log.Str(LogFieldCFRay, cfRay)
-	}
-	if flowID != "" {
-		log = log.Str(LogFieldFlowID, flowID).Int(management.EventTypeKey, int(management.TCP))
-	} else {
-		log = log.Int(management.EventTypeKey, int(management.HTTP))
-	}
-	if rule != "" {
-		log = log.Str(LogFieldRule, rule)
-	}
-	if service != "" {
-		log = log.Str(LogFieldOriginService, service)
-	}
-	log.Send()
-}
-
 func getDestFromRule(rule *ingress.Rule, req *http.Request) (string, error) {
 	switch rule.Service.String() {
 	case ingress.ServiceBastion:

proxy/proxy_test.go

@@ -162,7 +162,7 @@ func TestProxySingleOrigin(t *testing.T) {
 
 	require.NoError(t, ingressRule.StartOrigins(&log, ctx.Done()))
 
-	proxy := NewOriginProxy(ingressRule, noWarpRouting, testTags, &log)
+	proxy := NewOriginProxy(ingressRule, noWarpRouting, testTags, time.Duration(0), &log)
 	t.Run("testProxyHTTP", testProxyHTTP(proxy))
 	t.Run("testProxyWebsocket", testProxyWebsocket(proxy))
 	t.Run("testProxySSE", testProxySSE(proxy))
@@ -366,7 +366,7 @@ func runIngressTestScenarios(t *testing.T, unvalidatedIngress []config.Unvalidat
 	ctx, cancel := context.WithCancel(context.Background())
 	require.NoError(t, ingress.StartOrigins(&log, ctx.Done()))
 
-	proxy := NewOriginProxy(ingress, noWarpRouting, testTags, &log)
+	proxy := NewOriginProxy(ingress, noWarpRouting, testTags, time.Duration(0), &log)
 
 	for _, test := range tests {
 		responseWriter := newMockHTTPRespWriter()
@@ -414,7 +414,7 @@ func TestProxyError(t *testing.T) {
 
 	log := zerolog.Nop()
 
-	proxy := NewOriginProxy(ing, noWarpRouting, testTags, &log)
+	proxy := NewOriginProxy(ing, noWarpRouting, testTags, time.Duration(0), &log)
 
 	responseWriter := newMockHTTPRespWriter()
 	req, err := http.NewRequest(http.MethodGet, "http://127.0.0.1", nil)
@@ -530,7 +530,7 @@ func TestConnections(t *testing.T) {
 				originService:         runEchoTCPService,
 				eyeballResponseWriter: newTCPRespWriter(replayer),
 				eyeballRequestBody:    newTCPRequestBody([]byte("test2")),
-				warpRoutingService:    ingress.NewWarpRoutingService(testWarpRouting),
+				warpRoutingService:    ingress.NewWarpRoutingService(testWarpRouting, time.Duration(0)),
 				connectionType:        connection.TypeTCP,
 				requestHeaders: map[string][]string{
 					"Cf-Cloudflared-Proxy-Src": {"non-blank-value"},
@@ -548,7 +548,7 @@ func TestConnections(t *testing.T) {
 				originService:        runEchoWSService,
 				// eyeballResponseWriter gets set after roundtrip dial.
 				eyeballRequestBody: newPipedWSRequestBody([]byte("test3")),
-				warpRoutingService: ingress.NewWarpRoutingService(testWarpRouting),
+				warpRoutingService: ingress.NewWarpRoutingService(testWarpRouting, time.Duration(0)),
 				requestHeaders: map[string][]string{
 					"Cf-Cloudflared-Proxy-Src": {"non-blank-value"},
 				},
@@ -675,7 +675,7 @@ func TestConnections(t *testing.T) {
 
 			ingressRule := createSingleIngressConfig(t, test.args.ingressServiceScheme+ln.Addr().String())
 			ingressRule.StartOrigins(logger, ctx.Done())
-			proxy := NewOriginProxy(ingressRule, testWarpRouting, testTags, logger)
+			proxy := NewOriginProxy(ingressRule, testWarpRouting, testTags, time.Duration(0), logger)
 			proxy.warpRouting = test.args.warpRoutingService
 
 			dest := ln.Addr().String()

quic/conversion.go

@@ -7,17 +7,6 @@ import (
 	"github.com/quic-go/quic-go/logging"
 )
 
-func perspectiveString(p logging.Perspective) string {
-	switch p {
-	case logging.PerspectiveClient:
-		return "client"
-	case logging.PerspectiveServer:
-		return "server"
-	default:
-		return ""
-	}
-}
-
 // Helper to convert logging.ByteCount(alias for int64) to float64 used in prometheus
 func byteCountToPromCount(count logging.ByteCount) float64 {
 	return float64(count)

quic/metrics.go

@@ -1,6 +1,8 @@
 package quic
 
 import (
+	"reflect"
+	"strings"
 	"sync"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -16,10 +18,10 @@ var (
 	clientMetrics    = struct {
 		totalConnections  prometheus.Counter
 		closedConnections prometheus.Counter
-		sentPackets       *prometheus.CounterVec
+		sentFrames        *prometheus.CounterVec
 		sentBytes         *prometheus.CounterVec
-		receivePackets    *prometheus.CounterVec
-		receiveBytes      *prometheus.CounterVec
+		receivedFrames    *prometheus.CounterVec
+		receivedBytes     *prometheus.CounterVec
 		bufferedPackets   *prometheus.CounterVec
 		droppedPackets    *prometheus.CounterVec
 		lostPackets       *prometheus.CounterVec
@@ -28,43 +30,88 @@ var (
 		smoothedRTT       *prometheus.GaugeVec
 	}{
 		totalConnections: prometheus.NewCounter(
-			totalConnectionsOpts(logging.PerspectiveClient),
+			prometheus.CounterOpts{
+				Namespace: namespace,
+				Subsystem: "client",
+				Name:      "total_connections",
+				Help:      "Number of connections initiated",
+			},
 		),
 		closedConnections: prometheus.NewCounter(
-			closedConnectionsOpts(logging.PerspectiveClient),
+			prometheus.CounterOpts{
+				Namespace: namespace,
+				Subsystem: "client",
+				Name:      "closed_connections",
+				Help:      "Number of connections that has been closed",
+			},
 		),
-		sentPackets: prometheus.NewCounterVec(
-			sentPacketsOpts(logging.PerspectiveClient),
-			clientConnLabels,
+		sentFrames: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Namespace: namespace,
+				Subsystem: "client",
+				Name:      "sent_frames",
+				Help:      "Number of frames that have been sent through a connection",
+			},
+			append(clientConnLabels, "frame_type"),
 		),
 		sentBytes: prometheus.NewCounterVec(
-			sentBytesOpts(logging.PerspectiveClient),
+			prometheus.CounterOpts{
+				Namespace: namespace,
+				Subsystem: "client",
+				Name:      "sent_bytes",
+				Help:      "Number of bytes that have been sent through a connection",
+			},
 			clientConnLabels,
 		),
-		receivePackets: prometheus.NewCounterVec(
-			receivePacketsOpts(logging.PerspectiveClient),
-			clientConnLabels,
+		receivedFrames: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Namespace: namespace,
+				Subsystem: "client",
+				Name:      "received_frames",
+				Help:      "Number of frames that have been received through a connection",
+			},
+			append(clientConnLabels, "frame_type"),
 		),
-		receiveBytes: prometheus.NewCounterVec(
-			receiveBytesOpts(logging.PerspectiveClient),
+		receivedBytes: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Namespace: namespace,
+				Subsystem: "client",
+				Name:      "receive_bytes",
+				Help:      "Number of bytes that have been received through a connection",
+			},
 			clientConnLabels,
 		),
 		bufferedPackets: prometheus.NewCounterVec(
-			bufferedPacketsOpts(logging.PerspectiveClient),
+			prometheus.CounterOpts{
+				Namespace: namespace,
+				Subsystem: "client",
+				Name:      "buffered_packets",
+				Help:      "Number of bytes that have been buffered on a connection",
+			},
 			append(clientConnLabels, "packet_type"),
 		),
 		droppedPackets: prometheus.NewCounterVec(
-			droppedPacketsOpts(logging.PerspectiveClient),
+			prometheus.CounterOpts{
+				Namespace: namespace,
+				Subsystem: "client",
+				Name:      "dropped_packets",
+				Help:      "Number of bytes that have been dropped on a connection",
+			},
 			append(clientConnLabels, "packet_type", "reason"),
 		),
 		lostPackets: prometheus.NewCounterVec(
-			lostPacketsOpts(logging.PerspectiveClient),
+			prometheus.CounterOpts{
+				Namespace: namespace,
+				Subsystem: "client",
+				Name:      "lost_packets",
+				Help:      "Number of packets that have been lost from a connection",
+			},
 			append(clientConnLabels, "reason"),
 		),
 		minRTT: prometheus.NewGaugeVec(
 			prometheus.GaugeOpts{
 				Namespace: namespace,
-				Subsystem: perspectiveString(logging.PerspectiveClient),
+				Subsystem: "client",
 				Name:      "min_rtt",
 				Help:      "Lowest RTT measured on a connection in millisec",
 			},
@@ -73,7 +120,7 @@ var (
 		latestRTT: prometheus.NewGaugeVec(
 			prometheus.GaugeOpts{
 				Namespace: namespace,
-				Subsystem: perspectiveString(logging.PerspectiveClient),
+				Subsystem: "client",
 				Name:      "latest_rtt",
 				Help:      "Latest RTT measured on a connection",
 			},
@@ -82,188 +129,37 @@ var (
 		smoothedRTT: prometheus.NewGaugeVec(
 			prometheus.GaugeOpts{
 				Namespace: namespace,
-				Subsystem: perspectiveString(logging.PerspectiveClient),
+				Subsystem: "client",
 				Name:      "smoothed_rtt",
 				Help:      "Calculated smoothed RTT measured on a connection in millisec",
 			},
 			clientConnLabels,
 		),
 	}
-	// The server has many QUIC connections. Adding per connection label incurs high memory cost
-	serverMetrics = struct {
-		totalConnections  prometheus.Counter
-		closedConnections prometheus.Counter
-		sentPackets       prometheus.Counter
-		sentBytes         prometheus.Counter
-		receivePackets    prometheus.Counter
-		receiveBytes      prometheus.Counter
-		bufferedPackets   *prometheus.CounterVec
-		droppedPackets    *prometheus.CounterVec
-		lostPackets       *prometheus.CounterVec
-		rtt               prometheus.Histogram
-	}{
-		totalConnections: prometheus.NewCounter(
-			totalConnectionsOpts(logging.PerspectiveServer),
-		),
-		closedConnections: prometheus.NewCounter(
-			closedConnectionsOpts(logging.PerspectiveServer),
-		),
-		sentPackets: prometheus.NewCounter(
-			sentPacketsOpts(logging.PerspectiveServer),
-		),
-		sentBytes: prometheus.NewCounter(
-			sentBytesOpts(logging.PerspectiveServer),
-		),
-		receivePackets: prometheus.NewCounter(
-			receivePacketsOpts(logging.PerspectiveServer),
-		),
-		receiveBytes: prometheus.NewCounter(
-			receiveBytesOpts(logging.PerspectiveServer),
-		),
-		bufferedPackets: prometheus.NewCounterVec(
-			bufferedPacketsOpts(logging.PerspectiveServer),
-			[]string{"packet_type"},
-		),
-		droppedPackets: prometheus.NewCounterVec(
-			droppedPacketsOpts(logging.PerspectiveServer),
-			[]string{"packet_type", "reason"},
-		),
-		lostPackets: prometheus.NewCounterVec(
-			lostPacketsOpts(logging.PerspectiveServer),
-			[]string{"reason"},
-		),
-		rtt: prometheus.NewHistogram(
-			prometheus.HistogramOpts{
-				Namespace: namespace,
-				Subsystem: perspectiveString(logging.PerspectiveServer),
-				Name:      "rtt",
-				Buckets:   []float64{5, 10, 20, 30, 40, 50, 75, 100},
-			},
-		),
-	}
+
 	registerClient = sync.Once{}
-	registerServer = sync.Once{}
 
 	packetTooBigDropped = prometheus.NewCounter(prometheus.CounterOpts{
 		Namespace: namespace,
-		Subsystem: perspectiveString(logging.PerspectiveClient),
+		Subsystem: "client",
 		Name:      "packet_too_big_dropped",
 		Help:      "Count of packets received from origin that are too big to send to the edge and are dropped as a result",
 	})
 )
 
-// MetricsCollector abstracts the difference between client and server metrics from connTracer
-type MetricsCollector interface {
-	startedConnection()
-	closedConnection(err error)
-	sentPackets(logging.ByteCount)
-	receivedPackets(logging.ByteCount)
-	bufferedPackets(logging.PacketType)
-	droppedPackets(logging.PacketType, logging.ByteCount, logging.PacketDropReason)
-	lostPackets(logging.PacketLossReason)
-	updatedRTT(*logging.RTTStats)
-}
-
-func totalConnectionsOpts(p logging.Perspective) prometheus.CounterOpts {
-	var help string
-	if p == logging.PerspectiveClient {
-		help = "Number of connections initiated. For all quic metrics, client means the side initiating the connection"
-	} else {
-		help = "Number of connections accepted. For all quic metrics, server means the side accepting connections"
-	}
-	return prometheus.CounterOpts{
-		Namespace: namespace,
-		Subsystem: perspectiveString(p),
-		Name:      "total_connections",
-		Help:      help,
-	}
-}
-
-func closedConnectionsOpts(p logging.Perspective) prometheus.CounterOpts {
-	return prometheus.CounterOpts{
-		Namespace: namespace,
-		Subsystem: perspectiveString(p),
-		Name:      "closed_connections",
-		Help:      "Number of connections that has been closed",
-	}
-}
-
-func sentPacketsOpts(p logging.Perspective) prometheus.CounterOpts {
-	return prometheus.CounterOpts{
-		Namespace: namespace,
-		Subsystem: perspectiveString(p),
-		Name:      "sent_packets",
-		Help:      "Number of packets that have been sent through a connection",
-	}
-}
-
-func sentBytesOpts(p logging.Perspective) prometheus.CounterOpts {
-	return prometheus.CounterOpts{
-		Namespace: namespace,
-		Subsystem: perspectiveString(p),
-		Name:      "sent_bytes",
-		Help:      "Number of bytes that have been sent through a connection",
-	}
-}
-
-func receivePacketsOpts(p logging.Perspective) prometheus.CounterOpts {
-	return prometheus.CounterOpts{
-		Namespace: namespace,
-		Subsystem: perspectiveString(p),
-		Name:      "receive_packets",
-		Help:      "Number of packets that have been received through a connection",
-	}
-}
-
-func receiveBytesOpts(p logging.Perspective) prometheus.CounterOpts {
-	return prometheus.CounterOpts{
-		Namespace: namespace,
-		Subsystem: perspectiveString(p),
-		Name:      "receive_bytes",
-		Help:      "Number of bytes that have been received through a connection",
-	}
-}
-
-func bufferedPacketsOpts(p logging.Perspective) prometheus.CounterOpts {
-	return prometheus.CounterOpts{
-		Namespace: namespace,
-		Subsystem: perspectiveString(p),
-		Name:      "buffered_packets",
-		Help:      "Number of bytes that have been buffered on a connection",
-	}
-}
-
-func droppedPacketsOpts(p logging.Perspective) prometheus.CounterOpts {
-	return prometheus.CounterOpts{
-		Namespace: namespace,
-		Subsystem: perspectiveString(p),
-		Name:      "dropped_packets",
-		Help:      "Number of bytes that have been dropped on a connection",
-	}
-}
-
-func lostPacketsOpts(p logging.Perspective) prometheus.CounterOpts {
-	return prometheus.CounterOpts{
-		Namespace: namespace,
-		Subsystem: perspectiveString(p),
-		Name:      "lost_packets",
-		Help:      "Number of packets that have been lost from a connection",
-	}
-}
-
 type clientCollector struct {
 	index string
 }
 
-func newClientCollector(index uint8) MetricsCollector {
+func newClientCollector(index uint8) *clientCollector {
 	registerClient.Do(func() {
 		prometheus.MustRegister(
 			clientMetrics.totalConnections,
 			clientMetrics.closedConnections,
-			clientMetrics.sentPackets,
+			clientMetrics.sentFrames,
 			clientMetrics.sentBytes,
-			clientMetrics.receivePackets,
-			clientMetrics.receiveBytes,
+			clientMetrics.receivedFrames,
+			clientMetrics.receivedBytes,
 			clientMetrics.bufferedPackets,
 			clientMetrics.droppedPackets,
 			clientMetrics.lostPackets,
@@ -286,14 +182,12 @@ func (cc *clientCollector) closedConnection(err error) {
 	clientMetrics.closedConnections.Inc()
 }
 
-func (cc *clientCollector) sentPackets(size logging.ByteCount) {
-	clientMetrics.sentPackets.WithLabelValues(cc.index).Inc()
-	clientMetrics.sentBytes.WithLabelValues(cc.index).Add(byteCountToPromCount(size))
+func (cc *clientCollector) sentPackets(size logging.ByteCount, frames []logging.Frame) {
+	cc.collectPackets(size, frames, clientMetrics.sentFrames, clientMetrics.sentBytes)
 }
 
-func (cc *clientCollector) receivedPackets(size logging.ByteCount) {
-	clientMetrics.receivePackets.WithLabelValues(cc.index).Inc()
-	clientMetrics.receiveBytes.WithLabelValues(cc.index).Add(byteCountToPromCount(size))
+func (cc *clientCollector) receivedPackets(size logging.ByteCount, frames []logging.Frame) {
+	cc.collectPackets(size, frames, clientMetrics.receivedFrames, clientMetrics.receivedBytes)
 }
 
 func (cc *clientCollector) bufferedPackets(packetType logging.PacketType) {
@@ -318,63 +212,18 @@ func (cc *clientCollector) updatedRTT(rtt *logging.RTTStats) {
 	clientMetrics.smoothedRTT.WithLabelValues(cc.index).Set(durationToPromGauge(rtt.SmoothedRTT()))
 }
 
-type serverCollector struct{}
-
-func newServiceCollector() MetricsCollector {
-	registerServer.Do(func() {
-		prometheus.MustRegister(
-			serverMetrics.totalConnections,
-			serverMetrics.closedConnections,
-			serverMetrics.sentPackets,
-			serverMetrics.sentBytes,
-			serverMetrics.receivePackets,
-			serverMetrics.receiveBytes,
-			serverMetrics.bufferedPackets,
-			serverMetrics.droppedPackets,
-			serverMetrics.lostPackets,
-			serverMetrics.rtt,
-		)
-	})
-	return &serverCollector{}
+func (cc *clientCollector) collectPackets(size logging.ByteCount, frames []logging.Frame, counter, bandwidth *prometheus.CounterVec) {
+	for _, frame := range frames {
+		counter.WithLabelValues(cc.index, frameName(frame)).Inc()
+	}
+	bandwidth.WithLabelValues(cc.index).Add(byteCountToPromCount(size))
 }
 
-func (sc *serverCollector) startedConnection() {
-	serverMetrics.totalConnections.Inc()
-}
-
-func (sc *serverCollector) closedConnection(err error) {
-	serverMetrics.closedConnections.Inc()
-}
-
-func (sc *serverCollector) sentPackets(size logging.ByteCount) {
-	serverMetrics.sentPackets.Inc()
-	serverMetrics.sentBytes.Add(byteCountToPromCount(size))
-}
-
-func (sc *serverCollector) receivedPackets(size logging.ByteCount) {
-	serverMetrics.receivePackets.Inc()
-	serverMetrics.receiveBytes.Add(byteCountToPromCount(size))
-}
-
-func (sc *serverCollector) bufferedPackets(packetType logging.PacketType) {
-	serverMetrics.bufferedPackets.WithLabelValues(packetTypeString(packetType)).Inc()
-}
-
-func (sc *serverCollector) droppedPackets(packetType logging.PacketType, size logging.ByteCount, reason logging.PacketDropReason) {
-	serverMetrics.droppedPackets.WithLabelValues(
-		packetTypeString(packetType),
-		packetDropReasonString(reason),
-	).Add(byteCountToPromCount(size))
-}
-
-func (sc *serverCollector) lostPackets(reason logging.PacketLossReason) {
-	serverMetrics.lostPackets.WithLabelValues(packetLossReasonString(reason)).Inc()
-}
-
-func (sc *serverCollector) updatedRTT(rtt *logging.RTTStats) {
-	latestRTT := rtt.LatestRTT()
-	// May return 0 if no valid updates have occurred
-	if latestRTT > 0 {
-		serverMetrics.rtt.Observe(durationToPromGauge(latestRTT))
+func frameName(frame logging.Frame) string {
+	if frame == nil {
+		return "nil"
+	} else {
+		name := reflect.TypeOf(frame).Elem().Name()
+		return strings.TrimSuffix(name, "Frame")
 	}
 }

quic/safe_stream.go

@@ -1,20 +1,28 @@
 package quic
 
 import (
+	"errors"
+	"net"
 	"sync"
 	"time"
 
 	"github.com/quic-go/quic-go"
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
 )
 
 type SafeStreamCloser struct {
-	lock   sync.Mutex
-	stream quic.Stream
+	lock         sync.Mutex
+	stream       quic.Stream
+	writeTimeout time.Duration
+	log          *zerolog.Logger
 }
 
-func NewSafeStreamCloser(stream quic.Stream) *SafeStreamCloser {
+func NewSafeStreamCloser(stream quic.Stream, writeTimeout time.Duration, log *zerolog.Logger) *SafeStreamCloser {
 	return &SafeStreamCloser{
-		stream: stream,
+		stream:       stream,
+		writeTimeout: writeTimeout,
+		log:          log,
 	}
 }
 
@@ -25,7 +33,29 @@ func (s *SafeStreamCloser) Read(p []byte) (n int, err error) {
 func (s *SafeStreamCloser) Write(p []byte) (n int, err error) {
 	s.lock.Lock()
 	defer s.lock.Unlock()
-	return s.stream.Write(p)
+	if s.writeTimeout > 0 {
+		err = s.stream.SetWriteDeadline(time.Now().Add(s.writeTimeout))
+		if err != nil {
+			log.Err(err).Msg("Error setting write deadline for QUIC stream")
+		}
+	}
+	nBytes, err := s.stream.Write(p)
+	if err != nil {
+		s.handleTimeout(err)
+	}
+
+	return nBytes, err
+}
+
+// Handles the timeout error in case it happened, by canceling the stream write.
+func (s *SafeStreamCloser) handleTimeout(err error) {
+	var netErr net.Error
+	if errors.As(err, &netErr) {
+		if netErr.Timeout() {
+			s.log.Error().Err(netErr).Msg("Closing quic stream due to timeout while writing")
+			s.stream.CancelWrite(0)
+		}
+	}
 }
 
 func (s *SafeStreamCloser) Close() error {

quic/safe_stream_test.go

@@ -9,6 +9,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/rs/zerolog"
+
 	"github.com/quic-go/quic-go"
 	"github.com/stretchr/testify/require"
 )
@@ -70,7 +72,8 @@ func quicClient(t *testing.T, addr net.Addr) {
 		go func(iter int) {
 			defer wg.Done()
 
-			stream := NewSafeStreamCloser(quicStream)
+			log := zerolog.Nop()
+			stream := NewSafeStreamCloser(quicStream, 30*time.Second, &log)
 			defer stream.Close()
 
 			// Do a bunch of round trips over this stream that should work.
@@ -107,7 +110,8 @@ func quicServer(t *testing.T, serverReady *sync.WaitGroup, conn net.PacketConn)
 		go func(iter int) {
 			defer wg.Done()
 
-			stream := NewSafeStreamCloser(quicStream)
+			log := zerolog.Nop()
+			stream := NewSafeStreamCloser(quicStream, 30*time.Second, &log)
 			defer stream.Close()
 
 			// Do a bunch of round trips over this stream that should work.

quic/tracing.go

@@ -3,7 +3,6 @@ package quic
 import (
 	"context"
 	"net"
-	"time"
 
 	"github.com/quic-go/quic-go/logging"
 	"github.com/rs/zerolog"
@@ -16,8 +15,6 @@ type tracer struct {
 }
 
 type tracerConfig struct {
-	isClient bool
-	// Only client has an index
 	index uint8
 }
 
@@ -25,67 +22,36 @@ func NewClientTracer(logger *zerolog.Logger, index uint8) func(context.Context,
 	t := &tracer{
 		logger: logger,
 		config: &tracerConfig{
-			isClient: true,
-			index:    index,
+			index: index,
 		},
 	}
 	return t.TracerForConnection
 }
 
-func NewServerTracer(logger *zerolog.Logger) *logging.Tracer {
-	return &logging.Tracer{
-		SentPacket:                   func(net.Addr, *logging.Header, logging.ByteCount, []logging.Frame) {},
-		SentVersionNegotiationPacket: func(_ net.Addr, dest, src logging.ArbitraryLenConnectionID, _ []logging.VersionNumber) {},
-		DroppedPacket:                func(net.Addr, logging.PacketType, logging.ByteCount, logging.PacketDropReason) {},
-	}
-}
-
 func (t *tracer) TracerForConnection(_ctx context.Context, _p logging.Perspective, _odcid logging.ConnectionID) *logging.ConnectionTracer {
-	if t.config.isClient {
-		return newConnTracer(newClientCollector(t.config.index))
-	}
-	return newConnTracer(newServiceCollector())
+	return newConnTracer(newClientCollector(t.config.index))
 }
 
 // connTracer collects connection level metrics
 type connTracer struct {
-	metricsCollector MetricsCollector
+	metricsCollector *clientCollector
 }
 
-func newConnTracer(metricsCollector MetricsCollector) *logging.ConnectionTracer {
+func newConnTracer(metricsCollector *clientCollector) *logging.ConnectionTracer {
 	tracer := connTracer{
 		metricsCollector: metricsCollector,
 	}
 	return &logging.ConnectionTracer{
-		StartedConnection:                tracer.StartedConnection,
-		NegotiatedVersion:                tracer.NegotiatedVersion,
-		ClosedConnection:                 tracer.ClosedConnection,
-		SentTransportParameters:          tracer.SentTransportParameters,
-		ReceivedTransportParameters:      tracer.ReceivedTransportParameters,
-		RestoredTransportParameters:      tracer.RestoredTransportParameters,
-		SentLongHeaderPacket:             tracer.SentLongHeaderPacket,
-		SentShortHeaderPacket:            tracer.SentShortHeaderPacket,
-		ReceivedVersionNegotiationPacket: tracer.ReceivedVersionNegotiationPacket,
-		ReceivedRetry:                    tracer.ReceivedRetry,
-		ReceivedLongHeaderPacket:         tracer.ReceivedLongHeaderPacket,
-		ReceivedShortHeaderPacket:        tracer.ReceivedShortHeaderPacket,
-		BufferedPacket:                   tracer.BufferedPacket,
-		DroppedPacket:                    tracer.DroppedPacket,
-		UpdatedMetrics:                   tracer.UpdatedMetrics,
-		AcknowledgedPacket:               tracer.AcknowledgedPacket,
-		LostPacket:                       tracer.LostPacket,
-		UpdatedCongestionState:           tracer.UpdatedCongestionState,
-		UpdatedPTOCount:                  tracer.UpdatedPTOCount,
-		UpdatedKeyFromTLS:                tracer.UpdatedKeyFromTLS,
-		UpdatedKey:                       tracer.UpdatedKey,
-		DroppedEncryptionLevel:           tracer.DroppedEncryptionLevel,
-		DroppedKey:                       tracer.DroppedKey,
-		SetLossTimer:                     tracer.SetLossTimer,
-		LossTimerExpired:                 tracer.LossTimerExpired,
-		LossTimerCanceled:                tracer.LossTimerCanceled,
-		ECNStateUpdated:                  tracer.ECNStateUpdated,
-		Close:                            tracer.Close,
-		Debug:                            tracer.Debug,
+		StartedConnection:         tracer.StartedConnection,
+		ClosedConnection:          tracer.ClosedConnection,
+		SentLongHeaderPacket:      tracer.SentLongHeaderPacket,
+		SentShortHeaderPacket:     tracer.SentShortHeaderPacket,
+		ReceivedLongHeaderPacket:  tracer.ReceivedLongHeaderPacket,
+		ReceivedShortHeaderPacket: tracer.ReceivedShortHeaderPacket,
+		BufferedPacket:            tracer.BufferedPacket,
+		DroppedPacket:             tracer.DroppedPacket,
+		UpdatedMetrics:            tracer.UpdatedMetrics,
+		LostPacket:                tracer.LostPacket,
 	}
 }
 
@@ -97,14 +63,6 @@ func (ct *connTracer) ClosedConnection(err error) {
 	ct.metricsCollector.closedConnection(err)
 }
 
-func (ct *connTracer) SentPacket(hdr *logging.ExtendedHeader, packetSize logging.ByteCount, ack *logging.AckFrame, frames []logging.Frame) {
-	ct.metricsCollector.sentPackets(packetSize)
-}
-
-func (ct *connTracer) ReceivedPacket(hdr *logging.ExtendedHeader, size logging.ByteCount, frames []logging.Frame) {
-	ct.metricsCollector.receivedPackets(size)
-}
-
 func (ct *connTracer) BufferedPacket(pt logging.PacketType, size logging.ByteCount) {
 	ct.metricsCollector.bufferedPackets(pt)
 }
@@ -121,74 +79,20 @@ func (ct *connTracer) UpdatedMetrics(rttStats *logging.RTTStats, cwnd, bytesInFl
 	ct.metricsCollector.updatedRTT(rttStats)
 }
 
-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) SentLongHeaderPacket(hdr *logging.ExtendedHeader, size logging.ByteCount, ecn logging.ECN, ack *logging.AckFrame, frames []logging.Frame) {
+	ct.metricsCollector.sentPackets(size, frames)
 }
 
 func (ct *connTracer) SentShortHeaderPacket(hdr *logging.ShortHeader, size logging.ByteCount, ecn logging.ECN, ack *logging.AckFrame, frames []logging.Frame) {
-}
-
-func (ct *connTracer) ReceivedVersionNegotiationPacket(dest, src logging.ArbitraryLenConnectionID, _ []logging.VersionNumber) {
-}
-
-func (ct *connTracer) ReceivedRetry(header *logging.Header) {
+	ct.metricsCollector.sentPackets(size, frames)
 }
 
 func (ct *connTracer) ReceivedLongHeaderPacket(hdr *logging.ExtendedHeader, size logging.ByteCount, ecn logging.ECN, frames []logging.Frame) {
+	ct.metricsCollector.receivedPackets(size, frames)
 }
 
 func (ct *connTracer) ReceivedShortHeaderPacket(hdr *logging.ShortHeader, size logging.ByteCount, ecn logging.ECN, frames []logging.Frame) {
-}
-
-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) ECNStateUpdated(state logging.ECNState, trigger logging.ECNStateTrigger) {
-
-}
-
-func (ct *connTracer) Close() {
-}
-
-func (ct *connTracer) Debug(name, msg string) {
+	ct.metricsCollector.receivedPackets(size, frames)
 }
 
 type quicLogger struct {

supervisor/tunnel.go

@@ -66,6 +66,7 @@ type TunnelConfig struct {
 	PacketConfig     *ingress.GlobalRouterConfig
 
 	UDPUnregisterSessionTimeout time.Duration
+	WriteStreamTimeout          time.Duration
 
 	DisableQUICPathMTUDiscovery bool
 
@@ -614,6 +615,7 @@ func (e *EdgeTunnelServer) serveQUIC(
 		connLogger.Logger(),
 		e.config.PacketConfig,
 		e.config.UDPUnregisterSessionTimeout,
+		e.config.WriteStreamTimeout,
 	)
 	if err != nil {
 		connLogger.ConnAwareLogger().Err(err).Msgf("Failed to create new quic connection")