chore: import upstream snapshot with attribution
This commit is contained in:
wehub-resource-sync
@@ -0,0 +1,114 @@
|
||||
package endpointrelation
|
||||
|
||||
import (
|
||||
"sync"
|
||||
|
||||
portainer "github.com/portainer/portainer/api"
|
||||
"github.com/portainer/portainer/api/dataservices"
|
||||
"github.com/portainer/portainer/api/internal/edge/cache"
|
||||
)
|
||||
|
||||
// BucketName represents the name of the bucket where this service stores data.
|
||||
const BucketName = "endpoint_relations"
|
||||
|
||||
// Service represents a service for managing environment(endpoint) relation data.
|
||||
type Service struct {
|
||||
connection portainer.Connection
|
||||
updateStackFnTx func(tx portainer.Transaction, ID portainer.EdgeStackID, updateFunc func(edgeStack *portainer.EdgeStack)) error
|
||||
endpointRelationsCache []portainer.EndpointRelation
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
var _ dataservices.EndpointRelationService = &Service{}
|
||||
|
||||
func (service *Service) BucketName() string {
|
||||
return BucketName
|
||||
}
|
||||
|
||||
func (service *Service) RegisterUpdateStackFunction(
|
||||
updateFuncTx func(portainer.Transaction, portainer.EdgeStackID, func(*portainer.EdgeStack)) error,
|
||||
) {
|
||||
service.mu.Lock()
|
||||
defer service.mu.Unlock()
|
||||
|
||||
service.updateStackFnTx = updateFuncTx
|
||||
}
|
||||
|
||||
// NewService creates a new instance of a service.
|
||||
func NewService(connection portainer.Connection) (*Service, error) {
|
||||
if err := connection.SetServiceName(BucketName); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &Service{
|
||||
connection: connection,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (service *Service) Tx(tx portainer.Transaction) ServiceTx {
|
||||
return ServiceTx{
|
||||
service: service,
|
||||
tx: tx,
|
||||
}
|
||||
}
|
||||
|
||||
// EndpointRelations returns an array of all EndpointRelations
|
||||
func (service *Service) EndpointRelations() ([]portainer.EndpointRelation, error) {
|
||||
var all = make([]portainer.EndpointRelation, 0)
|
||||
|
||||
return all, service.connection.GetAll(
|
||||
BucketName,
|
||||
&portainer.EndpointRelation{},
|
||||
dataservices.AppendFn(&all),
|
||||
)
|
||||
}
|
||||
|
||||
// EndpointRelation returns a Environment(Endpoint) relation object by EndpointID
|
||||
func (service *Service) EndpointRelation(endpointID portainer.EndpointID) (*portainer.EndpointRelation, error) {
|
||||
var endpointRelation portainer.EndpointRelation
|
||||
identifier := service.connection.ConvertToKey(int(endpointID))
|
||||
|
||||
if err := service.connection.GetObject(BucketName, identifier, &endpointRelation); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &endpointRelation, nil
|
||||
}
|
||||
|
||||
// CreateEndpointRelation saves endpointRelation
|
||||
func (service *Service) Create(endpointRelation *portainer.EndpointRelation) error {
|
||||
err := service.connection.CreateObjectWithId(BucketName, int(endpointRelation.EndpointID), endpointRelation)
|
||||
cache.Del(endpointRelation.EndpointID)
|
||||
|
||||
service.mu.Lock()
|
||||
service.endpointRelationsCache = nil
|
||||
service.mu.Unlock()
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
// UpdateEndpointRelation updates an Environment(Endpoint) relation object
|
||||
func (service *Service) UpdateEndpointRelation(endpointID portainer.EndpointID, endpointRelation *portainer.EndpointRelation) error {
|
||||
return service.connection.UpdateTx(func(tx portainer.Transaction) error {
|
||||
return service.Tx(tx).UpdateEndpointRelation(endpointID, endpointRelation)
|
||||
})
|
||||
}
|
||||
|
||||
func (service *Service) AddEndpointRelationsForEdgeStack(endpointIDs []portainer.EndpointID, edgeStack *portainer.EdgeStack) error {
|
||||
return service.connection.UpdateTx(func(tx portainer.Transaction) error {
|
||||
return service.Tx(tx).AddEndpointRelationsForEdgeStack(endpointIDs, edgeStack)
|
||||
})
|
||||
}
|
||||
|
||||
func (service *Service) RemoveEndpointRelationsForEdgeStack(endpointIDs []portainer.EndpointID, edgeStackID portainer.EdgeStackID) error {
|
||||
return service.connection.UpdateTx(func(tx portainer.Transaction) error {
|
||||
return service.Tx(tx).RemoveEndpointRelationsForEdgeStack(endpointIDs, edgeStackID)
|
||||
})
|
||||
}
|
||||
|
||||
// DeleteEndpointRelation deletes an Environment(Endpoint) relation object
|
||||
func (service *Service) DeleteEndpointRelation(endpointID portainer.EndpointID) error {
|
||||
return service.connection.UpdateTx(func(tx portainer.Transaction) error {
|
||||
return service.Tx(tx).DeleteEndpointRelation(endpointID)
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,144 @@
|
||||
package endpointrelation
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
portainer "github.com/portainer/portainer/api"
|
||||
"github.com/portainer/portainer/api/database/boltdb"
|
||||
"github.com/portainer/portainer/api/dataservices/edgestack"
|
||||
"github.com/portainer/portainer/api/internal/edge/cache"
|
||||
"github.com/portainer/portainer/api/logs"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestUpdateRelation(t *testing.T) {
|
||||
t.Parallel()
|
||||
const endpointID = 1
|
||||
const edgeStackID1 = 1
|
||||
const edgeStackID2 = 2
|
||||
|
||||
var conn portainer.Connection = &boltdb.DbConnection{Path: t.TempDir()}
|
||||
err := conn.Open()
|
||||
require.NoError(t, err)
|
||||
|
||||
defer logs.CloseAndLogErr(conn)
|
||||
|
||||
service, err := NewService(conn)
|
||||
require.NoError(t, err)
|
||||
|
||||
updateStackFnTxCalled := false
|
||||
|
||||
edgeStacks := make(map[portainer.EdgeStackID]portainer.EdgeStack)
|
||||
edgeStacks[edgeStackID1] = portainer.EdgeStack{ID: edgeStackID1}
|
||||
edgeStacks[edgeStackID2] = portainer.EdgeStack{ID: edgeStackID2}
|
||||
|
||||
service.RegisterUpdateStackFunction(func(tx portainer.Transaction, ID portainer.EdgeStackID, updateFunc func(edgeStack *portainer.EdgeStack)) error {
|
||||
updateStackFnTxCalled = true
|
||||
|
||||
s, ok := edgeStacks[ID]
|
||||
require.True(t, ok)
|
||||
|
||||
updateFunc(&s)
|
||||
edgeStacks[ID] = s
|
||||
|
||||
return nil
|
||||
})
|
||||
|
||||
// Nil relation
|
||||
|
||||
cache.Set(endpointID, []byte("value"))
|
||||
|
||||
err = service.UpdateEndpointRelation(endpointID, nil)
|
||||
_, cacheKeyExists := cache.Get(endpointID)
|
||||
require.NoError(t, err)
|
||||
require.False(t, updateStackFnTxCalled)
|
||||
require.False(t, cacheKeyExists)
|
||||
|
||||
// Add a relation to two edge stacks
|
||||
|
||||
cache.Set(endpointID, []byte("value"))
|
||||
|
||||
err = service.UpdateEndpointRelation(endpointID, &portainer.EndpointRelation{
|
||||
EndpointID: endpointID,
|
||||
EdgeStacks: map[portainer.EdgeStackID]bool{
|
||||
edgeStackID1: true,
|
||||
edgeStackID2: true,
|
||||
},
|
||||
})
|
||||
_, cacheKeyExists = cache.Get(endpointID)
|
||||
require.NoError(t, err)
|
||||
require.True(t, updateStackFnTxCalled)
|
||||
require.False(t, cacheKeyExists)
|
||||
require.Equal(t, 1, edgeStacks[edgeStackID1].NumDeployments)
|
||||
require.Equal(t, 1, edgeStacks[edgeStackID2].NumDeployments)
|
||||
|
||||
// Remove a relation to one edge stack
|
||||
|
||||
updateStackFnTxCalled = false
|
||||
cache.Set(endpointID, []byte("value"))
|
||||
|
||||
err = service.UpdateEndpointRelation(endpointID, &portainer.EndpointRelation{
|
||||
EndpointID: endpointID,
|
||||
EdgeStacks: map[portainer.EdgeStackID]bool{
|
||||
2: true,
|
||||
},
|
||||
})
|
||||
_, cacheKeyExists = cache.Get(endpointID)
|
||||
require.NoError(t, err)
|
||||
require.True(t, updateStackFnTxCalled)
|
||||
require.False(t, cacheKeyExists)
|
||||
require.Equal(t, 0, edgeStacks[edgeStackID1].NumDeployments)
|
||||
require.Equal(t, 1, edgeStacks[edgeStackID2].NumDeployments)
|
||||
|
||||
// Delete the relation
|
||||
|
||||
updateStackFnTxCalled = false
|
||||
cache.Set(endpointID, []byte("value"))
|
||||
|
||||
err = service.DeleteEndpointRelation(endpointID)
|
||||
|
||||
_, cacheKeyExists = cache.Get(endpointID)
|
||||
require.NoError(t, err)
|
||||
require.True(t, updateStackFnTxCalled)
|
||||
require.False(t, cacheKeyExists)
|
||||
require.Equal(t, 0, edgeStacks[edgeStackID1].NumDeployments)
|
||||
require.Equal(t, 0, edgeStacks[edgeStackID2].NumDeployments)
|
||||
}
|
||||
|
||||
func TestAddEndpointRelationsForEdgeStack(t *testing.T) {
|
||||
t.Parallel()
|
||||
var conn portainer.Connection = &boltdb.DbConnection{Path: t.TempDir()}
|
||||
err := conn.Open()
|
||||
require.NoError(t, err)
|
||||
|
||||
defer logs.CloseAndLogErr(conn)
|
||||
|
||||
service, err := NewService(conn)
|
||||
require.NoError(t, err)
|
||||
|
||||
edgeStackService, err := edgestack.NewService(conn, func(t portainer.Transaction, esi portainer.EdgeStackID) {})
|
||||
require.NoError(t, err)
|
||||
|
||||
service.RegisterUpdateStackFunction(edgeStackService.UpdateEdgeStackFuncTx)
|
||||
require.NoError(t, edgeStackService.Create(1, &portainer.EdgeStack{}))
|
||||
require.NoError(t, service.Create(&portainer.EndpointRelation{EndpointID: 1, EdgeStacks: map[portainer.EdgeStackID]bool{}}))
|
||||
require.NoError(t, service.AddEndpointRelationsForEdgeStack([]portainer.EndpointID{1}, &portainer.EdgeStack{ID: 1}))
|
||||
}
|
||||
|
||||
func TestEndpointRelations(t *testing.T) {
|
||||
t.Parallel()
|
||||
var conn portainer.Connection = &boltdb.DbConnection{Path: t.TempDir()}
|
||||
err := conn.Open()
|
||||
require.NoError(t, err)
|
||||
|
||||
defer logs.CloseAndLogErr(conn)
|
||||
|
||||
service, err := NewService(conn)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.NoError(t, service.Create(&portainer.EndpointRelation{EndpointID: 1}))
|
||||
rels, err := service.EndpointRelations()
|
||||
require.NoError(t, err)
|
||||
require.Len(t, rels, 1)
|
||||
}
|
||||
@@ -0,0 +1,238 @@
|
||||
package endpointrelation
|
||||
|
||||
import (
|
||||
portainer "github.com/portainer/portainer/api"
|
||||
"github.com/portainer/portainer/api/dataservices"
|
||||
"github.com/portainer/portainer/api/internal/edge/cache"
|
||||
|
||||
"github.com/rs/zerolog/log"
|
||||
)
|
||||
|
||||
type ServiceTx struct {
|
||||
service *Service
|
||||
tx portainer.Transaction
|
||||
}
|
||||
|
||||
var _ dataservices.EndpointRelationService = &ServiceTx{}
|
||||
|
||||
func (service ServiceTx) BucketName() string {
|
||||
return BucketName
|
||||
}
|
||||
|
||||
// EndpointRelations returns an array of all EndpointRelations
|
||||
func (service ServiceTx) EndpointRelations() ([]portainer.EndpointRelation, error) {
|
||||
var all = make([]portainer.EndpointRelation, 0)
|
||||
|
||||
return all, service.tx.GetAll(
|
||||
BucketName,
|
||||
&portainer.EndpointRelation{},
|
||||
dataservices.AppendFn(&all),
|
||||
)
|
||||
}
|
||||
|
||||
// EndpointRelation returns an Environment(Endpoint) relation object by EndpointID
|
||||
func (service ServiceTx) EndpointRelation(endpointID portainer.EndpointID) (*portainer.EndpointRelation, error) {
|
||||
var endpointRelation portainer.EndpointRelation
|
||||
identifier := service.service.connection.ConvertToKey(int(endpointID))
|
||||
|
||||
if err := service.tx.GetObject(BucketName, identifier, &endpointRelation); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &endpointRelation, nil
|
||||
}
|
||||
|
||||
// CreateEndpointRelation saves endpointRelation
|
||||
func (service ServiceTx) Create(endpointRelation *portainer.EndpointRelation) error {
|
||||
err := service.tx.CreateObjectWithId(BucketName, int(endpointRelation.EndpointID), endpointRelation)
|
||||
cache.Del(endpointRelation.EndpointID)
|
||||
|
||||
service.service.mu.Lock()
|
||||
service.service.endpointRelationsCache = nil
|
||||
service.service.mu.Unlock()
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
// UpdateEndpointRelation updates an Environment(Endpoint) relation object
|
||||
func (service ServiceTx) UpdateEndpointRelation(endpointID portainer.EndpointID, endpointRelation *portainer.EndpointRelation) error {
|
||||
previousRelationState, _ := service.EndpointRelation(endpointID)
|
||||
|
||||
identifier := service.service.connection.ConvertToKey(int(endpointID))
|
||||
err := service.tx.UpdateObject(BucketName, identifier, endpointRelation)
|
||||
cache.Del(endpointID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
updatedRelationState, _ := service.EndpointRelation(endpointID)
|
||||
|
||||
service.service.mu.Lock()
|
||||
service.service.endpointRelationsCache = nil
|
||||
service.service.mu.Unlock()
|
||||
|
||||
service.updateEdgeStacksAfterRelationChange(previousRelationState, updatedRelationState)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (service ServiceTx) AddEndpointRelationsForEdgeStack(endpointIDs []portainer.EndpointID, edgeStack *portainer.EdgeStack) error {
|
||||
for _, endpointID := range endpointIDs {
|
||||
rel, err := service.EndpointRelation(endpointID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
rel.EdgeStacks[edgeStack.ID] = true
|
||||
|
||||
identifier := service.service.connection.ConvertToKey(int(endpointID))
|
||||
err = service.tx.UpdateObject(BucketName, identifier, rel)
|
||||
cache.Del(endpointID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
service.service.mu.Lock()
|
||||
service.service.endpointRelationsCache = nil
|
||||
service.service.mu.Unlock()
|
||||
|
||||
if err := service.service.updateStackFnTx(service.tx, edgeStack.ID, func(es *portainer.EdgeStack) {
|
||||
es.NumDeployments += len(endpointIDs)
|
||||
|
||||
// sync changes in `edgeStack` in case it is re-persisted after `AddEndpointRelationsForEdgeStack` call
|
||||
// to avoid overriding with the previous values
|
||||
edgeStack.NumDeployments = es.NumDeployments
|
||||
}); err != nil {
|
||||
log.Error().Err(err).Msg("could not update the number of deployments")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (service ServiceTx) RemoveEndpointRelationsForEdgeStack(endpointIDs []portainer.EndpointID, edgeStackID portainer.EdgeStackID) error {
|
||||
for _, endpointID := range endpointIDs {
|
||||
rel, err := service.EndpointRelation(endpointID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
delete(rel.EdgeStacks, edgeStackID)
|
||||
|
||||
identifier := service.service.connection.ConvertToKey(int(endpointID))
|
||||
err = service.tx.UpdateObject(BucketName, identifier, rel)
|
||||
cache.Del(endpointID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
service.service.mu.Lock()
|
||||
service.service.endpointRelationsCache = nil
|
||||
service.service.mu.Unlock()
|
||||
|
||||
if err := service.service.updateStackFnTx(service.tx, edgeStackID, func(edgeStack *portainer.EdgeStack) {
|
||||
edgeStack.NumDeployments -= len(endpointIDs)
|
||||
}); err != nil {
|
||||
log.Error().Err(err).Msg("could not update the number of deployments")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// DeleteEndpointRelation deletes an Environment(Endpoint) relation object
|
||||
func (service ServiceTx) DeleteEndpointRelation(endpointID portainer.EndpointID) error {
|
||||
deletedRelation, _ := service.EndpointRelation(endpointID)
|
||||
|
||||
identifier := service.service.connection.ConvertToKey(int(endpointID))
|
||||
err := service.tx.DeleteObject(BucketName, identifier)
|
||||
cache.Del(endpointID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
service.service.mu.Lock()
|
||||
service.service.endpointRelationsCache = nil
|
||||
service.service.mu.Unlock()
|
||||
|
||||
service.updateEdgeStacksAfterRelationChange(deletedRelation, nil)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (service ServiceTx) InvalidateEdgeCacheForEdgeStack(edgeStackID portainer.EdgeStackID) {
|
||||
rels, err := service.cachedEndpointRelations()
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msg("cannot retrieve endpoint relations")
|
||||
return
|
||||
}
|
||||
|
||||
for _, rel := range rels {
|
||||
if _, ok := rel.EdgeStacks[edgeStackID]; ok {
|
||||
cache.Del(rel.EndpointID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (service ServiceTx) cachedEndpointRelations() ([]portainer.EndpointRelation, error) {
|
||||
service.service.mu.Lock()
|
||||
defer service.service.mu.Unlock()
|
||||
|
||||
if service.service.endpointRelationsCache == nil {
|
||||
var err error
|
||||
service.service.endpointRelationsCache, err = service.EndpointRelations()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
return service.service.endpointRelationsCache, nil
|
||||
}
|
||||
|
||||
func (service ServiceTx) updateEdgeStacksAfterRelationChange(previousRelationState *portainer.EndpointRelation, updatedRelationState *portainer.EndpointRelation) {
|
||||
if previousRelationState != nil {
|
||||
for stackId, enabled := range previousRelationState.EdgeStacks {
|
||||
// flag stack for update if stack is not in the updated relation state
|
||||
// = stack has been removed for this relation
|
||||
// or this relation has been deleted
|
||||
if enabled && (updatedRelationState == nil || !updatedRelationState.EdgeStacks[stackId]) {
|
||||
if err := service.service.updateStackFnTx(service.tx, stackId, func(edgeStack *portainer.EdgeStack) {
|
||||
// Sanity check
|
||||
if edgeStack.NumDeployments <= 0 {
|
||||
log.Error().
|
||||
Int("edgestack_id", int(edgeStack.ID)).
|
||||
Int("endpoint_id", int(previousRelationState.EndpointID)).
|
||||
Int("num_deployments", edgeStack.NumDeployments).
|
||||
Msg("cannot decrement the number of deployments for an edge stack with zero deployments")
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
edgeStack.NumDeployments--
|
||||
}); err != nil {
|
||||
log.Error().Err(err).Msg("could not update the number of deployments")
|
||||
}
|
||||
|
||||
cache.Del(previousRelationState.EndpointID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if updatedRelationState == nil {
|
||||
return
|
||||
}
|
||||
|
||||
for stackId, enabled := range updatedRelationState.EdgeStacks {
|
||||
// flag stack for update if stack is not in the previous relation state
|
||||
// = stack has been added for this relation
|
||||
if enabled && (previousRelationState == nil || !previousRelationState.EdgeStacks[stackId]) {
|
||||
if err := service.service.updateStackFnTx(service.tx, stackId, func(edgeStack *portainer.EdgeStack) {
|
||||
edgeStack.NumDeployments++
|
||||
}); err != nil {
|
||||
log.Error().Err(err).Msg("could not update the number of deployments")
|
||||
}
|
||||
|
||||
cache.Del(updatedRelationState.EndpointID)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user