public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Alerting: Support PDC in Grafana-managed recording rules (#106677)

Browse Source
This commit is contained in:
Alexander Akhmetov
2025-06-17 11:46:34 +02:00
committed by GitHub
parent 80df9fab4d
commit e92baba748
7 changed files with 218 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/cloudmigration/cloudmigrationimpl/cloudmigration_test.go
+1 -1
View File
@@ -934,7 +934,7 @@ func setUpServiceTest(t *testing.T, withDashboardMock bool, cfgOverrides ...conf
cfg, featureToggles, nil, nil, rr, sqlStore, kvStore, nil, nil, quotatest.New(false, nil),
secretsService, nil, alertMetrics, mockFolder, accessControl, dashboardService, nil, bus, fakeAccessControlService,
annotationstest.NewFakeAnnotationsRepo(), &pluginstore.FakePluginStore{}, tracer, ruleStore,
httpclient.NewProvider(), ngalertfakes.NewFakeReceiverPermissionsService(), usertest.NewUserServiceFake(),
httpclient.NewProvider(), nil, ngalertfakes.NewFakeReceiverPermissionsService(), usertest.NewUserServiceFake(),
)
require.NoError(t, err)
pkg/services/ngalert/ngalert.go
+58 -54
View File
@@ -46,6 +46,7 @@ import (
"github.com/grafana/grafana/pkg/services/ngalert/store"
"github.com/grafana/grafana/pkg/services/ngalert/writer"
"github.com/grafana/grafana/pkg/services/notifications"
"github.com/grafana/grafana/pkg/services/pluginsintegration/plugincontext"
"github.com/grafana/grafana/pkg/services/pluginsintegration/pluginstore"
"github.com/grafana/grafana/pkg/services/quota"
"github.com/grafana/grafana/pkg/services/rendering"
@@ -79,37 +80,39 @@ func ProvideService(
tracer tracing.Tracer,
ruleStore *store.DBstore,
httpClientProvider httpclient.Provider,
pluginContextProvider *plugincontext.Provider,
resourcePermissions accesscontrol.ReceiverPermissionsService,
userService user.Service,
) (*AlertNG, error) {
ng := &AlertNG{
Cfg: cfg,
FeatureToggles: featureToggles,
DataSourceCache: dataSourceCache,
DataSourceService: dataSourceService,
RouteRegister: routeRegister,
SQLStore: sqlStore,
KVStore: kvStore,
ExpressionService: expressionService,
DataProxy: dataProxy,
QuotaService: quotaService,
SecretsService: secretsService,
Metrics: m,
Log: log.New("ngalert"),
NotificationService: notificationService,
folderService: folderService,
accesscontrol: ac,
dashboardService: dashboardService,
renderService: renderService,
bus: bus,
AccesscontrolService: accesscontrolService,
annotationsRepo: annotationsRepo,
pluginsStore: pluginsStore,
tracer: tracer,
store: ruleStore,
httpClientProvider: httpClientProvider,
ResourcePermissions: resourcePermissions,
userService: userService,
Cfg: cfg,
FeatureToggles: featureToggles,
DataSourceCache: dataSourceCache,
DataSourceService: dataSourceService,
RouteRegister: routeRegister,
SQLStore: sqlStore,
KVStore: kvStore,
ExpressionService: expressionService,
DataProxy: dataProxy,
QuotaService: quotaService,
SecretsService: secretsService,
Metrics: m,
Log: log.New("ngalert"),
NotificationService: notificationService,
folderService: folderService,
accesscontrol: ac,
dashboardService: dashboardService,
renderService: renderService,
bus: bus,
AccesscontrolService: accesscontrolService,
annotationsRepo: annotationsRepo,
pluginsStore: pluginsStore,
tracer: tracer,
store: ruleStore,
httpClientProvider: httpClientProvider,
pluginContextProvider: pluginContextProvider,
ResourcePermissions: resourcePermissions,
userService: userService,
}
if ng.IsDisabled() {
@@ -125,30 +128,31 @@ func ProvideService(
// AlertNG is the service for evaluating the condition of an alert definition.
type AlertNG struct {
Cfg *setting.Cfg
FeatureToggles featuremgmt.FeatureToggles
DataSourceCache datasources.CacheService
DataSourceService datasources.DataSourceService
RouteRegister routing.RouteRegister
SQLStore db.DB
KVStore kvstore.KVStore
ExpressionService *expr.Service
DataProxy *datasourceproxy.DataSourceProxyService
QuotaService quota.Service
SecretsService secrets.Service
Metrics *metrics.NGAlert
NotificationService notifications.Service
Log log.Logger
renderService rendering.Service
ImageService image.ImageService
RecordingWriter schedule.RecordingWriter
schedule schedule.ScheduleService
stateManager *state.Manager
folderService folder.Service
dashboardService dashboards.DashboardService
Api *api.API
httpClientProvider httpclient.Provider
InstanceStore state.InstanceStore
Cfg *setting.Cfg
FeatureToggles featuremgmt.FeatureToggles
DataSourceCache datasources.CacheService
DataSourceService datasources.DataSourceService
RouteRegister routing.RouteRegister
SQLStore db.DB
KVStore kvstore.KVStore
ExpressionService *expr.Service
DataProxy *datasourceproxy.DataSourceProxyService
QuotaService quota.Service
SecretsService secrets.Service
Metrics *metrics.NGAlert
NotificationService notifications.Service
Log log.Logger
renderService rendering.Service
ImageService image.ImageService
RecordingWriter schedule.RecordingWriter
schedule schedule.ScheduleService
stateManager *state.Manager
folderService folder.Service
dashboardService dashboards.DashboardService
Api *api.API
httpClientProvider httpclient.Provider
pluginContextProvider *plugincontext.Provider
InstanceStore state.InstanceStore
// StartupInstanceReader is used to fetch the state of alerts on startup.
StartupInstanceReader state.InstanceReader
@@ -333,7 +337,7 @@ func (ng *AlertNG) init() error {
evalFactory := eval.NewEvaluatorFactory(ng.Cfg.UnifiedAlerting, ng.DataSourceCache, ng.ExpressionService)
conditionValidator := eval.NewConditionValidator(ng.DataSourceCache, ng.ExpressionService, ng.pluginsStore)
recordingWriter, err := createRecordingWriter(ng.Cfg.UnifiedAlerting.RecordingRules, ng.httpClientProvider, ng.DataSourceService, clk, ng.Metrics.GetRemoteWriterMetrics())
recordingWriter, err := createRecordingWriter(ng.Cfg.UnifiedAlerting.RecordingRules, ng.httpClientProvider, ng.DataSourceService, ng.pluginContextProvider, clk, ng.Metrics.GetRemoteWriterMetrics())
if err != nil {
return fmt.Errorf("failed to initialize recording writer: %w", err)
}
@@ -665,7 +669,7 @@ func createRemoteAlertmanager(ctx context.Context, cfg remote.AlertmanagerConfig
return remote.NewAlertmanager(ctx, cfg, notifier.NewFileStore(cfg.OrgID, kvstore), decryptFn, autogenFn, m, tracer)
}
func createRecordingWriter(settings setting.RecordingRuleSettings, httpClientProvider httpclient.Provider, datasourceService datasources.DataSourceService, clock clock.Clock, m *metrics.RemoteWriter) (schedule.RecordingWriter, error) {
func createRecordingWriter(settings setting.RecordingRuleSettings, httpClientProvider httpclient.Provider, datasourceService datasources.DataSourceService, pluginContextProvider *plugincontext.Provider, clock clock.Clock, m *metrics.RemoteWriter) (schedule.RecordingWriter, error) {
logger := log.New("ngalert.writer")
if settings.Enabled {
@@ -678,7 +682,7 @@ func createRecordingWriter(settings setting.RecordingRuleSettings, httpClientPro
logger.Info("Setting up remote write using data sources",
"timeout", cfg.Timeout, "default_datasource_uid", cfg.DefaultDatasourceUID)
return writer.NewDatasourceWriter(cfg, datasourceService, httpClientProvider, clock, logger, m), nil
return writer.NewDatasourceWriter(cfg, datasourceService, httpClientProvider, pluginContextProvider, clock, logger, m), nil
}
return writer.NoopWriter{}, nil
pkg/services/ngalert/schedule/recording_rule_test.go
+10 -1
View File
@@ -17,8 +17,10 @@ import (
"github.com/stretchr/testify/require"
"go.uber.org/atomic"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana-plugin-sdk-go/backend/httpclient"
"github.com/grafana/grafana/pkg/apimachinery/identity"
"github.com/grafana/grafana/pkg/components/simplejson"
"github.com/grafana/grafana/pkg/expr"
"github.com/grafana/grafana/pkg/infra/log"
@@ -808,7 +810,8 @@ func setupDatasourceWriter(t *testing.T, target *writer.TestRemoteWriteTarget, r
DefaultDatasourceUID: "",
}
return writer.NewDatasourceWriter(cfg, dss, provider, clock.NewMock(),
mockPluginConfig := &mockPluginContextProvider{}
return writer.NewDatasourceWriter(cfg, dss, provider, mockPluginConfig, clock.NewMock(),
log.New("test"), m.GetRemoteWriterMetrics())
}
@@ -817,3 +820,9 @@ type testClientProvider struct{}
func (t testClientProvider) New(options ...httpclient.Options) (*http.Client, error) {
return &http.Client{}, nil
}
type mockPluginContextProvider struct{}
func (m *mockPluginContextProvider) GetWithDataSource(ctx context.Context, pluginID string, user identity.Requester, ds *datasources.DataSource) (backend.PluginContext, error) {
return backend.PluginContext{}, nil
}
pkg/services/ngalert/tests/util.go
+1 -1
View File
@@ -91,7 +91,7 @@ func SetupTestEnv(tb testing.TB, baseInterval time.Duration, opts ...TestEnvOpti
ng, err := ngalert.ProvideService(
cfg, options.featureToggles, nil, nil, routing.NewRouteRegister(), sqlStore, kvstore.NewFakeKVStore(), nil, nil, quotatest.New(false, nil),
secretsService, nil, m, folderService, ac, &dashboards.FakeDashboardService{}, nil, bus, ac,
annotationstest.NewFakeAnnotationsRepo(), &pluginstore.FakePluginStore{}, tracer, ruleStore, httpclient.NewProvider(), ngalertfakes.NewFakeReceiverPermissionsService(), usertest.NewUserServiceFake(),
annotationstest.NewFakeAnnotationsRepo(), &pluginstore.FakePluginStore{}, tracer, ruleStore, httpclient.NewProvider(), nil, ngalertfakes.NewFakeReceiverPermissionsService(), usertest.NewUserServiceFake(),
)
require.NoError(tb, err)
pkg/services/ngalert/writer/datasourcewriter.go
+40 -18
View File
@@ -11,10 +11,12 @@ import (
"time"
"github.com/benbjohnson/clock"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana-plugin-sdk-go/backend/httpclient"
"github.com/grafana/grafana-plugin-sdk-go/data"
gocache "github.com/patrickmn/go-cache"
"github.com/grafana/grafana/pkg/apimachinery/identity"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/services/datasources"
"github.com/grafana/grafana/pkg/services/ngalert/metrics"
@@ -44,13 +46,18 @@ type DatasourceWriterConfig struct {
CustomHeaders map[string]string
}
type PluginContextProvider interface {
GetWithDataSource(ctx context.Context, pluginID string, user identity.Requester, ds *datasources.DataSource) (backend.PluginContext, error)
}
type DatasourceWriter struct {
cfg DatasourceWriterConfig
datasources datasources.DataSourceService
httpClientProvider HttpClientProvider
clock clock.Clock
l log.Logger
metrics *metrics.RemoteWriter
cfg DatasourceWriterConfig
datasources datasources.DataSourceService
httpClientProvider HttpClientProvider
pluginContextProvider PluginContextProvider
clock clock.Clock
l log.Logger
metrics *metrics.RemoteWriter
writers *gocache.Cache
}
@@ -59,18 +66,20 @@ func NewDatasourceWriter(
cfg DatasourceWriterConfig,
datasources datasources.DataSourceService,
httpClientProvider HttpClientProvider,
pluginContextProvider PluginContextProvider,
clock clock.Clock,
l log.Logger,
metrics *metrics.RemoteWriter,
) *DatasourceWriter {
return &DatasourceWriter{
cfg: cfg,
datasources: datasources,
httpClientProvider: httpClientProvider,
clock: clock,
l: l,
metrics: metrics,
writers: gocache.New(cacheExpiration, cacheCleanupInterval),
cfg: cfg,
datasources: datasources,
httpClientProvider: httpClientProvider,
pluginContextProvider: pluginContextProvider,
clock: clock,
l: l,
metrics: metrics,
writers: gocache.New(cacheExpiration, cacheCleanupInterval),
}
}
@@ -167,7 +176,19 @@ func (w *DatasourceWriter) makeWriter(ctx context.Context, orgID int64, dsUID st
return nil, err
}
ho, err := is.HTTPClientOptions(ctx)
httpClientCtx := ctx
if w.pluginContextProvider != nil {
pluginCtx, err := w.pluginContextProvider.GetWithDataSource(ctx, ds.Type, nil, ds)
if err != nil {
return nil, fmt.Errorf("failed to get plugin context: %w", err)
}
httpClientCtx = backend.WithGrafanaConfig(ctx, pluginCtx.GrafanaConfig)
} else {
// This should not happen, but if the plugin context provider is not set, log a warning.
w.l.Warn("Plugin context provider is not set for the data source writer, PDC-enabled data sources may not work correctly", "datasource_uid", dsUID, "datasource_type", ds.Type)
}
ho, err := is.HTTPClientOptions(httpClientCtx)
if err != nil {
return nil, err
}
@@ -185,10 +206,11 @@ func (w *DatasourceWriter) makeWriter(ctx context.Context, orgID int64, dsUID st
cfg := PrometheusWriterConfig{
URL: u.String(),
HTTPOptions: httpclient.Options{
Timeouts: ho.Timeouts,
TLS: ho.TLS,
BasicAuth: ho.BasicAuth,
Header: headers,
Timeouts: ho.Timeouts,
TLS: ho.TLS,
BasicAuth: ho.BasicAuth,
Header: headers,
ProxyOptions: ho.ProxyOptions,
},
Timeout: w.cfg.Timeout,
}
pkg/services/ngalert/writer/datasourcewriter_test.go
+107 -4
View File
@@ -2,15 +2,19 @@ package writer
import (
"context"
"net/http"
"testing"
"time"
"github.com/benbjohnson/clock"
sdkhttpclient "github.com/grafana/grafana-plugin-sdk-go/backend/httpclient"
"github.com/grafana/grafana-plugin-sdk-go/data"
"github.com/prometheus/client_golang/prometheus"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana/pkg/apimachinery/identity"
"github.com/grafana/grafana/pkg/components/simplejson"
"github.com/grafana/grafana/pkg/infra/httpclient"
"github.com/grafana/grafana/pkg/infra/log"
@@ -19,21 +23,50 @@ import (
"github.com/grafana/grafana/pkg/services/ngalert/metrics"
)
type mockHTTPClientProvider struct {
client *http.Client
lastOptions *sdkhttpclient.Options
callCount int
}
type mockPluginContextProvider struct{}
func (m *mockPluginContextProvider) GetWithDataSource(ctx context.Context, pluginID string, user identity.Requester, ds *datasources.DataSource) (backend.PluginContext, error) {
return backend.PluginContext{}, nil
}
func newMockHTTPClientProvider() *mockHTTPClientProvider {
return &mockHTTPClientProvider{
client: &http.Client{},
}
}
func (m *mockHTTPClientProvider) New(options ...sdkhttpclient.Options) (*http.Client, error) {
m.callCount++
if len(options) > 0 {
opt := options[0]
m.lastOptions = &opt
}
return m.client, nil
}
type testDataSources struct {
dsfakes.FakeDataSourceService
prom1, prom2 *TestRemoteWriteTarget
prom1, prom2, prom3 *TestRemoteWriteTarget
}
func (t *testDataSources) Reset() {
t.prom1.Reset()
t.prom2.Reset()
t.prom3.Reset()
}
func setupDataSources(t *testing.T) *testDataSources {
res := &testDataSources{
prom1: NewTestRemoteWriteTarget(t),
prom2: NewTestRemoteWriteTarget(t),
prom3: NewTestRemoteWriteTarget(t),
}
t.Cleanup(func() {
@@ -42,8 +75,12 @@ func setupDataSources(t *testing.T) *testDataSources {
t.Cleanup(func() {
res.prom2.Close()
})
t.Cleanup(func() {
res.prom3.Close()
})
p1, _ := res.AddDataSource(context.Background(), &datasources.AddDataSourceCommand{
Name: "prom-1",
UID: "prom-1",
Type: datasources.DS_PROMETHEUS,
JsonData: simplejson.MustJson([]byte(`{"prometheusType":"Prometheus"}`)),
@@ -52,6 +89,7 @@ func setupDataSources(t *testing.T) *testDataSources {
res.prom1.ExpectedPath = "/api/v1/write"
p2, _ := res.AddDataSource(context.Background(), &datasources.AddDataSourceCommand{
Name: "prom-2",
UID: "prom-2",
Type: datasources.DS_PROMETHEUS,
JsonData: simplejson.MustJson([]byte(`{"prometheusType":"Mimir"}`)),
@@ -61,10 +99,27 @@ func setupDataSources(t *testing.T) *testDataSources {
// Add a non-Prometheus datasource.
_, _ = res.AddDataSource(context.Background(), &datasources.AddDataSourceCommand{
Name: "loki-1",
UID: "loki-1",
Type: datasources.DS_LOKI,
})
// Add a third Prometheus datasource that uses PDC
p3, _ := res.AddDataSource(context.Background(), &datasources.AddDataSourceCommand{
Name: "prom-3",
UID: "prom-3",
Type: datasources.DS_PROMETHEUS,
JsonData: simplejson.MustJson([]byte(`{
"prometheusType": "Prometheus",
"enableSecureSocksProxy": true,
"secureSocksProxyUsername": "testuser"
}`)),
})
p3.URL = res.prom3.srv.URL
res.prom3.ExpectedPath = "/api/v1/write"
require.True(t, p3.IsSecureSocksDSProxyEnabled())
return res
}
@@ -80,7 +135,8 @@ func TestDatasourceWriter(t *testing.T) {
}
met := metrics.NewRemoteWriterMetrics(prometheus.NewRegistry())
writer := NewDatasourceWriter(cfg, datasources, httpclient.NewProvider(), clock.New(), log.New("test"), met)
pluginContextProvider := &mockPluginContextProvider{}
writer := NewDatasourceWriter(cfg, datasources, httpclient.NewProvider(), pluginContextProvider, clock.New(), log.New("test"), met)
t.Run("when writing a prometheus datasource then the request is made to the expected endpoint", func(t *testing.T) {
datasources.Reset()
@@ -101,7 +157,7 @@ func TestDatasourceWriter(t *testing.T) {
t.Run("when writing an unknown datasource then an error is returned", func(t *testing.T) {
datasources.Reset()
err := writer.WriteDatasource(context.Background(), "prom-3", "metric", time.Now(), frames, 1, map[string]string{})
err := writer.WriteDatasource(context.Background(), "prom-unknown", "metric", time.Now(), frames, 1, map[string]string{})
require.Error(t, err)
require.EqualError(t, err, "data source not found")
})
@@ -136,7 +192,7 @@ func TestDatasourceWriter(t *testing.T) {
DefaultDatasourceUID: "prom-2",
CustomHeaders: headers,
}
writer = NewDatasourceWriter(cfg, datasources, httpclient.NewProvider(), clock.New(), log.New("test"), met)
writer = NewDatasourceWriter(cfg, datasources, httpclient.NewProvider(), pluginContextProvider, clock.New(), log.New("test"), met)
err := writer.WriteDatasource(context.Background(), "prom-1", "metric", time.Now(), frames, 1, map[string]string{})
require.NoError(t, err)
@@ -144,6 +200,53 @@ func TestDatasourceWriter(t *testing.T) {
assert.Equal(t, headers[header1], datasources.prom1.LastHeaders.Get(header1))
assert.Equal(t, headers[header2], datasources.prom1.LastHeaders.Get(header2))
})
t.Run("when PDC is enabled proxy options are passed to HTTP client provider", func(t *testing.T) {
datasources.Reset()
mockProvider := newMockHTTPClientProvider()
cfg := DatasourceWriterConfig{
Timeout: time.Second * 5,
DefaultDatasourceUID: "prom-3",
}
met := metrics.NewRemoteWriterMetrics(prometheus.NewRegistry())
writer := NewDatasourceWriter(cfg, datasources, mockProvider, &mockPluginContextProvider{}, clock.New(), log.New("test"), met)
err := writer.WriteDatasource(context.Background(), "prom-3", "metric", time.Now(), frames, 1, map[string]string{})
require.NoError(t, err)
assert.Equal(t, 1, mockProvider.callCount)
require.NotNil(t, mockProvider.lastOptions)
// Verify that proxy options were configured
require.NotNil(t, mockProvider.lastOptions.ProxyOptions)
require.True(t, mockProvider.lastOptions.ProxyOptions.Enabled)
require.Equal(t, "prom-3", mockProvider.lastOptions.ProxyOptions.DatasourceName)
require.Equal(t, "prometheus", mockProvider.lastOptions.ProxyOptions.DatasourceType)
})
t.Run("when PDC is disabled proxy options are not set", func(t *testing.T) {
datasources.Reset()
mockProvider := newMockHTTPClientProvider()
cfg := DatasourceWriterConfig{
Timeout: time.Second * 5,
DefaultDatasourceUID: "prom-1",
}
met := metrics.NewRemoteWriterMetrics(prometheus.NewRegistry())
writer := NewDatasourceWriter(cfg, datasources, mockProvider, &mockPluginContextProvider{}, clock.New(), log.New("test"), met)
err := writer.WriteDatasource(context.Background(), "prom-1", "metric", time.Now(), frames, 1, map[string]string{})
require.NoError(t, err)
require.Equal(t, 1, mockProvider.callCount)
require.NotNil(t, mockProvider.lastOptions)
require.Nil(t, mockProvider.lastOptions.ProxyOptions)
})
}
func TestDatasourceWriterGetRemoteWriteURL(t *testing.T) {
pkg/services/quota/quotaimpl/quota_test.go
+1 -1
View File
@@ -524,7 +524,7 @@ func setupEnv(t *testing.T, sqlStore db.DB, cfg *setting.Cfg, b bus.Bus, quotaSe
_, err = ngalert.ProvideService(
cfg, featuremgmt.WithFeatures(), nil, nil, routing.NewRouteRegister(), sqlStore, ngalertfakes.NewFakeKVStore(t), nil, nil, quotaService,
secretsService, nil, m, &foldertest.FakeService{}, &acmock.Mock{}, &dashboards.FakeDashboardService{}, nil, b, &acmock.Mock{},
annotationstest.NewFakeAnnotationsRepo(), &pluginstore.FakePluginStore{}, tracer, ruleStore, httpclient.NewProvider(), ngalertfakes.NewFakeReceiverPermissionsService(), usertest.NewUserServiceFake(),
annotationstest.NewFakeAnnotationsRepo(), &pluginstore.FakePluginStore{}, tracer, ruleStore, httpclient.NewProvider(), nil, ngalertfakes.NewFakeReceiverPermissionsService(), usertest.NewUserServiceFake(),
)
require.NoError(t, err)
_, err = storesrv.ProvideService(sqlStore, featuremgmt.WithFeatures(), cfg, quotaService, storesrv.ProvideSystemUsersService())