From 805fc1847ee5798f3a45d676971d6342e9df857e Mon Sep 17 00:00:00 2001 From: Yonatan Romero Date: Tue, 4 Aug 2026 17:20:02 -0300 Subject: [PATCH] fix(deployments): stop range scans and coalesce cache misses Issue Deployment range queries scanned every GitHub page, and concurrent cache misses duplicated scans. Solution Paginate until deployments predate the requested range and coalesce identical deployment cache misses. --- go.mod | 2 +- pkg/github/deployments.go | 65 ++++++++++++-------- pkg/github/deployments_test.go | 55 +++++++++++++++++ pkg/plugin/datasource_caching.go | 20 ++++++- pkg/plugin/datasource_caching_test.go | 85 +++++++++++++++++++++++++++ 5 files changed, 200 insertions(+), 27 deletions(-) diff --git a/go.mod b/go.mod index b900abc6..02dafca1 100644 --- a/go.mod +++ b/go.mod @@ -84,7 +84,7 @@ require ( go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/exp v0.0.0-20260112195511-716be5621a96 // indirect golang.org/x/mod v0.37.0 // indirect - golang.org/x/sync v0.22.0 // indirect + golang.org/x/sync v0.22.0 golang.org/x/telemetry v0.0.0-20260625142307-59b4966ccb57 // indirect golang.org/x/tools v0.47.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57 // indirect diff --git a/pkg/github/deployments.go b/pkg/github/deployments.go index 9a557bcd..e0b91a68 100644 --- a/pkg/github/deployments.go +++ b/pkg/github/deployments.go @@ -67,23 +67,7 @@ func GetAllDeployments(ctx context.Context, client models.Client, opts models.Li deployments := []*googlegithub.Deployment{} - // Build the list options with filters - listOpts := &googlegithub.DeploymentsListOptions{ - ListOptions: googlegithub.ListOptions{PerPage: 100}, - } - - if opts.SHA != "" { - listOpts.SHA = opts.SHA - } - if opts.GitRef != "" { - listOpts.Ref = opts.GitRef - } - if opts.Task != "" { - listOpts.Task = opts.Task - } - if opts.Environment != "" { - listOpts.Environment = opts.Environment - } + listOpts := toDeploymentsListOptions(opts) page := 1 for page != 0 { @@ -104,20 +88,53 @@ func GetAllDeployments(ctx context.Context, client models.Client, opts models.Li return DeploymentsWrapper(deployments), nil } -// GetDeploymentsInRange retrieves every deployment from the repository and then returns the ones that fall within the given time range. +// GetDeploymentsInRange retrieves deployments in the given time range. func GetDeploymentsInRange(ctx context.Context, client models.Client, opts models.ListDeploymentsOptions, from time.Time, to time.Time) (DeploymentsWrapper, error) { - deployments, err := GetAllDeployments(ctx, client, opts) - if err != nil { - return nil, err + if opts.Owner == "" || opts.Repository == "" { + return nil, nil } filtered := []*googlegithub.Deployment{} + listOpts := toDeploymentsListOptions(opts) - for _, deployment := range deployments { - if createdAt := deployment.CreatedAt.GetTime(); createdAt != nil && !createdAt.Before(from) && !createdAt.After(to) { - filtered = append(filtered, deployment) + page := 1 + for page != 0 { + listOpts.Page = page + deployments, resp, err := client.ListDeployments(ctx, opts.Owner, opts.Repository, listOpts) + if err != nil { + return nil, fmt.Errorf("listing deployments: opts=%+v: %v", opts, err) + } + + olderDeployment := false + for _, deployment := range deployments { + createdAt := deployment.CreatedAt.GetTime() + if createdAt == nil { + continue + } + if !createdAt.Before(from) && !createdAt.After(to) { + filtered = append(filtered, deployment) + } + if createdAt.Before(from) { + olderDeployment = true + } + } + + // GitHub lists deployments newest-first, so later pages cannot be in range. + if olderDeployment || resp == nil { + break } + page = resp.NextPage } return DeploymentsWrapper(filtered), nil } + +func toDeploymentsListOptions(opts models.ListDeploymentsOptions) *googlegithub.DeploymentsListOptions { + return &googlegithub.DeploymentsListOptions{ + SHA: opts.SHA, + Ref: opts.GitRef, + Task: opts.Task, + Environment: opts.Environment, + ListOptions: googlegithub.ListOptions{PerPage: 100}, + } +} diff --git a/pkg/github/deployments_test.go b/pkg/github/deployments_test.go index 93a8238e..69251847 100644 --- a/pkg/github/deployments_test.go +++ b/pkg/github/deployments_test.go @@ -13,6 +13,8 @@ import ( type mockDeploymentsClient struct { mockDeployments []*googlegithub.Deployment mockResponse *googlegithub.Response + pages map[int]deploymentPage + requests []*googlegithub.DeploymentsListOptions expectedOwner string expectedRepo string t *testing.T @@ -42,11 +44,22 @@ func (m *mockDeploymentsClient) ListAlertsForOrg(ctx context.Context, owner stri return nil, nil, nil } +type deploymentPage struct { + deployments []*googlegithub.Deployment + nextPage int +} + func (m *mockDeploymentsClient) ListDeployments(ctx context.Context, owner, repo string, opts *googlegithub.DeploymentsListOptions) ([]*googlegithub.Deployment, *googlegithub.Response, error) { if owner != m.expectedOwner || repo != m.expectedRepo { m.t.Errorf("Expected owner/repo to be %s/%s, got %s/%s", m.expectedOwner, m.expectedRepo, owner, repo) } + request := *opts + m.requests = append(m.requests, &request) + if m.pages != nil { + page := m.pages[opts.Page] + return page.deployments, &googlegithub.Response{NextPage: page.nextPage}, nil + } return m.mockDeployments, m.mockResponse, nil } @@ -239,6 +252,48 @@ func TestGetDeploymentsInRange(t *testing.T) { } } +func TestGetDeploymentsInRangePaginatesUntilOlderDeployment(t *testing.T) { + from := time.Date(2025, 1, 10, 0, 0, 0, 0, time.UTC) + to := time.Date(2025, 1, 20, 0, 0, 0, 0, time.UTC) + createdAt := func(id int64, at *time.Time) *googlegithub.Deployment { + deployment := &googlegithub.Deployment{ID: googlegithub.Ptr(id)} + if at != nil { + deployment.CreatedAt = &googlegithub.Timestamp{Time: *at} + } + return deployment + } + afterTo := to.Add(time.Hour) + inRange := from.Add(time.Hour) + beforeFrom := from.Add(-time.Hour) + opts := models.ListDeploymentsOptions{ + Repository: "repo", Owner: "owner", SHA: "sha", GitRef: "main", Task: "deploy", Environment: "production", + } + client := &mockDeploymentsClient{ + expectedOwner: "owner", expectedRepo: "repo", t: t, + pages: map[int]deploymentPage{ + 1: {deployments: []*googlegithub.Deployment{createdAt(1, &afterTo), createdAt(2, nil), createdAt(3, &inRange)}, nextPage: 2}, + 2: {deployments: []*googlegithub.Deployment{createdAt(4, &inRange), createdAt(5, &beforeFrom)}, nextPage: 3}, + 3: {deployments: []*googlegithub.Deployment{createdAt(6, &inRange)}}, + }, + } + + deployments, err := GetDeploymentsInRange(context.Background(), client, opts, from, to) + if err != nil { + t.Fatal(err) + } + if len(deployments) != 2 || deployments[0].GetID() != 3 || deployments[1].GetID() != 4 { + t.Fatalf("expected in-range deployments 3 and 4, got %+v", deployments) + } + if len(client.requests) != 2 || client.requests[0].Page != 1 || client.requests[1].Page != 2 { + t.Fatalf("expected requests for pages 1 and 2 only, got %+v", client.requests) + } + for _, request := range client.requests { + if request.PerPage != 100 || request.SHA != opts.SHA || request.Ref != opts.GitRef || request.Task != opts.Task || request.Environment != opts.Environment { + t.Errorf("deployment filters were not preserved: %+v", request) + } + } +} + func TestDeploymentsWrapperFrames(t *testing.T) { // Create test data createdAt := &googlegithub.Timestamp{Time: time.Now().Add(-48 * time.Hour)} diff --git a/pkg/plugin/datasource_caching.go b/pkg/plugin/datasource_caching.go index 87ab4045..6c47d8f8 100644 --- a/pkg/plugin/datasource_caching.go +++ b/pkg/plugin/datasource_caching.go @@ -11,6 +11,7 @@ import ( "github.com/grafana/grafana-plugin-sdk-go/backend" "github.com/pkg/errors" + "golang.org/x/sync/singleflight" "github.com/grafana/github-datasource/pkg/dfutil" "github.com/grafana/github-datasource/pkg/models" @@ -51,6 +52,8 @@ type CachedDatasource struct { mu sync.RWMutex // protects the cache map against concurrent access cache map[string]CachedResult + + deploymentRequests singleflight.Group } func (c *CachedDatasource) getCache(req backend.DataQuery) (dfutil.Framer, error) { @@ -268,8 +271,21 @@ func (c *CachedDatasource) HandleDeploymentsQuery(ctx context.Context, q *models return value, err } - f, err := c.datasource.HandleDeploymentsQuery(ctx, q, req) - return c.saveCache(req, f, err) + key, err := getCacheKey(req) + if err != nil { + return nil, err + } + value, err, _ := c.deploymentRequests.Do(key, func() (interface{}, error) { + if value, err := c.getCache(req); err == nil { + return value, nil + } + f, err := c.datasource.HandleDeploymentsQuery(context.WithoutCancel(ctx), q, req) + return c.saveCache(req, f, err) + }) + if value == nil { + return nil, err + } + return value.(dfutil.Framer), err } // HandleOrganizationsQuery is the cache wrapper for the organizations query handler diff --git a/pkg/plugin/datasource_caching_test.go b/pkg/plugin/datasource_caching_test.go index 2f5aadd2..faa1197c 100644 --- a/pkg/plugin/datasource_caching_test.go +++ b/pkg/plugin/datasource_caching_test.go @@ -1,13 +1,19 @@ package plugin import ( + "context" "encoding/json" "sync" + "sync/atomic" "testing" + "time" "github.com/grafana/grafana-plugin-sdk-go/backend" "github.com/grafana/grafana-plugin-sdk-go/data" "github.com/stretchr/testify/assert" + + "github.com/grafana/github-datasource/pkg/dfutil" + "github.com/grafana/github-datasource/pkg/models" ) // mockFramer is a struct implementing the Framer interface that returns predefined frames for testing purposes @@ -25,6 +31,85 @@ var framesA = data.Frames{data.NewFrame("A", nil)} var dataQueryB = backend.DataQuery{JSON: json.RawMessage(`{"query": "B"}`)} var framesB = data.Frames{data.NewFrame("B", nil)} +type blockingDeploymentsDatasource struct { + Datasource + calls atomic.Int32 + started chan struct{} + release <-chan struct{} + checkContext bool + once sync.Once +} + +func (d *blockingDeploymentsDatasource) HandleDeploymentsQuery(ctx context.Context, _ *models.DeploymentsQuery, _ backend.DataQuery) (dfutil.Framer, error) { + d.calls.Add(1) + d.once.Do(func() { close(d.started) }) + <-d.release + if d.checkContext && ctx.Err() != nil { + return nil, ctx.Err() + } + return mockFramer{frames: framesA}, nil +} + +func TestCachedDeploymentsQueryCoalescesCacheMisses(t *testing.T) { + const queries = 10 + release := make(chan struct{}) + datasource := &blockingDeploymentsDatasource{started: make(chan struct{}), release: release} + cached := WithCaching(datasource) + var wg sync.WaitGroup + start := make(chan struct{}) + errs := make(chan error, queries) + + for range queries { + wg.Add(1) + go func() { + defer wg.Done() + <-start + _, err := cached.HandleDeploymentsQuery(context.Background(), &models.DeploymentsQuery{}, dataQueryA) + errs <- err + }() + } + close(start) + <-datasource.started + time.Sleep(10 * time.Millisecond) + if calls := datasource.calls.Load(); calls != 1 { + t.Fatalf("expected one underlying deployment query, got %d", calls) + } + close(release) + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatal(err) + } + } +} + +func TestCachedDeploymentsQueryIgnoresLeaderCancellation(t *testing.T) { + release := make(chan struct{}) + datasource := &blockingDeploymentsDatasource{started: make(chan struct{}), release: release, checkContext: true} + cached := WithCaching(datasource) + leaderCtx, cancel := context.WithCancel(context.Background()) + errs := make(chan error, 2) + + go func() { + _, err := cached.HandleDeploymentsQuery(leaderCtx, &models.DeploymentsQuery{}, dataQueryA) + errs <- err + }() + <-datasource.started + cancel() + go func() { + _, err := cached.HandleDeploymentsQuery(context.Background(), &models.DeploymentsQuery{}, dataQueryA) + errs <- err + }() + close(release) + + for range 2 { + if err := <-errs; err != nil { + t.Fatalf("expected shared deployment query to survive leader cancellation: %v", err) + } + } +} + func TestWithCaching(t *testing.T) { cachedDS := WithCaching(nil)