Correlations: Add CreateCorrelation API

pkg/api/api.go

@@ -317,6 +317,11 @@ func (hs *HTTPServer) registerRoutes() {
 			datasourceRoute.Get("/uid/:uid", authorize(reqOrgAdmin, ac.EvalPermission(datasources.ActionRead, uidScope)), routing.Wrap(hs.GetDataSourceByUID))
 			datasourceRoute.Get("/name/:name", authorize(reqOrgAdmin, ac.EvalPermission(datasources.ActionRead, nameScope)), routing.Wrap(hs.GetDataSourceByName))
 			datasourceRoute.Get("/id/:name", authorize(reqSignedIn, ac.EvalPermission(datasources.ActionIDRead, nameScope)), routing.Wrap(hs.GetDataSourceIdByName))
+
+			// Correlations
+			datasourceRoute.Group("/:uid/correlations", func(correlationsRoute routing.RouteRegister) {
+				correlationsRoute.Post("/", authorize(reqOrgAdmin, ac.EvalPermission(datasources.ActionWrite, uidScope)), routing.Wrap(hs.CreateCorrelation))
+			})
 		})
 
 		apiRoute.Get("/plugins", routing.Wrap(hs.GetPluginList))

pkg/api/datasources.go

@@ -429,6 +429,42 @@ func (hs *HTTPServer) GetDataSourceIdByName(c *models.ReqContext) response.Respo
 	return response.JSON(http.StatusOK, &dtos)
 }
 
