Fix backpressure metric capturing for Kubernetes API calls

Author: jwebster7

frontend/csi/interceptor.go

@@ -83,8 +83,8 @@ func incomingRequestMetricsInterceptor(
 	return handler(ctx, req)
 }
 
-// timeoutInterceptor applies a default request timeout for Node and AllInOne deployments.
-// All RPCs served by these roles receive the timeout; Controller deployments are unaffected.
+// timeoutInterceptor applies a default request timeout for Node deployments.
+// All RPCs served by the Node role receive the timeout; Controller and AIO deployments are unaffected.
 // It must be the second from outermost interceptor in the chain so that downstream interceptors
 // capture the timeout-aware context.
 func timeoutInterceptor(

logging/transport.go

@@ -43,45 +43,56 @@ func NewMetricsTransport(
 
 // RoundTrip captures the time on the wire of a request. It does not capture the time
 // an HTTP client may be waiting for a lock or token.
-//
-// err serves double duty: the deferred recorder observes it for metrics,
-// while the Kubernetes backpressure path returns nil to the caller without
-// clearing err, so backpressure is still recorded without violating
-// RoundTripper semantics.
-func (m *MetricsTransport) RoundTrip(req *http.Request) (res *http.Response, err error) {
+func (m *MetricsTransport) RoundTrip(req *http.Request) (*http.Response, error) {
 	// target: "kubernetes", "ontap", etc.,
 	// host:   "10.0.0.1", "10.0.0.2", etc.,
 	// method: "GET", "POST", etc.
+	// recErr is a separate local variable used by the deferred recorder so that the
+	// Kubernetes backpressure path can return nil to the caller (RoundTripper semantics)
+	// while still recording the backpressure error in metrics.
+	var recErr error
 	ctx, rec := NewContextBuilder(req.Context()).
 		WithOutgoingAPIMetrics(m.target, req.URL.Host, req.Method).
 		BuildContextAndTelemetry()
-	defer rec(&err)
+	defer rec(&recErr)
 
-	res, err = m.base.RoundTrip(req.WithContext(ctx))
+	res, err := m.base.RoundTrip(req.WithContext(ctx))
 	if err != nil {
-		// Handle the special case where ONTAP or any other API doesn't honor HTTP status codes.
-		// This is a workaround for ONTAP's EOF handling when the API gives up.
+		recErr = err
+		// Handle the special ONTAP case where the API may return EOF instead of
+		// an HTTP backpressure status when the server gives up.
 		if errors.Is(err, io.EOF) && m.target == ContextRequestTargetONTAP {
-			err = errors.WrapWithServerBackPressureError(err, "received EOF from server")
+			recErr = errors.WrapWithServerBackPressureError(err, "received EOF from server")
 		}
-		return res, err
+		return res, recErr
 	}
 
 	// Check for HTTP-level backpressure.
 	switch res.StatusCode {
 	case http.StatusTooManyRequests, http.StatusServiceUnavailable, http.StatusGatewayTimeout:
-		err = errors.ServerBackPressureError("received status: %d from the server", res.StatusCode)
+		// Set the recorder error to a backpressure error.
+		// This should be set for all outgoing API requests that experience backpressure.
+		// Some outgoing API requests handle backpressure differently, so switch on the target
+		// to decide what to return.
+		recErr = errors.ServerBackPressureError("received status: %d from the server", res.StatusCode)
 
-		// ONTAP callers (LimitedRetryTransport, go-openapi, azgo) gate on err != nil
-		// and do not inspect the response status code on the error path.
-		if m.target == ContextRequestTargetONTAP {
-			return res, err
+		switch m.target {
+		case ContextRequestTargetONTAP:
+			// This breaks the standard RoundTripper semantics but is necessary to handle
+			// the special case where ONTAP returns an EOF error when the API is too busy.
+			// ONTAP callers gate on err != nil and may not inspect the HTTP status code,
+			// so return a non-nil backpressure error.
+			return res, recErr
+		case ContextRequestTargetKubernetes:
+			// Kubernetes client-go depends on the response status, not the error, but
+			// the metrics telemetry still needs to record the backpressure error.
+			// Therefore, handle this case explicitly by returning nil to the caller
+			// and recording the backpressure error in metrics.
+			return res, nil
 		}
-
-		// Kubernetes client-go depends on the response status, not the error.
-		// Return the response with a nil error so client-go can inspect the status code.
-		return res, nil
 	}
 
+	// Preserve standard RoundTripper semantics: return the response with nil error
+	// and let callers inspect the status code. recErr is captured independently for metrics.
 	return res, nil
 }

logging/transport_test.go

@@ -7,6 +7,7 @@ import (
 	stdErrors "errors"
 	"io"
 	"net/http"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -33,12 +34,21 @@ func makeResponse(statusCode int) *http.Response {
 	return &http.Response{StatusCode: statusCode}
 }
 
-// ---- ONTAP EOF handling ----
+// roundTrip is a shorthand that constructs a MetricsTransport from a stub and calls RoundTrip.
+// It uses a unique host derived from the test name so Prometheus counters never collide across tests,
+// while preserving the real target value for code-path correctness.
+func roundTrip(
+	t *testing.T, target ContextRequestTarget, base *stubRoundTripper, method string,
+) (*http.Response, error, string, string, string) {
+	t.Helper()
+	host := strings.ReplaceAll(t.Name(), "/", "_")
+	mt := &MetricsTransport{base: base, target: target}
+	req := makeRequest(context.Background(), method, "https://"+host+"/path")
+	resp, err := mt.RoundTrip(req)
+	return resp, err, string(target), host, method
+}
 
 func TestMetricsTransport_ONTAP_EOF_WrapsAsServerBackPressure(t *testing.T) {
-	// ONTAP signals overload with a bare EOF instead of an HTTP status code.
-	// MetricsTransport must wrap it as a ServerBackPressureError so the telemeter
-	// increments the backpressure counter correctly.
 	base := &stubRoundTripper{resp: nil, err: io.EOF}
 	mt := NewMetricsTransport(base, WithMetricsTransportTarget(ContextRequestTargetONTAP))
 	req := makeRequest(context.Background(), http.MethodGet, "https://ontap.local/cluster")
@@ -51,8 +61,6 @@ func TestMetricsTransport_ONTAP_EOF_WrapsAsServerBackPressure(t *testing.T) {
 }
 
 func TestMetricsTransport_ONTAP_EOF_PreservesEOFInChain(t *testing.T) {
-	// LimitedRetryTransport uses errors.Is(err, io.EOF) to decide whether to retry.
-	// The wrapping must preserve io.EOF in the chain so retries still trigger.
 	base := &stubRoundTripper{resp: nil, err: io.EOF}
 	mt := NewMetricsTransport(base, WithMetricsTransportTarget(ContextRequestTargetONTAP))
 	req := makeRequest(context.Background(), http.MethodGet, "https://ontap.local/cluster")
@@ -64,7 +72,6 @@ func TestMetricsTransport_ONTAP_EOF_PreservesEOFInChain(t *testing.T) {
 }
 
 func TestMetricsTransport_NonONTAP_EOF_PassesThrough(t *testing.T) {
-	// Only ONTAP gets EOF-to-backpressure wrapping; other targets pass EOF through unchanged.
 	base := &stubRoundTripper{resp: nil, err: io.EOF}
 	mt := NewMetricsTransport(base, WithMetricsTransportTarget(ContextRequestTargetKubernetes))
 	req := makeRequest(context.Background(), http.MethodGet, "https://k8s.local/api/v1/pods")
@@ -166,10 +173,7 @@ func TestMetricsTransport_HTTP200_Success(t *testing.T) {
 	assert.NoError(t, gotErr, "HTTP 200 should return no error")
 }
 
-// ---- Non-backpressure transport errors pass through unchanged ----
-
 func TestMetricsTransport_TransportError_PassesThrough(t *testing.T) {
-	// Connection-level errors that are not EOF pass through as-is for all targets.
 	someErr := stdErrors.New("connection refused")
 	base := &stubRoundTripper{resp: nil, err: someErr}
 	mt := NewMetricsTransport(base, WithMetricsTransportTarget(ContextRequestTargetUnknown))
@@ -181,3 +185,195 @@ func TestMetricsTransport_TransportError_PassesThrough(t *testing.T) {
 		"non-EOF transport errors should be returned unchanged")
 	assert.False(t, utilsErrors.IsServerBackPressureError(gotErr))
 }
+
+func TestMetricsTransport_ONTAP_HTTP429_RecordsBackpressureMetric(t *testing.T) {
+	_, _, target, host, method := roundTrip(
+		t, ContextRequestTargetONTAP,
+		&stubRoundTripper{resp: makeResponse(http.StatusTooManyRequests)},
+		http.MethodGet,
+	)
+
+	assert.Equal(t, float64(1),
+		readCounter(outgoingAPIRequestBackpressureTotal, target, host, method),
+		"backpressure counter must increment for ONTAP 429",
+	)
+}
+
+func TestMetricsTransport_Kubernetes_HTTP429_RecordsBackpressureMetric(t *testing.T) {
+	resp, err, target, host, method := roundTrip(
+		t, ContextRequestTargetKubernetes,
+		&stubRoundTripper{resp: makeResponse(http.StatusTooManyRequests)},
+		http.MethodGet,
+	)
+
+	assert.NoError(t, err, "caller must get nil error per RoundTripper semantics")
+	assert.Equal(t, http.StatusTooManyRequests, resp.StatusCode)
+	assert.Equal(t, float64(1),
+		readCounter(outgoingAPIRequestBackpressureTotal, target, host, method),
+		"backpressure counter must increment for Kubernetes 429 even though caller gets nil error",
+	)
+}
+
+func TestMetricsTransport_Kubernetes_HTTP503_RecordsBackpressureMetric(t *testing.T) {
+	resp, err, target, host, method := roundTrip(
+		t, ContextRequestTargetKubernetes,
+		&stubRoundTripper{resp: makeResponse(http.StatusServiceUnavailable)},
+		http.MethodGet,
+	)
+
+	assert.NoError(t, err)
+	assert.Equal(t, http.StatusServiceUnavailable, resp.StatusCode)
+	assert.Equal(t, float64(1),
+		readCounter(outgoingAPIRequestBackpressureTotal, target, host, method),
+		"backpressure counter must increment for Kubernetes 503",
+	)
+}
+
+func TestMetricsTransport_Kubernetes_HTTP504_RecordsBackpressureMetric(t *testing.T) {
+	resp, err, target, host, method := roundTrip(
+		t, ContextRequestTargetKubernetes,
+		&stubRoundTripper{resp: makeResponse(http.StatusGatewayTimeout)},
+		http.MethodGet,
+	)
+
+	assert.NoError(t, err)
+	assert.Equal(t, http.StatusGatewayTimeout, resp.StatusCode)
+	assert.Equal(t, float64(1),
+		readCounter(outgoingAPIRequestBackpressureTotal, target, host, method),
+		"backpressure counter must increment for Kubernetes 504",
+	)
+}
+
+func TestMetricsTransport_Unknown_HTTP429_ReturnsNilError(t *testing.T) {
+	resp, err, target, host, method := roundTrip(
+		t, ContextRequestTargetUnknown,
+		&stubRoundTripper{resp: makeResponse(http.StatusTooManyRequests)},
+		http.MethodGet,
+	)
+
+	assert.NoError(t, err,
+		"unknown targets must return nil error to preserve RoundTripper semantics")
+	assert.NotNil(t, resp)
+	assert.Equal(t, http.StatusTooManyRequests, resp.StatusCode)
+	assert.Equal(t, float64(1),
+		readCounter(outgoingAPIRequestBackpressureTotal, target, host, method),
+		"backpressure counter must still increment for unknown target 429",
+	)
+}
+
+func TestMetricsTransport_ONTAP_EOF_RecordsBackpressureMetric(t *testing.T) {
+	_, _, target, host, method := roundTrip(
+		t, ContextRequestTargetONTAP,
+		&stubRoundTripper{err: io.EOF},
+		http.MethodGet,
+	)
+
+	assert.Equal(t, float64(1),
+		readCounter(outgoingAPIRequestBackpressureTotal, target, host, method),
+		"backpressure counter must increment for ONTAP EOF",
+	)
+}
+
+func TestMetricsTransport_HTTP200_NoBackpressureMetric(t *testing.T) {
+	_, _, target, host, method := roundTrip(
+		t, ContextRequestTargetONTAP,
+		&stubRoundTripper{resp: makeResponse(http.StatusOK)},
+		http.MethodGet,
+	)
+
+	assert.Equal(t, float64(0),
+		readCounter(outgoingAPIRequestBackpressureTotal, target, host, method),
+		"backpressure counter must not increment for 200 OK",
+	)
+}
+
+func TestMetricsTransport_HTTP200_RecordsDurationAsSuccess(t *testing.T) {
+	_, _, target, host, method := roundTrip(
+		t, ContextRequestTargetONTAP,
+		&stubRoundTripper{resp: makeResponse(http.StatusOK)},
+		http.MethodGet,
+	)
+
+	assert.Equal(t, uint64(1),
+		readHistogramCount(outgoingAPIRequestDurationSeconds, metricStatusSuccess, target, host, method),
+	)
+}
+
+func TestMetricsTransport_ONTAP_HTTP429_RecordsDurationAsFailure(t *testing.T) {
+	_, _, target, host, method := roundTrip(
+		t, ContextRequestTargetONTAP,
+		&stubRoundTripper{resp: makeResponse(http.StatusTooManyRequests)},
+		http.MethodGet,
+	)
+
+	assert.Equal(t, uint64(1),
+		readHistogramCount(outgoingAPIRequestDurationSeconds, metricStatusFailure, target, host, method),
+		"ONTAP 429 must be recorded as failure in the duration histogram",
+	)
+}
+
+func TestMetricsTransport_Kubernetes_HTTP429_RecordsDurationAsFailure(t *testing.T) {
+	_, err, target, host, method := roundTrip(
+		t, ContextRequestTargetKubernetes,
+		&stubRoundTripper{resp: makeResponse(http.StatusTooManyRequests)},
+		http.MethodGet,
+	)
+
+	assert.NoError(t, err, "caller gets nil")
+	assert.Equal(t, uint64(1),
+		readHistogramCount(outgoingAPIRequestDurationSeconds, metricStatusFailure, target, host, method),
+		"Kubernetes 429 must be recorded as failure in the duration histogram even though caller gets nil",
+	)
+}
+
+func TestMetricsTransport_TransportError_RecordsDurationAsFailure(t *testing.T) {
+	_, _, target, host, method := roundTrip(
+		t, ContextRequestTargetUnknown,
+		&stubRoundTripper{err: stdErrors.New("connection refused")},
+		http.MethodGet,
+	)
+
+	assert.Equal(t, uint64(1),
+		readHistogramCount(outgoingAPIRequestDurationSeconds, metricStatusFailure, target, host, method),
+		"transport errors must be recorded as failure",
+	)
+}
+
+func TestMetricsTransport_InFlight_ReturnsToZeroAfterSuccess(t *testing.T) {
+	_, _, target, host, method := roundTrip(
+		t, ContextRequestTargetONTAP,
+		&stubRoundTripper{resp: makeResponse(http.StatusOK)},
+		http.MethodGet,
+	)
+
+	assert.Equal(t, float64(0),
+		readGauge(outgoingAPIRequestsInFlight, target, host, method),
+		"in-flight gauge must return to zero after a completed request",
+	)
+}
+
+func TestMetricsTransport_InFlight_ReturnsToZeroAfterError(t *testing.T) {
+	_, _, target, host, method := roundTrip(
+		t, ContextRequestTargetUnknown,
+		&stubRoundTripper{err: stdErrors.New("fail")},
+		http.MethodGet,
+	)
+
+	assert.Equal(t, float64(0),
+		readGauge(outgoingAPIRequestsInFlight, target, host, method),
+		"in-flight gauge must return to zero after an error",
+	)
+}
+
+func TestMetricsTransport_InFlight_ReturnsToZeroAfterKubernetesBackpressure(t *testing.T) {
+	_, _, target, host, method := roundTrip(
+		t, ContextRequestTargetKubernetes,
+		&stubRoundTripper{resp: makeResponse(http.StatusTooManyRequests)},
+		http.MethodGet,
+	)
+
+	assert.Equal(t, float64(0),
+		readGauge(outgoingAPIRequestsInFlight, target, host, method),
+		"in-flight gauge must return to zero after Kubernetes backpressure",
+	)
+}