TUN-6576: Consume cf-trace-id from incoming TCP requests to create root span

(cherry picked from commit f48a7cd3dd)

connection/connection.go

@@ -123,23 +123,24 @@ func (t Type) String() string {
 
 // OriginProxy is how data flows from cloudflared to the origin services running behind it.
 type OriginProxy interface {
-	ProxyHTTP(w ResponseWriter, tr *tracing.TracedRequest, isWebsocket bool) error
+	ProxyHTTP(w ResponseWriter, tr *tracing.TracedHTTPRequest, isWebsocket bool) error
 	ProxyTCP(ctx context.Context, rwa ReadWriteAcker, req *TCPRequest) error
 }
 
 // TCPRequest defines the input format needed to perform a TCP proxy.
 type TCPRequest struct {
-	Dest    string
-	CFRay   string
-	LBProbe bool
-	FlowID  string
+	Dest      string
+	CFRay     string
+	LBProbe   bool
+	FlowID    string
+	CfTraceID string
 }
 
 // ReadWriteAcker is a readwriter with the ability to Acknowledge to the downstream (edge) that the origin has
 // accepted the connection.
 type ReadWriteAcker interface {
 	io.ReadWriter
-	AckConnection() error
+	AckConnection(tracePropagation string) error
 }
 
 // HTTPResponseReadWriteAcker is an HTTP implementation of ReadWriteAcker.
@@ -168,7 +169,7 @@ func (h *HTTPResponseReadWriteAcker) Write(p []byte) (int, error) {
 
 // AckConnection acks an HTTP connection by sending a switch protocols status code that enables the caller to
 // upgrade to streams.
-func (h *HTTPResponseReadWriteAcker) AckConnection() error {
+func (h *HTTPResponseReadWriteAcker) AckConnection(tracePropagation string) error {
 	resp := &http.Response{
 		Status:        switchingProtocolText,
 		StatusCode:    http.StatusSwitchingProtocols,
@@ -179,6 +180,10 @@ func (h *HTTPResponseReadWriteAcker) AckConnection() error {
 		resp.Header = websocket.NewResponseHeader(h.req)
 	}
 
+	if tracePropagation != "" {
+		resp.Header.Add(tracing.CanonicalCloudflaredTracingHeader, tracePropagation)
+	}
+
 	return h.w.WriteRespHeaders(resp.StatusCode, resp.Header)
 }
 

connection/connection_test.go

@@ -30,6 +30,8 @@ var (
 	testLargeResp = make([]byte, largeFileSize)
 )
 
+var _ ReadWriteAcker = (*HTTPResponseReadWriteAcker)(nil)
+
 type testRequest struct {
 	name           string
 	endpoint       string
@@ -60,7 +62,7 @@ type mockOriginProxy struct{}
 
 func (moc *mockOriginProxy) ProxyHTTP(
 	w ResponseWriter,
-	tr *tracing.TracedRequest,
+	tr *tracing.TracedHTTPRequest,
 	isWebsocket bool,
 ) error {
 	req := tr.Request

connection/h2mux.go

@@ -69,6 +69,7 @@ func NewH2muxConnection(
 	connIndex uint8,
 	observer *Observer,
 	gracefulShutdownC <-chan struct{},
+	log *zerolog.Logger,
 ) (*h2muxConnection, error, bool) {
 	h := &h2muxConnection{
 		orchestrator:      orchestrator,
@@ -79,6 +80,7 @@ func NewH2muxConnection(
 		observer:          observer,
 		gracefulShutdownC: gracefulShutdownC,
 		newRPCClientFunc:  newRegistrationRPCClient,
+		log:               log,
 	}
 
 	// Establish a muxed connection with the edge
@@ -234,7 +236,7 @@ func (h *h2muxConnection) ServeStream(stream *h2mux.MuxedStream) error {
 		return err
 	}
 
-	err = originProxy.ProxyHTTP(respWriter, tracing.NewTracedRequest(req), sourceConnectionType == TypeWebsocket)
+	err = originProxy.ProxyHTTP(respWriter, tracing.NewTracedHTTPRequest(req, h.log), sourceConnectionType == TypeWebsocket)
 	if err != nil {
 		respWriter.WriteErrorResponse()
 	}

connection/h2mux_test.go

@@ -48,7 +48,7 @@ func newH2MuxConnection(t require.TestingT) (*h2muxConnection, *h2mux.Muxer) {
 	}()
 	var connIndex = uint8(0)
 	testObserver := NewObserver(&log, &log)
-	h2muxConn, err, _ := NewH2muxConnection(testOrchestrator, testGracePeriod, testMuxerConfig, originConn, connIndex, testObserver, nil)
+	h2muxConn, err, _ := NewH2muxConnection(testOrchestrator, testGracePeriod, testMuxerConfig, originConn, connIndex, testObserver, nil, &log)
 	require.NoError(t, err)
 	return h2muxConn, <-edgeMuxChan
 }

connection/http2.go

@@ -132,7 +132,7 @@ func (c *HTTP2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	case TypeWebsocket, TypeHTTP:
 		stripWebsocketUpgradeHeader(r)
 		// Check for tracing on request
-		tr := tracing.NewTracedRequest(r)
+		tr := tracing.NewTracedHTTPRequest(r, c.log)
 		if err := originProxy.ProxyHTTP(respWriter, tr, connType == TypeWebsocket); err != nil {
 			err := fmt.Errorf("Failed to proxy HTTP: %w", err)
 			c.log.Error().Err(err)

connection/quic.go

@@ -197,7 +197,7 @@ func (q *QUICConnection) dispatchRequest(ctx context.Context, stream *quicpogs.R
 
 	switch request.Type {
 	case quicpogs.ConnectionTypeHTTP, quicpogs.ConnectionTypeWebsocket:
-		tracedReq, err := buildHTTPRequest(ctx, request, stream)
+		tracedReq, err := buildHTTPRequest(ctx, request, stream, q.logger)
 		if err != nil {
 			return err
 		}
@@ -208,8 +208,9 @@ func (q *QUICConnection) dispatchRequest(ctx context.Context, stream *quicpogs.R
 		rwa := &streamReadWriteAcker{stream}
 		metadata := request.MetadataMap()
 		return originProxy.ProxyTCP(ctx, rwa, &TCPRequest{
-			Dest:   request.Dest,
-			FlowID: metadata[QUICMetadataFlowID],
+			Dest:      request.Dest,
+			FlowID:    metadata[QUICMetadataFlowID],
+			CfTraceID: metadata[tracing.TracerContextName],
 		})
 	}
 	return nil
@@ -296,8 +297,12 @@ type streamReadWriteAcker struct {
 }
 
 // AckConnection acks response back to the proxy.
-func (s *streamReadWriteAcker) AckConnection() error {
-	return s.WriteConnectResponseData(nil)
+func (s *streamReadWriteAcker) AckConnection(tracePropagation string) error {
+	metadata := quicpogs.Metadata{
+		Key: tracing.CanonicalCloudflaredTracingHeader,
+		Val: tracePropagation,
+	}
+	return s.WriteConnectResponseData(nil, metadata)
 }
 
 // httpResponseAdapter translates responses written by the HTTP Proxy into ones that can be used in QUIC.
@@ -325,7 +330,12 @@ func (hrw httpResponseAdapter) WriteErrorResponse(err error) {
 	hrw.WriteConnectResponseData(err, quicpogs.Metadata{Key: "HttpStatus", Val: strconv.Itoa(http.StatusBadGateway)})
 }
 
-func buildHTTPRequest(ctx context.Context, connectRequest *quicpogs.ConnectRequest, body io.ReadCloser) (*tracing.TracedRequest, error) {
+func buildHTTPRequest(
+	ctx context.Context,
+	connectRequest *quicpogs.ConnectRequest,
+	body io.ReadCloser,
+	log *zerolog.Logger,
+) (*tracing.TracedHTTPRequest, error) {
 	metadata := connectRequest.MetadataMap()
 	dest := connectRequest.Dest
 	method := metadata[HTTPMethodKey]
@@ -367,7 +377,7 @@ func buildHTTPRequest(ctx context.Context, connectRequest *quicpogs.ConnectReque
 	stripWebsocketUpgradeHeader(req)
 
 	// Check for tracing on request
-	tracedReq := tracing.NewTracedRequest(req)
+	tracedReq := tracing.NewTracedHTTPRequest(req, log)
 	return tracedReq, err
 }
 

connection/quic_test.go

@@ -36,6 +36,8 @@ var (
 	}
 )
 
+var _ ReadWriteAcker = (*streamReadWriteAcker)(nil)
+
 // TestQUICServer tests if a quic server accepts and responds to a quic client with the acceptance protocol.
 // It also serves as a demonstration for communication with the QUIC connection started by a cloudflared.
 func TestQUICServer(t *testing.T) {
@@ -220,7 +222,7 @@ func quicServer(
 
 type mockOriginProxyWithRequest struct{}
 
-func (moc *mockOriginProxyWithRequest) ProxyHTTP(w ResponseWriter, tr *tracing.TracedRequest, isWebsocket bool) error {
+func (moc *mockOriginProxyWithRequest) ProxyHTTP(w ResponseWriter, tr *tracing.TracedHTTPRequest, isWebsocket bool) error {
 	// These are a series of crude tests to ensure the headers and http related data is transferred from
 	// metadata.
 	r := tr.Request
@@ -475,9 +477,10 @@ func TestBuildHTTPRequest(t *testing.T) {
 		},
 	}
 
+	log := zerolog.Nop()
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			req, err := buildHTTPRequest(context.Background(), test.connectRequest, test.body)
+			req, err := buildHTTPRequest(context.Background(), test.connectRequest, test.body, &log)
 			assert.NoError(t, err)
 			test.req = test.req.WithContext(req.Context())
 			assert.Equal(t, test.req, req.Request)
@@ -486,7 +489,7 @@ func TestBuildHTTPRequest(t *testing.T) {
 }
 
 func (moc *mockOriginProxyWithRequest) ProxyTCP(ctx context.Context, rwa ReadWriteAcker, tcpRequest *TCPRequest) error {
-	rwa.AckConnection()
+	rwa.AckConnection("")
 	io.Copy(rwa, rwa)
 	return nil
 }