+// Post /api/datasources/:uid/correlations
+func (hs *HTTPServer) CreateCorrelation(c *models.ReqContext) response.Response {
+	cmd := datasources.CreateCorrelationCommand{
+		OrgID: c.OrgId,
+	}
+	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"]
+
+	err := hs.DataSourcesService.CreateCorrelation(c.Req.Context(), &cmd)
+
+	if err != nil {
+		// TODO: we may want to differentiate the following error between source and target DS
+		if errors.Is(err, datasources.ErrDataSourceNotFound) {
+			return response.Error(http.StatusNotFound, "Data source not found", nil)
+		}
+		if errors.Is(err, datasources.ErrDatasourceIsReadOnly) {
+			return response.Error(http.StatusForbidden, "Data source is read only", nil)
+		}
+		if errors.Is(err, datasources.ErrCorrelationExists) {
+			return response.Error(http.StatusConflict, fmt.Sprintf("Correlation to %s already exists", cmd.TargetUID), nil)
+		}
+
+		return response.Error(http.StatusInternalServerError, "Failed to query datasources", err)
+	}
+
+	// TODO: maybe this?
+	// hs.Live.HandleDatasourceUpdate(c.OrgId, cmd.SourceUID)
+
+	return response.JSON(http.StatusOK, util.DynMap{
+		"message":     "Correlation created",
+		"correlation": cmd.Result,
+	})
+}
+
 // /api/datasources/:id/resources/*
 func (hs *HTTPServer) CallDatasourceResource(c *models.ReqContext) {
 	datasourceID, err := strconv.ParseInt(web.Params(c.Req)[":id"], 10, 64)

pkg/api/docs/definitions/datasources.go

@@ -333,6 +333,19 @@ import (
 // 404: notFoundError
 // 500: internalServerError
 
+// swagger:route POST /datasources/{uid}/correlations correlations createCorrelation
+//
+// Creates a correlation.
+//
+// Responses:
+// 200: createCorrelationResponse
+// 400: badRequestError
+// 401: unauthorisedError
+// 403: forbiddenError
+// 404: notFoundError
+// 409: conflictError
+// 500: internalServerError
+
 // swagger:parameters updateDatasourceByID datasourceProxyDELETEcalls
 // swagger:parameters checkDatasourceHealthByID fetchDatasourceResourcesByID
 type DatasourceID struct {
@@ -366,7 +379,7 @@ type DatasourceProxyGETcallsParams struct {
 }
 
 // swagger:parameters datasourceProxyDELETEByUIDcalls
-// swagger:parameters checkDatasourceHealth fetchDatasourceResources
+// swagger:parameters checkDatasourceHealth fetchDatasourceResources createCorrelation
 type DatasourceUID struct {
 	// in:path
 	// required:true
@@ -480,6 +493,13 @@ type UpdateDatasourceByUIDParams struct {
 	DatasourceUID string `json:"uid"`
 }
 
+// swagger:parameters createCorrelation
+type CreateCorrelationParams struct {
+	// in:body
+	// required:true
+	Body datasources.CreateCorrelationCommand
+}
+
 // swagger:response getDatasourcesResponse
 type GetDatasourcesResponse struct {
 	// The response message
@@ -548,3 +568,18 @@ type DeleteDatasourceByNameResponse struct {
 		Message string `json:"message"`
 	} `json:"body"`
 }
+
+// swagger:response createCorrelationResponse
+type CreateCorrelationResponse struct {
+	// in: body
+	Body struct {
+		// Correlation properties
+		// required: true
+		Correlation datasources.Correlation `json:"correlation"`
+
+		// Message Message of the created correlation.
+		// required: true
+		// example: Correlation created
+		Message string `json:"message"`
+	} `json:"body"`
+}

pkg/services/datasources/datasources.go

@@ -33,6 +33,9 @@ type DataSourceService interface {
 	// GetDefaultDataSource gets the default datasource.
 	GetDefaultDataSource(ctx context.Context, query *GetDefaultDataSourceQuery) error
 
+	// CreateCorrelation adds a correlation between two datasources.
+	CreateCorrelation(ctx context.Context, cmd *CreateCorrelationCommand) error
+
 	// GetHTTPTransport gets a datasource specific HTTP transport.
 	GetHTTPTransport(ctx context.Context, ds *DataSource, provider httpclient.Provider, customMiddlewares ...sdkhttpclient.Middleware) (http.RoundTripper, error)
 

pkg/services/datasources/errors.go

@@ -10,4 +10,6 @@ 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")
+	ErrCorrelationExists                 = errors.New("correlation to the same datasource already exists")
 )

pkg/services/datasources/fakes/fake_datasource_service.go

@@ -94,6 +94,24 @@ func (s *FakeDataSourceService) UpdateDataSource(ctx context.Context, cmd *datas
 	return datasources.ErrDataSourceNotFound
 }
 
+func (s *FakeDataSourceService) CreateCorrelation(ctx context.Context, cmd *datasources.CreateCorrelationCommand) error {
+	for _, datasource := range s.DataSources {
+		if cmd.SourceUID != "" && cmd.SourceUID == datasource.Uid {
+			newCorrelation := datasources.Correlation{
+				Target:      cmd.TargetUID,
+				Label:       cmd.Label,
+				Description: cmd.Description,
+			}
+
+			datasource.Correlations = append(datasource.Correlations, newCorrelation)
+
+			cmd.Result = newCorrelation
+			return nil
+		}
+	}
+	return datasources.ErrDataSourceNotFound
+}
+
 func (s *FakeDataSourceService) GetDefaultDataSource(ctx context.Context, query *datasources.GetDefaultDataSourceQuery) error {
 	return nil
 }

pkg/services/datasources/models.go

@@ -30,6 +30,12 @@ const (
 
 type DsAccess string
 
+type Correlation struct {
+	Target      string `json:"target"`
+	Label       string `json:"label,omitempty"`
+	Description string `json:"description,omitempty"`
+}
+
 type DataSource struct {
 	Id      int64 `json:"id"`
 	OrgId   int64 `json:"orgId"`
@@ -49,6 +55,7 @@ type DataSource struct {
 	BasicAuthPassword string            `json:"-"`
 	WithCredentials   bool              `json:"withCredentials"`
 	IsDefault         bool              `json:"isDefault"`
+	Correlations      []Correlation     `json:"correlations"`
 	JsonData          *simplejson.Json  `json:"jsonData"`
 	SecureJsonData    map[string][]byte `json:"secureJsonData"`
 	ReadOnly          bool              `json:"readOnly"`
@@ -147,6 +154,28 @@ type DeleteDataSourceCommand struct {
 	UpdateSecretFn UpdateSecretFn
 }
 
+// Correlations are uniquely identified by a SourceUid+TargetUid pair.
+
+// CreateCorrelationCommand adds a correlation
+type CreateCorrelationCommand struct {
+	TargetUID   string `json:"targetUid" binding:"Required"`
+	Label       string `json:"label"`
+	Description string `json:"description"`
+
+	SourceUID string      `json:"-"`
+	OrgID     int64       `json:"-"`
+	Result    Correlation `json:"-"`
+}
+
+// UpdateCorrelationsCommand updates a correlation
+type UpdateCorrelationsCommand struct {
+	SourceUID    string
+	TargetUID    string
+	Correlations []Correlation
+	OrgId        int64
+	Result       []Correlation
+}
+
 // Function for updating secrets along with datasources, to ensure atomicity
 type UpdateSecretFn func() error
 

pkg/services/datasources/service/datasource_service.go

@@ -237,6 +237,66 @@ func (s *Service) GetDefaultDataSource(ctx context.Context, query *datasources.G
 	return s.SQLStore.GetDefaultDataSource(ctx, query)
 }
 
+// As correlations gets added in the data_source table as JSON, there's no way to delete an entry
+// from the correlations column when the target datasource gets deleted from grafana (or changes orgId).
+// We therefore need to either:
+// 	1) do a full table scan whenever we delete a datasource and delete correlations having the deleted DS as a target
+// 	2) setup a cleanup job that cleans the correlation column from stale data
+// 	3) do nothing and handle it in the FE
+func (s *Service) CreateCorrelation(ctx context.Context, cmd *datasources.CreateCorrelationCommand) error {
+	return s.SQLStore.InTransaction(ctx, func(ctx context.Context) error {
+		var err error
+
+		// TODO: the following is far from efficient, but should be enough to get a POC running
+		query := &datasources.GetDataSourceQuery{
+			OrgId: cmd.OrgID,
+			Uid:   cmd.SourceUID,
+		}
+
+		if err = s.SQLStore.GetDataSource(ctx, query); err != nil {
+			// Source datasource does not exist
+			return err
+		}
+
+		ds := query.Result
+		if ds.ReadOnly {
+			return datasources.ErrDatasourceIsReadOnly
+		}
+
+		if err = s.SQLStore.GetDataSource(ctx, &datasources.GetDataSourceQuery{
+			OrgId: cmd.OrgID,
+			Uid:   cmd.TargetUID,
+		}); err != nil {
+			// target datasource does not exist
+			return err
+		}
+
+		for _, correlation := range ds.Correlations {
+			if correlation.Target == cmd.TargetUID {
+				return datasources.ErrCorrelationExists
+			}
+		}
+
+		newCorrelation := datasources.Correlation{
+			Target:      cmd.TargetUID,
+			Label:       cmd.Label,
+			Description: cmd.Description,
+		}
+
+		ds.Correlations = append(ds.Correlations, newCorrelation)
+
+		if err = s.SQLStore.UpdateCorrelations(ctx, &datasources.UpdateCorrelationsCommand{
+			SourceUID:    ds.Uid,
+			OrgId:        cmd.OrgID,
+			Correlations: ds.Correlations,
+		}); err == nil {
+			cmd.Result = newCorrelation
+		}
+
+		return err
+	})
+}
+
 func (s *Service) GetHTTPClient(ctx context.Context, ds *datasources.DataSource, provider httpclient.Provider) (*http.Client, error) {
 	transport, err := s.GetHTTPTransport(ctx, ds, provider)
 	if err != nil {

pkg/services/sqlstore/datasource.go

@@ -289,6 +289,28 @@ func (ss *SQLStore) UpdateDataSource(ctx context.Context, cmd *datasources.Updat
 	})
 }
 
+func (ss *SQLStore) UpdateCorrelations(ctx context.Context, cmd *datasources.UpdateCorrelationsCommand) error {
+	return ss.WithTransactionalDbSession(ctx, func(sess *DBSession) error {
+		ds := &datasources.DataSource{
+			Correlations: cmd.Correlations,
+			Updated:      time.Now(),
+		}
+
+		// TODO: should we also check for version here instead (like UpdateDatasource)?
+		affected, err := sess.Where("uid=? and org_id=?", cmd.SourceUID, cmd.OrgId).Omit("json_data").Update(ds)
+		if err != nil {
+			return err
+		}
+
+		if affected == 0 {
+			return datasources.ErrDataSourceUpdatingOldVersion
+		}
+
+		cmd.Result = cmd.Correlations
+		return err
+	})
+}
+
 func generateNewDatasourceUid(sess *DBSession, orgId int64) (string, error) {
 	for i := 0; i < 3; i++ {
 		uid := generateNewUid()

public/api-merged.json

@@ -4039,6 +4039,52 @@
         }
       }
     },
+    "/datasources/{uid}/correlations": {
+      "post": {
+        "tags": ["correlations"],
+        "summary": "Creates a correlation.",
+        "operationId": "createCorrelation",
+        "parameters": [
+          {
+            "type": "string",
+            "name": "uid",
+            "in": "path",
+            "required": true
+          },
+          {
+            "name": "Body",
+            "in": "body",
+            "required": true,
+            "schema": {
+              "$ref": "#/definitions/CreateCorrelationCommand"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "$ref": "#/responses/createCorrelationResponse"
+          },
+          "400": {
+            "$ref": "#/responses/badRequestError"
+          },
+          "401": {
+            "$ref": "#/responses/unauthorisedError"
+          },
+          "403": {
+            "$ref": "#/responses/forbiddenError"
+          },
+          "404": {
+            "$ref": "#/responses/notFoundError"
+          },
+          "409": {
+            "$ref": "#/responses/conflictError"
+          },
+          "500": {
+            "$ref": "#/responses/internalServerError"
+          }
+        }
+      }
+    },
     "/ds/query": {
       "post": {
         "description": "If you are running Grafana Enterprise and have Fine-grained access control enabled\nyou need to have a permission with action: `datasources:query`.",
@@ -9865,6 +9911,20 @@
         "$ref": "#/definitions/EmbeddedContactPoint"
       }
     },
+    "Correlation": {
+      "type": "object",
+      "properties": {
+        "description": {
+          "type": "string"
+        },
+        "label": {
+          "type": "string"
+        },
+        "target": {
+          "type": "string"
+        }
+      }
+    },
     "CreateAlertNotificationCommand": {
       "type": "object",
       "properties": {
@@ -9900,6 +9960,21 @@
         }
       }
     },
+    "CreateCorrelationCommand": {
+      "description": "CreateCorrelationCommand adds a correlation",
+      "type": "object",
+      "properties": {
+        "description": {
+          "type": "string"
+        },
+        "label": {
+          "type": "string"
+        },
+        "targetUid": {
+          "type": "string"
+        }
+      }
+    },
     "CreateDashboardSnapshotCommand": {
       "type": "object",
       "required": ["dashboard"],
@@ -16277,6 +16352,23 @@
         }
       }
     },
+    "createCorrelationResponse": {
+      "description": "",
+      "schema": {
+        "type": "object",
+        "required": ["correlation", "message"],
+        "properties": {
+          "correlation": {
+            "$ref": "#/definitions/Correlation"
+          },
+          "message": {
+            "description": "Message Message of the created correlation.",
+            "type": "string",
+            "example": "Correlation created"
+          }
+        }
+      }
+    },
     "createOrUpdateDatasourceResponse": {
       "description": "",
       "schema": {

public/api-spec.json

@@ -4039,6 +4039,52 @@
         }
       }
     },
+    "/datasources/{uid}/correlations": {
+      "post": {
+        "tags": ["correlations"],
+        "summary": "Creates a correlation.",
+        "operationId": "createCorrelation",
+        "parameters": [
+          {
+            "type": "string",
+            "name": "uid",
+            "in": "path",
+            "required": true
+          },
+          {
+            "name": "Body",
+            "in": "body",
+            "required": true,
+            "schema": {
+              "$ref": "#/definitions/CreateCorrelationCommand"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "$ref": "#/responses/createCorrelationResponse"
+          },
+          "400": {
+            "$ref": "#/responses/badRequestError"
+          },
+          "401": {
+            "$ref": "#/responses/unauthorisedError"
+          },
+          "403": {
+            "$ref": "#/responses/forbiddenError"
+          },
+          "404": {
+            "$ref": "#/responses/notFoundError"
+          },
+          "409": {
+            "$ref": "#/responses/conflictError"
+          },
+          "500": {
+            "$ref": "#/responses/internalServerError"
+          }
+        }
+      }
+    },
     "/ds/query": {
       "post": {
         "description": "If you are running Grafana Enterprise and have Fine-grained access control enabled\nyou need to have a permission with action: `datasources:query`.",
@@ -8995,6 +9041,20 @@
         }
       }
     },
+    "Correlation": {
+      "type": "object",
+      "properties": {
+        "description": {
+          "type": "string"
+        },
+        "label": {
+          "type": "string"
+        },
+        "target": {
+          "type": "string"
+        }
+      }
+    },
     "CreateAlertNotificationCommand": {
       "type": "object",
       "properties": {
@@ -9030,6 +9090,21 @@
         }
       }
     },
+    "CreateCorrelationCommand": {
+      "description": "CreateCorrelationCommand adds a correlation",
+      "type": "object",
+      "properties": {
+        "description": {
+          "type": "string"
+        },
+        "label": {
+          "type": "string"
+        },
+        "targetUid": {
+          "type": "string"
+        }
+      }
+    },
     "CreateDashboardSnapshotCommand": {
       "type": "object",
       "required": ["dashboard"],
@@ -12736,6 +12811,23 @@
         }
       }
     },
+    "createCorrelationResponse": {
+      "description": "",
+      "schema": {
+        "type": "object",
+        "required": ["correlation", "message"],
+        "properties": {
+          "correlation": {
+            "$ref": "#/definitions/Correlation"
+          },
+          "message": {
+            "description": "Message Message of the created correlation.",
+            "type": "string",
+            "example": "Correlation created"
+          }
+        }
+      }
+    },
     "createOrUpdateDatasourceResponse": {
       "description": "",
       "schema": {