Correlations: Add CreateCorrelation HTTP API (#51630)

* Correlations: add migration

* Correlations: Add CreateCorrelation API

* Correlations: Make correlations work with provisioning

* Handle version changes

* Fix lining error

* lint fixes

* rebuild betterer results

* add a UID to each correlation

* Fix lint errors

* add docs

* better wording in API docs

* remove leftover comment

* handle ds updates

* Fix error message typo

* add bad data test

* make correlations a separate table

* skip readonly check when provisioning correlations

* delete stale correlations when datasources are deleted

* restore provisioned readonly ds

* publish deletion event with full data

* generate swagger and HTTP API docs

* apply source datasource permission to create correlation API

* Fix tests & lint errors

* ignore empty deletion events

* fix last lint errors

* fix more lint error

* Only publish deletion event if datasource was actually deleted

* delete DS provisioning deletes correlations, added & fixed tests

* Fix unmarshalling tests

* Fix linting errors

* Fix deltion event tests

* fix small linting error

* fix lint errors

* update betterer

* fix test

* make path singular

* Revert "make path singular"

This reverts commit 420c3d315e.

* add integration tests

* remove unneeded id from correlations table

* update spec

* update leftover references to CorrelationDTO

* fix tests

* cleanup tests

* fix lint error

pkg/services/correlations/api.go

@@ -0,0 +1,49 @@
+package correlations
+
+import (
+	"errors"
+	"net/http"
+
+	"github.com/grafana/grafana/pkg/api/response"
+	"github.com/grafana/grafana/pkg/api/routing"
+	"github.com/grafana/grafana/pkg/middleware"
+	"github.com/grafana/grafana/pkg/models"
+	ac "github.com/grafana/grafana/pkg/services/accesscontrol"
+	"github.com/grafana/grafana/pkg/services/datasources"
+
+	"github.com/grafana/grafana/pkg/web"
+)
+
+func (s *CorrelationsService) registerAPIEndpoints() {
+	uidScope := datasources.ScopeProvider.GetResourceScopeUID(ac.Parameter(":uid"))
+	authorize := ac.Middleware(s.AccessControl)
+
+	s.RouteRegister.Group("/api/datasources/uid/:uid/correlations", func(entities routing.RouteRegister) {
+		entities.Post("/", middleware.ReqSignedIn, authorize(ac.ReqOrgAdmin, ac.EvalPermission(datasources.ActionWrite, uidScope)), routing.Wrap(s.createHandler))
+	})
+}
+
+// createHandler handles POST /datasources/uid/:uid/correlations
+func (s *CorrelationsService) createHandler(c *models.ReqContext) response.Response {
+	cmd := CreateCorrelationCommand{}
+	if err := web.Bind(c.Req, &cmd); err != nil {
+		return response.Error(http.StatusBadRequest, "bad request data", err)
+	}
+	cmd.SourceUID = web.Params(c.Req)[":uid"]
+	cmd.OrgId = c.OrgId
+
+	correlation, err := s.CreateCorrelation(c.Req.Context(), cmd)
+	if err != nil {
+		if errors.Is(err, ErrSourceDataSourceDoesNotExists) || errors.Is(err, ErrTargetDataSourceDoesNotExists) {
+			return response.Error(http.StatusNotFound, "Data source not found", err)
+		}
+
+		if errors.Is(err, ErrSourceDataSourceReadOnly) {
+			return response.Error(http.StatusForbidden, "Data source is read only", err)
+		}
+
+		return response.Error(http.StatusInternalServerError, "Failed to add correlation", err)
+	}
+
+	return response.JSON(http.StatusOK, CreateCorrelationResponse{Result: correlation, Message: "Correlation created"})
+}

pkg/services/correlations/correlations.go

@@ -0,0 +1,74 @@
+package correlations
+
+import (
+	"context"
+
+	"github.com/grafana/grafana/pkg/api/routing"
+	"github.com/grafana/grafana/pkg/bus"
+	"github.com/grafana/grafana/pkg/events"
+
+	"github.com/grafana/grafana/pkg/infra/log"
+	"github.com/grafana/grafana/pkg/services/accesscontrol"
+	"github.com/grafana/grafana/pkg/services/datasources"
+	"github.com/grafana/grafana/pkg/services/sqlstore"
+)
+
+func ProvideService(sqlStore *sqlstore.SQLStore, routeRegister routing.RouteRegister, ds datasources.DataSourceService, ac accesscontrol.AccessControl, bus bus.Bus) *CorrelationsService {
+	s := &CorrelationsService{
+		SQLStore:          sqlStore,
+		RouteRegister:     routeRegister,
+		log:               log.New("correlations"),
+		DataSourceService: ds,
+		AccessControl:     ac,
+	}
+
+	s.registerAPIEndpoints()
+
+	bus.AddEventListener(s.handleDatasourceDeletion)
+
+	return s
+}
+
+type Service interface {
+	CreateCorrelation(ctx context.Context, cmd CreateCorrelationCommand) (Correlation, error)
+	DeleteCorrelationsBySourceUID(ctx context.Context, cmd DeleteCorrelationsBySourceUIDCommand) error
+	DeleteCorrelationsByTargetUID(ctx context.Context, cmd DeleteCorrelationsByTargetUIDCommand) error
+}
+
+type CorrelationsService struct {
+	SQLStore          *sqlstore.SQLStore
+	RouteRegister     routing.RouteRegister
+	log               log.Logger
+	DataSourceService datasources.DataSourceService
+	AccessControl     accesscontrol.AccessControl
+}
+
+func (s CorrelationsService) CreateCorrelation(ctx context.Context, cmd CreateCorrelationCommand) (Correlation, error) {
+	return s.createCorrelation(ctx, cmd)
+}
+
+func (s CorrelationsService) DeleteCorrelationsBySourceUID(ctx context.Context, cmd DeleteCorrelationsBySourceUIDCommand) error {
+	return s.deleteCorrelationsBySourceUID(ctx, cmd)
+}
+
+func (s CorrelationsService) DeleteCorrelationsByTargetUID(ctx context.Context, cmd DeleteCorrelationsByTargetUIDCommand) error {
+	return s.deleteCorrelationsByTargetUID(ctx, cmd)
+}
+
+func (s CorrelationsService) handleDatasourceDeletion(ctx context.Context, event *events.DataSourceDeleted) error {
+	return s.SQLStore.InTransaction(ctx, func(ctx context.Context) error {
+		if err := s.deleteCorrelationsBySourceUID(ctx, DeleteCorrelationsBySourceUIDCommand{
+			SourceUID: event.UID,
+		}); err != nil {
+			return err
+		}
+
+		if err := s.deleteCorrelationsByTargetUID(ctx, DeleteCorrelationsByTargetUIDCommand{
+			TargetUID: event.UID,
+		}); err != nil {
+			return err
+		}
+
+		return nil
+	})
+}

pkg/services/correlations/database.go

@@ -0,0 +1,70 @@
+package correlations
+
+import (
+	"context"
+
+	"github.com/grafana/grafana/pkg/services/datasources"
+	"github.com/grafana/grafana/pkg/services/sqlstore"
+	"github.com/grafana/grafana/pkg/util"
+)
+
+// createCorrelation adds a correlation
+func (s CorrelationsService) createCorrelation(ctx context.Context, cmd CreateCorrelationCommand) (Correlation, error) {
+	correlation := Correlation{
+		UID:         util.GenerateShortUID(),
+		SourceUID:   cmd.SourceUID,
+		TargetUID:   cmd.TargetUID,
+		Label:       cmd.Label,
+		Description: cmd.Description,
+	}
+
+	err := s.SQLStore.WithTransactionalDbSession(ctx, func(session *sqlstore.DBSession) error {
+		var err error
+
+		query := &datasources.GetDataSourceQuery{
+			OrgId: cmd.OrgId,
+			Uid:   cmd.SourceUID,
+		}
+		if err = s.DataSourceService.GetDataSource(ctx, query); err != nil {
+			return ErrSourceDataSourceDoesNotExists
+		}
+
+		if !cmd.SkipReadOnlyCheck && query.Result.ReadOnly {
+			return ErrSourceDataSourceReadOnly
+		}
+
+		if err = s.DataSourceService.GetDataSource(ctx, &datasources.GetDataSourceQuery{
+			OrgId: cmd.OrgId,
+			Uid:   cmd.TargetUID,
+		}); err != nil {
+			return ErrTargetDataSourceDoesNotExists
+		}
+
+		_, err = session.Insert(correlation)
+		if err != nil {
+			return err
+		}
+
+		return nil
+	})
+
+	if err != nil {
+		return Correlation{}, err
+	}
+
+	return correlation, nil
+}
+
+func (s CorrelationsService) deleteCorrelationsBySourceUID(ctx context.Context, cmd DeleteCorrelationsBySourceUIDCommand) error {
+	return s.SQLStore.WithDbSession(ctx, func(session *sqlstore.DBSession) error {
+		_, err := session.Delete(&Correlation{SourceUID: cmd.SourceUID})
+		return err
+	})
+}
+
+func (s CorrelationsService) deleteCorrelationsByTargetUID(ctx context.Context, cmd DeleteCorrelationsByTargetUIDCommand) error {
+	return s.SQLStore.WithDbSession(ctx, func(session *sqlstore.DBSession) error {
+		_, err := session.Delete(&Correlation{TargetUID: cmd.TargetUID})
+		return err
+	})
+}

pkg/services/correlations/models.go

@@ -0,0 +1,66 @@
+package correlations
+
+import (
+	"errors"
+)
+
+var (
+	ErrSourceDataSourceReadOnly           = errors.New("source data source is read only")
+	ErrSourceDataSourceDoesNotExists      = errors.New("source data source does not exist")
+	ErrTargetDataSourceDoesNotExists      = errors.New("target data source does not exist")
+	ErrCorrelationFailedGenerateUniqueUid = errors.New("failed to generate unique correlation UID")
+	ErrCorrelationIdentifierNotSet        = errors.New("source identifier and org id are needed to be able to edit correlations")
+)
+
+// Correlation is the model for correlations definitions
+type Correlation struct {
+	// Unique identifier of the correlation
+	// example: 50xhMlg9k
+	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'"`
+	// UID of the data source the correlation points to
+	// example:PE1C5CBDA0504A6A3
+	TargetUID string `json:"targetUid" xorm:"target_uid"`
+	// Label identifying the correlation
+	// example: My Label
+	Label string `json:"label" xorm:"label"`
+	// Description of the correlation
+	// example: Logs to Traces
+	Description string `json:"description" xorm:"description"`
+}
+
+// CreateCorrelationResponse is the response struct for CreateCorrelationCommand
+// swagger:model
+type CreateCorrelationResponse struct {
+	Result Correlation `json:"result"`
+	// example: Correlation created
+	Message string `json:"message"`
+}
+
+// CreateCorrelationCommand is the command for creating a correlation
+// swagger:model
+type CreateCorrelationCommand struct {
+	// UID of the data source for which correlation is created.
+	SourceUID         string `json:"-"`
+	OrgId             int64  `json:"-"`
+	SkipReadOnlyCheck bool   `json:"-"`
+	// Target data source UID to which the correlation is created
+	// example:PE1C5CBDA0504A6A3
+	TargetUID string `json:"targetUid" binding:"Required"`
+	// Optional label identifying the correlation
+	// example: My label
+	Label string `json:"label"`
+	// Optional description of the correlation
+	// example: Logs to Traces
+	Description string `json:"description"`
+}
+
+type DeleteCorrelationsBySourceUIDCommand struct {
+	SourceUID string
+}
+
+type DeleteCorrelationsByTargetUIDCommand struct {
+	TargetUID string
+}

pkg/services/datasources/errors.go

@@ -10,4 +10,5 @@ var (
 	ErrDataSourceAccessDenied            = errors.New("data source access denied")
 	ErrDataSourceFailedGenerateUniqueUid = errors.New("failed to generate unique datasource ID")
 	ErrDataSourceIdentifierNotSet        = errors.New("unique identifier and org id are needed to be able to get or delete a datasource")
+	ErrDatasourceIsReadOnly              = errors.New("data source is readonly, can only be updated from configuration")
 )

pkg/services/provisioning/datasources/config_reader_test.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/grafana/grafana/pkg/infra/log"
 	"github.com/grafana/grafana/pkg/models"
+	"github.com/grafana/grafana/pkg/services/correlations"
 	"github.com/grafana/grafana/pkg/services/datasources"
 	"github.com/grafana/grafana/pkg/util"
 )
