fix(edgegroups): fix data-race in edgeGroupCreate EE-4435 (#8477)

api/dataservices/edgegroup/edgegroup.go

@@ -1,11 +1,7 @@
 package edgegroup
 
 import (
-	"fmt"
-
 	portainer "github.com/portainer/portainer/api"
-
-	"github.com/rs/zerolog/log"
 )
 
 // BucketName represents the name of the bucket where this service stores data.
@@ -32,47 +28,46 @@ func NewService(connection portainer.Connection) (*Service, error) {
 	}, nil
 }
 
-// EdgeGroups return an array containing all the Edge groups.
+func (service *Service) Tx(tx portainer.Transaction) ServiceTx {
+	return ServiceTx{
+		service: service,
+		tx:      tx,
+	}
+}
+
+// EdgeGroups return a slice containing all the Edge groups.
 func (service *Service) EdgeGroups() ([]portainer.EdgeGroup, error) {
-	var groups = make([]portainer.EdgeGroup, 0)
+	var groups []portainer.EdgeGroup
+	var err error
 
-	err := service.connection.GetAllWithJsoniter(
-		BucketName,
-		&portainer.EdgeGroup{},
-		func(obj interface{}) (interface{}, error) {
-			group, ok := obj.(*portainer.EdgeGroup)
-			if !ok {
-				log.Debug().Str("obj", fmt.Sprintf("%#v", obj)).Msg("failed to convert to EdgeGroup object")
-				return nil, fmt.Errorf("Failed to convert to EdgeGroup object: %s", obj)
-			}
-			groups = append(groups, *group)
-
-			return &portainer.EdgeGroup{}, nil
-		})
+	err = service.connection.ViewTx(func(tx portainer.Transaction) error {
+		groups, err = service.Tx(tx).EdgeGroups()
+		return err
+	})
 
 	return groups, err
 }
 
 // EdgeGroup returns an Edge group by ID.
 func (service *Service) EdgeGroup(ID portainer.EdgeGroupID) (*portainer.EdgeGroup, error) {
-	var group portainer.EdgeGroup
-	identifier := service.connection.ConvertToKey(int(ID))
+	var group *portainer.EdgeGroup
+	var err error
 
-	err := service.connection.GetObject(BucketName, identifier, &group)
-	if err != nil {
-		return nil, err
-	}
+	err = service.connection.ViewTx(func(tx portainer.Transaction) error {
+		group, err = service.Tx(tx).EdgeGroup(ID)
+		return err
+	})
 
-	return &group, nil
+	return group, err
 }
 
-// Deprecated: Use UpdateEdgeGroupFunc instead.
+// UpdateEdgeGroup updates an edge group.
 func (service *Service) UpdateEdgeGroup(ID portainer.EdgeGroupID, group *portainer.EdgeGroup) error {
 	identifier := service.connection.ConvertToKey(int(ID))
 	return service.connection.UpdateObject(BucketName, identifier, group)
 }
 
