Add support for Thrift and HTTP client-side fault injection.

faults/common.go

@@ -0,0 +1,140 @@
+// Package faults provides common headers and client-side fault injection
+// functionality.
+package faults
+
+import (
+	"fmt"
+	"log/slog"
+	"math/rand/v2"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// GetHeaderFn is the function type to return the value of a protocol-specific
+// header with the given key.
+type GetHeaderFn func(key string) string
+
+// ResumeFn is the function type to continue processing the protocol-specific
+// request without injecting a fault.
+type ResumeFn[T any] func() (T, error)
+
+// ResponseFn is the function type to inject a protocol-specific fault with the
+// given code and message.
+type ResponseFn[T any] func(code int, message string) (T, error)
+
+// SleepFn is the function type to sleep for the given duration. It is only
+// exposed for testing purposes.
+type SleepFn func(d time.Duration)
+
+// Object to ensure a random number is only generated at most 1 time.
+type randSingleton struct {
+	randInt *int
+}
+
+func (r randSingleton) getRandInt() int {
+	if r.randInt == nil {
+		*r.randInt = rand.IntN(100)
+	}
+	return *r.randInt
+}
+
+func isSelected(percentageHeader string, GetHeaderFn func(key string) string, singleRand randSingleton) (bool, error) {
+	percentageStr := GetHeaderFn(percentageHeader)
+	if percentageStr == "" {
+		return true, nil
+	}
+	percentage, err := strconv.Atoi(percentageStr)
+	if err != nil {
+		return false, fmt.Errorf("provided percentage %q is not a valid integer: %w", percentageStr, err)
+	}
+	if percentage < 0 || percentage > 100 {
+		return false, fmt.Errorf("provided percentage %q is outside the valid range of [0-100]", percentage)
+	}
+	return singleRand.getRandInt() < percentage, nil
+}
+
+type InjectFaultParams[T any] struct {
+	CallerName string
+
+	Address, Method            string
+	AbortCodeMin, AbortCodeMax int
+
+	GetHeaderFn GetHeaderFn
+	ResumeFn    ResumeFn[T]
+	ResponseFn  ResponseFn[T]
+
+	// Exposed for tests.
+	RandInt *int
+	SleepFn *SleepFn
+}
+
+func InjectFault[T any](params InjectFaultParams[T]) (T, error) {
+	serverAddress := params.GetHeaderFn(FaultServerAddressHeader)
+
+	// The short address should just be <service>.<namespace>, without the
+	// local cluster suffix or port. Non-cluster-local addresses are not
+	// shortened.
+	shortAddress := params.Address
+	if i := strings.Index(params.Address, ".svc.cluster.local"); i != -1 {
+		shortAddress = params.Address[:i]
+	}
+	if serverAddress == "" || serverAddress != shortAddress {
+		return params.ResumeFn()
+	}
+
+	serverMethod := params.GetHeaderFn(FaultServerMethodHeader)
+	if serverMethod != "" && serverMethod != params.Method {
+		return params.ResumeFn()
+	}
+
+	singleRand := randSingleton{
+		randInt: params.RandInt,
+	}
+
+	delayMs := params.GetHeaderFn(FaultDelayMsHeader)
+	if delayMs != "" {
+		if selected, err := isSelected(FaultDelayPercentageHeader, params.GetHeaderFn, singleRand); !selected {
+			if err != nil {
+				slog.Warn(fmt.Sprintf("%s: %v", params.CallerName, err))
+			}
+			return params.ResumeFn()
+		}
+
+		delay, err := strconv.Atoi(delayMs)
+		if err != nil {
+			slog.Warn(fmt.Sprintf("%s: provided delay \"%s\" is not a valid integer", params.CallerName, delayMs))
+			return params.ResumeFn()
+		}
+
+		sleepFn := time.Sleep
+		if params.SleepFn != nil {
+			sleepFn = *params.SleepFn
+		}
+		sleepFn(time.Duration(delay) * time.Millisecond)
+	}
+
+	abortCode := params.GetHeaderFn(FaultAbortCodeHeader)
+	if abortCode != "" {
+		if selected, err := isSelected(FaultAbortPercentageHeader, params.GetHeaderFn, singleRand); !selected {
+			if err != nil {
+				slog.Warn(fmt.Sprintf("%s: %v", params.CallerName, err))
+			}
+			return params.ResumeFn()
+		}
+
+		code, err := strconv.Atoi(abortCode)
+		if err != nil {
+			slog.Warn(fmt.Sprintf("%s: provided abort code \"%s\" is not a valid integer", params.CallerName, abortCode))
+			return params.ResumeFn()
+		}
+		if code < params.AbortCodeMin || code > params.AbortCodeMax {
+			slog.Warn(fmt.Sprintf("%s: provided abort code \"%d\" is outside of the valid range", params.CallerName, code))
+			return params.ResumeFn()
+		}
+		abortMessage := params.GetHeaderFn(FaultAbortMessageHeader)
+		return params.ResponseFn(code, abortMessage)
+	}
+
+	return params.ResumeFn()
+}

faults/headers.go

@@ -0,0 +1,11 @@
+package faults
+
+const (
+	FaultServerAddressHeader   = "X-Bp-Fault-Server-Address"
+	FaultServerMethodHeader    = "X-Bp-Fault-Server-Method"
+	FaultDelayMsHeader         = "X-Bp-Fault-Delay-Ms"
+	FaultDelayPercentageHeader = "X-Bp-Fault-Delay-Percentage"
+	FaultAbortCodeHeader       = "X-Bp-Fault-Abort-Code"
+	FaultAbortMessageHeader    = "X-Bp-Fault-Abort-Message"
+	FaultAbortPercentageHeader = "X-Bp-Fault-Abort-Percentage"
+)

httpbp/client_middlewares.go

@@ -7,6 +7,7 @@ import (
 	"log/slog"
 	"net/http"
 	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -15,6 +16,8 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/reddit/baseplate.go/breakerbp"
+	"github.com/reddit/baseplate.go/faults"
+
 	//lint:ignore SA1019 This library is internal only, not actually deprecated
 	"github.com/reddit/baseplate.go/internalv2compat"
 	"github.com/reddit/baseplate.go/retrybp"
@@ -43,6 +46,8 @@ func (f roundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) {
 // plus any additional client middleware passed into this function. Default
 // middlewares are:
 //
+// * FaultInjection
+//
 // * MonitorClient with transport.WithRetrySlugSuffix
 //
 // * PrometheusClientMetrics with transport.WithRetrySlugSuffix
@@ -76,6 +81,7 @@ func NewClient(config ClientConfig, middleware ...ClientMiddleware) (*http.Clien
 	}
 
 	defaults := []ClientMiddleware{
+		FaultInjection(),
 		MonitorClient(config.Slug + transport.WithRetrySlugSuffix),
 		PrometheusClientMetrics(config.Slug + transport.WithRetrySlugSuffix),
 		Retries(config.MaxErrorReadAhead, config.RetryOptions...),
@@ -349,3 +355,48 @@ func PrometheusClientMetrics(serverSlug string) ClientMiddleware {
 		})
 	}
 }
+
+type ServiceAddressParts struct {
+	Name      string
+	Namespace string
+}
+
+func FaultInjection() ClientMiddleware {
+	return func(next http.RoundTripper) http.RoundTripper {
+		return roundTripperFunc(func(req *http.Request) (*http.Response, error) {
+			var resumeFn faults.ResumeFn[*http.Response] = func() (*http.Response, error) {
+				return next.RoundTrip(req)
+			}
+			var responseFn faults.ResponseFn[*http.Response] = func(code int, message string) (*http.Response, error) {
+				return &http.Response{
+					Status:     http.StatusText(code),
+					StatusCode: code,
+					Proto:      req.Proto,
+					ProtoMajor: req.ProtoMajor,
+					ProtoMinor: req.ProtoMinor,
+					Header: map[string][]string{
+						// Copied from the standard http.Error() function.
+						"Content-Type":           {"text/plain; charset=utf-8"},
+						"X-Content-Type-Options": {"nosniff"},
+					},
+					ContentLength:    0,
+					TransferEncoding: req.TransferEncoding,
+					Request:          req,
+					TLS:              req.TLS,
+				}, nil
+			}
+
+			resp, err := faults.InjectFault(faults.InjectFaultParams[*http.Response]{
+				CallerName:   "httpbp.FaultInjection",
+				Address:      req.URL.Host,
+				Method:       strings.TrimPrefix(req.URL.Path, "/"),
+				AbortCodeMin: 400,
+				AbortCodeMax: 599,
+				GetHeaderFn:  faults.GetHeaderFn(req.Header.Get),
+				ResumeFn:     resumeFn,
+				ResponseFn:   responseFn,
+			})
+			return resp, err
+		})
+	}
+}

httpbp/client_middlewares_test.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"net/url"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -16,6 +17,7 @@ import (
 	"github.com/sony/gobreaker"
 
 	"github.com/reddit/baseplate.go/breakerbp"
+	"github.com/reddit/baseplate.go/faults"
 )
 
 func TestNewClient(t *testing.T) {
@@ -395,3 +397,136 @@ func TestCircuitBreaker(t *testing.T) {
 		t.Errorf("Expected the third request to return %v, got %v", gobreaker.ErrOpenState, err)
 	}
 }
+
+func TestFaultInjection(t *testing.T) {
+	testCases := []struct {
+		name                       string
+		faultServerAddrMatch       bool
+		faultServerMethodHeader    string
+		faultDelayMsHeader         string
+		faultDelayPercentageHeader string
+		faultAbortCodeHeader       string
+		faultAbortMessageHeader    string
+		faultAbortPercentageHeader string
+
+		wantResp *http.Response
+	}{
+		{
+			name: "no fault specified",
+			wantResp: &http.Response{
+				StatusCode: http.StatusOK,
+			},
+		},
+		{
+			name: "abort",
+
+			faultServerAddrMatch:    true,
+			faultServerMethodHeader: "testMethod",
+			faultAbortCodeHeader:    "500",
+
+			wantResp: &http.Response{
+				StatusCode: http.StatusInternalServerError,
+			},
+		},
+		{
+			name: "service does not match",
+
+			faultServerAddrMatch:    false,
+			faultServerMethodHeader: "testMethod",
+			faultAbortCodeHeader:    "500",
+
+			wantResp: &http.Response{
+				StatusCode: http.StatusOK,
+			},
+		},
+		{
+			name: "method does not match",
+
+			faultServerAddrMatch:    true,
+			faultServerMethodHeader: "fooMethod",
+			faultAbortCodeHeader:    "500",
+
+			wantResp: &http.Response{
+				StatusCode: http.StatusOK,
+			},
+		},
+		{
+			name: "less than min abort code",
+
+			faultServerAddrMatch:    true,
+			faultServerMethodHeader: "testMethod",
+			faultAbortCodeHeader:    "99",
+
+			wantResp: &http.Response{
+				StatusCode: http.StatusOK,
+			},
+		},
+		{
+			name: "greater than max abort code",
+
+			faultServerAddrMatch:    true,
+			faultServerMethodHeader: "testMethod",
+			faultAbortCodeHeader:    "600",
+
+			wantResp: &http.Response{
+				StatusCode: http.StatusOK,
+			},
+		},
+	}
+
+	for _, tt := range testCases {
+		t.Run(tt.name, func(t *testing.T) {
+
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				io.WriteString(w, "Success!")
+			}))
+			defer server.Close()
+
+			client, err := NewClient(ClientConfig{
+				Slug: "test",
+			})
+			if err != nil {
+				t.Fatalf("NewClient returned error: %v", err)
+			}
+
+			req, err := http.NewRequest("GET", server.URL+"/testMethod", nil)
+			if err != nil {
+				t.Fatalf("unexpected error when creating request: %v", err)
+			}
+
+			if tt.faultServerAddrMatch {
+				// We can't set a specific address here because the middleware
+				// relies on the DNS address, which is not customizeable when making
+				// real requests to a local HTTP test server.
+				req.Header.Set(faults.FaultServerAddressHeader, strings.TrimPrefix(server.URL, "http://"))
+			}
+			if tt.faultServerMethodHeader != "" {
+				req.Header.Set(faults.FaultServerMethodHeader, tt.faultServerMethodHeader)
+			}
+			if tt.faultDelayMsHeader != "" {
+				req.Header.Set(faults.FaultDelayMsHeader, tt.faultDelayMsHeader)
+			}
+			if tt.faultDelayPercentageHeader != "" {
+				req.Header.Set(faults.FaultDelayPercentageHeader, tt.faultDelayPercentageHeader)
+			}
+			if tt.faultAbortCodeHeader != "" {
+				req.Header.Set(faults.FaultAbortCodeHeader, tt.faultAbortCodeHeader)
+			}
+			if tt.faultAbortMessageHeader != "" {
+				req.Header.Set(faults.FaultAbortMessageHeader, tt.faultAbortMessageHeader)
+			}
+			if tt.faultAbortPercentageHeader != "" {
+				req.Header.Set(faults.FaultAbortPercentageHeader, tt.faultAbortPercentageHeader)
+			}
+
+			resp, err := client.Do(req)
+
+			if err != nil {
+				t.Fatalf("expected no error, got %v", err)
+			}
+			if tt.wantResp.StatusCode != resp.StatusCode {
+				t.Fatalf("expected response code %v, got %v", tt.wantResp.StatusCode, resp.StatusCode)
+			}
+		})
+	}
+}