mirror of
https://github.com/cloudflare/cloudflared.git
synced 2025-07-27 20:59:58 +00:00
Revert "TUN-6576: Consume cf-trace-id from incoming TCP requests to create root span"
This reverts commit f48a7cd3dd
.
This commit is contained in:
@@ -123,24 +123,23 @@ 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.TracedHTTPRequest, isWebsocket bool) error
|
||||
ProxyHTTP(w ResponseWriter, tr *tracing.TracedRequest, 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
|
||||
CfTraceID string
|
||||
Dest string
|
||||
CFRay string
|
||||
LBProbe bool
|
||||
FlowID 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(tracePropagation string) error
|
||||
AckConnection() error
|
||||
}
|
||||
|
||||
// HTTPResponseReadWriteAcker is an HTTP implementation of ReadWriteAcker.
|
||||
@@ -169,7 +168,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(tracePropagation string) error {
|
||||
func (h *HTTPResponseReadWriteAcker) AckConnection() error {
|
||||
resp := &http.Response{
|
||||
Status: switchingProtocolText,
|
||||
StatusCode: http.StatusSwitchingProtocols,
|
||||
@@ -180,10 +179,6 @@ func (h *HTTPResponseReadWriteAcker) AckConnection(tracePropagation string) erro
|
||||
resp.Header = websocket.NewResponseHeader(h.req)
|
||||
}
|
||||
|
||||
if tracePropagation != "" {
|
||||
resp.Header.Add(tracing.CanonicalCloudflaredTracingHeader, tracePropagation)
|
||||
}
|
||||
|
||||
return h.w.WriteRespHeaders(resp.StatusCode, resp.Header)
|
||||
}
|
||||
|
||||
|
@@ -30,8 +30,6 @@ var (
|
||||
testLargeResp = make([]byte, largeFileSize)
|
||||
)
|
||||
|
||||
var _ ReadWriteAcker = (*HTTPResponseReadWriteAcker)(nil)
|
||||
|
||||
type testRequest struct {
|
||||
name string
|
||||
endpoint string
|
||||
@@ -62,7 +60,7 @@ type mockOriginProxy struct{}
|
||||
|
||||
func (moc *mockOriginProxy) ProxyHTTP(
|
||||
w ResponseWriter,
|
||||
tr *tracing.TracedHTTPRequest,
|
||||
tr *tracing.TracedRequest,
|
||||
isWebsocket bool,
|
||||
) error {
|
||||
req := tr.Request
|
||||
|
@@ -69,7 +69,6 @@ func NewH2muxConnection(
|
||||
connIndex uint8,
|
||||
observer *Observer,
|
||||
gracefulShutdownC <-chan struct{},
|
||||
log *zerolog.Logger,
|
||||
) (*h2muxConnection, error, bool) {
|
||||
h := &h2muxConnection{
|
||||
orchestrator: orchestrator,
|
||||
@@ -80,7 +79,6 @@ func NewH2muxConnection(
|
||||
observer: observer,
|
||||
gracefulShutdownC: gracefulShutdownC,
|
||||
newRPCClientFunc: newRegistrationRPCClient,
|
||||
log: log,
|
||||
}
|
||||
|
||||
// Establish a muxed connection with the edge
|
||||
@@ -236,7 +234,7 @@ func (h *h2muxConnection) ServeStream(stream *h2mux.MuxedStream) error {
|
||||
return err
|
||||
}
|
||||
|
||||
err = originProxy.ProxyHTTP(respWriter, tracing.NewTracedHTTPRequest(req, h.log), sourceConnectionType == TypeWebsocket)
|
||||
err = originProxy.ProxyHTTP(respWriter, tracing.NewTracedRequest(req), sourceConnectionType == TypeWebsocket)
|
||||
if err != nil {
|
||||
respWriter.WriteErrorResponse()
|
||||
}
|
||||
|
@@ -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, &log)
|
||||
h2muxConn, err, _ := NewH2muxConnection(testOrchestrator, testGracePeriod, testMuxerConfig, originConn, connIndex, testObserver, nil)
|
||||
require.NoError(t, err)
|
||||
return h2muxConn, <-edgeMuxChan
|
||||
}
|
||||
|
@@ -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.NewTracedHTTPRequest(r, c.log)
|
||||
tr := tracing.NewTracedRequest(r)
|
||||
if err := originProxy.ProxyHTTP(respWriter, tr, connType == TypeWebsocket); err != nil {
|
||||
err := fmt.Errorf("Failed to proxy HTTP: %w", err)
|
||||
c.log.Error().Err(err)
|
||||
|
@@ -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, q.logger)
|
||||
tracedReq, err := buildHTTPRequest(ctx, request, stream)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -208,9 +208,8 @@ 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],
|
||||
CfTraceID: metadata[tracing.TracerContextName],
|
||||
Dest: request.Dest,
|
||||
FlowID: metadata[QUICMetadataFlowID],
|
||||
})
|
||||
}
|
||||
return nil
|
||||
@@ -297,12 +296,8 @@ type streamReadWriteAcker struct {
|
||||
}
|
||||
|
||||
// AckConnection acks response back to the proxy.
|
||||
func (s *streamReadWriteAcker) AckConnection(tracePropagation string) error {
|
||||
metadata := quicpogs.Metadata{
|
||||
Key: tracing.CanonicalCloudflaredTracingHeader,
|
||||
Val: tracePropagation,
|
||||
}
|
||||
return s.WriteConnectResponseData(nil, metadata)
|
||||
func (s *streamReadWriteAcker) AckConnection() error {
|
||||
return s.WriteConnectResponseData(nil)
|
||||
}
|
||||
|
||||
// httpResponseAdapter translates responses written by the HTTP Proxy into ones that can be used in QUIC.
|
||||
@@ -330,12 +325,7 @@ 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,
|
||||
log *zerolog.Logger,
|
||||
) (*tracing.TracedHTTPRequest, error) {
|
||||
func buildHTTPRequest(ctx context.Context, connectRequest *quicpogs.ConnectRequest, body io.ReadCloser) (*tracing.TracedRequest, error) {
|
||||
metadata := connectRequest.MetadataMap()
|
||||
dest := connectRequest.Dest
|
||||
method := metadata[HTTPMethodKey]
|
||||
@@ -377,7 +367,7 @@ func buildHTTPRequest(
|
||||
stripWebsocketUpgradeHeader(req)
|
||||
|
||||
// Check for tracing on request
|
||||
tracedReq := tracing.NewTracedHTTPRequest(req, log)
|
||||
tracedReq := tracing.NewTracedRequest(req)
|
||||
return tracedReq, err
|
||||
}
|
||||
|
||||
|
@@ -36,8 +36,6 @@ 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) {
|
||||
@@ -222,7 +220,7 @@ func quicServer(
|
||||
|
||||
type mockOriginProxyWithRequest struct{}
|
||||
|
||||
func (moc *mockOriginProxyWithRequest) ProxyHTTP(w ResponseWriter, tr *tracing.TracedHTTPRequest, isWebsocket bool) error {
|
||||
func (moc *mockOriginProxyWithRequest) ProxyHTTP(w ResponseWriter, tr *tracing.TracedRequest, 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
|
||||
@@ -477,10 +475,9 @@ 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, &log)
|
||||
req, err := buildHTTPRequest(context.Background(), test.connectRequest, test.body)
|
||||
assert.NoError(t, err)
|
||||
test.req = test.req.WithContext(req.Context())
|
||||
assert.Equal(t, test.req, req.Request)
|
||||
@@ -489,7 +486,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
|
||||
}
|
||||
|
Reference in New Issue
Block a user