From c61624ad3ce6e18fdc80f635d8090d284c94dff6 Mon Sep 17 00:00:00 2001 From: Bruno Date: Mon, 15 Sep 2025 12:07:47 -0300 Subject: [PATCH] Query caching: add request deduplication middleware (#110892) * secrets: update test to accept []byte(nil) and []byte{} (#110630) Co-authored-by: Matheus Macabu * Query caching: add request deduplication middleware * log error if unable to build cache key * remove TODO * always use req.PluginContext.DataSourceInstanceSettings.UID * make update-workspace --------- Co-authored-by: Matheus Macabu --- apps/shorturl/go.mod | 7 +- apps/shorturl/go.sum | 14 ++-- .../src/types/featureToggles.gen.ts | 5 ++ pkg/services/caching/service.go | 71 +++++++++++++++++++ pkg/services/featuremgmt/registry.go | 7 ++ pkg/services/featuremgmt/toggles_gen.csv | 1 + pkg/services/featuremgmt/toggles_gen.go | 4 ++ pkg/services/featuremgmt/toggles_gen.json | 13 ++++ .../clientmiddleware/caching_middleware.go | 63 +++++++++++++++- .../caching_middleware_test.go | 51 +++++++++++++ 10 files changed, 224 insertions(+), 12 deletions(-) diff --git a/apps/shorturl/go.mod b/apps/shorturl/go.mod index d2011cd76c2..c62fd0d469a 100644 --- a/apps/shorturl/go.mod +++ b/apps/shorturl/go.mod @@ -22,6 +22,7 @@ require ( github.com/fxamacker/cbor/v2 v2.7.0 // indirect github.com/getkin/kin-openapi v0.132.0 // indirect github.com/go-jose/go-jose/v3 v3.0.4 // indirect + github.com/go-jose/go-jose/v4 v4.1.2 // indirect github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/go-openapi/jsonpointer v0.21.0 // indirect @@ -33,9 +34,9 @@ require ( github.com/google/go-cmp v0.7.0 // indirect github.com/google/pprof v0.0.0-20250403155104-27863c87afa6 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/grafana/authlib v0.0.0-20250710201142-9542f2f28d43 // indirect - github.com/grafana/authlib/types v0.0.0-20250710201142-9542f2f28d43 // indirect - github.com/grafana/dskit v0.0.0-20250611075409-46f51e1ce914 // indirect + github.com/grafana/authlib v0.0.0-20250909101823-1b466dbd19a1 // indirect + github.com/grafana/authlib/types v0.0.0-20250721184729-1593a38e4933 // indirect + github.com/grafana/dskit v0.0.0-20250908063411-6b6da59b5cc4 // indirect github.com/grafana/regexp v0.0.0-20240518133315-a468a5bfb3bc // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 // indirect github.com/hashicorp/errwrap v1.1.0 // indirect diff --git a/apps/shorturl/go.sum b/apps/shorturl/go.sum index fc332cc9112..79b654c2dce 100644 --- a/apps/shorturl/go.sum +++ b/apps/shorturl/go.sum @@ -20,6 +20,8 @@ github.com/getkin/kin-openapi v0.132.0 h1:3ISeLMsQzcb5v26yeJrBcdTCEQTag36ZjaGk7M github.com/getkin/kin-openapi v0.132.0/go.mod h1:3OlG51PCYNsPByuiMB0t4fjnNlIDnaEDsjiKUV8nL58= github.com/go-jose/go-jose/v3 v3.0.4 h1:Wp5HA7bLQcKnf6YYao/4kpRpVMp/yf6+pJKV8WFSaNY= github.com/go-jose/go-jose/v3 v3.0.4/go.mod h1:5b+7YgP7ZICgJDBdfjZaIt+H/9L9T/YQrVfLAMboGkQ= +github.com/go-jose/go-jose/v4 v4.1.2 h1:TK/7NqRQZfgAh+Td8AlsrvtPoUyiHh0LqVvokh+1vHI= +github.com/go-jose/go-jose/v4 v4.1.2/go.mod h1:22cg9HWM1pOlnRiY+9cQYJ9XHmya1bYW8OeDM6Ku6Oo= github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= @@ -49,12 +51,12 @@ github.com/google/pprof v0.0.0-20250403155104-27863c87afa6 h1:BHT72Gu3keYf3ZEu2J github.com/google/pprof v0.0.0-20250403155104-27863c87afa6/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/grafana/authlib v0.0.0-20250710201142-9542f2f28d43 h1:vVPT0i5Y1vI6qzecYStV2yk7cHKrC3Pc7AgvwT5KydQ= -github.com/grafana/authlib v0.0.0-20250710201142-9542f2f28d43/go.mod h1:1fWkOiL+m32NBgRHZtlZGz2ji868tPZACYbqP3nBRJI= -github.com/grafana/authlib/types v0.0.0-20250710201142-9542f2f28d43 h1:NlkGMnVi/oUn6Cr90QbJYpQJ4FnjyAIG9Ex5GtTZIzw= -github.com/grafana/authlib/types v0.0.0-20250710201142-9542f2f28d43/go.mod h1:qeWYbnWzaYGl88JlL9+DsP1GT2Cudm58rLtx13fKZdw= -github.com/grafana/dskit v0.0.0-20250611075409-46f51e1ce914 h1:qcSGhr691f1mmPHwg2svGyO40Ex92G02aOyHzP6XHCE= -github.com/grafana/dskit v0.0.0-20250611075409-46f51e1ce914/go.mod h1:OiN4P4aC6LwLzLbEupH3Ue83VfQoNMfG48rsna8jI/E= +github.com/grafana/authlib v0.0.0-20250909101823-1b466dbd19a1 h1:qdH5s5FV+0Dyja8O1tBJq7MGd8nPCfxfsMimcYq5cRI= +github.com/grafana/authlib v0.0.0-20250909101823-1b466dbd19a1/go.mod h1:C6CmTG6vfiqebjJswKsc6zes+1F/OtTCi6aAtL5Um6A= +github.com/grafana/authlib/types v0.0.0-20250721184729-1593a38e4933 h1:GjiMR5NIO1/bYSCnt8x7VUeOMaupv2qXJkeLDVAddxQ= +github.com/grafana/authlib/types v0.0.0-20250721184729-1593a38e4933/go.mod h1:qeWYbnWzaYGl88JlL9+DsP1GT2Cudm58rLtx13fKZdw= +github.com/grafana/dskit v0.0.0-20250908063411-6b6da59b5cc4 h1:jSojuc7njleS3UOz223WDlXOinmuLAIPI0z2vtq8EgI= +github.com/grafana/dskit v0.0.0-20250908063411-6b6da59b5cc4/go.mod h1:VahT+GtfQIM+o8ht2StR6J9g+Ef+C2Vokh5uuSmOD/4= github.com/grafana/grafana-app-sdk v0.40.3 h1:JFo7uAfbAJUfZ9neD7/4sODKm1xgu9zhckclH/N4DYU= github.com/grafana/grafana-app-sdk v0.40.3/go.mod h1:j0KzHo3Sa6kd+lnwSScBNoV9Vobkg/YY9HtEjxpyPrk= github.com/grafana/grafana-app-sdk/logging v0.40.3 h1:2VXsXXEQiqAavRP8wusRDB6rDqf5lufP7A6NfjELqPE= diff --git a/packages/grafana-data/src/types/featureToggles.gen.ts b/packages/grafana-data/src/types/featureToggles.gen.ts index 1da6c2612cc..f102e308fa0 100644 --- a/packages/grafana-data/src/types/featureToggles.gen.ts +++ b/packages/grafana-data/src/types/featureToggles.gen.ts @@ -210,6 +210,11 @@ export interface FeatureToggles { */ awsAsyncQueryCaching?: boolean; /** + * Enable request deduplication when query caching is enabled. Requests issuing the same query will be deduplicated, only the first request to arrive will be executed and the response will be shared with requests arriving while there is a request in-flight + * @default false + */ + queryCacheRequestDeduplication?: boolean; + /** * Alternative permission filter implementation that does not use subqueries for fetching the dashboard folder */ permissionsFilterRemoveSubquery?: boolean; diff --git a/pkg/services/caching/service.go b/pkg/services/caching/service.go index c9436e6a2c3..c3da3d22222 100644 --- a/pkg/services/caching/service.go +++ b/pkg/services/caching/service.go @@ -1,7 +1,13 @@ package caching import ( + "bytes" "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "io" + "strings" "github.com/grafana/grafana-plugin-sdk-go/backend" ) @@ -62,3 +68,68 @@ func (s *OSSCachingService) HandleResourceRequest(ctx context.Context, req *back } var _ CachingService = &OSSCachingService{} + +// GetKey creates a prefixed cache key and uses the internal `encoder` to encode the query into a string +func GetKey(prefix string, query interface{}) (string, error) { + keybuf := bytes.NewBuffer(nil) + + encoder := &JSONEncoder{} + + if err := encoder.Encode(keybuf, query); err != nil { + return "", err + } + + key, err := SHA256KeyFunc(keybuf) + if err != nil { + return "", err + } + + return strings.Join([]string{prefix, key}, ":"), nil +} + +// SHA256KeyFunc copies the data from `r` into a sha256.Hash, and returns the encoded Sum. +func SHA256KeyFunc(r io.Reader) (string, error) { + hash := sha256.New() + + // Read all data from the provided reader + if _, err := io.Copy(hash, r); err != nil { + return "", err + } + + // Encode the written values to SHA256 + return hex.EncodeToString(hash.Sum(nil)), nil +} + +// JSONEncoder encodes and decodes struct data to/from JSON +type JSONEncoder struct{} + +// NewJSONEncoder creates a pointer to a new JSONEncoder, which implements the `Encoder` interface +func NewJSONEncoder() *JSONEncoder { + return &JSONEncoder{} +} + +func (e *JSONEncoder) EncodeBytes(w io.Writer, b []byte) error { + _, err := w.Write(b) + if err != nil { + return err + } + return nil +} + +func (e *JSONEncoder) DecodeBytes(r io.Reader) ([]byte, error) { + encBytes, err := io.ReadAll(r) + if err != nil { + return []byte{}, err + } + return encBytes, err +} + +// Encode encodes the `v` interface into `w` using a json.Encoder +func (e *JSONEncoder) Encode(w io.Writer, v interface{}) error { + return json.NewEncoder(w).Encode(v) +} + +// Decode encodes the io.Reader `r` into the interface `v` using a json.Decoder +func (e *JSONEncoder) Decode(r io.Reader, v interface{}) error { + return json.NewDecoder(r).Decode(v) +} diff --git a/pkg/services/featuremgmt/registry.go b/pkg/services/featuremgmt/registry.go index e347dcf9c1e..2ec29bbe8a7 100644 --- a/pkg/services/featuremgmt/registry.go +++ b/pkg/services/featuremgmt/registry.go @@ -339,6 +339,13 @@ var ( Expression: "true", // enabled by default Owner: awsDatasourcesSquad, }, + { + Name: "queryCacheRequestDeduplication", + Description: "Enable request deduplication when query caching is enabled. Requests issuing the same query will be deduplicated, only the first request to arrive will be executed and the response will be shared with requests arriving while there is a request in-flight", + Stage: FeatureStageExperimental, + Expression: "false", // enabled by default + Owner: grafanaOperatorExperienceSquad, + }, { Name: "permissionsFilterRemoveSubquery", Description: "Alternative permission filter implementation that does not use subqueries for fetching the dashboard folder", diff --git a/pkg/services/featuremgmt/toggles_gen.csv b/pkg/services/featuremgmt/toggles_gen.csv index 74b4de46251..eb62626d352 100644 --- a/pkg/services/featuremgmt/toggles_gen.csv +++ b/pkg/services/featuremgmt/toggles_gen.csv @@ -43,6 +43,7 @@ provisioning,experimental,@grafana/grafana-app-platform-squad,false,true,false grafanaAPIServerEnsureKubectlAccess,experimental,@grafana/grafana-app-platform-squad,true,true,false featureToggleAdminPage,experimental,@grafana/grafana-operator-experience-squad,false,true,false awsAsyncQueryCaching,GA,@grafana/aws-datasources,false,false,false +queryCacheRequestDeduplication,experimental,@grafana/grafana-operator-experience-squad,false,false,false permissionsFilterRemoveSubquery,experimental,@grafana/search-and-storage,false,false,false configurableSchedulerTick,experimental,@grafana/alerting-squad,false,true,false dashgpt,GA,@grafana/dashboards-squad,false,false,true diff --git a/pkg/services/featuremgmt/toggles_gen.go b/pkg/services/featuremgmt/toggles_gen.go index 5376d303c15..bbfbebf2e32 100644 --- a/pkg/services/featuremgmt/toggles_gen.go +++ b/pkg/services/featuremgmt/toggles_gen.go @@ -183,6 +183,10 @@ const ( // Enable caching for async queries for Redshift and Athena. Requires that the datasource has caching and async query support enabled FlagAwsAsyncQueryCaching = "awsAsyncQueryCaching" + // FlagQueryCacheRequestDeduplication + // Enable request deduplication when query caching is enabled. Requests issuing the same query will be deduplicated, only the first request to arrive will be executed and the response will be shared with requests arriving while there is a request in-flight + FlagQueryCacheRequestDeduplication = "queryCacheRequestDeduplication" + // FlagPermissionsFilterRemoveSubquery // Alternative permission filter implementation that does not use subqueries for fetching the dashboard folder FlagPermissionsFilterRemoveSubquery = "permissionsFilterRemoveSubquery" diff --git a/pkg/services/featuremgmt/toggles_gen.json b/pkg/services/featuremgmt/toggles_gen.json index 24f00874702..250b608ac31 100644 --- a/pkg/services/featuremgmt/toggles_gen.json +++ b/pkg/services/featuremgmt/toggles_gen.json @@ -2887,6 +2887,19 @@ "expression": "true" } }, + { + "metadata": { + "name": "queryCacheRequestDeduplication", + "resourceVersion": "1757521912495", + "creationTimestamp": "2025-09-10T16:31:52Z" + }, + "spec": { + "description": "Enable request deduplication when query caching is enabled. Requests issuing the same query will be deduplicated, only the first request to arrive will be executed and the response will be shared with requests arriving while there is a request in-flight", + "stage": "experimental", + "codeowner": "@grafana/grafana-operator-experience-squad", + "expression": "false" + } + }, { "metadata": { "name": "queryLibrary", diff --git a/pkg/services/pluginsintegration/clientmiddleware/caching_middleware.go b/pkg/services/pluginsintegration/clientmiddleware/caching_middleware.go index 80058c0a5e8..6d8535a4a35 100644 --- a/pkg/services/pluginsintegration/clientmiddleware/caching_middleware.go +++ b/pkg/services/pluginsintegration/clientmiddleware/caching_middleware.go @@ -2,12 +2,14 @@ package clientmiddleware import ( "context" + "fmt" "strconv" "time" "github.com/grafana/grafana-aws-sdk/pkg/awsds" "github.com/grafana/grafana-plugin-sdk-go/backend" "github.com/prometheus/client_golang/prometheus" + "golang.org/x/sync/singleflight" "github.com/grafana/grafana/pkg/infra/log" "github.com/grafana/grafana/pkg/services/caching" @@ -34,14 +36,20 @@ func NewCachingMiddlewareWithFeatureManager(cachingService caching.CachingServic if err := prometheus.Register(ResourceCachingRequestHistogram); err != nil { log.Error("Error registering prometheus collector 'ResourceRequestHistogram'", "error", err) } - return backend.HandlerMiddlewareFunc(func(next backend.Handler) backend.Handler { - return &CachingMiddleware{ + cachingMiddlewareHandler := func(next backend.Handler) backend.Handler { + cachingMiddleware := &CachingMiddleware{ BaseHandler: backend.NewBaseHandler(next), caching: cachingService, log: log, features: features, } - }) + if features != nil && features.IsEnabled(context.Background(), featuremgmt.FlagQueryCacheRequestDeduplication) { + return newRequestDeduplicationMiddleware(log, cachingMiddleware) + } + return cachingMiddleware + } + + return backend.HandlerMiddlewareFunc(cachingMiddlewareHandler) } type CachingMiddleware struct { @@ -164,3 +172,52 @@ func (m *CachingMiddleware) CallResource(ctx context.Context, req *backend.CallR return m.BaseHandler.CallResource(ctx, req, cacheSender) } + +// Given N requests happening at the same time and issuing the same query, only one request will execute +// and the other ones will wait for the response received by the request being executed. +type requestDeduplicationMiddleware struct { + backend.BaseHandler + log *log.ConcreteLogger + singleflight *singleflight.Group +} + +func newRequestDeduplicationMiddleware(log *log.ConcreteLogger, next backend.Handler) *requestDeduplicationMiddleware { + return &requestDeduplicationMiddleware{log: log, BaseHandler: backend.NewBaseHandler(next), singleflight: &singleflight.Group{}} +} + +func (m *requestDeduplicationMiddleware) QueryData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) { + if req.PluginContext.DataSourceInstanceSettings.UID == "" { + return m.BaseHandler.QueryData(ctx, req) + } + key, err := caching.GetKey(req.PluginContext.DataSourceInstanceSettings.UID, req) + if err != nil { + m.log.Error("error building cache key for request deduplication, skipping request deduplication", "error", err) + return m.BaseHandler.QueryData(ctx, req) + } + v, err, _ := m.singleflight.Do(key, func() (interface{}, error) { + return m.BaseHandler.QueryData(ctx, req) + }) + if err != nil { + return nil, fmt.Errorf("singleflight: calling next.QueryData: %w", err) + } + return v.(*backend.QueryDataResponse), nil +} + +func (m *requestDeduplicationMiddleware) CallResource(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error { + if req.PluginContext.DataSourceInstanceSettings.UID == "" { + return m.BaseHandler.CallResource(ctx, req, sender) + } + + key, err := caching.GetKey(req.PluginContext.DataSourceInstanceSettings.UID, req) + if err != nil { + m.log.Error("error building cache key for request deduplication, skipping request deduplication", "error", err) + return m.BaseHandler.CallResource(ctx, req, sender) + } + _, err, _ = m.singleflight.Do(key, func() (interface{}, error) { + return nil, m.BaseHandler.CallResource(ctx, req, sender) + }) + if err != nil { + return fmt.Errorf("singleflight: calling next.CallResource: %w", err) + } + return nil +} diff --git a/pkg/services/pluginsintegration/clientmiddleware/caching_middleware_test.go b/pkg/services/pluginsintegration/clientmiddleware/caching_middleware_test.go index 548f7a693bf..e1aa9cc32e0 100644 --- a/pkg/services/pluginsintegration/clientmiddleware/caching_middleware_test.go +++ b/pkg/services/pluginsintegration/clientmiddleware/caching_middleware_test.go @@ -4,7 +4,10 @@ import ( "context" "encoding/json" "net/http" + "sync" + "sync/atomic" "testing" + "time" "github.com/grafana/grafana-plugin-sdk-go/backend" "github.com/grafana/grafana-plugin-sdk-go/backend/handlertest" @@ -322,3 +325,51 @@ func TestCachingMiddleware(t *testing.T) { }) }) } + +func TestRequestDeduplicationMiddleware(t *testing.T) { + t.Run("deduplicates requests issuing the same query", func(t *testing.T) { + t.Parallel() + + handler := newMockMiddlewareHandler() + middleware := newRequestDeduplicationMiddleware(nil, handler) + + req := backend.QueryDataRequest{ + PluginContext: backend.PluginContext{ + DataSourceInstanceSettings: &backend.DataSourceInstanceSettings{ + UID: "uid", + }, + }, + } + + wg := &sync.WaitGroup{} + wg.Add(2) + + for range 2 { + go func() { + defer wg.Done() + resp, err := middleware.QueryData(t.Context(), &req) + require.NoError(t, err) + require.Equal(t, &backend.QueryDataResponse{}, resp) + }() + } + + wg.Wait() + + require.EqualValues(t, 1, handler.QueryDataCalls) + }) +} + +type mockMiddlewareHandler struct { + backend.BaseHandler + QueryDataCalls int32 +} + +func newMockMiddlewareHandler() *mockMiddlewareHandler { + return &mockMiddlewareHandler{} +} + +func (m *mockMiddlewareHandler) QueryData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) { + atomic.AddInt32(&m.QueryDataCalls, 1) + time.Sleep(10 * time.Millisecond) + return &backend.QueryDataResponse{}, nil +}