Release 2022.10.3

cloudflared's Makefile uses `shell go env GOOS` to determine the
LOCAL_OS regardless of it being provided. We therefore need pinned_go as
a dependency to run docker-generate-versions.

RELEASE_NOTES

@@ -1,3 +1,12 @@
+2022.10.3
+- 2022-10-24 TUN-6871: Add default feature to cloudflared to support EOF on QUIC connections
+- 2022-10-19 TUN-6876: Fix flaky TestTraceICMPRouterEcho by taking account request span can return before reply
+- 2022-10-18 TUN-6867: Clear spans right after they are serialized to avoid returning duplicate spans
+2022.10.2
+- 2022-10-18 TUN-6869: Fix Makefile complaining about missing GO packages
+- 2022-10-18 TUN-6864: Don't reuse port in quic unit tests
+- 2022-10-18 TUN-6868: Return left padded tracing ID when tracing identity is converted to string
+
 2022.10.1
 - 2022-10-16 TUN-6861: Trace ICMP on Windows
 - 2022-10-15 TUN-6860: Send access configuration keys to the edge

cfsetup.yaml

@@ -81,6 +81,7 @@ stretch: &stretch
   generate-versions-file:
     build_dir: *build_dir
     builddeps: 
+      - *pinned_go
       - build-essential
     post-cache:
       - make generate-docker-version

cmd/cloudflared/tunnel/configuration.go

@@ -46,7 +46,7 @@ var (
 	LogFieldHostname = "hostname"
 
 	secretFlags     = [2]*altsrc.StringFlag{credentialsContentsFlag, tunnelTokenFlag}
-	defaultFeatures = []string{supervisor.FeatureAllowRemoteConfig, supervisor.FeatureSerializedHeaders, supervisor.FeatureDatagramV2}
+	defaultFeatures = []string{supervisor.FeatureAllowRemoteConfig, supervisor.FeatureSerializedHeaders, supervisor.FeatureDatagramV2, supervisor.FeatureQUICSupportEOF}
 
 	configFlags = []string{"autoupdate-freq", "no-autoupdate", "retries", "protocol", "loglevel", "transport-loglevel", "origincert", "metrics", "metrics-update-freq", "edge-ip-version"}
 )

connection/quic_test.go

@@ -11,7 +11,6 @@ import (
 	"net/url"
 	"os"
 	"strings"
-	"sync"
 	"testing"
 	"time"
 
@@ -142,24 +141,33 @@ func TestQUICServer(t *testing.T) {
 		},
 	}
 
-	for _, test := range tests {
+	for i, test := range tests {
+		test := test // capture range variable
 		t.Run(test.desc, func(t *testing.T) {
 			ctx, cancel := context.WithCancel(context.Background())
-			var wg sync.WaitGroup
-			wg.Add(1)
+
+			quicListener, err := quic.Listen(udpListener, testTLSServerConfig, testQUICConfig)
+			require.NoError(t, err)
+
+			serverDone := make(chan struct{})
 			go func() {
-				defer wg.Done()
 				quicServer(
-					t, udpListener, testTLSServerConfig, testQUICConfig,
-					test.dest, test.connectionType, test.metadata, test.message, test.expectedResponse,
+					ctx, t, quicListener, test.dest, test.connectionType, test.metadata, test.message, test.expectedResponse,
 				)
+				close(serverDone)
 			}()
 
-			qc := testQUICConnection(udpListener.LocalAddr(), t)
-			go qc.Serve(ctx)
+			qc := testQUICConnection(udpListener.LocalAddr(), t, uint8(i))
 
-			wg.Wait()
+			connDone := make(chan struct{})
+			go func() {
+				qc.Serve(ctx)
+				close(connDone)
+			}()
+
+			<-serverDone
 			cancel()
+			<-connDone
 		})
 	}
 }
