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)