@@ -26,13 +27,16 @@ var (
 	multipleOrgsWithDefault         = "testdata/multiple-org-default"
 	withoutDefaults                 = "testdata/appliedDefaults"
 	invalidAccess                   = "testdata/invalid-access"
+
+	oneDatasourceWithTwoCorrelations = "testdata/one-datasource-two-correlations"
 )
 
 func TestDatasourceAsConfig(t *testing.T) {
 	t.Run("when some values missing should apply default on insert", func(t *testing.T) {
 		store := &spyStore{}
 		orgStore := &mockOrgStore{ExpectedOrg: &models.Org{Id: 1}}
-		dc := newDatasourceProvisioner(logger, store, orgStore)
+		correlationsStore := &mockCorrelationsStore{}
+		dc := newDatasourceProvisioner(logger, store, correlationsStore, orgStore)
 		err := dc.applyChanges(context.Background(), withoutDefaults)
 		if err != nil {
 			t.Fatalf("applyChanges return an error %v", err)
@@ -50,7 +54,8 @@ func TestDatasourceAsConfig(t *testing.T) {
 			items: []*datasources.DataSource{{Name: "My datasource name", OrgId: 1, Id: 1, Uid: util.GenerateShortUID()}},
 		}
 		orgStore := &mockOrgStore{}
-		dc := newDatasourceProvisioner(logger, store, orgStore)
+		correlationsStore := &mockCorrelationsStore{}
+		dc := newDatasourceProvisioner(logger, store, correlationsStore, orgStore)
 		err := dc.applyChanges(context.Background(), withoutDefaults)
 		if err != nil {
 			t.Fatalf("applyChanges return an error %v", err)
@@ -65,7 +70,8 @@ func TestDatasourceAsConfig(t *testing.T) {
 	t.Run("no datasource in database", func(t *testing.T) {
 		store := &spyStore{}
 		orgStore := &mockOrgStore{}
-		dc := newDatasourceProvisioner(logger, store, orgStore)
+		correlationsStore := &mockCorrelationsStore{}
+		dc := newDatasourceProvisioner(logger, store, correlationsStore, orgStore)
 		err := dc.applyChanges(context.Background(), twoDatasourcesConfig)
 		if err != nil {
 			t.Fatalf("applyChanges return an error %v", err)
@@ -79,7 +85,8 @@ func TestDatasourceAsConfig(t *testing.T) {
 	t.Run("One datasource in database with same name should update one datasource", func(t *testing.T) {
 		store := &spyStore{items: []*datasources.DataSource{{Name: "Graphite", OrgId: 1, Id: 1}}}
 		orgStore := &mockOrgStore{}
-		dc := newDatasourceProvisioner(logger, store, orgStore)
+		correlationsStore := &mockCorrelationsStore{}
+		dc := newDatasourceProvisioner(logger, store, correlationsStore, orgStore)
 		err := dc.applyChanges(context.Background(), twoDatasourcesConfig)
 		if err != nil {
 			t.Fatalf("applyChanges return an error %v", err)
@@ -93,7 +100,8 @@ func TestDatasourceAsConfig(t *testing.T) {
 	t.Run("Two datasources with is_default should raise error", func(t *testing.T) {
 		store := &spyStore{}
 		orgStore := &mockOrgStore{}
-		dc := newDatasourceProvisioner(logger, store, orgStore)
+		correlationsStore := &mockCorrelationsStore{}
+		dc := newDatasourceProvisioner(logger, store, correlationsStore, orgStore)
 		err := dc.applyChanges(context.Background(), doubleDatasourcesConfig)
 		require.Equal(t, err, ErrInvalidConfigToManyDefault)
 	})
@@ -101,7 +109,8 @@ func TestDatasourceAsConfig(t *testing.T) {
 	t.Run("Multiple datasources in different organizations with isDefault in each organization should not raise error", func(t *testing.T) {
 		store := &spyStore{}
 		orgStore := &mockOrgStore{}
-		dc := newDatasourceProvisioner(logger, store, orgStore)
+		correlationsStore := &mockCorrelationsStore{}
+		dc := newDatasourceProvisioner(logger, store, correlationsStore, orgStore)
 		err := dc.applyChanges(context.Background(), multipleOrgsWithDefault)
 		require.NoError(t, err)
 		require.Equal(t, len(store.inserted), 4)
@@ -114,7 +123,8 @@ func TestDatasourceAsConfig(t *testing.T) {
 	t.Run("Remove one datasource should have removed old datasource", func(t *testing.T) {
 		store := &spyStore{}
 		orgStore := &mockOrgStore{}
-		dc := newDatasourceProvisioner(logger, store, orgStore)
+		correlationsStore := &mockCorrelationsStore{}
+		dc := newDatasourceProvisioner(logger, store, correlationsStore, orgStore)
 		err := dc.applyChanges(context.Background(), deleteOneDatasource)
 		if err != nil {
 			t.Fatalf("applyChanges return an error %v", err)
@@ -130,7 +140,8 @@ func TestDatasourceAsConfig(t *testing.T) {
 	t.Run("Two configured datasource and purge others", func(t *testing.T) {
 		store := &spyStore{items: []*datasources.DataSource{{Name: "old-graphite", OrgId: 1, Id: 1}, {Name: "old-graphite2", OrgId: 1, Id: 2}}}
 		orgStore := &mockOrgStore{}
-		dc := newDatasourceProvisioner(logger, store, orgStore)
+		correlationsStore := &mockCorrelationsStore{}
+		dc := newDatasourceProvisioner(logger, store, correlationsStore, orgStore)
 		err := dc.applyChanges(context.Background(), twoDatasourcesConfigPurgeOthers)
 		if err != nil {
 			t.Fatalf("applyChanges return an error %v", err)
@@ -144,7 +155,8 @@ func TestDatasourceAsConfig(t *testing.T) {
 	t.Run("Two configured datasource and purge others = false", func(t *testing.T) {
 		store := &spyStore{items: []*datasources.DataSource{{Name: "Graphite", OrgId: 1, Id: 1}, {Name: "old-graphite2", OrgId: 1, Id: 2}}}
 		orgStore := &mockOrgStore{}
-		dc := newDatasourceProvisioner(logger, store, orgStore)
+		correlationsStore := &mockCorrelationsStore{}
+		dc := newDatasourceProvisioner(logger, store, correlationsStore, orgStore)
 		err := dc.applyChanges(context.Background(), twoDatasourcesConfig)
 		if err != nil {
 			t.Fatalf("applyChanges return an error %v", err)
@@ -224,6 +236,53 @@ func TestDatasourceAsConfig(t *testing.T) {
 		validateDatasource(t, dsCfg)
 		validateDeleteDatasources(t, dsCfg)
 	})
+
+	t.Run("Correlations", func(t *testing.T) {
+		t.Run("Creates two correlations", func(t *testing.T) {
+			store := &spyStore{}
+			orgStore := &mockOrgStore{}
+			correlationsStore := &mockCorrelationsStore{}
+			dc := newDatasourceProvisioner(logger, store, correlationsStore, orgStore)
+			err := dc.applyChanges(context.Background(), oneDatasourceWithTwoCorrelations)
+			if err != nil {
+				t.Fatalf("applyChanges return an error %v", err)
+			}
+
+			require.Equal(t, 2, len(correlationsStore.created))
+			require.Equal(t, 0, len(correlationsStore.deletedBySourceUID))
+			require.Equal(t, 0, len(correlationsStore.deletedByTargetUID))
+		})
+
+		t.Run("Updating existing datasource deletes existing correlations and creates two", func(t *testing.T) {
+			store := &spyStore{items: []*datasources.DataSource{{Name: "Graphite", OrgId: 1, Id: 1}}}
+			orgStore := &mockOrgStore{}
+			correlationsStore := &mockCorrelationsStore{}
+			dc := newDatasourceProvisioner(logger, store, correlationsStore, orgStore)
+			err := dc.applyChanges(context.Background(), oneDatasourceWithTwoCorrelations)
+			if err != nil {
+				t.Fatalf("applyChanges return an error %v", err)
+			}
+
+			require.Equal(t, 2, len(correlationsStore.created))
+			require.Equal(t, 1, len(correlationsStore.deletedBySourceUID))
+			require.Equal(t, 0, len(correlationsStore.deletedByTargetUID))
+		})
+
+		t.Run("Deleting datasource deletes existing correlations", func(t *testing.T) {
+			store := &spyStore{items: []*datasources.DataSource{{Name: "old-data-source", OrgId: 1, Id: 1, Uid: "some-uid"}}}
+			orgStore := &mockOrgStore{}
+			correlationsStore := &mockCorrelationsStore{items: []correlations.Correlation{{UID: "some-uid", SourceUID: "some-uid", TargetUID: "target-uid"}}}
+			dc := newDatasourceProvisioner(logger, store, correlationsStore, orgStore)
+			err := dc.applyChanges(context.Background(), deleteOneDatasource)
+			if err != nil {
+				t.Fatalf("applyChanges return an error %v", err)
+			}
+
+			require.Equal(t, 0, len(correlationsStore.created))
+			require.Equal(t, 1, len(correlationsStore.deletedBySourceUID))
+			require.Equal(t, 1, len(correlationsStore.deletedByTargetUID))
+		})
+	})
 }
 
 func validateDeleteDatasources(t *testing.T, dsCfg *configs) {
@@ -249,6 +308,12 @@ func validateDatasource(t *testing.T, dsCfg *configs) {
 	require.True(t, ds.Editable)
 	require.Equal(t, ds.Version, 10)
 
+	require.Equal(t, []map[string]interface{}{{
+		"targetUid":   "a target",
+		"label":       "a label",
+		"description": "a description",
+	}}, ds.Correlations)
+
 	require.Greater(t, len(ds.JSONData), 2)
 	require.Equal(t, ds.JSONData["graphiteVersion"], "1.1")
 	require.Equal(t, ds.JSONData["tlsAuth"], true)
@@ -273,6 +338,28 @@ func (m *mockOrgStore) GetOrgById(c context.Context, cmd *models.GetOrgByIdQuery
 	return nil
 }
 
+type mockCorrelationsStore struct {
+	created            []correlations.CreateCorrelationCommand
+	deletedBySourceUID []correlations.DeleteCorrelationsBySourceUIDCommand
+	deletedByTargetUID []correlations.DeleteCorrelationsByTargetUIDCommand
+	items              []correlations.Correlation
+}
+
+func (m *mockCorrelationsStore) CreateCorrelation(c context.Context, cmd correlations.CreateCorrelationCommand) (correlations.Correlation, error) {
+	m.created = append(m.created, cmd)
+	return correlations.Correlation{}, nil
+}
+
+func (m *mockCorrelationsStore) DeleteCorrelationsBySourceUID(c context.Context, cmd correlations.DeleteCorrelationsBySourceUIDCommand) error {
+	m.deletedBySourceUID = append(m.deletedBySourceUID, cmd)
+	return nil
+}
+
+func (m *mockCorrelationsStore) DeleteCorrelationsByTargetUID(c context.Context, cmd correlations.DeleteCorrelationsByTargetUIDCommand) error {
+	m.deletedByTargetUID = append(m.deletedByTargetUID, cmd)
+	return nil
+}
+
 type spyStore struct {
 	inserted []*datasources.AddDataSourceCommand
 	deleted  []*datasources.DeleteDataSourceCommand
@@ -292,11 +379,20 @@ func (s *spyStore) GetDataSource(ctx context.Context, query *datasources.GetData
 
 func (s *spyStore) DeleteDataSource(ctx context.Context, cmd *datasources.DeleteDataSourceCommand) error {
 	s.deleted = append(s.deleted, cmd)
+	for _, v := range s.items {
+		if cmd.Name == v.Name && cmd.OrgID == v.OrgId {
+			cmd.DeletedDatasourcesCount = 1
+			return nil
+		}
+	}
 	return nil
 }
 
 func (s *spyStore) AddDataSource(ctx context.Context, cmd *datasources.AddDataSourceCommand) error {
 	s.inserted = append(s.inserted, cmd)
+	cmd.Result = &datasources.DataSource{
+		Uid: cmd.Uid,
+	}
 	return nil
 }
 

pkg/services/provisioning/datasources/datasources.go

@@ -3,8 +3,10 @@ package datasources
 import (
 	"context"
 	"errors"
+	"fmt"
 
 	"github.com/grafana/grafana/pkg/infra/log"
+	"github.com/grafana/grafana/pkg/services/correlations"
 	"github.com/grafana/grafana/pkg/services/datasources"
 	"github.com/grafana/grafana/pkg/services/provisioning/utils"
 )
@@ -16,6 +18,12 @@ type Store interface {
 	DeleteDataSource(ctx context.Context, cmd *datasources.DeleteDataSourceCommand) error
 }
 
+type CorrelationsStore interface {
+	DeleteCorrelationsByTargetUID(ctx context.Context, cmd correlations.DeleteCorrelationsByTargetUIDCommand) error
+	DeleteCorrelationsBySourceUID(ctx context.Context, cmd correlations.DeleteCorrelationsBySourceUIDCommand) error
+	CreateCorrelation(ctx context.Context, cmd correlations.CreateCorrelationCommand) (correlations.Correlation, error)
+}
+
 var (
 	// ErrInvalidConfigToManyDefault indicates that multiple datasource in the provisioning files
 	// contains more than one datasource marked as default.
@@ -24,24 +32,26 @@ var (
 
 // Provision scans a directory for provisioning config files
 // and provisions the datasource in those files.
-func Provision(ctx context.Context, configDirectory string, store Store, orgStore utils.OrgStore) error {
-	dc := newDatasourceProvisioner(log.New("provisioning.datasources"), store, orgStore)
+func Provision(ctx context.Context, configDirectory string, store Store, correlationsStore CorrelationsStore, orgStore utils.OrgStore) error {
+	dc := newDatasourceProvisioner(log.New("provisioning.datasources"), store, correlationsStore, orgStore)
 	return dc.applyChanges(ctx, configDirectory)
 }
 
 // DatasourceProvisioner is responsible for provisioning datasources based on
 // configuration read by the `configReader`
 type DatasourceProvisioner struct {
-	log         log.Logger
-	cfgProvider *configReader
-	store       Store
+	log               log.Logger
+	cfgProvider       *configReader
+	store             Store
+	correlationsStore CorrelationsStore
 }
 
-func newDatasourceProvisioner(log log.Logger, store Store, orgStore utils.OrgStore) DatasourceProvisioner {
+func newDatasourceProvisioner(log log.Logger, store Store, correlationsStore CorrelationsStore, orgStore utils.OrgStore) DatasourceProvisioner {
 	return DatasourceProvisioner{
-		log:         log,
-		cfgProvider: &configReader{log: log, orgStore: orgStore},
-		store:       store,
+		log:               log,
+		cfgProvider:       &configReader{log: log, orgStore: orgStore},
+		store:             store,
+		correlationsStore: correlationsStore,
 	}
 }
 
@@ -50,6 +60,8 @@ func (dc *DatasourceProvisioner) apply(ctx context.Context, cfg *configs) error
 		return err
 	}
 
+	correlationsToInsert := make([]correlations.CreateCorrelationCommand, 0)
+
 	for _, ds := range cfg.Datasources {
 		cmd := &datasources.GetDataSourceQuery{OrgId: ds.OrgID, Name: ds.Name}
 		err := dc.store.GetDataSource(ctx, cmd)
@@ -63,12 +75,44 @@ func (dc *DatasourceProvisioner) apply(ctx context.Context, cfg *configs) error
 			if err := dc.store.AddDataSource(ctx, insertCmd); err != nil {
 				return err
 			}
+
+			for _, correlation := range ds.Correlations {
+				if insertCorrelationCmd, err := makeCreateCorrelationCommand(correlation, insertCmd.Result.Uid, insertCmd.OrgId); err == nil {
+					correlationsToInsert = append(correlationsToInsert, insertCorrelationCmd)
+				} else {
+					dc.log.Error("failed to parse correlation", "correlation", correlation)
+					return err
+				}
+			}
 		} else {
 			updateCmd := createUpdateCommand(ds, cmd.Result.Id)
 			dc.log.Debug("updating datasource from configuration", "name", updateCmd.Name, "uid", updateCmd.Uid)
 			if err := dc.store.UpdateDataSource(ctx, updateCmd); err != nil {
 				return err
 			}
+
+			if len(ds.Correlations) > 0 {
+				if err := dc.correlationsStore.DeleteCorrelationsBySourceUID(ctx, correlations.DeleteCorrelationsBySourceUIDCommand{
+					SourceUID: cmd.Result.Uid,
+				}); err != nil {
+					return err
+				}
+			}
+
+			for _, correlation := range ds.Correlations {
+				if insertCorrelationCmd, err := makeCreateCorrelationCommand(correlation, cmd.Result.Uid, updateCmd.OrgId); err == nil {
+					correlationsToInsert = append(correlationsToInsert, insertCorrelationCmd)
+				} else {
+					dc.log.Error("failed to parse correlation", "correlation", correlation)
+					return err
+				}
+			}
+		}
+	}
+
+	for _, createCorrelationCmd := range correlationsToInsert {
+		if _, err := dc.correlationsStore.CreateCorrelation(ctx, createCorrelationCmd); err != nil {
+			return fmt.Errorf("err=%s source=%s", err.Error(), createCorrelationCmd.SourceUID)
 		}
 	}
 
@@ -90,13 +134,50 @@ func (dc *DatasourceProvisioner) applyChanges(ctx context.Context, configPath st
 	return nil
 }
 
+func makeCreateCorrelationCommand(correlation map[string]interface{}, SourceUid string, OrgId int64) (correlations.CreateCorrelationCommand, error) {
+	targetUid, ok := correlation["targetUid"].(string)
+	if !ok {
+		return correlations.CreateCorrelationCommand{}, fmt.Errorf("correlation missing targetUid")
+	}
+
+	return correlations.CreateCorrelationCommand{
+		SourceUID:         SourceUid,
+		TargetUID:         targetUid,
+		Label:             correlation["label"].(string),
+		Description:       correlation["description"].(string),
+		OrgId:             OrgId,
+		SkipReadOnlyCheck: true,
+	}, nil
+}
+
 func (dc *DatasourceProvisioner) deleteDatasources(ctx context.Context, dsToDelete []*deleteDatasourceConfig) error {
 	for _, ds := range dsToDelete {
 		cmd := &datasources.DeleteDataSourceCommand{OrgID: ds.OrgID, Name: ds.Name}
+		getDsQuery := &datasources.GetDataSourceQuery{Name: ds.Name, OrgId: ds.OrgID}
+		if err := dc.store.GetDataSource(ctx, getDsQuery); err != nil && !errors.Is(err, datasources.ErrDataSourceNotFound) {
+			return err
+		}
+
 		if err := dc.store.DeleteDataSource(ctx, cmd); err != nil {
 			return err
 		}
 
+		if getDsQuery.Result != nil {
+			if err := dc.correlationsStore.DeleteCorrelationsBySourceUID(ctx, correlations.DeleteCorrelationsBySourceUIDCommand{
+				SourceUID: getDsQuery.Result.Uid,
+			}); err != nil {
+				return err
+			}
+
+			if err := dc.correlationsStore.DeleteCorrelationsByTargetUID(ctx, correlations.DeleteCorrelationsByTargetUIDCommand{
+				TargetUID: getDsQuery.Result.Uid,
+			}); err != nil {
+				return err
+			}
+
+			dc.log.Info("deleted correlations based on configuration", "ds_name", ds.Name)
+		}
+
 		if cmd.DeletedDatasourcesCount > 0 {
 			dc.log.Info("deleted datasource based on configuration", "name", ds.Name)
 		}

pkg/services/provisioning/datasources/testdata/all-properties/all-properties.yaml

@@ -12,6 +12,10 @@ datasources:
     basicAuthUser: basic_auth_user
     withCredentials: true
     isDefault: true
+    correlations:
+      - targetUid: a target
+        label: a label
+        description: a description
     jsonData:
       graphiteVersion: "1.1"
       tlsAuth: true

pkg/services/provisioning/datasources/testdata/one-datasource-two-correlations/one-datasource-two-correlations.yaml

@@ -0,0 +1,15 @@
+apiVersion: 1
+
+datasources:
+  - name: Graphite
+    type: graphite
+    uid: graphite
+    access: proxy
+    url: http://localhost:8080
+    correlations:
+      - targetUid: graphite
+        label: a label
+        description: a description
+      - targetUid: graphite
+        label: a second label
+        description: a second description

pkg/services/provisioning/datasources/testdata/version-0/version-0.yaml

@@ -10,6 +10,10 @@ datasources:
     basic_auth_user: basic_auth_user
     with_credentials: true
     is_default: true
+    correlations:
+      - targetUid: a target
+        label: a label
+        description: a description
     json_data:
       graphiteVersion: "1.1"
       tlsAuth: true

pkg/services/provisioning/datasources/types.go

@@ -42,6 +42,7 @@ type upsertDataSourceFromConfig struct {
 	BasicAuthUser   string
 	WithCredentials bool
 	IsDefault       bool
+	Correlations    []map[string]interface{}
 	JSONData        map[string]interface{}
 	SecureJSONData  map[string]string
 	Editable        bool
@@ -74,21 +75,22 @@ type deleteDatasourceConfigV1 struct {
 }
 
 type upsertDataSourceFromConfigV0 struct {
-	OrgID           int64                  `json:"org_id" yaml:"org_id"`
-	Version         int                    `json:"version" yaml:"version"`
-	Name            string                 `json:"name" yaml:"name"`
-	Type            string                 `json:"type" yaml:"type"`
-	Access          string                 `json:"access" yaml:"access"`
-	URL             string                 `json:"url" yaml:"url"`
-	User            string                 `json:"user" yaml:"user"`
-	Database        string                 `json:"database" yaml:"database"`
-	BasicAuth       bool                   `json:"basic_auth" yaml:"basic_auth"`
-	BasicAuthUser   string                 `json:"basic_auth_user" yaml:"basic_auth_user"`
-	WithCredentials bool                   `json:"with_credentials" yaml:"with_credentials"`
-	IsDefault       bool                   `json:"is_default" yaml:"is_default"`
-	JSONData        map[string]interface{} `json:"json_data" yaml:"json_data"`
-	SecureJSONData  map[string]string      `json:"secure_json_data" yaml:"secure_json_data"`
-	Editable        bool                   `json:"editable" yaml:"editable"`
+	OrgID           int64                    `json:"org_id" yaml:"org_id"`
+	Version         int                      `json:"version" yaml:"version"`
+	Name            string                   `json:"name" yaml:"name"`
+	Type            string                   `json:"type" yaml:"type"`
+	Access          string                   `json:"access" yaml:"access"`
+	URL             string                   `json:"url" yaml:"url"`
+	User            string                   `json:"user" yaml:"user"`
+	Database        string                   `json:"database" yaml:"database"`
+	BasicAuth       bool                     `json:"basic_auth" yaml:"basic_auth"`
+	BasicAuthUser   string                   `json:"basic_auth_user" yaml:"basic_auth_user"`
+	WithCredentials bool                     `json:"with_credentials" yaml:"with_credentials"`
+	IsDefault       bool                     `json:"is_default" yaml:"is_default"`
+	Correlations    []map[string]interface{} `json:"correlations" yaml:"correlations"`
+	JSONData        map[string]interface{}   `json:"json_data" yaml:"json_data"`
+	SecureJSONData  map[string]string        `json:"secure_json_data" yaml:"secure_json_data"`
+	Editable        bool                     `json:"editable" yaml:"editable"`
 }
 
 type upsertDataSourceFromConfigV1 struct {
@@ -104,6 +106,7 @@ type upsertDataSourceFromConfigV1 struct {
 	BasicAuthUser   values.StringValue    `json:"basicAuthUser" yaml:"basicAuthUser"`
 	WithCredentials values.BoolValue      `json:"withCredentials" yaml:"withCredentials"`
 	IsDefault       values.BoolValue      `json:"isDefault" yaml:"isDefault"`
+	Correlations    values.JSONSliceValue `json:"correlations" yaml:"correlations"`
 	JSONData        values.JSONValue      `json:"jsonData" yaml:"jsonData"`
 	SecureJSONData  values.StringMapValue `json:"secureJsonData" yaml:"secureJsonData"`
 	Editable        values.BoolValue      `json:"editable" yaml:"editable"`
@@ -132,6 +135,7 @@ func (cfg *configsV1) mapToDatasourceFromConfig(apiVersion int64) *configs {
 			BasicAuthUser:   ds.BasicAuthUser.Value(),
 			WithCredentials: ds.WithCredentials.Value(),
 			IsDefault:       ds.IsDefault.Value(),
+			Correlations:    ds.Correlations.Value(),
 			JSONData:        ds.JSONData.Value(),
 			SecureJSONData:  ds.SecureJSONData.Value(),
 			Editable:        ds.Editable.Value(),
@@ -172,6 +176,7 @@ func (cfg *configsV0) mapToDatasourceFromConfig(apiVersion int64) *configs {
 			BasicAuthUser:   ds.BasicAuthUser,
 			WithCredentials: ds.WithCredentials,
 			IsDefault:       ds.IsDefault,
+			Correlations:    ds.Correlations,
 			JSONData:        ds.JSONData,
 			SecureJSONData:  ds.SecureJSONData,
 			Editable:        ds.Editable,

pkg/services/provisioning/provisioning.go

@@ -11,6 +11,7 @@ import (
 	"github.com/grafana/grafana/pkg/registry"
 	"github.com/grafana/grafana/pkg/services/accesscontrol"
 	"github.com/grafana/grafana/pkg/services/alerting"
+	"github.com/grafana/grafana/pkg/services/correlations"
 	dashboardservice "github.com/grafana/grafana/pkg/services/dashboards"
 	datasourceservice "github.com/grafana/grafana/pkg/services/datasources"
 	"github.com/grafana/grafana/pkg/services/encryption"
@@ -39,6 +40,7 @@ func ProvideService(
 	notificatonService *notifications.NotificationService,
 	dashboardProvisioningService dashboardservice.DashboardProvisioningService,
 	datasourceService datasourceservice.DataSourceService,
+	correlationsService correlations.Service,
 	dashboardService dashboardservice.DashboardService,
 	folderService dashboardservice.FolderService,
 	alertingService *alerting.AlertNotificationService,
@@ -60,6 +62,7 @@ func ProvideService(
 		dashboardProvisioningService: dashboardProvisioningService,
 		dashboardService:             dashboardService,
 		datasourceService:            datasourceService,
+		correlationsService:          correlationsService,
 		alertingService:              alertingService,
 		pluginsSettings:              pluginSettings,
 		searchService:                searchService,
@@ -98,7 +101,7 @@ func NewProvisioningServiceImpl() *ProvisioningServiceImpl {
 func newProvisioningServiceImpl(
 	newDashboardProvisioner dashboards.DashboardProvisionerFactory,
 	provisionNotifiers func(context.Context, string, notifiers.Manager, notifiers.SQLStore, encryption.Internal, *notifications.NotificationService) error,
-	provisionDatasources func(context.Context, string, datasources.Store, utils.OrgStore) error,
+	provisionDatasources func(context.Context, string, datasources.Store, datasources.CorrelationsStore, utils.OrgStore) error,
 	provisionPlugins func(context.Context, string, plugins.Store, plugifaces.Store, pluginsettings.Service) error,
 ) *ProvisioningServiceImpl {
 	return &ProvisioningServiceImpl{
@@ -122,13 +125,14 @@ type ProvisioningServiceImpl struct {
 	newDashboardProvisioner      dashboards.DashboardProvisionerFactory
 	dashboardProvisioner         dashboards.DashboardProvisioner
 	provisionNotifiers           func(context.Context, string, notifiers.Manager, notifiers.SQLStore, encryption.Internal, *notifications.NotificationService) error
-	provisionDatasources         func(context.Context, string, datasources.Store, utils.OrgStore) error
+	provisionDatasources         func(context.Context, string, datasources.Store, datasources.CorrelationsStore, utils.OrgStore) error
 	provisionPlugins             func(context.Context, string, plugins.Store, plugifaces.Store, pluginsettings.Service) error
 	provisionRules               func(context.Context, string, dashboardservice.DashboardService, dashboardservice.DashboardProvisioningService, provisioning.AlertRuleService) error
 	mutex                        sync.Mutex
 	dashboardProvisioningService dashboardservice.DashboardProvisioningService
 	dashboardService             dashboardservice.DashboardService
 	datasourceService            datasourceservice.DataSourceService
+	correlationsService          correlations.Service
 	alertingService              *alerting.AlertNotificationService
 	pluginsSettings              pluginsettings.Service
 	searchService                searchV2.SearchService
@@ -193,7 +197,7 @@ func (ps *ProvisioningServiceImpl) Run(ctx context.Context) error {
 
 func (ps *ProvisioningServiceImpl) ProvisionDatasources(ctx context.Context) error {
 	datasourcePath := filepath.Join(ps.Cfg.ProvisioningPath, "datasources")
-	if err := ps.provisionDatasources(ctx, datasourcePath, ps.datasourceService, ps.SQLStore); err != nil {
+	if err := ps.provisionDatasources(ctx, datasourcePath, ps.datasourceService, ps.correlationsService, ps.SQLStore); err != nil {
 		err = fmt.Errorf("%v: %w", "Datasource provisioning error", err)
 		ps.log.Error("Failed to provision data sources", "error", err)
 		return err

pkg/services/provisioning/values/values.go

@@ -188,6 +188,47 @@ func (val *StringMapValue) Value() map[string]string {
 	return val.value
 }
 
+// JSONSliceValue represents a slice value in a YAML
+// config that can be overridden by environment variables
+
+type JSONSliceValue struct {
+	value []map[string]interface{}
+	Raw   []map[string]interface{}
+}
+
+// UnmarshalYAML converts YAML into an *JSONSliceValue
+func (val *JSONSliceValue) UnmarshalYAML(unmarshal func(interface{}) error) error {
+	unmarshaled := make([]interface{}, 0)
+	err := unmarshal(&unmarshaled)
+	if err != nil {
+		return err
+	}
+	interpolated := make([]map[string]interface{}, 0)
+	raw := make([]map[string]interface{}, 0)
+
+	for _, v := range unmarshaled {
+		i := make(map[string]interface{})
+		r := make(map[string]interface{})
+		for key, val := range v.(map[interface{}]interface{}) {
+			i[key.(string)], r[key.(string)], err = transformInterface(val)
+			if err != nil {
+				return err
+			}
+		}
+		interpolated = append(interpolated, i)
+		raw = append(raw, r)
+	}
+
+	val.Raw = raw
+	val.value = interpolated
+	return err
+}
+
+// Value returns the wrapped []interface{} value
+func (val *JSONSliceValue) Value() []map[string]interface{} {
+	return val.value
+}
+
 // transformInterface tries to transform any interface type into proper value with env expansion. It traverses maps and
 // slices and the actual interpolation is done on all simple string values in the structure. It returns a copy of any
 // map or slice value instead of modifying them in place and also return value without interpolation but with converted

pkg/services/provisioning/values/values_test.go

@@ -220,6 +220,54 @@ func TestValues(t *testing.T) {
 			})
 		})
 
+		t.Run("JSONSliceValue", func(t *testing.T) {
+			type Data struct {
+				Val JSONSliceValue `yaml:"val"`
+			}
+			d := &Data{}
+
+			t.Run("Should unmarshal top-level slices and nested structures", func(t *testing.T) {
+				doc := `
+                 val:
+                   - interpolatedString: $STRING
+                     interpolatedInt: $INT
+                     string: "just a string"
+                   - interpolatedString: $STRING
+                     interpolatedInt: $INT
+                     string: "just a string"
+               `
+				unmarshalingTest(t, doc, d)
+
+				type stringMap = map[string]interface{}
+
+				require.Equal(t, []stringMap{
+					{
+						"interpolatedString": "test",
+						"interpolatedInt":    "1",
+						"string":             "just a string",
+					},
+					{
+						"interpolatedString": "test",
+						"interpolatedInt":    "1",
+						"string":             "just a string",
+					},
+				}, d.Val.Value())
+
+				require.Equal(t, []stringMap{
+					{
+						"interpolatedString": "$STRING",
+						"interpolatedInt":    "$INT",
+						"string":             "just a string",
+					},
+					{
+						"interpolatedString": "$STRING",
+						"interpolatedInt":    "$INT",
+						"string":             "just a string",
+					},
+				}, d.Val.Raw)
+			})
+		})
+
 		t.Run("StringMapValue", func(t *testing.T) {
 			type Data struct {
 				Val StringMapValue `yaml:"val"`

pkg/services/sqlstore/datasource.go

@@ -134,13 +134,15 @@ func (ss *SQLStore) DeleteDataSource(ctx context.Context, cmd *datasources.Delet
 		}
 
 		// Publish data source deletion event
-		sess.publishAfterCommit(&events.DataSourceDeleted{
-			Timestamp: time.Now(),
-			Name:      cmd.Name,
-			ID:        cmd.ID,
-			UID:       cmd.UID,
-			OrgID:     cmd.OrgID,
-		})
+		if cmd.DeletedDatasourcesCount > 0 {
+			sess.publishAfterCommit(&events.DataSourceDeleted{
+				Timestamp: time.Now(),
+				Name:      ds.Name,
+				ID:        ds.Id,
+				UID:       ds.Uid,
+				OrgID:     ds.OrgId,
+			})
+		}
 
 		return nil
 	})

pkg/services/sqlstore/datasource_test.go

@@ -251,7 +251,7 @@ func TestIntegrationDataAccess(t *testing.T) {
 		})
 
 		err := sqlStore.DeleteDataSource(context.Background(),
-			&datasources.DeleteDataSourceCommand{ID: ds.Id, UID: "nisse-uid", Name: "nisse", OrgID: int64(123123)})
+			&datasources.DeleteDataSourceCommand{ID: ds.Id, UID: ds.Uid, Name: ds.Name, OrgID: ds.OrgId})
 		require.NoError(t, err)
 
 		require.Eventually(t, func() bool {
@@ -259,9 +259,27 @@ func TestIntegrationDataAccess(t *testing.T) {
 		}, time.Second, time.Millisecond)
 
 		require.Equal(t, ds.Id, deleted.ID)
-		require.Equal(t, int64(123123), deleted.OrgID)
-		require.Equal(t, "nisse", deleted.Name)
-		require.Equal(t, "nisse-uid", deleted.UID)
+		require.Equal(t, ds.OrgId, deleted.OrgID)
+		require.Equal(t, ds.Name, deleted.Name)
+		require.Equal(t, ds.Uid, deleted.UID)
+	})
+
+	t.Run("does not fire an event when the datasource is not deleted", func(t *testing.T) {
+		sqlStore := InitTestDB(t)
+
+		var called bool
+		sqlStore.bus.AddEventListener(func(ctx context.Context, e *events.DataSourceDeleted) error {
+			called = true
+			return nil
+		})
+
+		err := sqlStore.DeleteDataSource(context.Background(),
+			&datasources.DeleteDataSourceCommand{ID: 1, UID: "non-existing", Name: "non-existing", OrgID: int64(10)})
+		require.NoError(t, err)
+
+		require.Never(t, func() bool {
+			return called
+		}, time.Second, time.Millisecond)
 	})
 
 	t.Run("DeleteDataSourceByName", func(t *testing.T) {

pkg/services/sqlstore/migrations/correlations_mig.go

@@ -0,0 +1,21 @@
+package migrations
+
+import (
+	. "github.com/grafana/grafana/pkg/services/sqlstore/migrator"
+)
+
+func addCorrelationsMigrations(mg *Migrator) {
+	correlationsV1 := Table{
+		Name: "correlation",
+		Columns: []*Column{
+			{Name: "uid", Type: DB_NVarchar, Length: 40, Nullable: false, IsPrimaryKey: true},
+			{Name: "source_uid", Type: DB_NVarchar, Length: 40, Nullable: false, IsPrimaryKey: true},
+			// Nullable because in the future we want to have correlations to external resources
+			{Name: "target_uid", Type: DB_NVarchar, Length: 40, Nullable: true},
+			{Name: "label", Type: DB_Text, Nullable: false},
+			{Name: "description", Type: DB_Text, Nullable: false},
+		},
+	}
+
+	mg.AddMigration("create correlation table v1", NewAddTableMigration(correlationsV1))
+}

pkg/services/sqlstore/migrations/migrations.go

@@ -75,6 +75,8 @@ func (*OSSMigrations) AddMigration(mg *Migrator) {
 
 	addQueryHistoryStarMigrations(mg)
 
+	addCorrelationsMigrations(mg)
+
 	if mg.Cfg != nil && mg.Cfg.IsFeatureToggleEnabled != nil {
 		if mg.Cfg.IsFeatureToggleEnabled(featuremgmt.FlagDashboardComments) || mg.Cfg.IsFeatureToggleEnabled(featuremgmt.FlagAnnotationComments) {
 			addCommentGroupMigrations(mg)