feat(proxy): label request-outcome metric by destination node to isolate a drained/stale node

packages/client-proxy/internal/proxy/proxy.go

@@ -12,6 +12,7 @@ import (
 	"strings"
 	"time"
 
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
 	"go.uber.org/zap"
 	"google.golang.org/grpc/codes"
@@ -195,6 +196,20 @@ func NewClientProxy(meterProvider metric.MeterProvider, serviceName string, port
 		resumeCloser = flights
 	}
 
+	outcomeCounter, err := telemetry.GetCounter(meter, telemetry.ClientProxyRequestOutcomeCounterName)
+	if err != nil {
+		return nil, nil, fmt.Errorf("error registering client proxy request outcome metric (%s): %w", telemetry.ClientProxyRequestOutcomeCounterName, err)
+	}
+	recordOutcome := func(ctx context.Context, outcome pool.RequestOutcome, node string) {
+		if node == "" {
+			node = "unknown"
+		}
+		outcomeCounter.Add(ctx, 1, metric.WithAttributes(
+			attribute.String("outcome", string(outcome)),
+			attribute.String("node", node),
+		))
+	}
+
 	getTargetFromRequest := reverseproxy.GetTargetFromRequest()
 	proxy := reverseproxy.New(
 		port,
@@ -280,6 +295,9 @@ func NewClientProxy(meterProvider metric.MeterProvider, serviceName string, port
 				SandboxPort:   port,
 				ConnectionKey: pool.ClientProxyConnectionKey,
 				Url:           url,
+				// Label outcomes by the worker instance so an unreachable-upstream
+				// spike points straight at a drained/stale node.
+				Node: route.OrchestratorID,
 				MaskRequestHost: clientProxyMaskRequestHost(
 					ctx,
 					featureFlagsClient,
@@ -291,9 +309,10 @@ func NewClientProxy(meterProvider metric.MeterProvider, serviceName string, port
 		},
 		nil,
 		false,
+		recordOutcome,
 	)
 
-	_, err := telemetry.GetObservableUpDownCounter(
+	_, err = telemetry.GetObservableUpDownCounter(
 		meter, telemetry.ClientProxyPoolConnectionsMeterCounterName, func(_ context.Context, observer metric.Int64Observer) error {
 			observer.Observe(proxy.CurrentPoolConnections())
 

packages/orchestrator/pkg/proxy/proxy.go

@@ -13,6 +13,7 @@ import (
 	"sync"
 	"time"
 
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
 	"go.uber.org/zap"
 
@@ -116,6 +117,16 @@ func NewSandboxProxy(meterProvider metric.MeterProvider, port uint16, sandboxes
 		OnConnectionBlocked:  metrics.RecordConnectionBlocked,
 	}
 
+	outcomeCounter, err := telemetry.GetCounter(meter, telemetry.OrchestratorProxyRequestOutcomeCounterName)
+	if err != nil {
+		return nil, fmt.Errorf("error registering orchestrator proxy request outcome metric (%s): %w", telemetry.OrchestratorProxyRequestOutcomeCounterName, err)
+	}
+	// node is unused here: the orchestrator sandbox proxy's node is itself, so
+	// its destinations carry no Node and the label would always be "unknown".
+	recordOutcome := func(ctx context.Context, outcome pool.RequestOutcome, _ string) {
+		outcomeCounter.Add(ctx, 1, metric.WithAttributes(attribute.String("outcome", string(outcome))))
+	}
+
 	sandboxProxy := &SandboxProxy{limiter: limiter, sandboxes: sandboxes, owners: make(map[string]*pool.Owner), retirement: retirementTimeout}
 	proxy := reverseproxy.New(
 		port,
@@ -128,9 +139,10 @@ func NewSandboxProxy(meterProvider metric.MeterProvider, port uint16, sandboxes
 		// because the servers inside of the sandbox can be unstable (restarts),
 		// and we are also on the same host, so the overhead is minimal.
 		true,
+		recordOutcome,
 	)
 
-	_, err := telemetry.GetObservableUpDownCounter(meter, telemetry.OrchestratorProxyServerConnectionsMeterCounterName, func(_ context.Context, observer metric.Int64Observer) error {
+	_, err = telemetry.GetObservableUpDownCounter(meter, telemetry.OrchestratorProxyServerConnectionsMeterCounterName, func(_ context.Context, observer metric.Int64Observer) error {
 		observer.Observe(proxy.CurrentServerConnections())
 
 		return nil

packages/orchestrator/pkg/proxy/proxy_test.go

@@ -142,7 +142,7 @@ func TestSandboxProxyRefusesFullRouterIngress(t *testing.T) {
 
 			sandboxes := sandbox.NewSandboxesMap()
 			p := &SandboxProxy{sandboxes: sandboxes, owners: make(map[string]*pool.Owner), limiter: connlimit.NewConnectionLimiter(), retirement: retirementTimeout}
-			p.proxy = reverseproxy.New(0, 1, time.Minute, p.resolveDestination, nil, true)
+			p.proxy = reverseproxy.New(0, 1, time.Minute, p.resolveDestination, nil, true, nil)
 			defer p.proxy.Close()
 			frontend := httptest.NewServer(p.proxy.Handler)
 			defer frontend.Close()
@@ -207,7 +207,7 @@ func TestSandboxProxyResetAndLifecycleRetirement(t *testing.T) {
 		}
 
 		return 8
-	}}, true)
+	}}, true, nil)
 	defer p.proxy.Close()
 	hold := func() *limitGate {
 		g := &limitGate{reachedCh: make(chan struct{}), releaseCh: make(chan struct{})}

packages/orchestrator/pkg/proxy/shutdown_h2c_test.go

@@ -54,7 +54,7 @@ func TestSandboxProxyCloseInterruptsStalledH2CUpload(t *testing.T) {
 					Limiter:              connlimit.NewConnectionLimiter(),
 					GetMaxLimit:          func(context.Context) int { return 10 },
 					OnConnectionReleased: func(context.Context, int64) { close(handlerDone) },
-				}, true)}
+				}, true, nil)}
 			var listenConfig net.ListenConfig
 			listener, err := listenConfig.Listen(t.Context(), "tcp", "127.0.0.1:0")
 			require.NoError(t, err)

packages/orchestrator/pkg/proxy/shutdown_test.go

@@ -62,7 +62,7 @@ func TestSandboxProxyCloseInterruptsStalledUpload(t *testing.T) {
 					Limiter:              connlimit.NewConnectionLimiter(),
 					GetMaxLimit:          func(context.Context) int { return 10 },
 					OnConnectionReleased: func(context.Context, int64) { close(handlerDone) },
-				}, true)}
+				}, true, nil)}
 			var listenConfig net.ListenConfig
 			listener, err := listenConfig.Listen(t.Context(), "tcp", "127.0.0.1:0")
 			require.NoError(t, err)

