Release 2024.4.0

Signed-off-by: Steven Kreitzer <skre@skre.me>

.github/workflows/check.yaml

@@ -9,10 +9,10 @@ jobs:
     runs-on: ${{ matrix.os }}
     steps:
     - name: Install Go
-      uses: actions/setup-go@v3
+      uses: actions/setup-go@v5
       with:
         go-version: ${{ matrix.go-version }}
     - name: Checkout code
-      uses: actions/checkout@v3
+      uses: actions/checkout@v4
     - name: Test
       run: make test

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,31 @@
+2024.4.0
+- 2024-04-02 feat: provide short version (#1206)
+- 2024-04-02 Format code
+- 2024-01-18 feat: auto tls sni
+- 2023-12-24 fix checkInPingGroup bugs
+- 2023-12-15 Add environment variables for TCP tunnel hostname / destination / URL.
+
+2024.3.0
+- 2024-03-14 TUN-8281: Run cloudflared query list tunnels/routes endpoint in a paginated way
+- 2024-03-13 TUN-8297: Improve write timeout logging on safe_stream.go
+- 2024-03-07 TUN-8290: Remove `|| true` from postrm.sh
+- 2024-03-05 TUN-8275: Skip write timeout log on "no network activity"
+- 2024-01-23 Update postrm.sh to fix incomplete uninstall
+- 2024-01-05 fix typo in errcheck for response parsing logic in CreateTunnel routine
+- 2023-12-23 Update linux_service.go
+- 2023-12-07 ci: bump actions/checkout to v4
+- 2023-12-07 ci/check: bump actions/setup-go to v5
+- 2023-04-28 check.yaml: bump actions/setup-go to v4
+
+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
 

cfapi/base_client.go

@@ -109,20 +109,34 @@ func (r *RESTClient) sendRequest(method string, url url.URL, body interface{}) (
 	return r.client.Do(req)
 }
 
-func parseResponse(reader io.Reader, data interface{}) error {
+func parseResponseEnvelope(reader io.Reader) (*response, error) {
 	// Schema for Tunnelstore responses in the v1 API.
 	// Roughly, it's a wrapper around a particular result that adds failures/errors/etc
 	var result response
 	// First, parse the wrapper and check the API call succeeded
 	if err := json.NewDecoder(reader).Decode(&result); err != nil {
-		return errors.Wrap(err, "failed to decode response")
+		return nil, errors.Wrap(err, "failed to decode response")
 	}
 	if err := result.checkErrors(); err != nil {
-		return err
+		return nil, err
 	}
 	if !result.Success {
-		return ErrAPINoSuccess
+		return nil, ErrAPINoSuccess
 	}
+
+	return &result, nil
+}
+
+func parseResponse(reader io.Reader, data interface{}) error {
+	result, err := parseResponseEnvelope(reader)
+	if err != nil {
+		return err
+	}
+
+	return parseResponseBody(result, data)
+}
+
+func parseResponseBody(result *response, data interface{}) error {
 	// At this point we know the API call succeeded, so, parse out the inner
 	// result into the datatype provided as a parameter.
 	if err := json.Unmarshal(result.Result, &data); err != nil {
@@ -131,11 +145,58 @@ func parseResponse(reader io.Reader, data interface{}) error {
 	return nil
 }
 
+func fetchExhaustively[T any](requestFn func(int) (*http.Response, error)) ([]*T, error) {
+	page := 0
+	var fullResponse []*T
+
+	for {
+		page += 1
+		envelope, parsedBody, err := fetchPage[T](requestFn, page)
+
+		if err != nil {
+			return nil, errors.Wrap(err, fmt.Sprintf("Error Parsing page %d", page))
+		}
+
+		fullResponse = append(fullResponse, parsedBody...)
+		if envelope.Pagination.Count < envelope.Pagination.PerPage || len(fullResponse) >= envelope.Pagination.TotalCount {
+			break
+		}
+
+	}
+	return fullResponse, nil
+}
+
+func fetchPage[T any](requestFn func(int) (*http.Response, error), page int) (*response, []*T, error) {
+	pageResp, err := requestFn(page)
+	if err != nil {
+		return nil, nil, errors.Wrap(err, "REST request failed")
+	}
+	defer pageResp.Body.Close()
+	if pageResp.StatusCode == http.StatusOK {
+		envelope, err := parseResponseEnvelope(pageResp.Body)
+		if err != nil {
+			return nil, nil, err
+		}
+		var parsedRspBody []*T
+		return envelope, parsedRspBody, parseResponseBody(envelope, &parsedRspBody)
+
+	}
+	return nil, nil, errors.New(fmt.Sprintf("Failed to fetch page. Server returned: %d", pageResp.StatusCode))
+}
+
 type response struct {
-	Success  bool            `json:"success,omitempty"`
-	Errors   []apiErr        `json:"errors,omitempty"`
-	Messages []string        `json:"messages,omitempty"`
-	Result   json.RawMessage `json:"result,omitempty"`
+	Success    bool            `json:"success,omitempty"`
+	Errors     []apiErr        `json:"errors,omitempty"`
+	Messages   []string        `json:"messages,omitempty"`
+	Result     json.RawMessage `json:"result,omitempty"`
+	Pagination Pagination      `json:"result_info,omitempty"`
+}
+
+type Pagination struct {
+	Count      int `json:"count,omitempty"`
+	Page       int `json:"page,omitempty"`
+	PerPage    int `json:"per_page,omitempty"`
+	TotalCount int `json:"total_count,omitempty"`
 }
 
 func (r *response) checkErrors() error {

cfapi/ip_route.go

@@ -137,20 +137,24 @@ type GetRouteByIpParams struct {
 }
 
 // ListRoutes calls the Tunnelstore GET endpoint for all routes under an account.
+// Due to pagination on the server side it will call the endpoint multiple times if needed.
 func (r *RESTClient) ListRoutes(filter *IpRouteFilter) ([]*DetailedRoute, error) {
-	endpoint := r.baseEndpoints.accountRoutes
-	endpoint.RawQuery = filter.Encode()
-	resp, err := r.sendRequest("GET", endpoint, nil)
-	if err != nil {
-		return nil, errors.Wrap(err, "REST request failed")
-	}
-	defer resp.Body.Close()
+	fetchFn := func(page int) (*http.Response, error) {
+		endpoint := r.baseEndpoints.accountRoutes
+		filter.Page(page)
+		endpoint.RawQuery = filter.Encode()
+		rsp, err := r.sendRequest("GET", endpoint, nil)
 
-	if resp.StatusCode == http.StatusOK {
-		return parseListDetailedRoutes(resp.Body)
+		if err != nil {
+			return nil, errors.Wrap(err, "REST request failed")
+		}
+		if rsp.StatusCode != http.StatusOK {
+			rsp.Body.Close()
+			return nil, r.statusCodeToError("list routes", rsp)
+		}
+		return rsp, nil
 	}
-
-	return nil, r.statusCodeToError("list routes", resp)
+	return fetchExhaustively[DetailedRoute](fetchFn)
 }
 
 // AddRoute calls the Tunnelstore POST endpoint for a given route.
@@ -208,12 +212,6 @@ func (r *RESTClient) GetByIP(params GetRouteByIpParams) (DetailedRoute, error) {
 	return DetailedRoute{}, r.statusCodeToError("get route by IP", resp)
 }
 
-func parseListDetailedRoutes(body io.ReadCloser) ([]*DetailedRoute, error) {
-	var routes []*DetailedRoute
-	err := parseResponse(body, &routes)
-	return routes, err
-}
-
 func parseRoute(body io.ReadCloser) (Route, error) {
 	var route Route
 	err := parseResponse(body, &route)

cfapi/ip_route_filter.go

@@ -167,6 +167,10 @@ func (f *IpRouteFilter) MaxFetchSize(max uint) {
 	f.queryParams.Set("per_page", strconv.Itoa(int(max)))
 }
 
+func (f *IpRouteFilter) Page(page int) {
+	f.queryParams.Set("page", strconv.Itoa(page))
+}
+
 func (f IpRouteFilter) Encode() string {
 	return f.queryParams.Encode()
 }

cfapi/tunnel.go

@@ -93,7 +93,7 @@ func (r *RESTClient) CreateTunnel(name string, tunnelSecret []byte) (*TunnelWith
 	switch resp.StatusCode {
 	case http.StatusOK:
 		var tunnel TunnelWithToken
-		if serdeErr := parseResponse(resp.Body, &tunnel); err != nil {
+		if serdeErr := parseResponse(resp.Body, &tunnel); serdeErr != nil {
 			return nil, serdeErr
 		}
 		return &tunnel, nil
@@ -177,25 +177,22 @@ func (r *RESTClient) DeleteTunnel(tunnelID uuid.UUID, cascade bool) error {
 }
 
 func (r *RESTClient) ListTunnels(filter *TunnelFilter) ([]*Tunnel, error) {
-	endpoint := r.baseEndpoints.accountLevel
-	endpoint.RawQuery = filter.encode()
-	resp, err := r.sendRequest("GET", endpoint, nil)
-	if err != nil {
-		return nil, errors.Wrap(err, "REST request failed")
-	}
-	defer resp.Body.Close()
-
-	if resp.StatusCode == http.StatusOK {
-		return parseListTunnels(resp.Body)
+	fetchFn := func(page int) (*http.Response, error) {
+		endpoint := r.baseEndpoints.accountLevel
+		filter.Page(page)
+		endpoint.RawQuery = filter.encode()
+		rsp, err := r.sendRequest("GET", endpoint, nil)
+		if err != nil {
+			return nil, errors.Wrap(err, "REST request failed")
+		}
+		if rsp.StatusCode != http.StatusOK {
+			rsp.Body.Close()
+			return nil, r.statusCodeToError("list tunnels", rsp)
+		}
+		return rsp, nil
 	}
 
-	return nil, r.statusCodeToError("list tunnels", resp)
-}
-
-func parseListTunnels(body io.ReadCloser) ([]*Tunnel, error) {
-	var tunnels []*Tunnel
-	err := parseResponse(body, &tunnels)
-	return tunnels, err
+	return fetchExhaustively[Tunnel](fetchFn)
 }
 
 func (r *RESTClient) ListActiveClients(tunnelID uuid.UUID) ([]*ActiveClient, error) {

cfapi/tunnel_filter.go

@@ -50,6 +50,10 @@ func (f *TunnelFilter) MaxFetchSize(max uint) {
 	f.queryParams.Set("per_page", strconv.Itoa(int(max)))
 }
 
+func (f *TunnelFilter) Page(page int) {
+	f.queryParams.Set("page", strconv.Itoa(page))
+}
+
 func (f TunnelFilter) encode() string {
 	return f.queryParams.Encode()
 }

cfapi/tunnel_test.go

@@ -3,7 +3,6 @@ package cfapi
 import (
 	"bytes"
 	"fmt"
-	"io"
 	"net"
 	"reflect"
 	"strings"
@@ -16,52 +15,6 @@ import (
 
 var loc, _ = time.LoadLocation("UTC")
 
-func Test_parseListTunnels(t *testing.T) {
-	type args struct {
-		body string
-	}
-	tests := []struct {
-		name    string
-		args    args
-		want    []*Tunnel
-		wantErr bool
-	}{
-		{
-			name: "empty list",
-			args: args{body: `{"success": true, "result": []}`},
-			want: []*Tunnel{},
-		},
-		{
-			name:    "success is false",
-			args:    args{body: `{"success": false, "result": []}`},
-			wantErr: true,
-		},
-		{
-			name:    "errors are present",
-			args:    args{body: `{"errors": [{"code": 1003, "message":"An A, AAAA or CNAME record already exists with that host"}], "result": []}`},
-			wantErr: true,
-		},
-		{
-			name:    "invalid response",
-			args:    args{body: `abc`},
-			wantErr: true,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			body := io.NopCloser(bytes.NewReader([]byte(tt.args.body)))
-			got, err := parseListTunnels(body)
-			if (err != nil) != tt.wantErr {
-				t.Errorf("parseListTunnels() error = %v, wantErr %v", err, tt.wantErr)
-				return
-			}
-			if !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("parseListTunnels() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
 func Test_unmarshalTunnel(t *testing.T) {
 	type args struct {
 		body string

cmd/cloudflared/access/cmd.go

@@ -132,15 +132,18 @@ func Commands() []*cli.Command {
 							Name:    sshHostnameFlag,
 							Aliases: []string{"tunnel-host", "T"},
 							Usage:   "specify the hostname of your application.",
+							EnvVars: []string{"TUNNEL_SERVICE_HOSTNAME"},
 						},
 						&cli.StringFlag{
-							Name:  sshDestinationFlag,
-							Usage: "specify the destination address of your SSH server.",
+							Name:    sshDestinationFlag,
+							Usage:   "specify the destination address of your SSH server.",
+							EnvVars: []string{"TUNNEL_SERVICE_DESTINATION"},
 						},
 						&cli.StringFlag{
 							Name:    sshURLFlag,
 							Aliases: []string{"listener", "L"},
 							Usage:   "specify the host:port to forward data to Cloudflare edge.",
+							EnvVars: []string{"TUNNEL_SERVICE_URL"},
 						},
 						&cli.StringSliceFlag{
 							Name:    sshHeaderFlag,

cmd/cloudflared/linux_service.go

@@ -55,7 +55,8 @@ var systemdAllTemplates = map[string]ServiceTemplate{
 		Path: fmt.Sprintf("/etc/systemd/system/%s", cloudflaredService),
 		Content: `[Unit]
 Description=cloudflared
-After=network.target
+After=network-online.target
+Wants=network-online.target
 
 [Service]
 TimeoutStartSec=0
@@ -72,7 +73,8 @@ WantedBy=multi-user.target
 		Path: fmt.Sprintf("/etc/systemd/system/%s", cloudflaredUpdateService),
 		Content: `[Unit]
 Description=Update cloudflared
-After=network.target
+After=network-online.target
+Wants=network-online.target
 
 [Service]
 ExecStart=/bin/bash -c '{{ .Path }} update; code=$?; if [ $code -eq 11 ]; then systemctl restart cloudflared; exit 0; fi; exit $code'

cmd/cloudflared/main.go

@@ -134,11 +134,22 @@ To determine if an update happened in a script, check for error code 11.`,
 		{
 			Name: "version",
 			Action: func(c *cli.Context) (err error) {
+				if c.Bool("short") {
+					fmt.Println(strings.Split(c.App.Version, " ")[0])
+					return nil
+				}
 				version(c)
 				return nil
 			},
 			Usage:       versionText,
 			Description: versionText,
+			Flags: []cli.Flag{
+				&cli.BoolFlag{
+					Name:    "short",
+					Aliases: []string{"s"},
+					Usage:   "print just the version number",
+				},
+			},
 		},
 	}
 	cmds = append(cmds, tunnel.Commands()...)

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)

config/configuration.go

@@ -205,6 +205,8 @@ type OriginRequestConfig struct {
 	HTTPHostHeader *string `yaml:"httpHostHeader" json:"httpHostHeader,omitempty"`
 	// Hostname on the origin server certificate.
 	OriginServerName *string `yaml:"originServerName" json:"originServerName,omitempty"`
+	// Auto configure the Hostname on the origin server certificate.
+	MatchSNIToHost *bool `yaml:"matchSNItoHost" json:"matchSNItoHost,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,omitempty"`

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/config.go

@@ -32,6 +32,7 @@ const (
 	ProxyKeepAliveTimeoutFlag     = "proxy-keepalive-timeout"
 	HTTPHostHeaderFlag            = "http-host-header"
 	OriginServerNameFlag          = "origin-server-name"
+	MatchSNIToHostFlag            = "match-sni-to-host"
 	NoTLSVerifyFlag               = "no-tls-verify"
 	NoChunkedEncodingFlag         = "no-chunked-encoding"
 	ProxyAddressFlag              = "proxy-address"
@@ -118,6 +119,7 @@ func originRequestFromSingleRule(c *cli.Context) OriginRequestConfig {
 	var keepAliveTimeout = defaultKeepAliveTimeout
 	var httpHostHeader string
 	var originServerName string
+	var matchSNItoHost bool
 	var caPool string
 	var noTLSVerify bool
 	var disableChunkedEncoding bool
@@ -150,6 +152,9 @@ func originRequestFromSingleRule(c *cli.Context) OriginRequestConfig {
 	if flag := OriginServerNameFlag; c.IsSet(flag) {
 		originServerName = c.String(flag)
 	}
+	if flag := MatchSNIToHostFlag; c.IsSet(flag) {
+		matchSNItoHost = c.Bool(flag)
+	}
 	if flag := tlsconfig.OriginCAPoolFlag; c.IsSet(flag) {
 		caPool = c.String(flag)
 	}
@@ -185,6 +190,7 @@ func originRequestFromSingleRule(c *cli.Context) OriginRequestConfig {
 		KeepAliveTimeout:       keepAliveTimeout,
 		HTTPHostHeader:         httpHostHeader,
 		OriginServerName:       originServerName,
+		MatchSNIToHost:         matchSNItoHost,
 		CAPool:                 caPool,
 		NoTLSVerify:            noTLSVerify,
 		DisableChunkedEncoding: disableChunkedEncoding,
@@ -229,6 +235,9 @@ func originRequestFromConfig(c config.OriginRequestConfig) OriginRequestConfig {
 	if c.OriginServerName != nil {
 		out.OriginServerName = *c.OriginServerName
 	}
+	if c.MatchSNIToHost != nil {
+		out.MatchSNIToHost = *c.MatchSNIToHost
+	}
 	if c.CAPool != nil {
 		out.CAPool = *c.CAPool
 	}
@@ -287,6 +296,8 @@ type OriginRequestConfig struct {
 	HTTPHostHeader string `yaml:"httpHostHeader" json:"httpHostHeader"`
 	// Hostname on the origin server certificate.
 	OriginServerName string `yaml:"originServerName" json:"originServerName"`
+	// Auto configure the Hostname on the origin server certificate.
+	MatchSNIToHost bool `yaml:"matchSNItoHost" json:"matchSNItoHost"`
 	// 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"`
@@ -362,6 +373,12 @@ func (defaults *OriginRequestConfig) setOriginServerName(overrides config.Origin
 	}
 }
 
+func (defaults *OriginRequestConfig) setMatchSNIToHost(overrides config.OriginRequestConfig) {
+	if val := overrides.MatchSNIToHost; val != nil {
+		defaults.MatchSNIToHost = *val
+	}
+}
+
 func (defaults *OriginRequestConfig) setCAPool(overrides config.OriginRequestConfig) {
 	if val := overrides.CAPool; val != nil {
 		defaults.CAPool = *val
@@ -447,6 +464,7 @@ func setConfig(defaults OriginRequestConfig, overrides config.OriginRequestConfi
 	cfg.setTCPKeepAlive(overrides)
 	cfg.setHTTPHostHeader(overrides)
 	cfg.setOriginServerName(overrides)
+	cfg.setMatchSNIToHost(overrides)
 	cfg.setCAPool(overrides)
 	cfg.setNoTLSVerify(overrides)
 	cfg.setDisableChunkedEncoding(overrides)
@@ -501,6 +519,7 @@ func ConvertToRawOriginConfig(c OriginRequestConfig) config.OriginRequestConfig
 		KeepAliveTimeout:       keepAliveTimeout,
 		HTTPHostHeader:         emptyStringToNil(c.HTTPHostHeader),
 		OriginServerName:       emptyStringToNil(c.OriginServerName),
+		MatchSNIToHost:         defaultBoolToNil(c.MatchSNIToHost),
 		CAPool:                 emptyStringToNil(c.CAPool),
 		NoTLSVerify:            defaultBoolToNil(c.NoTLSVerify),
 		DisableChunkedEncoding: defaultBoolToNil(c.DisableChunkedEncoding),

ingress/constants_test.go

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

ingress/icmp_linux.go

@@ -78,19 +78,19 @@ func checkInPingGroup() error {
 	if err != nil {
 		return err
 	}
-	groupID := os.Getgid()
+	groupID := uint64(os.Getegid())
 	// Example content: 999	   59999
 	found := findGroupIDRegex.FindAll(file, 2)
 	if len(found) == 2 {
-		groupMin, err := strconv.ParseInt(string(found[0]), 10, 32)
+		groupMin, err := strconv.ParseUint(string(found[0]), 10, 32)
 		if err != nil {
 			return errors.Wrapf(err, "failed to determine minimum ping group ID")
 		}
-		groupMax, err := strconv.ParseInt(string(found[1]), 10, 32)
+		groupMax, err := strconv.ParseUint(string(found[1]), 10, 32)
 		if err != nil {
-			return errors.Wrapf(err, "failed to determine minimum ping group ID")
+			return errors.Wrapf(err, "failed to determine maximum ping group ID")
 		}
-		if groupID < int(groupMin) || groupID > int(groupMax) {
+		if groupID < groupMin || groupID > groupMax {
 			return fmt.Errorf("Group ID %d is not between ping group %d to %d", groupID, groupMin, groupMax)
 		}
 		return 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

@@ -2,8 +2,12 @@ package ingress
 
 import (
 	"context"
+	"crypto/tls"
 	"fmt"
+	"net"
 	"net/http"
+
+	"github.com/rs/zerolog"
 )
 
 // HTTPOriginProxy can be implemented by origin services that want to proxy http requests.
@@ -14,7 +18,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.
@@ -46,9 +50,28 @@ func (o *httpService) RoundTrip(req *http.Request) (*http.Response, error) {
 		req.Header.Set("X-Forwarded-Host", req.Host)
 		req.Host = o.hostHeader
 	}
+
+	if o.matchSNIToHost {
+		o.SetOriginServerName(req)
+	}
+
 	return o.transport.RoundTrip(req)
 }
 
+func (o *httpService) SetOriginServerName(req *http.Request) {
+	o.transport.DialTLSContext = func(ctx context.Context, network, addr string) (net.Conn, error) {
+		conn, err := o.transport.DialContext(ctx, network, addr)
+		if err != nil {
+			return nil, err
+		}
+		return tls.Client(conn, &tls.Config{
+			RootCAs:            o.transport.TLSClientConfig.RootCAs,
+			InsecureSkipVerify: o.transport.TLSClientConfig.InsecureSkipVerify,
+			ServerName:         req.Host,
+		}), nil
+	}
+}
+
 func (o *statusCode) RoundTrip(_ *http.Request) (*http.Response, error) {
 	if o.defaultResp {
 		o.log.Warn().Msgf(ErrNoIngressRulesCLI.Error())
@@ -62,19 +85,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 +117,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

@@ -68,9 +68,10 @@ func (o unixSocketPath) MarshalJSON() ([]byte, error) {
 }
 
 type httpService struct {
-	url        *url.URL
-	hostHeader string
-	transport  *http.Transport
+	url            *url.URL
+	hostHeader     string
+	transport      *http.Transport
+	matchSNIToHost bool
 }
 
 func (o *httpService) start(log *zerolog.Logger, _ <-chan struct{}, cfg OriginRequestConfig) error {
@@ -80,6 +81,7 @@ func (o *httpService) start(log *zerolog.Logger, _ <-chan struct{}, cfg OriginRe
 	}
 	o.hostHeader = cfg.HTTPHostHeader
 	o.transport = transport
+	o.matchSNIToHost = cfg.MatchSNIToHost
 	return nil
 }
 
@@ -94,15 +96,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 +289,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}

ingress/rule_test.go

@@ -204,25 +204,25 @@ func TestMarshalJSON(t *testing.T) {
 		{
 			name:     "Nil",
 			path:     nil,
-			expected: `{"hostname":"example.com","path":null,"service":"https://localhost:8000","Handlers":null,"originRequest":{"connectTimeout":30,"tlsTimeout":10,"tcpKeepAlive":30,"noHappyEyeballs":false,"keepAliveTimeout":90,"keepAliveConnections":100,"httpHostHeader":"","originServerName":"","caPool":"","noTLSVerify":false,"disableChunkedEncoding":false,"bastionMode":false,"proxyAddress":"127.0.0.1","proxyPort":0,"proxyType":"","ipRules":null,"http2Origin":false,"access":{"teamName":"","audTag":null}}}`,
+			expected: `{"hostname":"example.com","path":null,"service":"https://localhost:8000","Handlers":null,"originRequest":{"connectTimeout":30,"tlsTimeout":10,"tcpKeepAlive":30,"noHappyEyeballs":false,"keepAliveTimeout":90,"keepAliveConnections":100,"httpHostHeader":"","originServerName":"","matchSNItoHost":false,"caPool":"","noTLSVerify":false,"disableChunkedEncoding":false,"bastionMode":false,"proxyAddress":"127.0.0.1","proxyPort":0,"proxyType":"","ipRules":null,"http2Origin":false,"access":{"teamName":"","audTag":null}}}`,
 			want:     true,
 		},
 		{
 			name:     "Nil regex",
 			path:     &Regexp{Regexp: nil},
-			expected: `{"hostname":"example.com","path":null,"service":"https://localhost:8000","Handlers":null,"originRequest":{"connectTimeout":30,"tlsTimeout":10,"tcpKeepAlive":30,"noHappyEyeballs":false,"keepAliveTimeout":90,"keepAliveConnections":100,"httpHostHeader":"","originServerName":"","caPool":"","noTLSVerify":false,"disableChunkedEncoding":false,"bastionMode":false,"proxyAddress":"127.0.0.1","proxyPort":0,"proxyType":"","ipRules":null,"http2Origin":false,"access":{"teamName":"","audTag":null}}}`,
+			expected: `{"hostname":"example.com","path":null,"service":"https://localhost:8000","Handlers":null,"originRequest":{"connectTimeout":30,"tlsTimeout":10,"tcpKeepAlive":30,"noHappyEyeballs":false,"keepAliveTimeout":90,"keepAliveConnections":100,"httpHostHeader":"","originServerName":"","matchSNItoHost":false,"caPool":"","noTLSVerify":false,"disableChunkedEncoding":false,"bastionMode":false,"proxyAddress":"127.0.0.1","proxyPort":0,"proxyType":"","ipRules":null,"http2Origin":false,"access":{"teamName":"","audTag":null}}}`,
 			want:     true,
 		},
 		{
 			name:     "Empty",
 			path:     &Regexp{Regexp: regexp.MustCompile("")},
-			expected: `{"hostname":"example.com","path":"","service":"https://localhost:8000","Handlers":null,"originRequest":{"connectTimeout":30,"tlsTimeout":10,"tcpKeepAlive":30,"noHappyEyeballs":false,"keepAliveTimeout":90,"keepAliveConnections":100,"httpHostHeader":"","originServerName":"","caPool":"","noTLSVerify":false,"disableChunkedEncoding":false,"bastionMode":false,"proxyAddress":"127.0.0.1","proxyPort":0,"proxyType":"","ipRules":null,"http2Origin":false,"access":{"teamName":"","audTag":null}}}`,
+			expected: `{"hostname":"example.com","path":"","service":"https://localhost:8000","Handlers":null,"originRequest":{"connectTimeout":30,"tlsTimeout":10,"tcpKeepAlive":30,"noHappyEyeballs":false,"keepAliveTimeout":90,"keepAliveConnections":100,"httpHostHeader":"","originServerName":"","matchSNItoHost":false,"caPool":"","noTLSVerify":false,"disableChunkedEncoding":false,"bastionMode":false,"proxyAddress":"127.0.0.1","proxyPort":0,"proxyType":"","ipRules":null,"http2Origin":false,"access":{"teamName":"","audTag":null}}}`,
 			want:     true,
 		},
 		{
 			name:     "Basic",
 			path:     &Regexp{Regexp: regexp.MustCompile("/echo")},
-			expected: `{"hostname":"example.com","path":"/echo","service":"https://localhost:8000","Handlers":null,"originRequest":{"connectTimeout":30,"tlsTimeout":10,"tcpKeepAlive":30,"noHappyEyeballs":false,"keepAliveTimeout":90,"keepAliveConnections":100,"httpHostHeader":"","originServerName":"","caPool":"","noTLSVerify":false,"disableChunkedEncoding":false,"bastionMode":false,"proxyAddress":"127.0.0.1","proxyPort":0,"proxyType":"","ipRules":null,"http2Origin":false,"access":{"teamName":"","audTag":null}}}`,
+			expected: `{"hostname":"example.com","path":"/echo","service":"https://localhost:8000","Handlers":null,"originRequest":{"connectTimeout":30,"tlsTimeout":10,"tcpKeepAlive":30,"noHappyEyeballs":false,"keepAliveTimeout":90,"keepAliveConnections":100,"httpHostHeader":"","originServerName":"","matchSNItoHost":false,"caPool":"","noTLSVerify":false,"disableChunkedEncoding":false,"bastionMode":false,"proxyAddress":"127.0.0.1","proxyPort":0,"proxyType":"","ipRules":null,"http2Origin":false,"access":{"teamName":"","audTag":null}}}`,
 			want:     true,
 		},
 	}

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",

postrm.sh

@@ -1,4 +1,4 @@
 #!/bin/bash
 set -eu
-rm /usr/local/bin/cloudflared
-rm /usr/local/etc/cloudflared/.installedFromPackageManager || true
+rm -f /usr/local/bin/cloudflared
+rm -f /usr/local/etc/cloudflared/.installedFromPackageManager

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,33 @@
 package quic
 
 import (
+	"errors"
+	"net"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/quic-go/quic-go"
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
 )
 
+// The error that is throw by the writer when there is `no network activity`.
+var idleTimeoutError = quic.IdleTimeoutError{}
+
 type SafeStreamCloser struct {
-	lock   sync.Mutex
-	stream quic.Stream
+	lock         sync.Mutex
+	stream       quic.Stream
+	writeTimeout time.Duration
+	log          *zerolog.Logger
+	closing      atomic.Bool
 }
 
-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,10 +38,43 @@ 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.handleWriteError(err)
+	}
+
+	return nBytes, err
+}
+
+// Handles the timeout error in case it happened, by canceling the stream write.
+func (s *SafeStreamCloser) handleWriteError(err error) {
+	// If we are closing the stream we just ignore any write error.
+	if s.closing.Load() {
+		return
+	}
+	var netErr net.Error
+	if errors.As(err, &netErr) {
+		if netErr.Timeout() {
+			// We don't need to log if what cause the timeout was no network activity.
+			if !errors.Is(netErr, &idleTimeoutError) {
+				s.log.Error().Err(netErr).Msg("Closing quic stream due to timeout while writing")
+			}
+			// We need to explicitly cancel the write so that it frees all buffers.
+			s.stream.CancelWrite(0)
+		}
+	}
 }
 
 func (s *SafeStreamCloser) Close() error {
+	// Set this stream to a closing state.
+	s.closing.Store(true)
+
 	// Make sure a possible writer does not block the lock forever. We need it, so we can close the writer
 	// side of the stream safely.
 	_ = s.stream.SetWriteDeadline(time.Now())

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")