diff --git a/pkg/storage/unified/sql/notifier_sql.go b/pkg/storage/unified/sql/notifier_sql.go index 0972829ccc9..29888923c44 100644 --- a/pkg/storage/unified/sql/notifier_sql.go +++ b/pkg/storage/unified/sql/notifier_sql.go @@ -8,6 +8,7 @@ import ( "go.opentelemetry.io/otel/trace" "github.com/grafana/grafana-app-sdk/logging" + "github.com/grafana/grafana/pkg/storage/unified/resource" "github.com/grafana/grafana/pkg/storage/unified/sql/sqltemplate" ) @@ -138,7 +139,7 @@ func (p *pollingNotifier) poller(ctx context.Context, since groupResourceRV, str continue } for group, items := range grv { - for resource := range items { + for resource, latestRV := range items { // If we haven't seen this resource before, we start from 0. if _, ok := since[group]; !ok { since[group] = make(map[string]int64) @@ -147,7 +148,17 @@ func (p *pollingNotifier) poller(ctx context.Context, since groupResourceRV, str since[group][resource] = 0 } - // Poll for new events. + // We don't need to poll if the RV hasn't changed. + if since[group][resource] >= latestRV { + p.log.Debug("polling for resource skipped", + "group", group, + "resource", resource, + "latestKnownRV", since[group][resource], + "latestFetchedRV", latestRV) + continue + } + + // Poll for new events since the last known RV. next, err := p.poll(ctx, group, resource, since[group][resource], stream) if err != nil { p.log.Error("polling for resource", "err", err) diff --git a/pkg/storage/unified/sql/notifier_sql_test.go b/pkg/storage/unified/sql/notifier_sql_test.go index b2c8630f264..af4344c0c85 100644 --- a/pkg/storage/unified/sql/notifier_sql_test.go +++ b/pkg/storage/unified/sql/notifier_sql_test.go @@ -221,23 +221,28 @@ func TestPollingNotifier(t *testing.T) { Action: 1, } - var latestRVsCalled bool + var listLatestRVsCalledCounter int listLatestRVs := func(ctx context.Context) (groupResourceRV, error) { - latestRVsCalled = true + // On the first call return 0, then the highest known RV. + var value int64 = 0 + if listLatestRVsCalledCounter > 0 { + value = testEvent.ResourceVersion + } + listLatestRVsCalledCounter++ return groupResourceRV{ "test-group": map[string]int64{ - "test-resource": 0, + "test-resource": value, }, }, nil } - var historyPollCalled bool + var historyPollCalledCounter int once := sync.Once{} historyPoll := func(ctx context.Context, grp string, res string, since int64) ([]*historyPollResponse, error) { // only assert the first time - this may be called multiple times // depending on the host hardware etc, due to timing issues... + historyPollCalledCounter++ once.Do(func() { - historyPollCalled = true require.Equal(t, "test-group", grp) require.Equal(t, "test-resource", res) require.Equal(t, int64(0), since) @@ -274,8 +279,8 @@ func TestPollingNotifier(t *testing.T) { require.Equal(t, "test-name", event.Key.Name) require.Equal(t, int64(2), event.ResourceVersion) require.Equal(t, "test-folder", event.Folder) - require.True(t, latestRVsCalled, "listLatestRVs should be called") - require.True(t, historyPollCalled, "historyPoll should be called") + require.True(t, listLatestRVsCalledCounter > 0, "listLatestRVs should be called at least once") + require.True(t, historyPollCalledCounter == 1, "historyPoll should be called exactly once") case <-time.After(100 * time.Millisecond): t.Fatal("timeout waiting for event") }