packages/shared/pkg/proxy/cors_proxy_test.go

@@ -313,7 +313,7 @@ func TestNoRawHTTPErrorInProxyPackage(t *testing.T) {
 
 func TestRetiredOwnerAnswersPreflight(t *testing.T) {
 	t.Parallel()
-	p := pool.New(4, 1, time.Minute, false)
+	p := pool.New(4, 1, time.Minute, false, nil)
 	target := &url.URL{Scheme: "http", Host: "127.0.0.1:1"}
 	d, err := destinationFor(target, "retired")(nil)
 	require.NoError(t, err)

packages/shared/pkg/proxy/full_duplex_test.go

@@ -169,7 +169,7 @@ func startProxyWithIdleTimeout(t *testing.T, getDestination func(r *http.Request
 	limits := &ConnectionLimitConfig{Limiter: limiter, GetMaxLimit: func(context.Context) int { return -1 }}
 	serverLog := &lockedBuffer{}
 	port := listener.Addr().(*net.TCPAddr).Port
-	proxy := New(uint16(port), 1, idleTimeout, getDestination, limits, false)
+	proxy := New(uint16(port), 1, idleTimeout, getDestination, limits, false, nil)
 	proxy.IdleTimeout = 2 * idleTimeout
 	proxy.ErrorLog = log.New(serverLog, "", 0)
 	go func() {
@@ -733,7 +733,7 @@ func TestHandlerEnablesFullDuplex(t *testing.T) {
 	backendURL, err := url.Parse(backend.URL)
 	require.NoError(t, err)
 
-	p := pool.New(maxClientConns, 1, time.Second, false)
+	p := pool.New(maxClientConns, 1, time.Second, false, nil)
 
 	recorder := &fullDuplexRecorder{ResponseRecorder: httptest.NewRecorder()}
 	req := httptest.NewRequestWithContext(t.Context(), http.MethodGet, "http://49983-sbx-1-00000000.localhost/", nil)

packages/shared/pkg/proxy/handler.go

@@ -53,6 +53,7 @@ func handler(p *pool.ProxyPool, getDestination func(r *http.Request) (*pool.Dest
 		// request ended; a client that is gone reads nothing either way.
 		if ctxErr := ctx.Err(); err != nil && ctxErr != nil && errors.Is(err, ctxErr) {
 			l.Debug(ctx, "request ended before it was routed", zap.String("host", r.Host))
+			p.RecordOutcome(ctx, pool.OutcomeClientClosed, "")
 			cors.Error(w, "Request ended before it was routed", http.StatusRequestTimeout)
 
 			return
@@ -97,6 +98,8 @@ func handler(p *pool.ProxyPool, getDestination func(r *http.Request) (*pool.Dest
 				zap.String("host", r.Host),
 				logger.WithSandboxID(notFoundErr.SandboxId))
 
+			p.RecordOutcome(ctx, pool.OutcomeSandboxNotFound, "")
+
 			err := template.
 				NewSandboxNotFoundError(notFoundErr.SandboxId, r.Host).
 				HandleError(w, r)
@@ -250,6 +253,7 @@ func handler(p *pool.ProxyPool, getDestination func(r *http.Request) (*pool.Dest
 
 		if err != nil {
 			l.Error(ctx, "failed to route request", zap.Error(err), zap.String("host", r.Host))
+			p.RecordOutcome(ctx, pool.OutcomeRoutingFailed, "")
 			cors.Error(w, fmt.Sprintf("Unexpected error when routing request: %s", err), http.StatusInternalServerError)
 
 			return
@@ -279,6 +283,7 @@ func handler(p *pool.ProxyPool, getDestination func(r *http.Request) (*pool.Dest
 			if uploading {
 				closeUnreadBody(w)
 			}
+			p.RecordOutcome(ctx, pool.OutcomeUpstreamUnreachable, d.Node)
 			cors.Error(w, "Failed to route request to sandbox", http.StatusBadGateway)
 
 			return
@@ -322,6 +327,8 @@ func handler(p *pool.ProxyPool, getDestination func(r *http.Request) (*pool.Dest
 					connLimitConfig.OnConnectionBlocked(ctx)
 				}
 
+				p.RecordOutcome(ctx, pool.OutcomeTooManyConnections, d.Node)
+
 				// The sandbox is reachable, but the proxy is refusing to dial it,
 				// so the proxy is the only party that can answer this preflight.
 				if cors.HandlePreflight(w, r) {

packages/shared/pkg/proxy/outcome_test.go

@@ -0,0 +1,169 @@
+package proxy
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"net/http"
+	"net/url"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/e2b-dev/infra/packages/shared/pkg/logger"
+	"github.com/e2b-dev/infra/packages/shared/pkg/proxy/pool"
+)
+
+// outcomeRecorder collects the outcomes the proxy reports, concurrency-safe.
+type outcomeRecorder struct {
+	mu       sync.Mutex
+	outcomes []pool.RequestOutcome
+	nodes    []string
+}
+
+func (o *outcomeRecorder) add(_ context.Context, outcome pool.RequestOutcome, node string) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	o.outcomes = append(o.outcomes, outcome)
+	o.nodes = append(o.nodes, node)
+}
+
+func (o *outcomeRecorder) last() (pool.RequestOutcome, bool) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	if len(o.outcomes) == 0 {
+		return "", false
+	}
+
+	return o.outcomes[len(o.outcomes)-1], true
+}
+
+func (o *outcomeRecorder) lastNode() (string, bool) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	if len(o.nodes) == 0 {
+		return "", false
+	}
+
+	return o.nodes[len(o.nodes)-1], true
+}
+
+// startOutcomeProxy stands up a proxy whose destination resolver returns
+// resolve(), recording every request outcome into rec. Returns the proxy URL.
+func startOutcomeProxy(t *testing.T, rec *outcomeRecorder, resolve func() (*pool.Destination, error)) string {
+	t.Helper()
+
+	var lisCfg net.ListenConfig
+	proxyListener, err := lisCfg.Listen(t.Context(), "tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	proxyAddr := proxyListener.Addr().(*net.TCPAddr)
+	proxyPort := uint16(proxyAddr.Port) //nolint:gosec // test-only port fits
+
+	proxy := New(proxyPort, 1, time.Second, func(_ *http.Request) (*pool.Destination, error) {
+		return resolve()
+	}, nil, false, rec.add)
+
+	go func() {
+		serveErr := proxy.Serve(proxyListener)
+		assert.ErrorIs(t, serveErr, http.ErrServerClosed)
+	}()
+	t.Cleanup(func() { assert.NoError(t, proxy.Close()) })
+
+	return fmt.Sprintf("http://%s", proxyAddr.String())
+}
+
+func TestProxyRecordsOutcome_OK(t *testing.T) {
+	t.Parallel()
+
+	var lisCfg net.ListenConfig
+	backendListener, err := lisCfg.Listen(t.Context(), "tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	backendPort := uint64(backendListener.Addr().(*net.TCPAddr).Port) //nolint:gosec // test-only
+	backendURL, err := url.Parse(fmt.Sprintf("http://%s", backendListener.Addr().String()))
+	require.NoError(t, err)
+
+	backend := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	})}
+	go func() { assert.ErrorIs(t, backend.Serve(backendListener), http.ErrServerClosed) }()
+	t.Cleanup(func() { assert.NoError(t, backend.Close()) })
+
+	rec := &outcomeRecorder{}
+	proxyURL := startOutcomeProxy(t, rec, func() (*pool.Destination, error) {
+		return &pool.Destination{
+			Url:           backendURL,
+			SandboxId:     "sbx-ok",
+			SandboxPort:   backendPort,
+			ConnectionKey: "ok",
+			RequestLogger: logger.L(),
+			Node:          "node-42",
+		}, nil
+	})
+
+	resp, err := http.Get(proxyURL) //nolint:noctx // test
+	require.NoError(t, err)
+	_ = resp.Body.Close()
+
+	require.Eventually(t, func() bool {
+		o, ok := rec.last()
+
+		return ok && o == pool.OutcomeOK
+	}, 2*time.Second, 10*time.Millisecond, "expected an ok outcome")
+
+	// The destination's node is carried through to the recorder.
+	node, ok := rec.lastNode()
+	require.True(t, ok)
+	assert.Equal(t, "node-42", node)
+}
+
+func TestProxyRecordsOutcome_UpstreamUnreachable(t *testing.T) {
+	t.Parallel()
+
+	// Port 1 is reliably closed: the dial fails, so the request is
+	// upstream-unreachable, not a routing failure.
+	deadURL, err := url.Parse("http://127.0.0.1:1")
+	require.NoError(t, err)
+
+	rec := &outcomeRecorder{}
+	proxyURL := startOutcomeProxy(t, rec, func() (*pool.Destination, error) {
+		return &pool.Destination{
+			Url:           deadURL,
+			SandboxId:     "sbx-dead",
+			SandboxPort:   1,
+			ConnectionKey: "dead",
+			RequestLogger: logger.L(),
+		}, nil
+	})
+
+	resp, err := http.Get(proxyURL) //nolint:noctx // test
+	require.NoError(t, err)
+	_ = resp.Body.Close()
+
+	require.Eventually(t, func() bool {
+		o, ok := rec.last()
+
+		return ok && o == pool.OutcomeUpstreamUnreachable
+	}, 2*time.Second, 10*time.Millisecond, "expected an upstream_unreachable outcome")
+}
+
+func TestProxyRecordsOutcome_SandboxNotFound(t *testing.T) {
+	t.Parallel()
+
+	rec := &outcomeRecorder{}
+	proxyURL := startOutcomeProxy(t, rec, func() (*pool.Destination, error) {
+		return nil, NewErrSandboxNotFound("sbx-missing")
+	})
+
+	resp, err := http.Get(proxyURL) //nolint:noctx // test
+	require.NoError(t, err)
+	_ = resp.Body.Close()
+
+	require.Eventually(t, func() bool {
+		o, ok := rec.last()
+
+		return ok && o == pool.OutcomeSandboxNotFound
+	}, 2*time.Second, 10*time.Millisecond, "expected a sandbox_not_found outcome")
+}