datasources: change to optimisic concurrency

prerequisite for #9504

pkg/services/sqlstore/datasource.go

@@ -3,6 +3,8 @@ package sqlstore
 import (
 	"time"
 
+	"github.com/go-xorm/xorm"
+
 	"github.com/grafana/grafana/pkg/bus"
 	"github.com/grafana/grafana/pkg/components/securejsondata"
 	"github.com/grafana/grafana/pkg/metrics"
@@ -69,7 +71,6 @@ func DeleteDataSourceByName(cmd *m.DeleteDataSourceByNameCommand) error {
 }
 
 func AddDataSource(cmd *m.AddDataSourceCommand) error {
-
 	return inTransaction(func(sess *DBSession) error {
 		existing := m.DataSource{OrgId: cmd.OrgId, Name: cmd.Name}
 		has, _ := sess.Get(&existing)
@@ -96,6 +97,7 @@ func AddDataSource(cmd *m.AddDataSourceCommand) error {
 			SecureJsonData:    securejsondata.GetEncryptedJsonData(cmd.SecureJsonData),
 			Created:           time.Now(),
 			Updated:           time.Now(),
+			Version:           1,
 		}
 
 		if _, err := sess.Insert(ds); err != nil {
@@ -122,7 +124,6 @@ func updateIsDefaultFlag(ds *m.DataSource, sess *DBSession) error {
 }
 
 func UpdateDataSource(cmd *m.UpdateDataSourceCommand) error {
-
 	return inTransaction(func(sess *DBSession) error {
 		ds := &m.DataSource{
 			Id:                cmd.Id,
@@ -149,12 +150,25 @@ func UpdateDataSource(cmd *m.UpdateDataSourceCommand) error {
 		sess.UseBool("basic_auth")
 		sess.UseBool("with_credentials")
 
-		_, err := sess.Where("id=? and org_id=?", ds.Id, ds.OrgId).Update(ds)
+		var updateSession *xorm.Session
+		if cmd.Version != 0 {
+			updateSession = sess.Where("id=? and org_id=? and (version = ? or version < ?)", ds.Id, ds.OrgId, cmd.Version, cmd.Version)
+		} else {
+			updateSession = sess.Where("id=? and org_id=?", ds.Id, ds.OrgId)
+		}
+
+		affected, err := updateSession.Update(ds)
 		if err != nil {
 			return err
 		}
 
+		if affected == 0 {
+			return m.ErrDataSouceUpdatingOldVersion
+		}
+
 		err = updateIsDefaultFlag(ds, sess)
+
+		cmd.Result = ds
 		return err
 	})
 }

pkg/services/sqlstore/datasource_test.go

@@ -35,12 +35,9 @@ type Test struct {
 }
 
 func TestDataAccess(t *testing.T) {
-
 	Convey("Testing DB", t, func() {
 		InitTestDB(t)
-
 		Convey("Can add datasource", func() {
-
 			err := AddDataSource(&m.AddDataSourceCommand{
 				OrgId:    10,
 				Name:     "laban",
@@ -65,7 +62,6 @@ func TestDataAccess(t *testing.T) {
 		})
 
 		Convey("Given a datasource", func() {
-
 			err := AddDataSource(&m.AddDataSourceCommand{
 				OrgId:  10,
 				Name:   "nisse",
@@ -81,6 +77,89 @@ func TestDataAccess(t *testing.T) {
 
 			ds := query.Result[0]
 
+			Convey(" updated ", func() {
+				cmd := &m.UpdateDataSourceCommand{
+					Id:      ds.Id,
+					OrgId:   10,
+					Name:    "nisse",
+					Type:    m.DS_GRAPHITE,
+					Access:  m.DS_ACCESS_PROXY,
+					Url:     "http://test",
+					Version: ds.Version,
+				}
+
+				Convey("with same version as source", func() {
+					err := UpdateDataSource(cmd)
+					So(err, ShouldBeNil)
+				})
+
+				Convey("when someone else updated between read and update", func() {
+					query := m.GetDataSourcesQuery{OrgId: 10}
+					err = GetDataSources(&query)
+					So(err, ShouldBeNil)
+
+					ds := query.Result[0]
+					intendedUpdate := &m.UpdateDataSourceCommand{
+						Id:      ds.Id,
+						OrgId:   10,
+						Name:    "nisse",
+						Type:    m.DS_GRAPHITE,
+						Access:  m.DS_ACCESS_PROXY,
+						Url:     "http://test",
+						Version: ds.Version,
+					}
+
+					updateFromOtherUser := &m.UpdateDataSourceCommand{
+						Id:      ds.Id,
+						OrgId:   10,
+						Name:    "nisse",
+						Type:    m.DS_GRAPHITE,
+						Access:  m.DS_ACCESS_PROXY,
+						Url:     "http://test",
+						Version: ds.Version,
+					}
+
+					err := UpdateDataSource(updateFromOtherUser)
+					So(err, ShouldBeNil)
+
+					err = UpdateDataSource(intendedUpdate)
+					So(err, ShouldNotBeNil)
+				})
+
+				Convey("updating datasource without version", func() {
+					cmd := &m.UpdateDataSourceCommand{
+						Id:     ds.Id,
+						OrgId:  10,
+						Name:   "nisse",
+						Type:   m.DS_GRAPHITE,
+						Access: m.DS_ACCESS_PROXY,
+						Url:    "http://test",
+					}
+
+					Convey("should not raise errors", func() {
+						err := UpdateDataSource(cmd)
+						So(err, ShouldBeNil)
+					})
+				})
+
+				Convey("updating datasource without higher version", func() {
+					cmd := &m.UpdateDataSourceCommand{
+						Id:      ds.Id,
+						OrgId:   10,
+						Name:    "nisse",
+						Type:    m.DS_GRAPHITE,
+						Access:  m.DS_ACCESS_PROXY,
+						Url:     "http://test",
+						Version: 90000,
+					}
+
+					Convey("should not raise errors", func() {
+						err := UpdateDataSource(cmd)
+						So(err, ShouldBeNil)
+					})
+				})
+			})
+
 			Convey("Can delete datasource by id", func() {
 				err := DeleteDataSourceById(&m.DeleteDataSourceByIdCommand{Id: ds.Id, OrgId: ds.OrgId})
 				So(err, ShouldBeNil)
@@ -104,9 +183,6 @@ func TestDataAccess(t *testing.T) {
 				GetDataSources(&query)
 				So(len(query.Result), ShouldEqual, 1)
 			})
-
 		})
-
 	})
-
 }

pkg/services/sqlstore/migrations/datasource_mig.go

@@ -120,4 +120,10 @@ func addDataSourceMigration(mg *Migrator) {
 		{Name: "json_data", Type: DB_Text, Nullable: true},
 		{Name: "secure_json_data", Type: DB_Text, Nullable: true},
 	}))
+
+	const setVersionToOneWhereZero = `UPDATE data_source SET version = 1 WHERE version = 0`
+	mg.AddMigration("Update initial version to 1", new(RawSqlMigration).
+		Sqlite(setVersionToOneWhereZero).
+		Postgres(setVersionToOneWhereZero).
+		Mysql(setVersionToOneWhereZero))
 }