Correlations: Add legacy storage (#112038)

pkg/registry/apps/apps.go

@@ -34,7 +34,7 @@ func ProvideAppInstallers(
 	pluginsApplInstaller *plugins.PluginsAppInstaller,
 	shorturlAppInstaller *shorturl.ShortURLAppInstaller,
 	rulesAppInstaller *rules.AlertingRulesAppInstaller,
-	correlationsAppInstaller *correlations.CorrelationsAppInstaller,
+	correlationsAppInstaller *correlations.AppInstaller,
 	alertingNotificationAppInstaller *notifications.AlertingNotificationsAppInstaller,
 ) []appsdkapiserver.AppInstaller {
 	installers := []appsdkapiserver.AppInstaller{

pkg/registry/apps/apps_test.go

@@ -17,7 +17,7 @@ func TestProvideAppInstallers_Table(t *testing.T) {
 	playlistInstaller := &playlist.PlaylistAppInstaller{}
 	pluginsInstaller := &plugins.PluginsAppInstaller{}
 	rulesInstaller := &rules.AlertingRulesAppInstaller{}
-	correlationsAppInstaller := &correlations.CorrelationsAppInstaller{}
+	correlationsAppInstaller := &correlations.AppInstaller{}
 	notificationsAppInstaller := &notifications.AlertingNotificationsAppInstaller{}
 
 	tests := []struct {

pkg/registry/apps/correlations/legacy_storage.go

@@ -0,0 +1,195 @@
+package correlations
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"k8s.io/apimachinery/pkg/apis/meta/internalversion"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/selection"
+	"k8s.io/apiserver/pkg/registry/rest"
+
+	correlationsV0 "github.com/grafana/grafana/apps/correlations/pkg/apis/correlation/v0alpha1"
+	"github.com/grafana/grafana/pkg/services/apiserver/endpoints/request"
+	"github.com/grafana/grafana/pkg/services/correlations"
+)
+
+var (
+	_ rest.Scoper               = (*legacyStorage)(nil)
+	_ rest.SingularNameProvider = (*legacyStorage)(nil)
+	_ rest.Getter               = (*legacyStorage)(nil)
+	_ rest.Storage              = (*legacyStorage)(nil)
+	_ rest.Creater              = (*legacyStorage)(nil)
+	_ rest.Updater              = (*legacyStorage)(nil)
+	_ rest.GracefulDeleter      = (*legacyStorage)(nil)
+)
+
+type legacyStorage struct {
+	service        correlations.Service
+	namespacer     request.NamespaceMapper
+	tableConverter rest.TableConvertor
+}
+
+func (s *legacyStorage) New() runtime.Object {
+	return correlationsV0.CorrelationKind().ZeroValue()
+}
+
+func (s *legacyStorage) Destroy() {}
+
+func (s *legacyStorage) NamespaceScoped() bool {
+	return true // namespace == org
+}
+
+func (s *legacyStorage) GetSingularName() string {
+	return strings.ToLower(correlationsV0.CorrelationKind().Kind())
+}
+
+func (s *legacyStorage) NewList() runtime.Object {
+	return correlationsV0.CorrelationKind().ZeroListValue()
+}
+
+func (s *legacyStorage) ConvertToTable(ctx context.Context, object runtime.Object, tableOptions runtime.Object) (*metav1.Table, error) {
+	return s.tableConverter.ConvertToTable(ctx, object, tableOptions)
+}
+
+func (s *legacyStorage) List(ctx context.Context, options *internalversion.ListOptions) (runtime.Object, error) {
+	orgID, err := request.OrgIDForList(ctx)
+	if err != nil {
+		return nil, err
+	}
+	uids := []string{}
+	if options.FieldSelector != nil {
+		for _, r := range options.FieldSelector.Requirements() {
+			switch r.Field {
+			case "spec.datasource.name":
+				switch r.Operator {
+				case selection.Equals, selection.DoubleEquals:
+					uids = []string{r.Value}
+				case selection.In:
+					uids = strings.Split(r.Value, ";") // ??? not sure how/if this supports multiple values
+				default:
+					return nil, fmt.Errorf("unsupported operation")
+				}
+			default:
+				return nil, fmt.Errorf("unsupported field")
+			}
+		}
+	}
+
+	if options.Continue != "" {
+		return nil, fmt.Errorf("paging not yet supported")
+	}
+
+	rsp, err := s.service.GetCorrelations(ctx, correlations.GetCorrelationsQuery{
+		OrgId:      orgID,
+		Limit:      1000,
+		SourceUIDs: uids,
+	})
+	if err != nil {
+		return nil, err
+	}
+	list := &correlationsV0.CorrelationList{
+		Items: make([]correlationsV0.Correlation, len(rsp.Correlations)),
+	}
+	for i, orig := range rsp.Correlations {
+		c, err := correlations.ToResource(orig, s.namespacer)
+		if err != nil {
+			return nil, err
+		}
+		list.Items[i] = *c
+	}
+	return list, nil
+}
+
+func (s *legacyStorage) Get(ctx context.Context, name string, options *metav1.GetOptions) (runtime.Object, error) {
+	orgID, err := request.OrgIDForList(ctx)
+	if err != nil {
+		return nil, err
+	}
+	c, err := s.service.GetCorrelation(ctx, correlations.GetCorrelationQuery{
+		UID:   name,
+		OrgId: orgID,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return correlations.ToResource(c, s.namespacer)
+}
+
+func (s *legacyStorage) Create(ctx context.Context,
+	obj runtime.Object,
+	createValidation rest.ValidateObjectFunc,
+	options *metav1.CreateOptions,
+) (runtime.Object, error) {
+	resource, ok := obj.(*correlationsV0.Correlation)
+	if !ok {
+		return nil, fmt.Errorf("expected correlation")
+	}
+
+	cmd, err := correlations.ToCreateCorrelationCommand(resource)
+	if err != nil {
+		return nil, err
+	}
+
+	out, err := s.service.CreateCorrelation(ctx, *cmd)
+	if err != nil {
+		return nil, err
+	}
+	return s.Get(ctx, out.UID, &metav1.GetOptions{})
+}
+
+func (s *legacyStorage) Update(ctx context.Context,
+	name string,
+	objInfo rest.UpdatedObjectInfo,
+	createValidation rest.ValidateObjectFunc,
+	updateValidation rest.ValidateObjectUpdateFunc,
+	forceAllowCreate bool,
+	options *metav1.UpdateOptions,
+) (runtime.Object, bool, error) {
+	before, err := s.Get(ctx, name, &metav1.GetOptions{})
+	if err != nil {
+		return nil, false, err
+	}
+	obj, err := objInfo.UpdatedObject(ctx, before)
+	if err != nil {
+		return nil, false, err
+	}
+
+	resource, ok := obj.(*correlationsV0.Correlation)
+	if !ok {
+		return nil, false, fmt.Errorf("expected correlation")
+	}
+
+	cmd, err := correlations.ToUpdateCorrelationCommand(resource)
+	if err != nil {
+		return nil, false, err
+	}
+
+	out, err := s.service.UpdateCorrelation(ctx, *cmd)
+	if err != nil {
+		return nil, false, err
+	}
+	obj, err = s.Get(ctx, out.UID, &metav1.GetOptions{})
+	return obj, false, err
+}
+
+// GracefulDeleter
+func (s *legacyStorage) Delete(ctx context.Context, name string, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions) (runtime.Object, bool, error) {
+	orgID, err := request.OrgIDForList(ctx)
+	if err != nil {
+		return nil, false, err
+	}
+	err = s.service.DeleteCorrelation(ctx, correlations.DeleteCorrelationCommand{
+		OrgId: orgID,
+		UID:   name,
+	})
+	return nil, (err == nil), err
+}
+
+// CollectionDeleter
+func (s *legacyStorage) DeleteCollection(ctx context.Context, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions, listOptions *internalversion.ListOptions) (runtime.Object, error) {
+	return nil, fmt.Errorf("DeleteCollection for shorturl not implemented")
+}

pkg/registry/apps/correlations/register.go

@@ -1,33 +1,44 @@
 package correlations
 
 import (
+	"fmt"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	restclient "k8s.io/client-go/rest"
 
 	"github.com/grafana/grafana-app-sdk/app"
 	appsdkapiserver "github.com/grafana/grafana-app-sdk/k8s/apiserver"
 	"github.com/grafana/grafana-app-sdk/simple"
 	"github.com/grafana/grafana/apps/correlations/pkg/apis"
+	correlationsV0 "github.com/grafana/grafana/apps/correlations/pkg/apis/correlation/v0alpha1"
 	correlationsapp "github.com/grafana/grafana/apps/correlations/pkg/app"
+	"github.com/grafana/grafana/pkg/apimachinery/utils"
+	"github.com/grafana/grafana/pkg/apiserver/rest"
+	"github.com/grafana/grafana/pkg/services/apiserver/appinstaller"
+	"github.com/grafana/grafana/pkg/services/apiserver/endpoints/request"
+	"github.com/grafana/grafana/pkg/services/correlations"
 	"github.com/grafana/grafana/pkg/services/featuremgmt"
 	"github.com/grafana/grafana/pkg/setting"
 )
 
 var (
-	_ appsdkapiserver.AppInstaller = (*CorrelationsAppInstaller)(nil)
+	_ appsdkapiserver.AppInstaller       = (*AppInstaller)(nil)
+	_ appinstaller.LegacyStorageProvider = (*AppInstaller)(nil)
 )
 
-type CorrelationsAppInstaller struct {
+type AppInstaller struct {
 	appsdkapiserver.AppInstaller
-	cfg *setting.Cfg
+
+	legacy *legacyStorage
 }
 
 func RegisterAppInstaller(
 	cfg *setting.Cfg,
 	features featuremgmt.FeatureToggles,
-) (*CorrelationsAppInstaller, error) {
-	installer := &CorrelationsAppInstaller{
-		cfg: cfg,
-	}
+	service correlations.Service,
+) (*AppInstaller, error) {
+	installer := &AppInstaller{}
 	provider := simple.NewAppProvider(apis.LocalManifest(), nil, correlationsapp.New)
 
 	appConfig := app.Config{
@@ -40,5 +51,48 @@ func RegisterAppInstaller(
 	}
 	installer.AppInstaller = i
 
+	if service != nil {
+		installer.legacy = &legacyStorage{
+			service:    service,
+			namespacer: request.GetNamespaceMapper(cfg),
+		}
+	}
 	return installer, nil
 }
+
+func (a *AppInstaller) GetLegacyStorage(requested schema.GroupVersionResource) rest.Storage {
+	kind := correlationsV0.CorrelationKind()
+	gvr := schema.GroupVersionResource{
+		Group:    kind.Group(),
+		Version:  kind.Version(),
+		Resource: kind.Plural(),
+	}
+	if requested.String() != gvr.String() {
+		return nil
+	}
+	a.legacy.tableConverter = utils.NewTableConverter(
+		gvr.GroupResource(),
+		utils.TableColumns{
+			Definition: []metav1.TableColumnDefinition{
+				{Name: "Name", Type: "string", Format: "name"},
+				{Name: "Source", Type: "string", Format: "string"},
+				{Name: "Target", Type: "string", Format: "string"},
+				{Name: "Description", Type: "string", Format: "string"},
+			},
+			Reader: func(obj any) ([]any, error) {
+				m, ok := obj.(*correlationsV0.Correlation)
+				if !ok {
+					return nil, fmt.Errorf("expected Correlation")
+				}
+				return []any{
+					m.Name,
+					m.Spec.Source.Name,
+					m.Spec.Target.Name,
+					m.Spec.Description,
+				}, nil
+			},
+		},
+	)
+
+	return a.legacy
+}

pkg/server/wire_gen.go

@@ -761,7 +761,7 @@ func Initialize(ctx context.Context, cfg *setting.Cfg, opts Options, apiOpts api
 	if err != nil {
 		return nil, err
 	}
-	correlationsAppInstaller, err := correlations2.RegisterAppInstaller(cfg, featureToggles)
+	appInstaller, err := correlations2.RegisterAppInstaller(cfg, featureToggles, correlationsService)
 	if err != nil {
 		return nil, err
 	}
@@ -769,7 +769,7 @@ func Initialize(ctx context.Context, cfg *setting.Cfg, opts Options, apiOpts api
 	if err != nil {
 		return nil, err
 	}
-	v2 := appregistry.ProvideAppInstallers(featureToggles, playlistAppInstaller, pluginsAppInstaller, shortURLAppInstaller, alertingRulesAppInstaller, correlationsAppInstaller, alertingNotificationsAppInstaller)
+	v2 := appregistry.ProvideAppInstallers(featureToggles, playlistAppInstaller, pluginsAppInstaller, shortURLAppInstaller, alertingRulesAppInstaller, appInstaller, alertingNotificationsAppInstaller)
 	builderMetrics := builder.ProvideBuilderMetrics(registerer)
 	apiserverService, err := apiserver.ProvideService(cfg, featureToggles, routeRegisterImpl, tracingService, serverLockService, sqlStore, kvStore, middlewareHandler, scopedPluginDatasourceProvider, plugincontextProvider, pluginstoreService, dualwriteService, resourceClient, inlineSecureValueSupport, eventualRestConfigProvider, v, eventualRestConfigProvider, registerer, aggregatorRunner, v2, builderMetrics)
 	if err != nil {
@@ -1365,7 +1365,7 @@ func InitializeForTest(ctx context.Context, t sqlutil.ITestDB, testingT interfac
 	if err != nil {
 		return nil, err
 	}
-	correlationsAppInstaller, err := correlations2.RegisterAppInstaller(cfg, featureToggles)
+	appInstaller, err := correlations2.RegisterAppInstaller(cfg, featureToggles, correlationsService)
 	if err != nil {
 		return nil, err
 	}
@@ -1373,7 +1373,7 @@ func InitializeForTest(ctx context.Context, t sqlutil.ITestDB, testingT interfac
 	if err != nil {
 		return nil, err
 	}
-	v2 := appregistry.ProvideAppInstallers(featureToggles, playlistAppInstaller, pluginsAppInstaller, shortURLAppInstaller, alertingRulesAppInstaller, correlationsAppInstaller, alertingNotificationsAppInstaller)
+	v2 := appregistry.ProvideAppInstallers(featureToggles, playlistAppInstaller, pluginsAppInstaller, shortURLAppInstaller, alertingRulesAppInstaller, appInstaller, alertingNotificationsAppInstaller)
 	builderMetrics := builder.ProvideBuilderMetrics(registerer)
 	apiserverService, err := apiserver.ProvideService(cfg, featureToggles, routeRegisterImpl, tracingService, serverLockService, sqlStore, kvStore, middlewareHandler, scopedPluginDatasourceProvider, plugincontextProvider, pluginstoreService, dualwriteService, resourceClient, inlineSecureValueSupport, eventualRestConfigProvider, v, eventualRestConfigProvider, registerer, aggregatorRunner, v2, builderMetrics)
 	if err != nil {

pkg/services/correlations/conversions.go

@@ -0,0 +1,151 @@
+package correlations
+
+import (
+	"encoding/json"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"
+
+	authlib "github.com/grafana/authlib/types"
+	correlationsV0 "github.com/grafana/grafana/apps/correlations/pkg/apis/correlation/v0alpha1"
+	"github.com/grafana/grafana/pkg/apimachinery/utils"
+)
+
+func ToResource(orig Correlation, namespacer authlib.NamespaceFormatter) (*correlationsV0.Correlation, error) {
+	cfg, err := ToSpecConfig(orig.Config)
+	if err != nil {
+		return nil, err
+	}
+	obj := &correlationsV0.Correlation{
+		ObjectMeta: v1.ObjectMeta{
+			Name:      orig.UID,
+			Namespace: namespacer(orig.OrgID),
+		},
+		Spec: correlationsV0.CorrelationSpec{
+			Label: orig.Label,
+			Type:  correlationsV0.CorrelationCorrelationType(orig.Type),
+			Source: correlationsV0.CorrelationDataSourceRef{
+				Group: ptr.Deref(orig.SourceType, ""),
+				Name:  orig.SourceUID,
+			},
+			Config: *cfg,
+		},
+	}
+	if orig.TargetUID != nil {
+		obj.Spec.Target = &correlationsV0.CorrelationDataSourceRef{
+			Group: ptr.Deref(orig.TargetType, ""),
+			Name:  *orig.TargetUID,
+		}
+	}
+	if orig.Description != "" {
+		obj.Spec.Description = &orig.Description
+	}
+	if orig.Provisioned {
+		tmp, _ := utils.MetaAccessor(obj)
+		tmp.SetManagerProperties(utils.ManagerProperties{
+			Kind: utils.ManagerKindClassicFP, // nolint:staticcheck
+		})
+	}
+	return obj, nil
+}
+
+func ToCorrelation(obj *correlationsV0.Correlation) (*Correlation, error) {
+	ns, err := authlib.ParseNamespace(obj.Namespace)
+	if err != nil {
+		return nil, err
+	}
+	cfg, err := ToConfig(obj.Spec.Config)
+	if err != nil {
+		return nil, err
+	}
+	result := &Correlation{
+		UID:         obj.Name,
+		OrgID:       ns.OrgID,
+		Label:       obj.Spec.Label,
+		Description: ptr.Deref(obj.Spec.Description, ""),
+		SourceUID:   obj.Spec.Source.Name,
+		SourceType:  ptr.To(obj.Spec.Source.Group),
+		Type:        CorrelationType(obj.Spec.Type),
+		Config:      *cfg,
+	}
+	if obj.Annotations[utils.AnnoKeyManagerKind] != "" {
+		result.Provisioned = true
+	}
+	if obj.Spec.Target != nil {
+		result.TargetUID = &obj.Spec.Target.Name
+		result.TargetType = ptr.To(obj.Spec.Target.Group)
+	}
+	return result, nil
+}
+
+func ToSpecConfig(orig CorrelationConfig) (*correlationsV0.CorrelationConfigSpec, error) {
+	out := &correlationsV0.CorrelationConfigSpec{}
+	raw, err := json.Marshal(orig)
+	if err != nil {
+		return nil, err
+	}
+	err = json.Unmarshal(raw, out)
+	if err != nil {
+		return nil, err
+	}
+	if len(out.Target) == 0 {
+		out.Target = nil
+	}
+	return out, err
+}
+
+func ToConfig(orig correlationsV0.CorrelationConfigSpec) (*CorrelationConfig, error) {
+	out := &CorrelationConfig{}
+	raw, err := json.Marshal(orig)
+	if err != nil {
+		return nil, err
+	}
+	err = json.Unmarshal(raw, out)
+	if err != nil {
+		return nil, err
+	}
+	if len(out.Target) == 0 {
+		out.Target = nil
+	}
+	return out, err
+}
+
+func ToUpdateCorrelationCommand(obj *correlationsV0.Correlation) (*UpdateCorrelationCommand, error) {
+	tmp, err := ToCorrelation(obj)
+	if err != nil {
+		return nil, err
+	}
+	if tmp.Config.Target == nil {
+		tmp.Config.Target = map[string]any{} // replace it
+	}
+	return &UpdateCorrelationCommand{
+		UID:         tmp.UID,
+		OrgId:       tmp.OrgID,
+		SourceUID:   tmp.SourceUID,
+		Label:       &tmp.Label,
+		Description: &tmp.Description,
+		Type:        &tmp.Type,
+		Config: &CorrelationConfigUpdateDTO{
+			Field:           &tmp.Config.Field,
+			Target:          &tmp.Config.Target,
+			Transformations: tmp.Config.Transformations,
+		},
+	}, nil
+}
+
+func ToCreateCorrelationCommand(obj *correlationsV0.Correlation) (*CreateCorrelationCommand, error) {
+	tmp, err := ToCorrelation(obj)
+	if err != nil {
+		return nil, err
+	}
+	return &CreateCorrelationCommand{
+		OrgId:       tmp.OrgID,
+		SourceUID:   tmp.SourceUID,
+		TargetUID:   tmp.TargetUID,
+		Label:       tmp.Label,
+		Description: tmp.Description,
+		Config:      tmp.Config,
+		Type:        tmp.Type,
+		Provisioned: tmp.Provisioned,
+	}, nil
+}

pkg/services/correlations/conversions_test.go

@@ -0,0 +1,112 @@
+package correlations
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"
+
+	authlib "github.com/grafana/authlib/types"
+	correlationsV0 "github.com/grafana/grafana/apps/correlations/pkg/apis/correlation/v0alpha1"
+)
+
+func TestConversion(t *testing.T) {
+	namespacer := authlib.OrgNamespaceFormatter
+
+	tests := []struct {
+		name   string
+		input  Correlation
+		expect correlationsV0.Correlation
+		create CreateCorrelationCommand
+		update UpdateCorrelationCommand
+	}{
+		{
+			name: "Basic fields",
+			input: Correlation{
+				UID:         "uid",
+				OrgID:       2,
+				Label:       "Test Label",
+				Type:        query,
+				SourceUID:   "source",
+				SourceType:  ptr.To("source-type"),
+				TargetUID:   ptr.To("target"),
+				TargetType:  ptr.To("target-type"),
+				Description: "A test correlation",
+				Provisioned: true,
+				Config: CorrelationConfig{
+					Field: "test-field",
+				},
+			},
+			expect: correlationsV0.Correlation{
+				ObjectMeta: v1.ObjectMeta{
+					Name:      "uid",
+					Namespace: "org-2",
+					Annotations: map[string]string{
+						"grafana.app/managedBy": "classic-file-provisioning",
+					},
+				},
+				Spec: correlationsV0.CorrelationSpec{
+					Description: ptr.To("A test correlation"),
+					Label:       "Test Label",
+					Type:        correlationsV0.CorrelationCorrelationTypeQuery,
+					Source: correlationsV0.CorrelationDataSourceRef{
+						Group: "source-type",
+						Name:  "source",
+					},
+					Target: &correlationsV0.CorrelationDataSourceRef{
+						Group: "target-type",
+						Name:  "target",
+					},
+					Config: correlationsV0.CorrelationConfigSpec{
+						Field: "test-field",
+					},
+				},
+			},
+			create: CreateCorrelationCommand{
+				OrgId:       2,
+				Label:       "Test Label",
+				Type:        query,
+				SourceUID:   "source",
+				TargetUID:   ptr.To("target"),
+				Description: "A test correlation",
+				Provisioned: true,
+				Config: CorrelationConfig{
+					Field: "test-field",
+				},
+			},
+			update: UpdateCorrelationCommand{
+				UID:         "uid",
+				OrgId:       2,
+				Label:       ptr.To("Test Label"),
+				Type:        ptr.To(query),
+				SourceUID:   "source",
+				Description: ptr.To("A test correlation"),
+				Config: &CorrelationConfigUpdateDTO{
+					Field:  ptr.To("test-field"),
+					Target: &map[string]any{},
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			res, err := ToResource(tt.input, namespacer)
+			require.NoError(t, err)
+			require.Equal(t, &tt.expect, res, "conversion")
+
+			roundtrip, err := ToCorrelation(res)
+			require.NoError(t, err)
+			require.Equal(t, &tt.input, roundtrip, "roundtrip")
+
+			create, err := ToCreateCorrelationCommand(res)
+			require.NoError(t, err)
+			require.Equal(t, &tt.create, create, "create")
+
+			update, err := ToUpdateCorrelationCommand(res)
+			require.NoError(t, err)
+			require.Equal(t, &tt.update, update, "update")
+		})
+	}
+}

pkg/services/correlations/correlations.go

@@ -50,7 +50,10 @@ func ProvideService(sqlStore db.DB, routeRegister routing.RouteRegister, ds data
 }
 
 type Service interface {
+	GetCorrelation(ctx context.Context, cmd GetCorrelationQuery) (Correlation, error)
+	GetCorrelations(ctx context.Context, cmd GetCorrelationsQuery) (GetCorrelationsResponseBody, error)
 	CreateCorrelation(ctx context.Context, cmd CreateCorrelationCommand) (Correlation, error)
+	UpdateCorrelation(ctx context.Context, cmd UpdateCorrelationCommand) (Correlation, error)
 	CreateOrUpdateCorrelation(ctx context.Context, cmd CreateCorrelationCommand) error
 	DeleteCorrelation(ctx context.Context, cmd DeleteCorrelationCommand) error
 	DeleteCorrelationsBySourceUID(ctx context.Context, cmd DeleteCorrelationsBySourceUIDCommand) error

pkg/services/correlations/database.go

@@ -54,7 +54,7 @@ func (s CorrelationsService) createCorrelation(ctx context.Context, cmd CreateCo
 			}
 		}
 
-		_, err = session.Insert(correlation)
+		_, err = session.Omit("source_type", "target_type").Insert(correlation)
 		if err != nil {
 			return err
 		}
@@ -121,13 +121,13 @@ func (s CorrelationsService) updateCorrelation(ctx context.Context, cmd UpdateCo
 			return ErrSourceDataSourceDoesNotExists
 		}
 
-		found, err := session.Get(&correlation)
-		if !found {
-			return ErrCorrelationNotFound
-		}
+		found, err := session.Omit("source_type", "target_type").Get(&correlation)
 		if err != nil {
 			return err
 		}
+		if !found {
+			return ErrCorrelationNotFound
+		}
 		if correlation.Provisioned {
 			return ErrCorrelationReadOnly
 		}
@@ -156,7 +156,11 @@ func (s CorrelationsService) updateCorrelation(ctx context.Context, cmd UpdateCo
 			}
 		}
 
-		updateCount, err := session.Where("uid = ? AND source_uid = ?", correlation.UID, correlation.SourceUID).Limit(1).Update(correlation)
+		updateCount, err := session.
+			Where("uid = ? AND source_uid = ?", correlation.UID, correlation.SourceUID).
+			Limit(1).
+			Omit("source_type", "target_type").
+			Update(correlation)
 
 		if err != nil {
 			return err
@@ -179,20 +183,31 @@ func (s CorrelationsService) updateCorrelation(ctx context.Context, cmd UpdateCo
 func (s CorrelationsService) getCorrelation(ctx context.Context, cmd GetCorrelationQuery) (Correlation, error) {
 	correlation := Correlation{
 		UID:       cmd.UID,
+		OrgID:     cmd.OrgId,
 		SourceUID: cmd.SourceUID,
 	}
 
-	err := s.SQLStore.WithTransactionalDbSession(ctx, func(session *db.Session) error {
-		query := &datasources.GetDataSourceQuery{
+	if cmd.SourceUID != "" {
+		if _, err := s.DataSourceService.GetDataSource(ctx, &datasources.GetDataSourceQuery{
+			UID:   correlation.SourceUID,
 			OrgID: cmd.OrgId,
-			UID:   cmd.SourceUID,
-		}
-		if _, err := s.DataSourceService.GetDataSource(ctx, query); err != nil {
-			return ErrSourceDataSourceDoesNotExists
+		}); err != nil {
+			return Correlation{}, ErrSourceDataSourceDoesNotExists
 		}
+	}
 
+	err := s.SQLStore.WithTransactionalDbSession(ctx, func(session *db.Session) error {
 		// Correlations created before the fix #72498 may have org_id = 0, but it's deprecated and will be removed in #72325
-		found, err := session.Select("correlation.*").Join("", "data_source AS dss", "correlation.source_uid = dss.uid and (correlation.org_id = 0 or dss.org_id = correlation.org_id) and dss.org_id = ?", cmd.OrgId).Join("LEFT OUTER", "data_source AS dst", "correlation.target_uid = dst.uid and dst.org_id = ?", cmd.OrgId).Where("correlation.uid = ?", correlation.UID).And("correlation.source_uid = ?", correlation.SourceUID).And(VALID_TYPE_FILTER).Get(&correlation)
+		sql := session.Select("correlation.*, dss.type as source_type, dst.type as target_type").
+			Join("", "data_source AS dss", "correlation.source_uid = dss.uid and (correlation.org_id = 0 or dss.org_id = correlation.org_id) and dss.org_id = ?", cmd.OrgId).
+			Join("LEFT OUTER", "data_source AS dst", "correlation.target_uid = dst.uid and dst.org_id = ?", cmd.OrgId).
+			Where("correlation.uid = ?", correlation.UID).
+			And("correlation.org_id = ?", correlation.OrgID).
+			And(VALID_TYPE_FILTER)
+		if correlation.SourceUID != "" {
+			sql = sql.And("correlation.source_uid = ?", correlation.SourceUID)
+		}
+		found, err := sql.Get(&correlation)
 		if !found {
 			return ErrCorrelationNotFound
 		}
@@ -264,7 +279,9 @@ func (s CorrelationsService) getCorrelations(ctx context.Context, cmd GetCorrela
 		offset := cmd.Limit * (cmd.Page - 1)
 
 		// Correlations created before the fix #72498 may have org_id = 0, but it's deprecated and will be removed in #72325
-		q := session.Select("correlation.*").Join("", "data_source AS dss", "correlation.source_uid = dss.uid and (correlation.org_id = 0 or dss.org_id = correlation.org_id) and dss.org_id = ? ", cmd.OrgId).Join("LEFT OUTER", "data_source AS dst", "correlation.target_uid = dst.uid and dst.org_id = ?", cmd.OrgId)
+		q := session.Select("correlation.*, dss.type as source_type, dst.type as target_type").
+			Join("", "data_source AS dss", "correlation.source_uid = dss.uid and (correlation.org_id = 0 or dss.org_id = correlation.org_id) and dss.org_id = ? ", cmd.OrgId).
+			Join("LEFT OUTER", "data_source AS dst", "correlation.target_uid = dst.uid and dst.org_id = ?", cmd.OrgId)
 
 		if len(cmd.SourceUIDs) > 0 {
 			q.In("dss.uid", cmd.SourceUIDs)
@@ -331,7 +348,7 @@ func (s CorrelationsService) createOrUpdateCorrelation(ctx context.Context, cmd
 
 	found := false
 	err := s.SQLStore.WithDbSession(ctx, func(session *db.Session) error {
-		has, err := session.Get(&correlation)
+		has, err := session.Omit("source_type", "target_type").Get(&correlation)
 		found = has
 		return err
 	})

pkg/services/correlations/models.go

@@ -110,13 +110,15 @@ type Correlation struct {
 	UID string `json:"uid" xorm:"pk 'uid'"`
 	// UID of the data source the correlation originates from
 	// example: d0oxYRg4z
-	SourceUID string `json:"sourceUID" xorm:"pk 'source_uid'"`
+	SourceUID  string  `json:"sourceUID" xorm:"pk 'source_uid'"`
+	SourceType *string `json:"-" xorm:"source_type"`
 	// OrgID of the data source the correlation originates from
 	// Example: 1
 	OrgID int64 `json:"orgId" xorm:"pk 'org_id'"`
 	// UID of the data source the correlation points to
 	// example: PE1C5CBDA0504A6A3
-	TargetUID *string `json:"targetUID" xorm:"target_uid"`
+	TargetUID  *string `json:"targetUID" xorm:"target_uid"`
+	TargetType *string `json:"-" xorm:"target_type"`
 	// Label identifying the correlation
 	// example: My Label
 	Label string `json:"label" xorm:"label"`

pkg/tests/api/correlations/correlations_read_test.go

@@ -109,7 +109,7 @@ func TestIntegrationReadCorrelation(t *testing.T) {
 	var created int64 = 0
 	err := ctx.env.SQLStore.WithDbSession(context.Background(), func(sess *db.Session) error {
 		var innerError error
-		created, innerError = sess.InsertMulti(&[]correlations.Correlation{
+		created, innerError = sess.Omit("source_type", "target_type").InsertMulti(&[]correlations.Correlation{
 			{
 				UID:       "uid-1",
 				SourceUID: dsWithoutCorrelations.UID,

pkg/tests/apis/correlations/correlations_test.go

@@ -0,0 +1,113 @@
+package correlations
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"
+
+	correlationsV0 "github.com/grafana/grafana/apps/correlations/pkg/apis/correlation/v0alpha1"
+	grafanarest "github.com/grafana/grafana/pkg/apiserver/rest"
+	"github.com/grafana/grafana/pkg/services/correlations"
+	"github.com/grafana/grafana/pkg/services/datasources"
+	"github.com/grafana/grafana/pkg/services/featuremgmt"
+	"github.com/grafana/grafana/pkg/setting"
+	"github.com/grafana/grafana/pkg/tests/apis"
+	"github.com/grafana/grafana/pkg/tests/testinfra"
+	"github.com/grafana/grafana/pkg/tests/testsuite"
+	"github.com/grafana/grafana/pkg/util/testutil"
+)
+
+func TestMain(m *testing.M) {
+	testsuite.Run(m)
+}
+
+func TestIntegrationCorrelations(t *testing.T) {
+	testutil.SkipIntegrationTestInShortMode(t)
+
+	for _, mode := range []grafanarest.DualWriterMode{
+		grafanarest.Mode0, // Only legacy for now
+		// grafanarest.Mode2,
+		// grafanarest.Mode3,
+		// grafanarest.Mode5,
+	} {
+		t.Run(fmt.Sprintf("correlations (mode:%d)", mode), func(t *testing.T) {
+			helper := apis.NewK8sTestHelper(t, testinfra.GrafanaOpts{
+				DisableAnonymous:     true,
+				EnableFeatureToggles: []string{featuremgmt.FlagKubernetesCorrelations},
+				UnifiedStorageConfig: map[string]setting.UnifiedStorageConfig{
+					"correlation.correlations.grafana.app": {
+						DualWriterMode: mode,
+					},
+				},
+			})
+			helper.CreateDS(&datasources.AddDataSourceCommand{
+				OrgID: helper.Org1.OrgID,
+				Name:  "test-A",
+				UID:   "test-A",
+				Type:  "testdata",
+			})
+			helper.CreateDS(&datasources.AddDataSourceCommand{
+				OrgID: helper.Org1.OrgID,
+				Name:  "test-B",
+				UID:   "test-B",
+				Type:  "testdata",
+			})
+
+			ctx := context.Background()
+			kind := correlationsV0.CorrelationKind()
+			correlationsClient := helper.GetResourceClient(apis.ResourceClientArgs{
+				User: helper.Org1.Admin,
+				GVR:  kind.GroupVersionResource(),
+			})
+
+			cmd := &correlations.CreateCorrelationCommand{
+				// Loaded from the request, not the payload
+				// SourceUID:   "test-A",
+				// OrgId:       correlationsClient.Args.User.Identity.GetOrgID(),
+				TargetUID:   ptr.To("test-B"),
+				Label:       "hello",
+				Description: "test test test",
+				Type:        correlations.CorrelationType("query"),
+				Config: correlations.CorrelationConfig{
+					Field:  "a",
+					Target: map[string]any{},
+					Transformations: correlations.Transformations{{
+						Type:       "logfmt",
+						Expression: "aaaa",
+						Field:      "f0",
+						MapValue:   "mapped",
+					}},
+				},
+			}
+			body, err := json.Marshal(cmd)
+			require.NoError(t, err)
+
+			createAtoB := apis.DoRequest(helper, apis.RequestParams{
+				User:   correlationsClient.Args.User,
+				Method: http.MethodPost,
+				Path:   "/api/datasources/uid/test-A/correlations",
+				Body:   body,
+			}, &correlations.CreateCorrelationResponseBody{})
+			require.Equal(t, http.StatusOK, createAtoB.Response.StatusCode, "create correlation")
+			require.NotEmpty(t, createAtoB.Result.Result.UID, "a to b")
+			uidAtoB := createAtoB.Result.Result.UID
+
+			// List the value
+			listResults, err := correlationsClient.Resource.List(ctx, v1.ListOptions{})
+			require.NoError(t, err)
+			require.Len(t, listResults.Items, 1)
+			require.Equal(t, uidAtoB, listResults.Items[0].GetName())
+
+			// Get the value
+			getResults, err := correlationsClient.Resource.Get(ctx, uidAtoB, v1.GetOptions{})
+			require.NoError(t, err)
+			require.Equal(t, uidAtoB, getResults.GetName())
+		})
+	}
+}

pkg/tests/apis/helper.go

@@ -784,8 +784,12 @@ func (c *K8sTestHelper) GetGroupVersionInfoJSON(group string) string {
 func (c *K8sTestHelper) CreateDS(cmd *datasources.AddDataSourceCommand) *datasources.DataSource {
 	c.t.Helper()
 
+	require.NotZero(c.t, cmd.OrgID, "requires a non zero orgId")
 	dataSource, err := c.env.Server.HTTPServer.DataSourcesService.AddDataSource(context.Background(), cmd)
 	require.NoError(c.t, err)
+	if cmd.UID != "" {
+		require.Equal(c.t, cmd.UID, dataSource.UID)
+	}
 	return dataSource
 }