CloudMigration: Show warning message for successfully migrated non-core data sources (#91545)

* minor performance improvement

* apply a warning to any non-core plugins that successfully migrate

* commit frontend wip while I refactor some stuff

* update api

* repurpose error dialog to be a generic details dialog

* whitespace

* add unit test

* fixes from testing

* fix migration summary

* add comment

* fix localization stuff

* fix backend test

* reduce number of queries to the db

* some PR feedback

* whitespace

pkg/services/cloudmigration/cloudmigrationimpl/cloudmigration.go

@@ -27,6 +27,7 @@ import (
 	"github.com/grafana/grafana/pkg/services/featuremgmt"
 	"github.com/grafana/grafana/pkg/services/folder"
 	"github.com/grafana/grafana/pkg/services/gcom"
+	"github.com/grafana/grafana/pkg/services/pluginsintegration/pluginstore"
 	"github.com/grafana/grafana/pkg/services/secrets"
 	secretskv "github.com/grafana/grafana/pkg/services/secrets/kvstore"
 	"github.com/grafana/grafana/pkg/services/user"
@@ -57,6 +58,7 @@ type Service struct {
 	gcomService      gcom.Service
 	dashboardService dashboards.DashboardService
 	folderService    folder.Service
+	pluginStore      pluginstore.Store
 	secretsService   secrets.Service
 	kvStore          *kvstore.NamespacedKVStore
 
@@ -90,6 +92,7 @@ func ProvideService(
 	tracer tracing.Tracer,
 	dashboardService dashboards.DashboardService,
 	folderService folder.Service,
+	pluginStore pluginstore.Store,
 	kvStore kvstore.KVStore,
 ) (cloudmigration.Service, error) {
 	if !features.IsEnabledGlobally(featuremgmt.FlagOnPremToCloudMigrations) {
@@ -107,6 +110,7 @@ func ProvideService(
 		secretsService:   secretsService,
 		dashboardService: dashboardService,
 		folderService:    folderService,
+		pluginStore:      pluginStore,
 		kvStore:          kvstore.WithNamespace(kvStore, 0, "cloudmigration"),
 	}
 	s.api = api.RegisterApi(routeRegister, s, tracer)
@@ -590,12 +594,19 @@ func (s *Service) GetSnapshot(ctx context.Context, query cloudmigration.GetSnaps
 			return snapshot, nil
 		}
 
+		// For 11.2 we only support core data sources. Apply a warning for any non-core ones before storing.
+		resources, err := s.getResourcesWithPluginWarnings(ctx, snapshotMeta.Results)
+		if err != nil {
+			// treat this as non-fatal since the migration still succeeded
+			s.log.Error("error applying plugin warnings, please open a bug report: %w", err)
+		}
+
 		// We need to update the snapshot in our db before reporting anything
 		if err := s.store.UpdateSnapshot(ctx, cloudmigration.UpdateSnapshotCmd{
 			UID:       snapshot.UID,
 			SessionID: sessionUid,
 			Status:    localStatus,
-			Resources: snapshotMeta.Results,
+			Resources: resources,
 		}); err != nil {
 			return nil, fmt.Errorf("error updating snapshot status: %w", err)
 		}
@@ -777,3 +788,39 @@ func (s *Service) getLocalEventId(ctx context.Context) (string, error) {
 
 	return anonId, nil
 }
+
+// getResourcesWithPluginWarnings iterates through each resource and, if a non-core datasource, applies a warning that we only support core
+func (s *Service) getResourcesWithPluginWarnings(ctx context.Context, results []cloudmigration.CloudMigrationResource) ([]cloudmigration.CloudMigrationResource, error) {
+	dsList, err := s.dsService.GetAllDataSources(ctx, &datasources.GetAllDataSourcesQuery{})
+	if err != nil {
+		return nil, fmt.Errorf("getting all data sources: %w", err)
+	}
+	dsMap := make(map[string]*datasources.DataSource, len(dsList))
+	for i := 0; i < len(dsList); i++ {
+		dsMap[dsList[i].UID] = dsList[i]
+	}
+
+	for i := 0; i < len(results); i++ {
+		r := results[i]
+
+		if r.Type == cloudmigration.DatasourceDataType &&
+			r.Error == "" { // any error returned by GMS takes priority
+			ds, ok := dsMap[r.RefID]
+			if !ok {
+				s.log.Error("data source with id %s was not found in data sources list", r.RefID)
+				continue
+			}
+
+			p, found := s.pluginStore.Plugin(ctx, ds.Type)
+			// if the plugin is not found, it means it was uninstalled, meaning it wasn't core
+			if !p.IsCorePlugin() || !found {
+				r.Status = cloudmigration.ItemStatusWarning
+				r.Error = "Only core data sources are supported. Please ensure the plugin is installed on the cloud stack."
+			}
+
+			results[i] = r
+		}
+	}
+
+	return results, nil
+}

pkg/services/cloudmigration/cloudmigrationimpl/cloudmigration_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/grafana/grafana/pkg/infra/db"
 	"github.com/grafana/grafana/pkg/infra/kvstore"
 	"github.com/grafana/grafana/pkg/infra/tracing"
+	"github.com/grafana/grafana/pkg/plugins"
 	"github.com/grafana/grafana/pkg/services/cloudmigration"
 	"github.com/grafana/grafana/pkg/services/cloudmigration/gmsclient"
 	"github.com/grafana/grafana/pkg/services/contexthandler/ctxkey"
@@ -23,6 +24,7 @@ import (
 	"github.com/grafana/grafana/pkg/services/featuremgmt"
 	"github.com/grafana/grafana/pkg/services/folder"
 	"github.com/grafana/grafana/pkg/services/folder/foldertest"
+	"github.com/grafana/grafana/pkg/services/pluginsintegration/pluginstore"
 	secretsfakes "github.com/grafana/grafana/pkg/services/secrets/fakes"
 	secretskv "github.com/grafana/grafana/pkg/services/secrets/kvstore"
 	"github.com/grafana/grafana/pkg/services/user"
@@ -399,6 +401,122 @@ func Test_SortFolders(t *testing.T) {
 	require.Equal(t, expected, sortedFolders)
 }
 
+func Test_NonCoreDataSourcesHaveWarning(t *testing.T) {
+	s := setUpServiceTest(t, false).(*Service)
+
+	// Insert a processing snapshot into the database before we start so we query GMS
+	sess, err := s.store.CreateMigrationSession(context.Background(), cloudmigration.CloudMigrationSession{})
+	require.NoError(t, err)
+	snapshotUid, err := s.store.CreateSnapshot(context.Background(), cloudmigration.CloudMigrationSnapshot{
+		UID:            uuid.NewString(),
+		SessionUID:     sess.UID,
+		Status:         cloudmigration.SnapshotStatusProcessing,
+		GMSSnapshotUID: "gms uid",
+	})
+	require.NoError(t, err)
+
+	// GMS should return: a core ds, a non-core ds, a non-core ds with an error, and a ds that has been uninstalled
+	gmsClientMock := &gmsClientMock{
+		getSnapshotResponse: &cloudmigration.GetSnapshotStatusResponse{
+			State: cloudmigration.SnapshotStateFinished,
+			Results: []cloudmigration.CloudMigrationResource{
+				{
+					Type:        cloudmigration.DatasourceDataType,
+					RefID:       "1", // this will be core
+					Status:      cloudmigration.ItemStatusOK,
+					SnapshotUID: snapshotUid,
+				},
+				{
+					Type:        cloudmigration.DatasourceDataType,
+					RefID:       "2", // this will be non-core
+					Status:      cloudmigration.ItemStatusOK,
+					SnapshotUID: snapshotUid,
+				},
+				{
+					Type:        cloudmigration.DatasourceDataType,
+					RefID:       "3", // this will be non-core with an error
+					Status:      cloudmigration.ItemStatusError,
+					Error:       "please don't overwrite me",
+					SnapshotUID: snapshotUid,
+				},
+				{
+					Type:        cloudmigration.DatasourceDataType,
+					RefID:       "4", // this will be deleted
+					Status:      cloudmigration.ItemStatusOK,
+					SnapshotUID: snapshotUid,
+				},
+			},
+		},
+	}
+	s.gmsClient = gmsClientMock
+
+	// Update the internal plugin store and ds store with seed data matching the descriptions above
+	s.pluginStore = pluginstore.NewFakePluginStore([]pluginstore.Plugin{
+		{
+			JSONData: plugins.JSONData{
+				ID: "1",
+			},
+			Class: plugins.ClassCore,
+		},
+		{
+			JSONData: plugins.JSONData{
+				ID: "2",
+			},
+			Class: plugins.ClassExternal,
+		},
+		{
+			JSONData: plugins.JSONData{
+				ID: "3",
+			},
+			Class: plugins.ClassExternal,
+		},
+	}...)
+
+	s.dsService = &datafakes.FakeDataSourceService{
+		DataSources: []*datasources.DataSource{
+			{UID: "1", Type: "1"},
+			{UID: "2", Type: "2"},
+			{UID: "3", Type: "3"},
+			{UID: "4", Type: "4"},
+		},
+	}
+
+	// Retrieve the snapshot with results
+	snapshot, err := s.GetSnapshot(ctxWithSignedInUser(), cloudmigration.GetSnapshotsQuery{
+		SnapshotUID: snapshotUid,
+		SessionUID:  sess.UID,
+		ResultPage:  1,
+		ResultLimit: 10,
+	})
+	assert.NoError(t, err)
+	assert.Len(t, snapshot.Resources, 4)
+
+	findRef := func(id string) *cloudmigration.CloudMigrationResource {
+		for _, r := range snapshot.Resources {
+			if r.RefID == id {
+				return &r
+			}
+		}
+		return nil
+	}
+
+	shouldBeUnaltered := findRef("1")
+	assert.Equal(t, cloudmigration.ItemStatusOK, shouldBeUnaltered.Status)
+	assert.Empty(t, shouldBeUnaltered.Error)
+
+	shouldBeAltered := findRef("2")
+	assert.Equal(t, cloudmigration.ItemStatusWarning, shouldBeAltered.Status)
+	assert.Equal(t, shouldBeAltered.Error, "Only core data sources are supported. Please ensure the plugin is installed on the cloud stack.")
+
+	shouldHaveOriginalError := findRef("3")
+	assert.Equal(t, cloudmigration.ItemStatusError, shouldHaveOriginalError.Status)
+	assert.Equal(t, shouldHaveOriginalError.Error, "please don't overwrite me")
+
+	uninstalledAltered := findRef("4")
+	assert.Equal(t, cloudmigration.ItemStatusWarning, uninstalledAltered.Status)
+	assert.Equal(t, uninstalledAltered.Error, "Only core data sources are supported. Please ensure the plugin is installed on the cloud stack.")
+}
+
 func ctxWithSignedInUser() context.Context {
 	c := &contextmodel.ReqContext{
 		SignedInUser: &user.SignedInUser{OrgID: 1},
@@ -461,6 +579,7 @@ func setUpServiceTest(t *testing.T, withDashboardMock bool) cloudmigration.Servi
 		tracer,
 		dashboardService,
 		mockFolder,
+		&pluginstore.FakePluginStore{},
 		kvstore.ProvideService(sqlStore),
 	)
 	require.NoError(t, err)

pkg/services/cloudmigration/cloudmigrationimpl/xorm_store.go

@@ -294,11 +294,6 @@ func (ss *sqlStore) GetSnapshotList(ctx context.Context, query cloudmigration.Li
 // CreateUpdateSnapshotResources either updates a migration resource for a snapshot, or creates it if it does not exist
 // If the uid is not known, it uses snapshot_uid + resource_uid as a lookup
 func (ss *sqlStore) CreateUpdateSnapshotResources(ctx context.Context, snapshotUid string, resources []cloudmigration.CloudMigrationResource) error {
-	// ensure snapshot_uids are consistent so that we can use them to query when uid isn't known
-	for i := 0; i < len(resources); i++ {
-		resources[i].SnapshotUID = snapshotUid
-	}
-
 	return ss.db.InTransaction(ctx, func(ctx context.Context) error {
 		sql := "UPDATE cloud_migration_resource SET status=?, error_string=? WHERE uid=? OR (snapshot_uid=? AND resource_uid=?)"
 		err := ss.db.WithDbSession(ctx, func(sess *sqlstore.DBSession) error {
@@ -314,6 +309,8 @@ func (ss *sqlStore) CreateUpdateSnapshotResources(ctx context.Context, snapshotU
 					return err
 				} else if n == 0 {
 					r.UID = util.GenerateShortUID()
+					// ensure snapshot_uids are consistent so that we can use them to query when uid isn't known
+					r.SnapshotUID = snapshotUid
 					_, err := sess.Insert(r)
 					if err != nil {
 						return err