feat: circuit breaker implementation

router-tests/go.mod

@@ -51,6 +51,7 @@ require (
 	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-tests/testenv/testenv.go

@@ -260,15 +260,17 @@ type EngineStatOptions struct {
 }
 
 type MetricOptions struct {
-	MetricExclusions                  MetricExclusions
-	EnableRuntimeMetrics              bool
-	EnableOTLPRouterCache             bool
-	EnablePrometheusRouterCache       bool
-	OTLPEngineStatsOptions            EngineStatOptions
-	PrometheusEngineStatsOptions      EngineStatOptions
-	PrometheusSchemaFieldUsage        PrometheusSchemaFieldUsage
-	EnableOTLPConnectionMetrics       bool
-	EnablePrometheusConnectionMetrics bool
+	MetricExclusions                      MetricExclusions
+	EnableRuntimeMetrics                  bool
+	EnableOTLPRouterCache                 bool
+	EnablePrometheusRouterCache           bool
+	OTLPEngineStatsOptions                EngineStatOptions
+	PrometheusEngineStatsOptions          EngineStatOptions
+	PrometheusSchemaFieldUsage            PrometheusSchemaFieldUsage
+	EnableOTLPConnectionMetrics           bool
+	EnableOTLPCircuitBreakerMetrics       bool
+	EnablePrometheusConnectionMetrics     bool
+	EnablePrometheusCircuitBreakerMetrics bool
 }
 
 type PrometheusSchemaFieldUsage struct {
@@ -1416,6 +1418,7 @@ func configureRouter(listenerAddr string, testConfig *Config, routerConfig *node
 		core.WithQueryPlans(true),
 		core.WithEvents(eventsConfiguration),
 	}
+
 	routerOpts = append(routerOpts, testConfig.RouterOptions...)
 
 	if testConfig.RouterConfig != nil {
@@ -1496,6 +1499,7 @@ func configureRouter(listenerAddr string, testConfig *Config, routerConfig *node
 			EngineStats: rmetric.EngineStatsConfig{
 				Subscription: testConfig.MetricOptions.PrometheusEngineStatsOptions.EnableSubscription,
 			},
+			CircuitBreaker:      testConfig.MetricOptions.EnablePrometheusCircuitBreakerMetrics,
 			ExcludeMetrics:      testConfig.MetricOptions.MetricExclusions.ExcludedPrometheusMetrics,
 			ExcludeMetricLabels: testConfig.MetricOptions.MetricExclusions.ExcludedPrometheusMetricLabels,
 			ExcludeScopeInfo:    testConfig.MetricOptions.MetricExclusions.ExcludeScopeInfo,
@@ -1524,6 +1528,7 @@ func configureRouter(listenerAddr string, testConfig *Config, routerConfig *node
 					EngineStats: config.EngineStats{
 						Subscriptions: testConfig.MetricOptions.OTLPEngineStatsOptions.EnableSubscription,
 					},
+					CircuitBreaker:      testConfig.MetricOptions.EnableOTLPCircuitBreakerMetrics,
 					ExcludeMetrics:      testConfig.MetricOptions.MetricExclusions.ExcludedOTLPMetrics,
 					ExcludeMetricLabels: testConfig.MetricOptions.MetricExclusions.ExcludedOTLPMetricLabels,
 				},

router/core/context.go

@@ -3,6 +3,7 @@ package core
 import (
 	"context"
 	"errors"
+	rcontext "github.com/wundergraph/cosmo/router/internal/context"
 	"net/http"
 	"net/url"
 	"strconv"
@@ -27,14 +28,6 @@ import (
 	ctrace "github.com/wundergraph/cosmo/router/pkg/trace"
 )
 
-type contextKey int
-
-const (
-	requestContextKey contextKey = iota
-	subgraphResolverContextKey
-	engineLoaderHooksContextKey
-)
-
 var _ RequestContext = (*requestContext)(nil)
 
 type Subgraph struct {
@@ -288,14 +281,14 @@ func (c *requestContext) Request() *http.Request {
 }
 
 func withRequestContext(ctx context.Context, operation *requestContext) context.Context {
-	return context.WithValue(ctx, requestContextKey, operation)
+	return context.WithValue(ctx, rcontext.RequestContextKey, operation)
 }
 
 func getRequestContext(ctx context.Context) *requestContext {
 	if ctx == nil {
 		return nil
 	}
-	op := ctx.Value(requestContextKey)
+	op := ctx.Value(rcontext.RequestContextKey)
 	if op == nil {
 		return nil
 	}
@@ -664,6 +657,8 @@ func NewSubgraphResolver(subgraphs []Subgraph) *SubgraphResolver {
 			Url:       subgraphs[i].Url,
 			UrlString: subgraphs[i].UrlString,
 		}
+		// TODO: In case there are multiple subgraphs with the same URL, the previous
+		// one will be overwritten. To investigate if this causes an issue.
 		if sg.UrlString != "" {
 			resolver.subgraphsByURL[sg.UrlString] = &sg
 		}
@@ -691,11 +686,11 @@ func (s *SubgraphResolver) BySubgraphURL(u string) *Subgraph {
 }
 
 func withSubgraphResolver(ctx context.Context, resolver *SubgraphResolver) context.Context {
-	return context.WithValue(ctx, subgraphResolverContextKey, resolver)
+	return context.WithValue(ctx, rcontext.SubgraphResolverContextKey, resolver)
 }
 
 func subgraphResolverFromContext(ctx context.Context) *SubgraphResolver {
-	resolver, _ := ctx.Value(subgraphResolverContextKey).(*SubgraphResolver)
+	resolver, _ := ctx.Value(rcontext.SubgraphResolverContextKey).(*SubgraphResolver)
 	return resolver
 }
 

router/core/engine_loader_hooks.go

@@ -4,8 +4,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	rcontext "github.com/wundergraph/cosmo/router/internal/context"
 	"github.com/wundergraph/cosmo/router/internal/requestlogger"
-	"github.com/wundergraph/cosmo/router/internal/traceclient"
 	"github.com/wundergraph/cosmo/router/internal/unique"
 	"github.com/wundergraph/cosmo/router/pkg/metric"
 	rotel "github.com/wundergraph/cosmo/router/pkg/otel"
@@ -86,7 +86,7 @@ func (f *engineLoaderHooks) OnLoad(ctx context.Context, ds resolve.DataSourceInf
 
 	start := time.Now()
 
-	ctx = context.WithValue(ctx, traceclient.CurrentSubgraphContextKey{}, ds.Name)
+	ctx = context.WithValue(ctx, rcontext.CurrentSubgraphContextKey{}, ds.Name)
 
 	reqContext := getRequestContext(ctx)
 	if reqContext == nil {
@@ -100,7 +100,7 @@ func (f *engineLoaderHooks) OnLoad(ctx context.Context, ds resolve.DataSourceInf
 		}...),
 	)
 
-	return context.WithValue(ctx, engineLoaderHooksContextKey, &engineLoaderHooksRequestContext{
+	return context.WithValue(ctx, rcontext.EngineLoaderHooksContextKey, &engineLoaderHooksRequestContext{
 		startTime: start,
 	})
 }
@@ -117,7 +117,7 @@ func (f *engineLoaderHooks) OnFinished(ctx context.Context, ds resolve.DataSourc
 		return
 	}
 
-	hookCtx, ok := ctx.Value(engineLoaderHooksContextKey).(*engineLoaderHooksRequestContext)
+	hookCtx, ok := ctx.Value(rcontext.EngineLoaderHooksContextKey).(*engineLoaderHooksRequestContext)
 	if !ok {
 		return
 	}