Grafana Advisor: Datasource checks (#99313)

apps/advisor/kinds/check.cue

@@ -10,6 +10,12 @@ check: {
 				frontend: false
 				backend:  true
 			}
+			validation: {
+				operations: [
+					"CREATE",
+					"UPDATE",
+				]
+			}
 			schema: {
 				spec: {
 					// Generic data input that a check can receive

apps/advisor/pkg/apis/advisor_manifest.go

@@ -27,7 +27,15 @@ var appManifestData = app.ManifestData{
 			Conversion: false,
 			Versions: []app.ManifestKindVersion{
 				{
-					Name:   "v0alpha1",
+					Name: "v0alpha1",
+					Admission: &app.AdmissionCapabilities{
+						Validation: &app.ValidationCapability{
+							Operations: []app.AdmissionOperation{
+								app.AdmissionOperationCreate,
+								app.AdmissionOperationUpdate,
+							},
+						},
+					},
 					Schema: &versionSchemaCheckv0alpha1,
 				},
 			},

apps/advisor/pkg/app/app.go

@@ -2,16 +2,44 @@ package app
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/grafana/grafana-app-sdk/app"
+	"github.com/grafana/grafana-app-sdk/k8s"
 	"github.com/grafana/grafana-app-sdk/resource"
 	"github.com/grafana/grafana-app-sdk/simple"
 	advisorv0alpha1 "github.com/grafana/grafana/apps/advisor/pkg/apis/advisor/v0alpha1"
+	"github.com/grafana/grafana/apps/advisor/pkg/app/checkregistry"
+	"github.com/grafana/grafana/apps/advisor/pkg/app/checks"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/klog/v2"
 )
 
+const (
+	typeLabel        = "advisor.grafana.app/type"
+	statusAnnotation = "advisor.grafana.app/status"
+)
+
 func New(cfg app.Config) (app.App, error) {
+	// Read config
+	checkRegistry, ok := cfg.SpecificConfig.(checkregistry.CheckService)
+	if !ok {
+		return nil, fmt.Errorf("invalid config type")
+	}
+
+	// Prepare storage client
+	clientGenerator := k8s.NewClientRegistry(cfg.KubeConfig, k8s.ClientConfig{})
+	client, err := clientGenerator.ClientFor(advisorv0alpha1.CheckKind())
+	if err != nil {
+		return nil, err
+	}
+
+	// Initialize checks
+	checkMap := map[string]checks.Check{}
+	for _, c := range checkRegistry.Checks() {
+		checkMap[c.Type()] = c
+	}
+
 	simpleConfig := simple.AppConfig{
 		Name:       "advisor",
 		KubeConfig: cfg.KubeConfig,
@@ -23,6 +51,24 @@ func New(cfg app.Config) (app.App, error) {
 		ManagedKinds: []simple.AppManagedKind{
 			{
 				Kind: advisorv0alpha1.CheckKind(),
+				Validator: &simple.Validator{
+					ValidateFunc: func(ctx context.Context, req *app.AdmissionRequest) error {
+						if req.Object != nil {
+							_, err := getCheck(req.Object, checkMap)
+							return err
+						}
+						return nil
+					},
+				},
+				Watcher: &simple.Watcher{
+					AddFunc: func(ctx context.Context, obj resource.Object) error {
+						check, err := getCheck(obj, checkMap)
+						if err != nil {
+							return err
+						}
+						return processCheck(ctx, client, obj, check)
+					},
+				},
 			},
 		},
 	}

apps/advisor/pkg/app/checkregistry/checkregistry.go

@@ -0,0 +1,42 @@
+package checkregistry
+
+import (
+	"github.com/grafana/grafana/apps/advisor/pkg/app/checks"
+	"github.com/grafana/grafana/apps/advisor/pkg/app/checks/datasourcecheck"
+	"github.com/grafana/grafana/pkg/plugins"
+	"github.com/grafana/grafana/pkg/registry/apis/datasource"
+	"github.com/grafana/grafana/pkg/services/datasources"
+	"github.com/grafana/grafana/pkg/services/pluginsintegration/pluginstore"
+)
+
+type CheckService interface {
+	Checks() []checks.Check
+}
+
+type Service struct {
+	datasourceSvc         datasources.DataSourceService
+	pluginStore           pluginstore.Store
+	pluginContextProvider datasource.PluginContextWrapper
+	pluginClient          plugins.Client
+}
+
+func ProvideService(datasourceSvc datasources.DataSourceService, pluginStore pluginstore.Store,
+	pluginContextProvider datasource.PluginContextWrapper, pluginClient plugins.Client) *Service {
+	return &Service{
+		datasourceSvc:         datasourceSvc,
+		pluginStore:           pluginStore,
+		pluginContextProvider: pluginContextProvider,
+		pluginClient:          pluginClient,
+	}
+}
+
+func (s *Service) Checks() []checks.Check {
+	return []checks.Check{
+		datasourcecheck.New(
+			s.datasourceSvc,
+			s.pluginStore,
+			s.pluginContextProvider,
+			s.pluginClient,
+		),
+	}
+}

apps/advisor/pkg/app/checks/datasourcecheck/check.go

@@ -0,0 +1,96 @@
+package datasourcecheck
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	advisor "github.com/grafana/grafana/apps/advisor/pkg/apis/advisor/v0alpha1"
+	"github.com/grafana/grafana/apps/advisor/pkg/app/checks"
+	"github.com/grafana/grafana/pkg/plugins"
+	"github.com/grafana/grafana/pkg/registry/apis/datasource"
+	"github.com/grafana/grafana/pkg/services/datasources"
+	"github.com/grafana/grafana/pkg/services/pluginsintegration/pluginstore"
+	"github.com/grafana/grafana/pkg/util"
+	"k8s.io/klog/v2"
+)
+
+func New(
+	datasourceSvc datasources.DataSourceService,
+	pluginStore pluginstore.Store,
+	pluginContextProvider datasource.PluginContextWrapper,
+	pluginClient plugins.Client,
+) checks.Check {
+	return &check{
+		DatasourceSvc:         datasourceSvc,
+		PluginStore:           pluginStore,
+		PluginContextProvider: pluginContextProvider,
+		PluginClient:          pluginClient,
+	}
+}
+
+type check struct {
+	DatasourceSvc         datasources.DataSourceService
+	PluginStore           pluginstore.Store
+	PluginContextProvider datasource.PluginContextWrapper
+	PluginClient          plugins.Client
+}
+
+func (c *check) Type() string {
+	return "datasource"
+}
+
+func (c *check) Run(ctx context.Context, obj *advisor.CheckSpec) (*advisor.CheckV0alpha1StatusReport, error) {
+	// Optionally read the check input encoded in the object
+	// fmt.Println(obj.Data)
+
+	dss, err := c.DatasourceSvc.GetAllDataSources(ctx, &datasources.GetAllDataSourcesQuery{})
+	if err != nil {
+		return nil, err
+	}
+
+	dsErrs := []advisor.CheckV0alpha1StatusReportErrors{}
+	for _, ds := range dss {
+		// Data source UID validation
+		err := util.ValidateUID(ds.UID)
+		if err != nil {
+			dsErrs = append(dsErrs, advisor.CheckV0alpha1StatusReportErrors{
+				Severity: advisor.CheckStatusSeverityLow,
+				Reason:   fmt.Sprintf("Invalid UID: %s", ds.UID),
+				Action:   "Change UID",
+			})
+		}
+
+		// Health check execution
+		pCtx, err := c.PluginContextProvider.PluginContextForDataSource(ctx, &backend.DataSourceInstanceSettings{
+			Type:       ds.Type,
+			UID:        ds.UID,
+			APIVersion: ds.APIVersion,
+		})
+		if err != nil {
+			klog.ErrorS(err, "Error creating plugin context", "datasource", ds.Name)
+			continue
+		}
+		req := &backend.CheckHealthRequest{
+			PluginContext: pCtx,
+			Headers:       map[string]string{},
+		}
+		resp, err := c.PluginClient.CheckHealth(ctx, req)
+		if err != nil {
+			fmt.Println("Error checking health", err)
+			continue
+		}
+		if resp.Status != backend.HealthStatusOk {
+			dsErrs = append(dsErrs, advisor.CheckV0alpha1StatusReportErrors{
+				Severity: advisor.CheckStatusSeverityHigh,
+				Reason:   fmt.Sprintf("Health check failed: %s", ds.Name),
+				Action:   "Check datasource",
+			})
+		}
+	}
+
+	return &advisor.CheckV0alpha1StatusReport{
+		Count:  int64(len(dss)),
+		Errors: dsErrs,
+	}, nil
+}

apps/advisor/pkg/app/checks/datasourcecheck/check_test.go

@@ -0,0 +1,114 @@
+package datasourcecheck
+
+import (
+	"context"
+	"testing"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	advisor "github.com/grafana/grafana/apps/advisor/pkg/apis/advisor/v0alpha1"
+	"github.com/grafana/grafana/pkg/plugins"
+	"github.com/grafana/grafana/pkg/registry/apis/datasource"
+	"github.com/grafana/grafana/pkg/services/datasources"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestCheck_Run(t *testing.T) {
+	t.Run("should return no errors when all datasources are healthy", func(t *testing.T) {
+		datasources := []*datasources.DataSource{
+			{UID: "valid-uid-1", Type: "prometheus", Name: "Prometheus"},
+			{UID: "valid-uid-2", Type: "mysql", Name: "MySQL"},
+		}
+
+		mockDatasourceSvc := &MockDatasourceSvc{dss: datasources}
+		mockPluginContextProvider := &MockPluginContextProvider{pCtx: backend.PluginContext{}}
+		mockPluginClient := &MockPluginClient{res: &backend.CheckHealthResult{Status: backend.HealthStatusOk}}
+
+		check := &check{
+			DatasourceSvc:         mockDatasourceSvc,
+			PluginContextProvider: mockPluginContextProvider,
+			PluginClient:          mockPluginClient,
+		}
+
+		report, err := check.Run(context.Background(), &advisor.CheckSpec{})
+
+		assert.NoError(t, err)
+		assert.Equal(t, int64(2), report.Count)
+		assert.Empty(t, report.Errors)
+	})
+
+	t.Run("should return errors when datasource UID is invalid", func(t *testing.T) {
+		datasources := []*datasources.DataSource{
+			{UID: "invalid uid", Type: "prometheus", Name: "Prometheus"},
+		}
+
+		mockDatasourceSvc := &MockDatasourceSvc{dss: datasources}
+		mockPluginContextProvider := &MockPluginContextProvider{pCtx: backend.PluginContext{}}
+		mockPluginClient := &MockPluginClient{res: &backend.CheckHealthResult{Status: backend.HealthStatusOk}}
+
+		check := &check{
+			DatasourceSvc:         mockDatasourceSvc,
+			PluginContextProvider: mockPluginContextProvider,
+			PluginClient:          mockPluginClient,
+		}
+
+		report, err := check.Run(context.Background(), &advisor.CheckSpec{})
+
+		assert.NoError(t, err)
+		assert.Equal(t, int64(1), report.Count)
+		assert.Len(t, report.Errors, 1)
+		assert.Equal(t, "Invalid UID: invalid uid", report.Errors[0].Reason)
+	})
+
+	t.Run("should return errors when datasource health check fails", func(t *testing.T) {
+		datasources := []*datasources.DataSource{
+			{UID: "valid-uid-1", Type: "prometheus", Name: "Prometheus"},
+		}
+
+		mockDatasourceSvc := &MockDatasourceSvc{dss: datasources}
+		mockPluginContextProvider := &MockPluginContextProvider{pCtx: backend.PluginContext{}}
+		mockPluginClient := &MockPluginClient{res: &backend.CheckHealthResult{Status: backend.HealthStatusError}}
+
+		check := &check{
+			DatasourceSvc:         mockDatasourceSvc,
+			PluginContextProvider: mockPluginContextProvider,
+			PluginClient:          mockPluginClient,
+		}
+
+		report, err := check.Run(context.Background(), &advisor.CheckSpec{})
+
+		assert.NoError(t, err)
+		assert.Equal(t, int64(1), report.Count)
+		assert.Len(t, report.Errors, 1)
+		assert.Equal(t, "Health check failed: Prometheus", report.Errors[0].Reason)
+	})
+}
+
+type MockDatasourceSvc struct {
+	datasources.DataSourceService
+
+	dss []*datasources.DataSource
+}
+
+func (m *MockDatasourceSvc) GetAllDataSources(ctx context.Context, query *datasources.GetAllDataSourcesQuery) ([]*datasources.DataSource, error) {
+	return m.dss, nil
+}
+
+type MockPluginContextProvider struct {
+	datasource.PluginContextWrapper
+
+	pCtx backend.PluginContext
+}
+
+func (m *MockPluginContextProvider) PluginContextForDataSource(ctx context.Context, datasourceSettings *backend.DataSourceInstanceSettings) (backend.PluginContext, error) {
+	return m.pCtx, nil
+}
+
+type MockPluginClient struct {
+	plugins.Client
+
+	res *backend.CheckHealthResult
+}
+
+func (m *MockPluginClient) CheckHealth(ctx context.Context, req *backend.CheckHealthRequest) (*backend.CheckHealthResult, error) {
+	return m.res, nil
+}

apps/advisor/pkg/app/checks/ifaces.go

@@ -0,0 +1,13 @@
+package checks
+
+import (
+	"context"
+
+	advisorv0alpha1 "github.com/grafana/grafana/apps/advisor/pkg/apis/advisor/v0alpha1"
+)
+
+// Check defines the methods that a check must implement to be executed.
+type Check interface {
+	Run(ctx context.Context, obj *advisorv0alpha1.CheckSpec) (*advisorv0alpha1.CheckV0alpha1StatusReport, error)
+	Type() string
+}

apps/advisor/pkg/app/utils.go

@@ -0,0 +1,95 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	claims "github.com/grafana/authlib/types"
+	"github.com/grafana/grafana-app-sdk/resource"
+	advisorv0alpha1 "github.com/grafana/grafana/apps/advisor/pkg/apis/advisor/v0alpha1"
+	"github.com/grafana/grafana/apps/advisor/pkg/app/checks"
+	"github.com/grafana/grafana/pkg/apimachinery/identity"
+	"github.com/grafana/grafana/pkg/apimachinery/utils"
+	"github.com/grafana/grafana/pkg/services/user"
+)
+
+func getCheck(obj resource.Object, checks map[string]checks.Check) (checks.Check, error) {
+	labels := obj.GetLabels()
+	objTypeLabel, ok := labels[typeLabel]
+	if !ok {
+		return nil, errors.New("missing check type as label")
+	}
+	c, ok := checks[objTypeLabel]
+	if !ok {
+		supportedTypes := ""
+		for k := range checks {
+			supportedTypes += k + ", "
+		}
+		return nil, fmt.Errorf("unknown check type %s. Supported types are: %s", objTypeLabel, supportedTypes)
+	}
+
+	return c, nil
+}
+
+func getStatusAnnotation(obj resource.Object) string {
+	return obj.GetAnnotations()[statusAnnotation]
+}
+
+func setStatusAnnotation(ctx context.Context, client resource.Client, obj resource.Object, status string) error {
+	annotations := obj.GetAnnotations()
+	annotations[statusAnnotation] = status
+	return client.PatchInto(ctx, obj.GetStaticMetadata().Identifier(), resource.PatchRequest{
+		Operations: []resource.PatchOperation{{
+			Operation: resource.PatchOpAdd,
+			Path:      "/metadata/annotations",
+			Value:     annotations,
+		}},
+	}, resource.PatchOptions{}, obj)
+}
+
+func processCheck(ctx context.Context, client resource.Client, obj resource.Object, check checks.Check) error {
+	status := getStatusAnnotation(obj)
+	if status != "" {
+		// Check already processed
+		return nil
+	}
+	c, ok := obj.(*advisorv0alpha1.Check)
+	if !ok {
+		return fmt.Errorf("invalid object type")
+	}
+	// Populate ctx with the user that created the check
+	meta, err := utils.MetaAccessor(obj)
+	if err != nil {
+		return err
+	}
+	createdBy := meta.GetCreatedBy()
+	typ, uid, err := claims.ParseTypeID(createdBy)
+	if err != nil {
+		return err
+	}
+	ctx = identity.WithRequester(ctx, &user.SignedInUser{
+		UserUID:      uid,
+		FallbackType: typ,
+	})
+	// Run the checks
+	report, err := check.Run(ctx, &c.Spec)
+	if err != nil {
+		setErr := setStatusAnnotation(ctx, client, obj, "error")
+		if setErr != nil {
+			return setErr
+		}
+		return err
+	}
+	err = setStatusAnnotation(ctx, client, obj, "processed")
+	if err != nil {
+		return err
+	}
+	return client.PatchInto(ctx, obj.GetStaticMetadata().Identifier(), resource.PatchRequest{
+		Operations: []resource.PatchOperation{{
+			Operation: resource.PatchOpAdd,
+			Path:      "/status/report",
+			Value:     *report,
+		}},
+	}, resource.PatchOptions{}, obj)
+}

apps/advisor/pkg/app/utils_test.go

@@ -0,0 +1,124 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/grafana/grafana-app-sdk/resource"
+	advisorv0alpha1 "github.com/grafana/grafana/apps/advisor/pkg/apis/advisor/v0alpha1"
+	"github.com/grafana/grafana/apps/advisor/pkg/app/checks"
+	"github.com/grafana/grafana/pkg/apimachinery/utils"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGetCheck(t *testing.T) {
+	obj := &advisorv0alpha1.Check{}
+	obj.SetLabels(map[string]string{typeLabel: "testType"})
+
+	checkMap := map[string]checks.Check{
+		"testType": &mockCheck{},
+	}
+
+	check, err := getCheck(obj, checkMap)
+	assert.NoError(t, err)
+	assert.NotNil(t, check)
+}
+
+func TestGetCheck_MissingLabel(t *testing.T) {
+	obj := &advisorv0alpha1.Check{}
+	checkMap := map[string]checks.Check{}
+
+	_, err := getCheck(obj, checkMap)
+	assert.Error(t, err)
+	assert.Equal(t, "missing check type as label", err.Error())
+}
+
+func TestGetCheck_UnknownType(t *testing.T) {
+	obj := &advisorv0alpha1.Check{}
+	obj.SetLabels(map[string]string{typeLabel: "unknownType"})
+
+	checkMap := map[string]checks.Check{
+		"testType": &mockCheck{},
+	}
+
+	_, err := getCheck(obj, checkMap)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "unknown check type unknownType")
+}
+
+func TestSetStatusAnnotation(t *testing.T) {
+	obj := &advisorv0alpha1.Check{}
+	obj.SetAnnotations(map[string]string{})
+	client := &mockClient{}
+	ctx := context.TODO()
+
+	err := setStatusAnnotation(ctx, client, obj, "processed")
+	assert.NoError(t, err)
+	assert.Equal(t, "processed", obj.GetAnnotations()[statusAnnotation])
+}
+
+func TestProcessCheck(t *testing.T) {
+	obj := &advisorv0alpha1.Check{}
+	obj.SetAnnotations(map[string]string{})
+	meta, err := utils.MetaAccessor(obj)
+	if err != nil {
+		t.Fatal(err)
+	}
+	meta.SetCreatedBy("user:1")
+	client := &mockClient{}
+	ctx := context.TODO()
+	check := &mockCheck{}
+
+	err = processCheck(ctx, client, obj, check)
+	assert.NoError(t, err)
+	assert.Equal(t, "processed", obj.GetAnnotations()[statusAnnotation])
+}
+
+func TestProcessCheck_AlreadyProcessed(t *testing.T) {
+	obj := &advisorv0alpha1.Check{}
+	obj.SetAnnotations(map[string]string{statusAnnotation: "processed"})
+	client := &mockClient{}
+	ctx := context.TODO()
+	check := &mockCheck{}
+
+	err := processCheck(ctx, client, obj, check)
+	assert.NoError(t, err)
+}
+
+func TestProcessCheck_RunError(t *testing.T) {
+	obj := &advisorv0alpha1.Check{}
+	obj.SetAnnotations(map[string]string{})
+	meta, err := utils.MetaAccessor(obj)
+	if err != nil {
+		t.Fatal(err)
+	}
+	meta.SetCreatedBy("user:1")
+	client := &mockClient{}
+	ctx := context.TODO()
+
+	check := &mockCheck{
+		err: errors.New("run error"),
+	}
+
+	err = processCheck(ctx, client, obj, check)
+	assert.Error(t, err)
+	assert.Equal(t, "error", obj.GetAnnotations()[statusAnnotation])
+}
+
+type mockClient struct {
+	resource.Client
+}
+
+func (m *mockClient) PatchInto(ctx context.Context, id resource.Identifier, req resource.PatchRequest, opts resource.PatchOptions, obj resource.Object) error {
+	return nil
+}
+
+type mockCheck struct {
+	checks.Check
+	err error
+}
+
+func (m *mockCheck) Run(ctx context.Context, spec *advisorv0alpha1.CheckSpec) (*advisorv0alpha1.CheckV0alpha1StatusReport, error) {
+	return &advisorv0alpha1.CheckV0alpha1StatusReport{}, m.err
+}

pkg/registry/apps/advisor/register.go

@@ -6,6 +6,7 @@ import (
 	"github.com/grafana/grafana/apps/advisor/pkg/apis"
 	advisorv0alpha1 "github.com/grafana/grafana/apps/advisor/pkg/apis/advisor/v0alpha1"
 	advisorapp "github.com/grafana/grafana/apps/advisor/pkg/app"
+	"github.com/grafana/grafana/apps/advisor/pkg/app/checkregistry"
 	"github.com/grafana/grafana/pkg/services/apiserver/builder/runner"
 )
 
@@ -13,11 +14,14 @@ type AdvisorAppProvider struct {
 	app.Provider
 }
 
-func RegisterApp() *AdvisorAppProvider {
+func RegisterApp(
+	checkRegistry checkregistry.CheckService,
+) *AdvisorAppProvider {
 	provider := &AdvisorAppProvider{}
 	appCfg := &runner.AppBuilderConfig{
 		OpenAPIDefGetter: advisorv0alpha1.GetOpenAPIDefinitions,
 		ManagedKinds:     advisorapp.GetKinds(),
+		CustomConfig:     any(checkRegistry),
 	}
 	provider.Provider = simple.NewAppProvider(apis.LocalManifest(), appCfg, advisorapp.New)
 	return provider

pkg/registry/apps/wireset.go

@@ -3,6 +3,7 @@ package appregistry
 import (
 	"github.com/google/wire"
 
+	"github.com/grafana/grafana/apps/advisor/pkg/app/checkregistry"
 	"github.com/grafana/grafana/pkg/registry/apps/advisor"
 	"github.com/grafana/grafana/pkg/registry/apps/investigation"
 	"github.com/grafana/grafana/pkg/registry/apps/playlist"
@@ -13,4 +14,6 @@ var WireSet = wire.NewSet(
 	playlist.RegisterApp,
 	investigation.RegisterApp,
 	advisor.RegisterApp,
+	checkregistry.ProvideService,
+	wire.Bind(new(checkregistry.CheckService), new(*checkregistry.Service)),
 )