@@ -177,23 +185,16 @@ func (fakeControlStream) IsStopped() bool {
 }
 
 func quicServer(
+	ctx context.Context,
 	t *testing.T,
-	conn net.PacketConn,
-	tlsConf *tls.Config,
-	config *quic.Config,
+	listener quic.Listener,
 	dest string,
 	connectionType quicpogs.ConnectionType,
 	metadata []quicpogs.Metadata,
 	message []byte,
 	expectedResponse []byte,
 ) {
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	earlyListener, err := quic.Listen(conn, tlsConf, config)
-	require.NoError(t, err)
-
-	session, err := earlyListener.Accept(ctx)
+	session, err := listener.Accept(ctx)
 	require.NoError(t, err)
 
 	quicStream, err := session.OpenStreamSync(context.Background())
@@ -482,6 +483,7 @@ func TestBuildHTTPRequest(t *testing.T) {
 
 	log := zerolog.Nop()
 	for _, test := range tests {
+		test := test // capture range variable
 		t.Run(test.name, func(t *testing.T) {
 			req, err := buildHTTPRequest(context.Background(), test.connectRequest, test.body, &log)
 			assert.NoError(t, err)
@@ -519,7 +521,8 @@ func TestServeUDPSession(t *testing.T) {
 		edgeQUICSessionChan <- edgeQUICSession
 	}()
 
-	qc := testQUICConnection(val, t)
+	// Random index to avoid reusing port
+	qc := testQUICConnection(val, t, 28)
 	go qc.Serve(ctx)
 
 	edgeQUICSession := <-edgeQUICSessionChan
@@ -703,7 +706,7 @@ func (s mockSessionRPCServer) UnregisterUdpSession(ctx context.Context, sessionI
 	return nil
 }
 
-func testQUICConnection(udpListenerAddr net.Addr, t *testing.T) *QUICConnection {
+func testQUICConnection(udpListenerAddr net.Addr, t *testing.T, index uint8) *QUICConnection {
 	tlsClientConfig := &tls.Config{
 		InsecureSkipVerify: true,
 		NextProtos:         []string{"argotunnel"},
@@ -713,7 +716,7 @@ func testQUICConnection(udpListenerAddr net.Addr, t *testing.T) *QUICConnection
 	qc, err := NewQUICConnection(
 		testQUICConfig,
 		udpListenerAddr,
-		0,
+		index,
 		tlsClientConfig,
 		&mockOrchestrator{originProxy: &mockOriginProxyWithRequest{}},
 		&tunnelpogs.ConnectionOptions{},

ingress/icmp_posix_test.go

@@ -57,13 +57,13 @@ func TestFunnelIdleTimeout(t *testing.T) {
 		datagramMuxer: muxer,
 	}
 	require.NoError(t, proxy.Request(ctx, &pk, &responder))
-	validateEchoFlow(t, muxer, &pk)
+	validateEchoFlow(t, <-muxer.cfdToEdge, &pk)
 
 	// Send second request, should reuse the funnel
 	require.NoError(t, proxy.Request(ctx, &pk, &packetResponder{
 		datagramMuxer: nil,
 	}))
-	validateEchoFlow(t, muxer, &pk)
+	validateEchoFlow(t, <-muxer.cfdToEdge, &pk)
 
 	time.Sleep(idleTimeout * 2)
 	newMuxer := newMockMuxer(0)
@@ -71,7 +71,7 @@ func TestFunnelIdleTimeout(t *testing.T) {
 		datagramMuxer: newMuxer,
 	}
 	require.NoError(t, proxy.Request(ctx, &pk, &newResponder))
-	validateEchoFlow(t, newMuxer, &pk)
+	validateEchoFlow(t, <-newMuxer.cfdToEdge, &pk)
 
 	cancel()
 	<-proxyDone

ingress/origin_icmp_proxy_test.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"net"
 	"net/netip"
-	"runtime"
 	"strings"
 	"sync"
 	"testing"
@@ -95,7 +94,7 @@ func testICMPRouterEcho(t *testing.T, sendIPv4 bool) {
 				},
 			}
 			require.NoError(t, router.Request(ctx, &pk, &responder))
-			validateEchoFlow(t, muxer, &pk)
+			validateEchoFlow(t, <-muxer.cfdToEdge, &pk)
 		}
 	}
 	cancel()
@@ -148,25 +147,35 @@ func TestTraceICMPRouterEcho(t *testing.T) {
 	}
 
 	require.NoError(t, router.Request(ctx, &pk, &responder))
-	// On Windows, request span is returned before reply
-	if runtime.GOOS != "windows" {
-		validateEchoFlow(t, muxer, &pk)
+	firstPK := <-muxer.cfdToEdge
+	var requestSpan *quicpogs.TracingSpanPacket
+	// The order of receiving reply or request span is not deterministic
+	switch firstPK.Type() {
+	case quicpogs.DatagramTypeIP:
+		// reply packet
+		validateEchoFlow(t, firstPK, &pk)
+	case quicpogs.DatagramTypeTracingSpan:
+		// Request span
+		requestSpan = firstPK.(*quicpogs.TracingSpanPacket)
+		require.NotEmpty(t, requestSpan.Spans)
+		require.True(t, bytes.Equal(serializedIdentity, requestSpan.TracingIdentity))
+	default:
+		panic(fmt.Sprintf("received unexpected packet type %d", firstPK.Type()))
 	}
 
-	// Request span
-	receivedPacket := <-muxer.cfdToEdge
-	requestSpan, ok := receivedPacket.(*quicpogs.TracingSpanPacket)
-	require.True(t, ok)
-	require.NotEmpty(t, requestSpan.Spans)
-	require.True(t, bytes.Equal(serializedIdentity, requestSpan.TracingIdentity))
-
-	if runtime.GOOS == "windows" {
-		validateEchoFlow(t, muxer, &pk)
+	secondPK := <-muxer.cfdToEdge
+	if requestSpan != nil {
+		// If first packet is request span, second packet should be the reply
+		validateEchoFlow(t, secondPK, &pk)
+	} else {
+		requestSpan = secondPK.(*quicpogs.TracingSpanPacket)
+		require.NotEmpty(t, requestSpan.Spans)
+		require.True(t, bytes.Equal(serializedIdentity, requestSpan.TracingIdentity))
 	}
 
 	// Reply span
-	receivedPacket = <-muxer.cfdToEdge
-	replySpan, ok := receivedPacket.(*quicpogs.TracingSpanPacket)
+	thirdPacket := <-muxer.cfdToEdge
+	replySpan, ok := thirdPacket.(*quicpogs.TracingSpanPacket)
 	require.True(t, ok)
 	require.NotEmpty(t, replySpan.Spans)
 	require.True(t, bytes.Equal(serializedIdentity, replySpan.TracingIdentity))
@@ -179,10 +188,10 @@ func TestTraceICMPRouterEcho(t *testing.T) {
 		datagramMuxer: muxer,
 	}
 	require.NoError(t, router.Request(ctx, &pk, &newResponder))
-	validateEchoFlow(t, muxer, &pk)
+	validateEchoFlow(t, <-muxer.cfdToEdge, &pk)
 
 	select {
-	case receivedPacket = <-muxer.cfdToEdge:
+	case receivedPacket := <-muxer.cfdToEdge:
 		panic(fmt.Sprintf("Receive unexpected packet %+v", receivedPacket))
 	default:
 	}
@@ -240,7 +249,7 @@ func TestConcurrentRequestsToSameDst(t *testing.T) {
 					},
 				}
 				require.NoError(t, router.Request(ctx, pk, &responder))
-				validateEchoFlow(t, muxer, pk)
+				validateEchoFlow(t, <-muxer.cfdToEdge, pk)
 			}
 		}()
 		go func() {
@@ -268,7 +277,7 @@ func TestConcurrentRequestsToSameDst(t *testing.T) {
 					},
 				}
 				require.NoError(t, router.Request(ctx, pk, &responder))
-				validateEchoFlow(t, muxer, pk)
+				validateEchoFlow(t, <-muxer.cfdToEdge, pk)
 			}
 		}()
 	}
@@ -357,33 +366,10 @@ func testICMPRouterRejectNotEcho(t *testing.T, srcDstIP netip.Addr, msgs []icmp.
 	}
 }
 
-type echoFlowResponder struct {
-	lock     sync.Mutex
-	decoder  *packet.ICMPDecoder
-	respChan chan []byte
-}
-
-func (efr *echoFlowResponder) SendPacket(dst netip.Addr, pk packet.RawPacket) error {
-	efr.lock.Lock()
-	defer efr.lock.Unlock()
-	copiedPacket := make([]byte, len(pk.Data))
-	copy(copiedPacket, pk.Data)
-	efr.respChan <- copiedPacket
-	return nil
-}
-
-func (efr *echoFlowResponder) Close() error {
-	efr.lock.Lock()
-	defer efr.lock.Unlock()
-	close(efr.respChan)
-	return nil
-}
-
-func validateEchoFlow(t *testing.T, muxer *mockMuxer, echoReq *packet.ICMP) {
-	pk := <-muxer.cfdToEdge
+func validateEchoFlow(t *testing.T, pk quicpogs.Packet, echoReq *packet.ICMP) {
 	decoder := packet.NewICMPDecoder()
 	decoded, err := decoder.Decode(packet.RawPacket{Data: pk.Payload()})
-	require.NoError(t, err)
+	require.NoError(t, err, pk)
 	require.Equal(t, decoded.Src, echoReq.Dst)
 	require.Equal(t, decoded.Dst, echoReq.Src)
 	require.Equal(t, echoReq.Protocol, decoded.Protocol)

ingress/packet_router.go

@@ -176,8 +176,6 @@ func (pr *packetResponder) exportSpan() {
 	}
 	spans := pr.tracedCtx.GetProtoSpans()
 	if len(spans) > 0 {
-		// Make sure spans are cleared after they are sent
-		defer pr.tracedCtx.ClearSpans()
 		pr.datagramMuxer.SendPacket(&quicpogs.TracingSpanPacket{
 			Spans:           spans,
 			TracingIdentity: pr.serializedIdentity,

supervisor/tunnel.go

@@ -37,6 +37,7 @@ const (
 	FeatureAllowRemoteConfig = "allow_remote_config"
 	FeatureDatagramV2        = "support_datagram_v2"
 	FeaturePostQuantum       = "postquantum"
+	FeatureQUICSupportEOF    = "support_quic_eof"
 )
 
 type TunnelConfig struct {

tracing/client.go

@@ -24,11 +24,9 @@ type InMemoryClient interface {
 	// Spans returns a copy of the list of in-memory stored spans as a base64
 	// encoded otlp protobuf string.
 	Spans() (string, error)
-	// ProtoSpans returns a copy of the list of in-memory stored spans as otlp
-	// protobuf byte array.
-	ProtoSpans() ([]byte, error)
-	// Clear spans removes all in-memory spans
-	ClearSpans()
+	// ExportProtoSpans returns a copy of the list of in-memory stored spans as otlp
+	// protobuf byte array and clears the in-memory spans.
+	ExportProtoSpans() ([]byte, error)
 }
 
 // InMemoryOtlpClient is a client implementation for otlptrace.Client
@@ -60,7 +58,7 @@ func (mc *InMemoryOtlpClient) UploadTraces(_ context.Context, protoSpans []*trac
 
 // Spans returns the list of in-memory stored spans as a base64 encoded otlp protobuf string.
 func (mc *InMemoryOtlpClient) Spans() (string, error) {
-	data, err := mc.ProtoSpans()
+	data, err := mc.ExportProtoSpans()
 	if err != nil {
 		return "", err
 	}
@@ -68,7 +66,7 @@ func (mc *InMemoryOtlpClient) Spans() (string, error) {
 }
 
 // ProtoSpans returns the list of in-memory stored spans as the protobuf byte array.
-func (mc *InMemoryOtlpClient) ProtoSpans() ([]byte, error) {
+func (mc *InMemoryOtlpClient) ExportProtoSpans() ([]byte, error) {
 	mc.mu.Lock()
 	defer mc.mu.Unlock()
 	if len(mc.spans) <= 0 {
@@ -77,13 +75,12 @@ func (mc *InMemoryOtlpClient) ProtoSpans() ([]byte, error) {
 	pbRequest := &coltracepb.ExportTraceServiceRequest{
 		ResourceSpans: mc.spans,
 	}
-	return proto.Marshal(pbRequest)
-}
-
-func (mc *InMemoryOtlpClient) ClearSpans() {
-	mc.mu.Lock()
-	defer mc.mu.Unlock()
+	serializedSpans, err := proto.Marshal(pbRequest)
+	if err != nil {
+		return nil, err
+	}
 	mc.spans = make([]*tracepb.ResourceSpans, 0)
+	return serializedSpans, nil
 }
 
 // NoopOtlpClient is a client implementation for otlptrace.Client that does nothing
@@ -107,7 +104,7 @@ func (mc *NoopOtlpClient) Spans() (string, error) {
 }
 
 // Spans always returns no traces error
-func (mc *NoopOtlpClient) ProtoSpans() ([]byte, error) {
+func (mc *NoopOtlpClient) ExportProtoSpans() ([]byte, error) {
 	return nil, errNoopTracer
 }
 

tracing/identity.go

@@ -24,7 +24,7 @@ type Identity struct {
 
 // TODO: TUN-6604 Remove this. To reconstruct into Jaeger propagation format, convert tracingContext to tracing.Identity
 func (tc *Identity) String() string {
-	return fmt.Sprintf("%x%x:%x:0:%x", tc.traceIDUpper, tc.traceIDLower, tc.spanID, tc.flags)
+	return fmt.Sprintf("%016x%016x:%x:0:%x", tc.traceIDUpper, tc.traceIDLower, tc.spanID, tc.flags)
 }
 
 func (tc *Identity) MarshalBinary() ([]byte, error) {
@@ -68,14 +68,9 @@ func NewIdentity(trace string) (*Identity, error) {
 		return nil, fmt.Errorf("trace '%s' doesn't have exactly 4 parts separated by %s", trace, separator)
 	}
 	const base = 16
-	tracingID := parts[0]
-	if len(tracingID) == 0 {
-		return nil, fmt.Errorf("missing tracing ID")
-	}
-	if len(tracingID) != 32 {
-		// Correctly left pad the trace to a length of 32
-		left := traceID128bitsWidth - len(tracingID)
-		tracingID = strings.Repeat("0", left) + tracingID
+	tracingID, err := padTracingID(parts[0])
+	if err != nil {
+		return nil, err
 	}
 	traceIDUpper, err := strconv.ParseUint(tracingID[:16], base, 64)
 	if err != nil {
@@ -100,3 +95,16 @@ func NewIdentity(trace string) (*Identity, error) {
 		flags:        uint8(flags),
 	}, nil
 }
+
+func padTracingID(tracingID string) (string, error) {
+	if len(tracingID) == 0 {
+		return "", fmt.Errorf("missing tracing ID")
+	}
+	if len(tracingID) == traceID128bitsWidth {
+		return tracingID, nil
+	}
+	// Correctly left pad the trace to a length of 32
+	left := traceID128bitsWidth - len(tracingID)
+	paddedTracingID := strings.Repeat("0", left) + tracingID
+	return paddedTracingID, nil
+}

tracing/identity_test.go

@@ -10,40 +10,55 @@ func TestNewIdentity(t *testing.T) {
 	testCases := []struct {
 		testCase string
 		trace    string
-		valid    bool
+		expected string
 	}{
 		{
 			testCase: "full length trace",
 			trace:    "ec31ad8a01fde11fdcabe2efdce36873:52726f6cabc144f5:0:1",
-			valid:    true,
+			expected: "ec31ad8a01fde11fdcabe2efdce36873:52726f6cabc144f5:0:1",
 		},
 		{
 			testCase: "short trace ID",
 			trace:    "ad8a01fde11fdcabe2efdce36873:52726f6cabc144f5:0:1",
-			valid:    true,
+			expected: "0000ad8a01fde11fdcabe2efdce36873:52726f6cabc144f5:0:1",
+		},
+		{
+			testCase: "short trace ID with 0s in the middle",
+			trace:    "ad8a01fde11f000002efdce36873:52726f6cabc144f5:0:1",
+			expected: "0000ad8a01fde11f000002efdce36873:52726f6cabc144f5:0:1",
+		},
+		{
+			testCase: "short trace ID with 0s in the beginning and middle",
+			trace:    "001ad8a01fde11fdcabe2efdce36873:52726f6cabc144f5:0:1",
+			expected: "0001ad8a01fde11fdcabe2efdce36873:52726f6cabc144f5:0:1",
 		},
 		{
 			testCase: "no trace",
 			trace:    "",
-			valid:    false,
 		},
 		{
 			testCase: "missing flags",
 			trace:    "ec31ad8a01fde11fdcabe2efdce36873:52726f6cabc144f5:0",
-			valid:    false,
 		},
 		{
 			testCase: "missing separator",
 			trace:    "ec31ad8a01fde11fdcabe2efdce3687352726f6cabc144f501",
-			valid:    false,
 		},
 	}
 
 	for _, testCase := range testCases {
 		identity, err := NewIdentity(testCase.trace)
-		if testCase.valid {
-			require.NoError(t, err, testCase.testCase)
-			require.Equal(t, testCase.trace, identity.String())
+		if testCase.expected != "" {
+			require.NoError(t, err)
+			require.Equal(t, testCase.expected, identity.String())
+
+			serializedIdentity, err := identity.MarshalBinary()
+			require.NoError(t, err)
+			deserializedIdentity := new(Identity)
+			err = deserializedIdentity.UnmarshalBinary(serializedIdentity)
+			require.NoError(t, err)
+			require.Equal(t, identity, deserializedIdentity)
+
 		} else {
 			require.Error(t, err)
 			require.Nil(t, identity)

tracing/tracing.go

@@ -157,7 +157,7 @@ func (cft *cfdTracer) GetSpans() (enc string) {
 
 // GetProtoSpans returns the spans as the otlp traces in protobuf byte array.
 func (cft *cfdTracer) GetProtoSpans() (proto []byte) {
-	proto, err := cft.exporter.ProtoSpans()
+	proto, err := cft.exporter.ExportProtoSpans()
 	switch err {
 	case nil:
 		break
@@ -189,10 +189,6 @@ func (cft *cfdTracer) AddSpans(headers http.Header) {
 	headers[CanonicalCloudflaredTracingHeader] = []string{enc}
 }
 
-func (cft *cfdTracer) ClearSpans() {
-	cft.exporter.ClearSpans()
-}
-
 // End will set the OK status for the span and then end it.
 func End(span trace.Span) {
 	endSpan(span, -1, codes.Ok, nil)