diff --git a/go.mod b/go.mod index f11c75f3..0d154fa1 100644 --- a/go.mod +++ b/go.mod @@ -30,6 +30,7 @@ require ( go.uber.org/mock v0.5.2 go.uber.org/zap v1.25.0 golang.org/x/oauth2 v0.35.0 + golang.org/x/sync v0.22.0 google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9 google.golang.org/grpc v1.80.0 google.golang.org/protobuf v1.36.11 diff --git a/go.sum b/go.sum index 67a86079..ec555aa5 100644 --- a/go.sum +++ b/go.sum @@ -666,8 +666,8 @@ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= -golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= +golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.0.0-20170830134202-bb24a47a89ea/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= diff --git a/pkg/evaluators/authorization.go b/pkg/evaluators/authorization.go index 765d0d2c..e693c3be 100644 --- a/pkg/evaluators/authorization.go +++ b/pkg/evaluators/authorization.go @@ -4,6 +4,8 @@ import ( "context" "fmt" + "golang.org/x/sync/singleflight" + "github.com/kuadrant/authorino/pkg/auth" "github.com/kuadrant/authorino/pkg/evaluators/authorization" "github.com/kuadrant/authorino/pkg/jsonexp" @@ -23,6 +25,7 @@ type AuthorizationConfig struct { Conditions jsonexp.Expression `yaml:"conditions"` Metrics bool `yaml:"metrics"` Cache EvaluatorCache + flight singleflight.Group OPA *authorization.OPA `yaml:"opa,omitempty"` JSON *authorization.JSONPatternMatching `yaml:"json,omitempty"` @@ -48,35 +51,49 @@ func (config *AuthorizationConfig) GetAuthConfigEvaluator() auth.AuthConfigEvalu // impl:AuthConfigEvaluator func (config *AuthorizationConfig) Call(pipeline auth.AuthPipeline, ctx context.Context) (interface{}, error) { - if evaluator := config.GetAuthConfigEvaluator(); evaluator == nil { + evaluator := config.GetAuthConfigEvaluator() + if evaluator == nil { return nil, fmt.Errorf("invalid authorization config") - } else { - logger := log.FromContext(ctx).WithName("authorization") + } - cache := config.Cache - var cacheKey interface{} + logger := log.FromContext(ctx).WithName("authorization") - if cache != nil { - cacheKey, _ = cache.ResolveKeyFor(pipeline.GetAuthorizationJSON()) - if cacheKey != nil { - if cachedObj, err := cache.Get(cacheKey); err != nil { - logger.V(1).Error(err, "failed to retrieve data from the cache") - } else if cachedObj != nil { - return cachedObj, nil - } + cache := config.Cache + var cacheKey interface{} + + if cache != nil { + cacheKey, _ = cache.ResolveKeyFor(pipeline.GetAuthorizationJSON()) + if cacheKey != nil { + if cachedObj, err := cache.Get(cacheKey); err != nil { + logger.V(1).Error(err, "failed to retrieve data from the cache") + } else if cachedObj != nil { + return cachedObj, nil } } + } - obj, err := evaluator.Call(pipeline, log.IntoContext(ctx, logger)) - - if err == nil && cacheKey != nil { - if err := cache.Set(cacheKey, obj); err != nil { - logger.V(1).Info("unable to store data in the cache", "err", err) + if cache != nil && cacheKey != nil { + flightKey := fmt.Sprintf("%s/%v", config.Name, cacheKey) + result, err, shared := config.flight.Do(flightKey, func() (interface{}, error) { + if cachedObj, _ := cache.Get(cacheKey); cachedObj != nil { + return cachedObj, nil + } + obj, err := evaluator.Call(pipeline, log.IntoContext(ctx, logger)) + if err == nil { + if setErr := cache.Set(cacheKey, obj); setErr != nil { + logger.V(1).Info("unable to store data in the cache", "err", setErr) + } } + return obj, err + }) + if shared { + logger.V(1).Info("singleflight: coalesced duplicate authorization call", "key", flightKey) } - - return obj, err + return result, err } + + obj, err := evaluator.Call(pipeline, log.IntoContext(ctx, logger)) + return obj, err } // impl:NamedEvaluator diff --git a/pkg/evaluators/metadata.go b/pkg/evaluators/metadata.go index 02fc2602..1e8cfda3 100644 --- a/pkg/evaluators/metadata.go +++ b/pkg/evaluators/metadata.go @@ -4,6 +4,8 @@ import ( "context" "fmt" + "golang.org/x/sync/singleflight" + "github.com/kuadrant/authorino/pkg/auth" "github.com/kuadrant/authorino/pkg/evaluators/metadata" "github.com/kuadrant/authorino/pkg/jsonexp" @@ -22,6 +24,7 @@ type MetadataConfig struct { Conditions jsonexp.Expression `yaml:"conditions"` Metrics bool `yaml:"metrics"` Cache EvaluatorCache + flight singleflight.Group UserInfo *metadata.UserInfo `yaml:"userinfo,omitempty"` UMA *metadata.UMA `yaml:"uma,omitempty"` @@ -44,35 +47,49 @@ func (config *MetadataConfig) GetAuthConfigEvaluator() auth.AuthConfigEvaluator // impl:AuthConfigEvaluator func (config *MetadataConfig) Call(pipeline auth.AuthPipeline, ctx context.Context) (interface{}, error) { - if evaluator := config.GetAuthConfigEvaluator(); evaluator == nil { + evaluator := config.GetAuthConfigEvaluator() + if evaluator == nil { return nil, fmt.Errorf("invalid metadata config") - } else { - logger := log.FromContext(ctx).WithName("metadata").WithValues("config", config.Name) - - cache := config.Cache - var cacheKey interface{} - - if cache != nil { - cacheKey, _ = cache.ResolveKeyFor(pipeline.GetAuthorizationJSON()) - if cacheKey != nil { - if cachedObj, err := cache.Get(cacheKey); err != nil { - logger.V(1).Error(err, "failed to retrieve data from the cache") - } else if cachedObj != nil { - return cachedObj, nil - } - } - } + } + + logger := log.FromContext(ctx).WithName("metadata").WithValues("config", config.Name) - obj, err := evaluator.Call(pipeline, log.IntoContext(ctx, logger)) + cache := config.Cache + var cacheKey interface{} - if err == nil && cacheKey != nil { - if err := cache.Set(cacheKey, obj); err != nil { - logger.V(1).Info("unable to store data in the cache", "err", err) + if cache != nil { + cacheKey, _ = cache.ResolveKeyFor(pipeline.GetAuthorizationJSON()) + if cacheKey != nil { + if cachedObj, err := cache.Get(cacheKey); err != nil { + logger.V(1).Error(err, "failed to retrieve data from the cache") + } else if cachedObj != nil { + return cachedObj, nil } } + } - return obj, err + if cache != nil && cacheKey != nil { + flightKey := fmt.Sprintf("%s/%v", config.Name, cacheKey) + result, err, shared := config.flight.Do(flightKey, func() (interface{}, error) { + if cachedObj, _ := cache.Get(cacheKey); cachedObj != nil { + return cachedObj, nil + } + obj, err := evaluator.Call(pipeline, log.IntoContext(ctx, logger)) + if err == nil { + if setErr := cache.Set(cacheKey, obj); setErr != nil { + logger.V(1).Info("unable to store data in the cache", "err", setErr) + } + } + return obj, err + }) + if shared { + logger.V(1).Info("singleflight: coalesced duplicate metadata call", "key", flightKey) + } + return result, err } + + obj, err := evaluator.Call(pipeline, log.IntoContext(ctx, logger)) + return obj, err } // impl:NamedEvaluator diff --git a/pkg/evaluators/metadata_singleflight_test.go b/pkg/evaluators/metadata_singleflight_test.go new file mode 100644 index 00000000..7cec07dd --- /dev/null +++ b/pkg/evaluators/metadata_singleflight_test.go @@ -0,0 +1,193 @@ +package evaluators + +import ( + "context" + "fmt" + "sync" + "sync/atomic" + "testing" + + "github.com/kuadrant/authorino/pkg/auth" + mock_auth "github.com/kuadrant/authorino/pkg/auth/mocks" + "github.com/kuadrant/authorino/pkg/evaluators/metadata" + "github.com/kuadrant/authorino/pkg/httptest" + "github.com/kuadrant/authorino/pkg/json" + + "go.uber.org/mock/gomock" + "gotest.tools/assert" +) + +const singleflightTestHost = "127.0.0.1:9018" + +func TestMetadataSingleflight_ConcurrentMissesCoalesce(t *testing.T) { + var callCount atomic.Int32 + + server := httptest.NewHttpServerMock(singleflightTestHost, map[string]httptest.HttpServerMockResponseFunc{ + "/metadata": func() httptest.HttpServerMockResponse { + callCount.Add(1) + return httptest.HttpServerMockResponse{ + Status: 200, + Headers: map[string]string{"Content-Type": "application/json"}, + Body: `{"valid":true}`, + } + }, + }) + defer server.Close() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + evaluator := &metadata.GenericHttp{ + Endpoint: fmt.Sprintf("http://%s/metadata", singleflightTestHost), + Method: "GET", + AuthCredentials: auth.NewAuthCredential("", "authorization_header"), + } + + cache := NewEvaluatorCache(&json.JSONValue{Static: "test-key"}, 60) + + config := MetadataConfig{ + Name: "test-metadata", + Cache: cache, + GenericHTTP: evaluator, + } + defer config.Clean(context.Background()) + + const concurrency = 50 + var wg sync.WaitGroup + wg.Add(concurrency) + errors := make([]error, concurrency) + results := make([]interface{}, concurrency) + + pipelineMock := mock_auth.NewMockAuthPipeline(ctrl) + pipelineMock.EXPECT().GetAuthorizationJSON().AnyTimes().Return(`{}`) + + for i := 0; i < concurrency; i++ { + go func(idx int) { + defer wg.Done() + results[idx], errors[idx] = config.Call(pipelineMock, context.Background()) + }(i) + } + + wg.Wait() + + for i, err := range errors { + assert.NilError(t, err, "goroutine %d returned error", i) + } + + for i, res := range results { + assert.Assert(t, res != nil, "goroutine %d returned nil result", i) + } + + actual := callCount.Load() + assert.Assert(t, actual == 1, "expected 1 HTTP call, got %d (singleflight did not coalesce)", actual) +} + +func TestMetadataSingleflight_DifferentKeysDoNotCoalesce(t *testing.T) { + var callCount atomic.Int32 + + const testHost2 = "127.0.0.1:9019" + server := httptest.NewHttpServerMock(testHost2, map[string]httptest.HttpServerMockResponseFunc{ + "/metadata": func() httptest.HttpServerMockResponse { + callCount.Add(1) + return httptest.HttpServerMockResponse{ + Status: 200, + Headers: map[string]string{"Content-Type": "application/json"}, + Body: `{"valid":true}`, + } + }, + }) + defer server.Close() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + evaluator := &metadata.GenericHttp{ + Endpoint: fmt.Sprintf("http://%s/metadata", testHost2), + Method: "GET", + AuthCredentials: auth.NewAuthCredential("", "authorization_header"), + } + + cacheA := NewEvaluatorCache(&json.JSONValue{Static: "key-A"}, 60) + cacheB := NewEvaluatorCache(&json.JSONValue{Static: "key-B"}, 60) + + configA := MetadataConfig{ + Name: "test-metadata", + Cache: cacheA, + GenericHTTP: evaluator, + } + configB := MetadataConfig{ + Name: "test-metadata", + Cache: cacheB, + GenericHTTP: evaluator, + } + defer configA.Clean(context.Background()) + defer configB.Clean(context.Background()) + + pipelineMockA := mock_auth.NewMockAuthPipeline(ctrl) + pipelineMockA.EXPECT().GetAuthorizationJSON().AnyTimes().Return(`{}`) + pipelineMockB := mock_auth.NewMockAuthPipeline(ctrl) + pipelineMockB.EXPECT().GetAuthorizationJSON().AnyTimes().Return(`{}`) + + var wg sync.WaitGroup + wg.Add(2) + + go func() { + defer wg.Done() + configA.Call(pipelineMockA, context.Background()) + }() + go func() { + defer wg.Done() + configB.Call(pipelineMockB, context.Background()) + }() + + wg.Wait() + + actual := callCount.Load() + assert.Assert(t, actual == 2, "expected 2 HTTP calls (one per distinct cache), got %d", actual) +} + +func TestMetadataSingleflight_CachePopulatedAfterCoalescedCall(t *testing.T) { + var callCount atomic.Int32 + + const testHost3 = "127.0.0.1:9020" + server := httptest.NewHttpServerMock(testHost3, map[string]httptest.HttpServerMockResponseFunc{ + "/metadata": func() httptest.HttpServerMockResponse { + callCount.Add(1) + return httptest.HttpServerMockResponse{ + Status: 200, + Headers: map[string]string{"Content-Type": "application/json"}, + Body: `{"cached":"value"}`, + } + }, + }) + defer server.Close() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + evaluator := &metadata.GenericHttp{ + Endpoint: fmt.Sprintf("http://%s/metadata", testHost3), + Method: "GET", + AuthCredentials: auth.NewAuthCredential("", "authorization_header"), + } + + cache := NewEvaluatorCache(&json.JSONValue{Static: "populate-key"}, 60) + + config := MetadataConfig{ + Name: "test-metadata-populate", + Cache: cache, + GenericHTTP: evaluator, + } + defer config.Clean(context.Background()) + + pipelineMock := mock_auth.NewMockAuthPipeline(ctrl) + pipelineMock.EXPECT().GetAuthorizationJSON().AnyTimes().Return(`{}`) + + _, err := config.Call(pipelineMock, context.Background()) + assert.NilError(t, err) + assert.Assert(t, callCount.Load() == 1, "first call should invoke evaluator") + + _, err = config.Call(pipelineMock, context.Background()) + assert.NilError(t, err) + assert.Assert(t, callCount.Load() == 1, "second call should hit cache, not invoke evaluator again") +}