hoelee/cloudflared
1
0
Fork 0
mirror of https://github.com/cloudflare/cloudflared.git synced 2025-07-27 15:39:58 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

TUN-8224: Count and collect metrics on stream connect successes/errors

Browse Source
This commit is contained in:
chungthuang
2024-02-07 14:32:39 +00:00
parent 98e043d17d
commit 638203f9f1
2 changed files with 45 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
proxy/proxy.go
View File

@@ -6,6 +6,7 @@ import (
"io"
"net/http"
"strconv"
"time"
"github.com/pkg/errors"
"github.com/rs/zerolog"
@@ -139,7 +140,11 @@ func (p *Proxy) ProxyHTTP(
return fmt.Errorf("response writer is not a flusher")
}
rws := connection.NewHTTPResponseReadWriterAcker(w, flusher, req)
if err := p.proxyStream(tr.ToTracedContext(), rws, dest, originProxy); err != nil {
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)
return err
@@ -173,24 +178,22 @@ func (p *Proxy) ProxyTCP(
tracedCtx := tracing.NewTracedContext(serveCtx, req.CfTraceID, p.log)
p.log.Debug().
Int(management.EventTypeKey, int(management.TCP)).
Str(LogFieldFlowID, req.FlowID).
Str(LogFieldDestAddr, req.Dest).
Uint8(LogFieldConnIndex, req.ConnIndex).
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)
}
if err := p.proxyStream(tracedCtx, rwa, req.Dest, p.warpRouting.Proxy); err != nil {
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)
return err
}
p.log.Debug().
Int(management.EventTypeKey, int(management.TCP)).
Str(LogFieldFlowID, req.FlowID).
Str(LogFieldDestAddr, req.Dest).
Uint8(LogFieldConnIndex, req.ConnIndex).
Msg("tcp proxy stream finished successfully")
debugLogger().Msg("tcp proxy stream finished successfully")
return nil
}
@@ -294,16 +297,21 @@ func (p *Proxy) proxyHTTPRequest(
// proxyStream proxies type TCP and other underlying types if the connection is defined as a stream oriented
// ingress rule.
// connectedLogger is used to log when the connection is acknowledged
func (p *Proxy) proxyStream(
tr *tracing.TracedContext,
rwa connection.ReadWriteAcker,
dest string,
connectionProxy ingress.StreamBasedOriginProxy,
connectedLogger *zerolog.Event,
) error {
ctx := tr.Context
_, connectSpan := tr.Tracer().Start(ctx, "stream-connect")
start := time.Now()
originConn, err := connectionProxy.EstablishConnection(ctx, dest)
if err != nil {
connectStreamErrors.Inc()
tracing.EndWithErrorStatus(connectSpan, err)
return err
}
@@ -313,9 +321,13 @@ func (p *Proxy) proxyStream(
encodedSpans := tr.GetSpans()
if err := rwa.AckConnection(encodedSpans); err != nil {
connectStreamErrors.Inc()
return err
}
connectLatency.Observe(float64(time.Since(start).Milliseconds()))
connectedLogger.Msg("proxy stream established")
originConn.Stream(ctx, rwa, p.log)
return nil
}