feat: circuit breaker implementation

router-tests/go.mod

@@ -5,6 +5,7 @@ go 1.23.0
 require (
 	github.com/MicahParks/jwkset v0.9.0
 	github.com/buger/jsonparser v1.1.1
+	github.com/caarlos0/env/v11 v11.3.1
 	github.com/cloudflare/backoff v0.0.0-20240920015135-e46b80a3a7d0
 	github.com/golang-jwt/jwt/v5 v5.2.2
 	github.com/google/uuid v1.6.0
@@ -48,8 +49,8 @@ require (
 	github.com/benbjohnson/clock v1.3.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bufbuild/protocompile v0.14.1 // indirect
-	github.com/caarlos0/env/v11 v11.3.1 // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
+	github.com/cep21/circuit/v4 v4.0.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cilium/ebpf v0.16.0 // indirect
 	github.com/coder/websocket v1.8.13 // indirect

router-tests/go.sum

@@ -41,6 +41,8 @@ github.com/caarlos0/env/v11 v11.3.1 h1:cArPWC15hWmEt+gWk7YBi7lEXTXCvpaSdCiZE2X5m
 github.com/caarlos0/env/v11 v11.3.1/go.mod h1:qupehSf/Y0TUTsxKywqRt/vJjN5nz6vauiYEUUr8P4U=
 github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=
 github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
+github.com/cep21/circuit/v4 v4.0.0 h1:g1AzMmRLuwCst0eccy1nGsD/CL2XKbDnaPUHVHDvVmI=
+github.com/cep21/circuit/v4 v4.0.0/go.mod h1:Bb1fHpuiRu+AIgKf7DTM1c5U94qTZtKouKcDwtZYCXk=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cilium/ebpf v0.16.0 h1:+BiEnHL6Z7lXnlGUsXQPPAE7+kenAd4ES8MQ5min0Ok=

router/core/graph_server.go

@@ -5,6 +5,8 @@ import (
 	"crypto/ecdsa"
 	"errors"
 	"fmt"
+	"github.com/wundergraph/cosmo/router/internal/circuit"
+	"github.com/wundergraph/cosmo/router/internal/traceclient"
 	"net/http"
 	"net/url"
 	"path/filepath"
@@ -82,17 +84,18 @@ type (
 		mux                     *chi.Mux
 		// inFlightRequests is used to track the number of requests currently being processed
 		// does not include websocket (hijacked) connections
-		inFlightRequests        *atomic.Uint64
-		graphMuxList            []*graphMux
-		graphMuxListLock        sync.Mutex
-		runtimeMetrics          *rmetric.RuntimeMetrics
-		otlpEngineMetrics       *rmetric.EngineMetrics
-		prometheusEngineMetrics *rmetric.EngineMetrics
-		connectionMetrics       *rmetric.ConnectionMetrics
-		instanceData            InstanceData
-		pubSubProviders         []datasource.Provider
-		traceDialer             *TraceDialer
-		pluginHost              *routerplugin.Host
+		inFlightRequests              *atomic.Uint64
+		graphMuxList                  []*graphMux
+		graphMuxListLock              sync.Mutex
+		runtimeMetrics                *rmetric.RuntimeMetrics
+		otlpEngineMetrics             *rmetric.EngineMetrics
+		prometheusEngineMetrics       *rmetric.EngineMetrics
+		connectionMetrics             *rmetric.ConnectionMetrics
+		instanceData                  InstanceData
+		pubSubProviders               []datasource.Provider
+		traceDialer                   *TraceDialer
+		pluginHost                    *routerplugin.Host
+		subgraphCircuitBreakerOptions *SubgraphCircuitBreakerOptions
 	}
 )
 
@@ -141,7 +144,8 @@ func newGraphServer(ctx context.Context, r *Router, routerConfig *nodev1.RouterC
 			HostName:      r.hostName,
 			ListenAddress: r.listenAddr,
 		},
-		storageProviders: &r.storageProviders,
+		storageProviders:              &r.storageProviders,
+		subgraphCircuitBreakerOptions: r.subgraphCircuitBreakerOptions,
 	}
 
 	baseOtelAttributes := []attribute.KeyValue{
@@ -753,6 +757,21 @@ func (s *graphServer) buildGraphMux(ctx context.Context,
 		gm.metricStore = m
 	}
 
+	// Feature flags can fail independently of the base subgraph, thus we should have circuit breakers per feature flag + subgraph
+	managerOpts := circuit.ManagerOpts{}
+	if s.subgraphCircuitBreakerOptions != nil {
+		managerOpts = circuit.ManagerOpts{
+			BaseConfig:              s.subgraphCircuitBreakerOptions.CircuitBreaker,
+			SubgraphCircuitBreakers: s.subgraphCircuitBreakerOptions.SubgraphMap,
+			Subgraphs:               configSubgraphs,
+			FeatureFlagName:         featureFlagName,
+			MetricStore:             gm.metricStore,
+			UseMetrics:              metricsEnabled,
+			BaseOtelAttributes:      baseMetricAttributes,
+		}
+	}
+	circuitBreakerManager := circuit.NewManager(managerOpts)
+
 	subgraphs, err := configureSubgraphOverwrites(
 		engineConfig,
 		configSubgraphs,
@@ -798,7 +817,12 @@ func (s *graphServer) buildGraphMux(ctx context.Context,
 	subgraphResolver := NewSubgraphResolver(subgraphs)
 	httpRouter.Use(func(h http.Handler) http.Handler {
 		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if featureFlagName != "" {
+				r = r.WithContext(context.WithValue(r.Context(), traceclient.CurrentFeatureFlagContextKey{}, featureFlagName))
+			}
+
 			r = r.WithContext(withSubgraphResolver(r.Context(), subgraphResolver))
+
 			requestLogger := s.logger.With(logging.WithRequestID(middleware.GetReqID(r.Context())))
 			// If this is a batched request attach id to the logger
 			if batchedOperationId, ok := r.Context().Value(BatchedOperationId{}).(string); ok {
@@ -1032,6 +1056,7 @@ func (s *graphServer) buildGraphMux(ctx context.Context,
 			LocalhostFallbackInsideDocker: s.localhostFallbackInsideDocker,
 			Logger:                        s.logger,
 			EnableTraceClient:             enableTraceClient,
+			CircuitBreaker:                circuitBreakerManager,
 		},
 	}
 

router/core/router.go

@@ -116,6 +116,11 @@ type (
 		SubgraphMap map[string]*TransportRequestOptions
 	}
 
+	SubgraphCircuitBreakerOptions struct {
+		*config.CircuitBreaker
+		SubgraphMap map[string]*config.CircuitBreaker
+	}
+
 	GraphQLMetricsConfig struct {
 		Enabled           bool
 		CollectorEndpoint string
@@ -1728,6 +1733,12 @@ func WithSubgraphTransportOptions(opts *SubgraphTransportOptions) Option {
 	}
 }
 
+func WithSubgraphCircuitBreakerOptions(opts *SubgraphCircuitBreakerOptions) Option {
+	return func(r *Router) {
+		r.subgraphCircuitBreakerOptions = opts
+	}
+}
+
 func WithSubgraphRetryOptions(enabled bool, maxRetryCount int, retryMaxDuration, retryInterval time.Duration) Option {
 	return func(r *Router) {
 		r.retryOptions = retrytransport.RetryOptions{
@@ -1841,6 +1852,21 @@ func NewSubgraphTransportOptions(cfg config.TrafficShapingRules) *SubgraphTransp
 	return base
 }
 
+func NewSubgraphCircuitBreakerOptions(cfg config.TrafficShapingRules) *SubgraphCircuitBreakerOptions {
+	entry := &SubgraphCircuitBreakerOptions{
+		SubgraphMap: map[string]*config.CircuitBreaker{},
+	}
+	if cfg.All.CircuitBreaker.Enabled {
+		entry.CircuitBreaker = &cfg.All.CircuitBreaker
+	}
+	for k, v := range cfg.Subgraphs {
+		if v.CircuitBreaker.Enabled {
+			entry.SubgraphMap[k] = &v.CircuitBreaker
+		}
+	}
+	return entry
+}
+
 func DefaultSubgraphTransportOptions() *SubgraphTransportOptions {
 	return &SubgraphTransportOptions{
 		TransportRequestOptions: DefaultTransportRequestOptions(),

router/core/router_config.go

@@ -76,6 +76,7 @@ type Config struct {
 	postOriginHandlers              []TransportPostHandler
 	headerRules                     *config.HeaderRules
 	subgraphTransportOptions        *SubgraphTransportOptions
+	subgraphCircuitBreakerOptions   *SubgraphCircuitBreakerOptions
 	graphqlMetricsConfig            *GraphQLMetricsConfig
 	routerTrafficConfig             *config.RouterTrafficConfiguration
 	batchingConfig                  *BatchingConfig

router/core/supervisor_instance.go

@@ -191,6 +191,7 @@ func optionsFromResources(logger *zap.Logger, config *config.Config) []Option {
 		WithRouterTrafficConfig(&config.TrafficShaping.Router),
 		WithFileUploadConfig(&config.FileUpload),
 		WithSubgraphTransportOptions(NewSubgraphTransportOptions(config.TrafficShaping)),
+		WithSubgraphCircuitBreakerOptions(NewSubgraphCircuitBreakerOptions(config.TrafficShaping)),
 		WithSubgraphRetryOptions(
 			config.TrafficShaping.All.BackoffJitterRetry.Enabled,
 			config.TrafficShaping.All.BackoffJitterRetry.MaxAttempts,

router/core/transport.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"github.com/wundergraph/cosmo/router/internal/circuit"
 	"io"
 	"net/http"
 	"net/url"
@@ -36,7 +37,7 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/wundergraph/cosmo/router/internal/docker"
-	"github.com/wundergraph/cosmo/router/internal/retrytransport"
+	retrytransport "github.com/wundergraph/cosmo/router/internal/retrytransport"
 )
 
 type (
@@ -68,12 +69,23 @@ func NewCustomTransport(
 	metricStore metric.Store,
 	connectionMetricStore metric.ConnectionMetricStore,
 	enableSingleFlight bool,
+	breaker *circuit.Manager,
 	enableTraceClient bool,
 ) *CustomTransport {
 	ct := &CustomTransport{
 		metricStore: metricStore,
 	}
 
+	// The round trip method is almost always called via the http.Client roundTripper interface
+	// as a result we cannot pass in the request context logger directly, since this will break the interface
+	// The roundTripper is also not in the core package so it does not have access to the
+	// getRequestContext function since its private to only the core package
+	// As a workaround we pass in a function that can be used to get the logger from within the round tripper
+	getRequestContextLogger := func(req *http.Request) *zap.Logger {
+		reqContext := getRequestContext(req.Context())
+		return reqContext.Logger()
+	}
+
 	if enableTraceClient {
 		getExprContext := func(ctx context.Context) *expr.Context {
 			reqContext := getRequestContext(ctx)
@@ -85,16 +97,11 @@ func NewCustomTransport(
 		baseRoundTripper = traceclient.NewTraceInjectingRoundTripper(baseRoundTripper, connectionMetricStore, getExprContext)
 	}
 
+	if breaker.IsEnabled() {
+		baseRoundTripper = circuit.NewCircuitTripper(baseRoundTripper, breaker, getRequestContextLogger)
+	}
+
 	if retryOptions.Enabled {
-		// The round trip method is almost always called via the http.Client RoundTripper interface
-		// as a result we cannot pass in the request context logger directly, since this will break the interface
-		// The RoundTripper is also not in the core package so it does not have access to the
-		// getRequestContext function since its private to only the core package
-		// As a workaround we pass in a function that can be used to get the logger from within the round tripper
-		getRequestContextLogger := func(req *http.Request) *zap.Logger {
-			reqContext := getRequestContext(req.Context())
-			return reqContext.Logger()
-		}
 		ct.roundTripper = retrytransport.NewRetryHTTPTransport(baseRoundTripper, retryOptions, getRequestContextLogger)
 	} else {
 		ct.roundTripper = baseRoundTripper
@@ -335,6 +342,7 @@ type TransportFactory struct {
 	localhostFallbackInsideDocker bool
 	metricStore                   metric.Store
 	connectionMetricStore         metric.ConnectionMetricStore
+	circuitBreaker                *circuit.Manager
 	logger                        *zap.Logger
 	tracerProvider                *sdktrace.TracerProvider
 	tracePropagators              propagation.TextMapPropagator
@@ -351,6 +359,7 @@ type TransportOptions struct {
 	LocalhostFallbackInsideDocker bool
 	MetricStore                   metric.Store
 	ConnectionMetricStore         metric.ConnectionMetricStore
+	CircuitBreaker                *circuit.Manager
 	Logger                        *zap.Logger
 	TracerProvider                *sdktrace.TracerProvider
 	TracePropagators              propagation.TextMapPropagator
@@ -375,6 +384,7 @@ func NewTransport(opts *TransportOptions) *TransportFactory {
 		logger:                        opts.Logger,
 		tracerProvider:                opts.TracerProvider,
 		tracePropagators:              opts.TracePropagators,
+		circuitBreaker:                opts.CircuitBreaker,
 		enableTraceClient:             opts.EnableTraceClient,
 	}
 }
@@ -420,6 +430,7 @@ func (t TransportFactory) RoundTripper(enableSingleFlight bool, baseTransport ht
 		t.metricStore,
 		t.connectionMetricStore,
 		enableSingleFlight,
+		t.circuitBreaker,
 		t.enableTraceClient,
 	)
 

router/go.mod

@@ -62,6 +62,7 @@ require (
 	github.com/MicahParks/keyfunc/v3 v3.3.5
 	github.com/alicebob/miniredis/v2 v2.34.0
 	github.com/caarlos0/env/v11 v11.3.1
+	github.com/cep21/circuit/v4 v4.0.0
 	github.com/dgraph-io/ristretto/v2 v2.1.0
 	github.com/expr-lang/expr v1.17.3
 	github.com/goccy/go-json v0.10.3

router/go.sum

@@ -33,6 +33,8 @@ github.com/caarlos0/env/v11 v11.3.1 h1:cArPWC15hWmEt+gWk7YBi7lEXTXCvpaSdCiZE2X5m
 github.com/caarlos0/env/v11 v11.3.1/go.mod h1:qupehSf/Y0TUTsxKywqRt/vJjN5nz6vauiYEUUr8P4U=
 github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqylYbM=
 github.com/cenkalti/backoff/v4 v4.2.1/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
+github.com/cep21/circuit/v4 v4.0.0 h1:g1AzMmRLuwCst0eccy1nGsD/CL2XKbDnaPUHVHDvVmI=
+github.com/cep21/circuit/v4 v4.0.0/go.mod h1:Bb1fHpuiRu+AIgKf7DTM1c5U94qTZtKouKcDwtZYCXk=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cilium/ebpf v0.9.1 h1:64sn2K3UKw8NbP/blsixRpF3nXuyhz/VjRlRzvlBRu4=

router/internal/circuit/breaker.go

@@ -0,0 +1,65 @@
+package circuit
+
+import (
+	"context"
+	"fmt"
+	"github.com/wundergraph/cosmo/router/internal/traceclient"
+	"go.uber.org/zap"
+	"net/http"
+)
+
+type Breaker struct {
+	roundTripper   http.RoundTripper
+	loggerFunc     func(req *http.Request) *zap.Logger
+	circuitBreaker *Manager
+}
+
+func NewCircuitTripper(roundTripper http.RoundTripper, breaker *Manager, logger func(req *http.Request) *zap.Logger) *Breaker {
+	return &Breaker{
+		circuitBreaker: breaker,
+		loggerFunc:     logger,
+		roundTripper:   roundTripper,
+	}
+}
+
+func (rt *Breaker) RoundTrip(req *http.Request) (resp *http.Response, err error) {
+	ctx := req.Context()
+
+	var subgraph string
+	subgraphCtxVal := ctx.Value(traceclient.CurrentSubgraphContextKey{})
+	if subgraphCtxVal != nil {
+		subgraph = subgraphCtxVal.(string)
+	}
+
+	var prefix string
+	prefixCtxVal := ctx.Value(traceclient.CurrentFeatureFlagContextKey{})
+	if prefixCtxVal != nil {
+		prefix = prefixCtxVal.(string)
+	}
+
+	cbKey := fmt.Sprintf("%s::%s", prefix, subgraph)
+
+	// If there is no circuit defined for this subgraph
+	circuit := rt.circuitBreaker.GetCircuitBreaker(cbKey)
+	if circuit == nil {
+		return rt.roundTripper.RoundTrip(req)
+	}
+
+	preRunStatus := circuit.IsOpen()
+
+	err = circuit.Run(context.Background(), func(_ context.Context) error {
+		resp, err = rt.roundTripper.RoundTrip(req)
+		return err
+	})
+
+	postRunStatus := circuit.IsOpen()
+
+	logger := rt.loggerFunc(req)
+	if preRunStatus != postRunStatus {
+		logger.Debug("Circuit breaker status changed", zap.String("subgraph", subgraph), zap.Bool("isOpen", postRunStatus))
+	} else if preRunStatus {
+		logger.Debug("Circuit breaker open, request callback did not execute", zap.String("subgraph", subgraph))
+	}
+
+	return resp, err
+}