-// UpdateEdgeGroupFunc updates an edge group inside a transaction avoiding data races.
+// Deprecated: UpdateEdgeGroupFunc updates an edge group inside a transaction avoiding data races.
 func (service *Service) UpdateEdgeGroupFunc(ID portainer.EdgeGroupID, updateFunc func(edgeGroup *portainer.EdgeGroup)) error {
 	id := service.connection.ConvertToKey(int(ID))
 	edgeGroup := &portainer.EdgeGroup{}
@@ -84,17 +79,14 @@ func (service *Service) UpdateEdgeGroupFunc(ID portainer.EdgeGroupID, updateFunc
 
 // DeleteEdgeGroup deletes an Edge group.
 func (service *Service) DeleteEdgeGroup(ID portainer.EdgeGroupID) error {
-	identifier := service.connection.ConvertToKey(int(ID))
-	return service.connection.DeleteObject(BucketName, identifier)
+	return service.connection.UpdateTx(func(tx portainer.Transaction) error {
+		return service.Tx(tx).DeleteEdgeGroup(ID)
+	})
 }
 
 // CreateEdgeGroup assign an ID to a new Edge group and saves it.
 func (service *Service) Create(group *portainer.EdgeGroup) error {
-	return service.connection.CreateObject(
-		BucketName,
-		func(id uint64) (int, interface{}) {
-			group.ID = portainer.EdgeGroupID(id)
-			return int(group.ID), group
-		},
-	)
+	return service.connection.UpdateTx(func(tx portainer.Transaction) error {
+		return service.Tx(tx).Create(group)
+	})
 }

api/dataservices/edgegroup/tx.go

@@ -0,0 +1,80 @@
+package edgegroup
+
+import (
+	"errors"
+	"fmt"
+
+	portainer "github.com/portainer/portainer/api"
+
+	"github.com/rs/zerolog/log"
+)
+
+type ServiceTx struct {
+	service *Service
+	tx      portainer.Transaction
+}
+
+func (service ServiceTx) BucketName() string {
+	return BucketName
+}
+
+// EdgeGroups return a slice containing all the Edge groups.
+func (service ServiceTx) EdgeGroups() ([]portainer.EdgeGroup, error) {
+	var groups = make([]portainer.EdgeGroup, 0)
+
+	err := service.tx.GetAllWithJsoniter(
+		BucketName,
+		&portainer.EdgeGroup{},
+		func(obj interface{}) (interface{}, error) {
+			group, ok := obj.(*portainer.EdgeGroup)
+			if !ok {
+				log.Debug().Str("obj", fmt.Sprintf("%#v", obj)).Msg("failed to convert to EdgeGroup object")
+				return nil, fmt.Errorf("Failed to convert to EdgeGroup object: %s", obj)
+			}
+			groups = append(groups, *group)
+
+			return &portainer.EdgeGroup{}, nil
+		})
+
+	return groups, err
+}
+
+// EdgeGroup returns an Edge group by ID.
+func (service ServiceTx) EdgeGroup(ID portainer.EdgeGroupID) (*portainer.EdgeGroup, error) {
+	var group portainer.EdgeGroup
+	identifier := service.service.connection.ConvertToKey(int(ID))
+
+	err := service.tx.GetObject(BucketName, identifier, &group)
+	if err != nil {
+		return nil, err
+	}
+
+	return &group, nil
+}
+
+// UpdateEdgeGroup updates an edge group.
+func (service ServiceTx) UpdateEdgeGroup(ID portainer.EdgeGroupID, group *portainer.EdgeGroup) error {
+	identifier := service.service.connection.ConvertToKey(int(ID))
+	return service.tx.UpdateObject(BucketName, identifier, group)
+}
+
+// UpdateEdgeGroupFunc is a no-op inside a transaction.
+func (service ServiceTx) UpdateEdgeGroupFunc(ID portainer.EdgeGroupID, updateFunc func(edgeGroup *portainer.EdgeGroup)) error {
+	return errors.New("cannot be called inside a transaction")
+}
+
+// DeleteEdgeGroup deletes an Edge group.
+func (service ServiceTx) DeleteEdgeGroup(ID portainer.EdgeGroupID) error {
+	identifier := service.service.connection.ConvertToKey(int(ID))
+	return service.tx.DeleteObject(BucketName, identifier)
+}
+
+func (service ServiceTx) Create(group *portainer.EdgeGroup) error {
+	return service.tx.CreateObject(
+		BucketName,
+		func(id uint64) (int, interface{}) {
+			group.ID = portainer.EdgeGroupID(id)
+			return int(group.ID), group
+		},
+	)
+}

api/dataservices/endpoint/endpoint.go

@@ -1,20 +1,14 @@
 package endpoint
 
 import (
-	"fmt"
 	"sync"
 	"time"
 
 	portainer "github.com/portainer/portainer/api"
-	"github.com/portainer/portainer/api/internal/edge/cache"
-
-	"github.com/rs/zerolog/log"
 )
 
-const (
-	// BucketName represents the name of the bucket where this service stores data.
-	BucketName = "endpoints"
-)
+// BucketName represents the name of the bucket where this service stores data.
+const BucketName = "endpoints"
 
 // Service represents a service for managing environment(endpoint) data.
 type Service struct {
@@ -56,84 +50,54 @@ func NewService(connection portainer.Connection) (*Service, error) {
 	return s, nil
 }
 
+func (service *Service) Tx(tx portainer.Transaction) ServiceTx {
+	return ServiceTx{
+		service: service,
+		tx:      tx,
+	}
+}
+
 // Endpoint returns an environment(endpoint) by ID.
 func (service *Service) Endpoint(ID portainer.EndpointID) (*portainer.Endpoint, error) {
-	var endpoint portainer.Endpoint
-	identifier := service.connection.ConvertToKey(int(ID))
+	var endpoint *portainer.Endpoint
+	var err error
 
-	err := service.connection.GetObject(BucketName, identifier, &endpoint)
+	err = service.connection.ViewTx(func(tx portainer.Transaction) error {
+		endpoint, err = service.Tx(tx).Endpoint(ID)
+		return err
+	})
 	if err != nil {
 		return nil, err
 	}
 
 	endpoint.LastCheckInDate, _ = service.Heartbeat(ID)
 
-	return &endpoint, nil
+	return endpoint, nil
 }
 
 // UpdateEndpoint updates an environment(endpoint).
 func (service *Service) UpdateEndpoint(ID portainer.EndpointID, endpoint *portainer.Endpoint) error {
-	identifier := service.connection.ConvertToKey(int(ID))
-
-	err := service.connection.UpdateObject(BucketName, identifier, endpoint)
-	if err != nil {
-		return err
-	}
-
-	service.mu.Lock()
-	if len(endpoint.EdgeID) > 0 {
-		service.idxEdgeID[endpoint.EdgeID] = ID
-	}
-	service.heartbeats.Store(ID, endpoint.LastCheckInDate)
-	service.mu.Unlock()
-
-	cache.Del(endpoint.ID)
-
-	return nil
+	return service.connection.UpdateTx(func(tx portainer.Transaction) error {
+		return service.Tx(tx).UpdateEndpoint(ID, endpoint)
+	})
 }
 
 // DeleteEndpoint deletes an environment(endpoint).
 func (service *Service) DeleteEndpoint(ID portainer.EndpointID) error {
-	identifier := service.connection.ConvertToKey(int(ID))
-
-	err := service.connection.DeleteObject(BucketName, identifier)
-	if err != nil {
-		return err
-	}
-
-	service.mu.Lock()
-	for edgeID, endpointID := range service.idxEdgeID {
-		if endpointID == ID {
-			delete(service.idxEdgeID, edgeID)
-			break
-		}
-	}
-	service.heartbeats.Delete(ID)
-	service.mu.Unlock()
-
-	cache.Del(ID)
-
-	return nil
+	return service.connection.UpdateTx(func(tx portainer.Transaction) error {
+		return service.Tx(tx).DeleteEndpoint(ID)
+	})
 }
 
 // Endpoints return an array containing all the environments(endpoints).
 func (service *Service) Endpoints() ([]portainer.Endpoint, error) {
-	var endpoints = make([]portainer.Endpoint, 0)
+	var endpoints []portainer.Endpoint
+	var err error
 
-	err := service.connection.GetAllWithJsoniter(
-		BucketName,
-		&portainer.Endpoint{},
-		func(obj interface{}) (interface{}, error) {
-			endpoint, ok := obj.(*portainer.Endpoint)
-			if !ok {
-				log.Debug().Str("obj", fmt.Sprintf("%#v", obj)).Msg("failed to convert to Endpoint object")
-				return nil, fmt.Errorf("failed to convert to Endpoint object: %s", obj)
-			}
-
-			endpoints = append(endpoints, *endpoint)
-
-			return &portainer.Endpoint{}, nil
-		})
+	err = service.connection.ViewTx(func(tx portainer.Transaction) error {
+		endpoints, err = service.Tx(tx).Endpoints()
+		return err
+	})
 
 	if err != nil {
 		return endpoints, err
@@ -170,22 +134,20 @@ func (service *Service) UpdateHeartbeat(endpointID portainer.EndpointID) {
 
 // CreateEndpoint assign an ID to a new environment(endpoint) and saves it.
 func (service *Service) Create(endpoint *portainer.Endpoint) error {
-	err := service.connection.CreateObjectWithId(BucketName, int(endpoint.ID), endpoint)
-	if err != nil {
-		return err
-	}
-
-	service.mu.Lock()
-	if len(endpoint.EdgeID) > 0 {
-		service.idxEdgeID[endpoint.EdgeID] = endpoint.ID
-	}
-	service.heartbeats.Store(endpoint.ID, endpoint.LastCheckInDate)
-	service.mu.Unlock()
-
-	return nil
+	return service.connection.UpdateTx(func(tx portainer.Transaction) error {
+		return service.Tx(tx).Create(endpoint)
+	})
 }
 
 // GetNextIdentifier returns the next identifier for an environment(endpoint).
 func (service *Service) GetNextIdentifier() int {
-	return service.connection.GetNextIdentifier(BucketName)
+	var identifier int
+
+	service.connection.UpdateTx(func(tx portainer.Transaction) error {
+		identifier = service.Tx(tx).GetNextIdentifier()
+
+		return nil
+	})
+
+	return identifier
 }

api/dataservices/endpoint/tx.go

@@ -0,0 +1,137 @@
+package endpoint
+
+import (
+	"fmt"
+
+	portainer "github.com/portainer/portainer/api"
+	"github.com/portainer/portainer/api/internal/edge/cache"
+
+	"github.com/rs/zerolog/log"
+)
+
+type ServiceTx struct {
+	service *Service
+	tx      portainer.Transaction
+}
+
+func (service ServiceTx) BucketName() string {
+	return BucketName
+}
+
+// Endpoint returns an environment(endpoint) by ID.
+func (service ServiceTx) Endpoint(ID portainer.EndpointID) (*portainer.Endpoint, error) {
+	var endpoint portainer.Endpoint
+	identifier := service.service.connection.ConvertToKey(int(ID))
+
+	err := service.tx.GetObject(BucketName, identifier, &endpoint)
+	if err != nil {
+		return nil, err
+	}
+
+	return &endpoint, nil
+}
+
+// UpdateEndpoint updates an environment(endpoint).
+func (service ServiceTx) UpdateEndpoint(ID portainer.EndpointID, endpoint *portainer.Endpoint) error {
+	identifier := service.service.connection.ConvertToKey(int(ID))
+
+	err := service.tx.UpdateObject(BucketName, identifier, endpoint)
+	if err != nil {
+		return err
+	}
+
+	service.service.mu.Lock()
+	if len(endpoint.EdgeID) > 0 {
+		service.service.idxEdgeID[endpoint.EdgeID] = ID
+	}
+	service.service.heartbeats.Store(ID, endpoint.LastCheckInDate)
+	service.service.mu.Unlock()
+
+	cache.Del(endpoint.ID)
+
+	return nil
+}
+
+// DeleteEndpoint deletes an environment(endpoint).
+func (service ServiceTx) DeleteEndpoint(ID portainer.EndpointID) error {
+	identifier := service.service.connection.ConvertToKey(int(ID))
+
+	err := service.tx.DeleteObject(BucketName, identifier)
+	if err != nil {
+		return err
+	}
+
+	service.service.mu.Lock()
+	for edgeID, endpointID := range service.service.idxEdgeID {
+		if endpointID == ID {
+			delete(service.service.idxEdgeID, edgeID)
+			break
+		}
+	}
+	service.service.heartbeats.Delete(ID)
+	service.service.mu.Unlock()
+
+	cache.Del(ID)
+
+	return nil
+}
+
+// Endpoints return an array containing all the environments(endpoints).
+func (service ServiceTx) Endpoints() ([]portainer.Endpoint, error) {
+	var endpoints = make([]portainer.Endpoint, 0)
+
+	err := service.tx.GetAllWithJsoniter(
+		BucketName,
+		&portainer.Endpoint{},
+		func(obj interface{}) (interface{}, error) {
+			endpoint, ok := obj.(*portainer.Endpoint)
+			if !ok {
+				log.Debug().Str("obj", fmt.Sprintf("%#v", obj)).Msg("failed to convert to Endpoint object")
+				return nil, fmt.Errorf("failed to convert to Endpoint object: %s", obj)
+			}
+
+			endpoints = append(endpoints, *endpoint)
+
+			return &portainer.Endpoint{}, nil
+		})
+
+	return endpoints, err
+}
+
+func (service ServiceTx) EndpointIDByEdgeID(edgeID string) (portainer.EndpointID, bool) {
+	log.Error().Str("func", "EndpointIDByEdgeID").Msg("cannot be called inside a transaction")
+
+	return 0, false
+}
+
+func (service ServiceTx) Heartbeat(endpointID portainer.EndpointID) (int64, bool) {
+	log.Error().Str("func", "Heartbeat").Msg("cannot be called inside a transaction")
+
+	return 0, false
+}
+
+func (service ServiceTx) UpdateHeartbeat(endpointID portainer.EndpointID) {
+	log.Error().Str("func", "UpdateHeartbeat").Msg("cannot be called inside a transaction")
+}
+
+// CreateEndpoint assign an ID to a new environment(endpoint) and saves it.
+func (service ServiceTx) Create(endpoint *portainer.Endpoint) error {
+	err := service.tx.CreateObjectWithId(BucketName, int(endpoint.ID), endpoint)
+	if err != nil {
+		return err
+	}
+
+	service.service.mu.Lock()
+	if len(endpoint.EdgeID) > 0 {
+		service.service.idxEdgeID[endpoint.EdgeID] = endpoint.ID
+	}
+	service.service.heartbeats.Store(endpoint.ID, endpoint.LastCheckInDate)
+	service.service.mu.Unlock()
+
+	return nil
+}
+
+// GetNextIdentifier returns the next identifier for an environment(endpoint).
+func (service ServiceTx) GetNextIdentifier() int {
+	return service.tx.GetNextIdentifier(BucketName)
+}

api/dataservices/interface.go

@@ -13,16 +13,7 @@ import (
 )
 
 type (
-	// DataStore defines the interface to manage the data
-	DataStore interface {
-		Open() (newStore bool, err error)
-		Init() error
-		Close() error
-		MigrateData() error
-		Rollback(force bool) error
-		CheckCurrentEdition() error
-		BackupTo(w io.Writer) error
-		Export(filename string) (err error)
+	DataStoreTx interface {
 		IsErrObjectNotFound(err error) bool
 		CustomTemplate() CustomTemplateService
 		EdgeGroup() EdgeGroupService
@@ -50,6 +41,22 @@ type (
 		Webhook() WebhookService
 	}
 
+	// DataStore defines the interface to manage the data
+	DataStore interface {
+		Open() (newStore bool, err error)
+		Init() error
+		Close() error
+		UpdateTx(func(DataStoreTx) error) error
+		ViewTx(func(DataStoreTx) error) error
+		MigrateData() error
+		Rollback(force bool) error
+		CheckCurrentEdition() error
+		BackupTo(w io.Writer) error
+		Export(filename string) (err error)
+
+		DataStoreTx
+	}
+
 	// CustomTemplateService represents a service to manage custom templates
 	CustomTemplateService interface {
 		GetNextIdentifier() int