Release 2022.10.3

RELEASE_NOTES

@@ -1,3 +1,7 @@
+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

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

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