Docker/portainer
Docker/portainer
1
0
Fork 0
mirror of https://github.com/portainer/portainer.git synced 2025-07-19 05:19:39 +02:00
Code Issues Releases Activity

feat(edgestackstatus): optimize the Edge Stack structures BE-11740 (#756)

Browse source
This commit is contained in:
andres-portainer 2025-06-05 19:46:10 -03:00 • committed by GitHub
parent eaf0deb2f6
commit 75f165d1ff
33 changed files with 452 additions and 391 deletions
Download patch file Download diff file Expand all files Collapse all files

3
api/http/handler/edgestacks/edgestack_create_file.go
View file

@ -101,8 +101,7 @@ func (payload *edgeStackFromFileUploadPayload) Validate(r *http.Request) error {
// @router /edge_stacks/create/file [post]
func (handler *Handler) createEdgeStackFromFileUpload(r *http.Request, tx dataservices.DataStoreTx, dryrun bool) (*portainer.EdgeStack, error) {
payload := &edgeStackFromFileUploadPayload{}
err := payload.Validate(r)
if err != nil {
if err := payload.Validate(r); err != nil {
return nil, err
}

12
api/http/handler/edgestacks/edgestack_create_git.go
View file

@ -103,8 +103,7 @@ func (payload *edgeStackFromGitRepositoryPayload) Validate(r *http.Request) erro
// @router /edge_stacks/create/repository [post]
func (handler *Handler) createEdgeStackFromGitRepository(r *http.Request, tx dataservices.DataStoreTx, dryrun bool, userID portainer.UserID) (*portainer.EdgeStack, error) {
var payload edgeStackFromGitRepositoryPayload
err := request.DecodeAndValidateJSONPayload(r, &payload)
if err != nil {
if err := request.DecodeAndValidateJSONPayload(r, &payload); err != nil {
return nil, err
}
@ -137,11 +136,9 @@ func (handler *Handler) createEdgeStackFromGitRepository(r *http.Request, tx dat
}
func (handler *Handler) storeManifestFromGitRepository(tx dataservices.DataStoreTx, stackFolder string, relatedEndpointIds []portainer.EndpointID, deploymentType portainer.EdgeStackDeploymentType, currentUserID portainer.UserID, repositoryConfig gittypes.RepoConfig) (composePath, manifestPath, projectPath string, err error) {
hasWrongType, err := hasWrongEnvironmentType(tx.Endpoint(), relatedEndpointIds, deploymentType)
if err != nil {
if hasWrongType, err := hasWrongEnvironmentType(tx.Endpoint(), relatedEndpointIds, deploymentType); err != nil {
return "", "", "", fmt.Errorf("unable to check for existence of non fitting environments: %w", err)
}
if hasWrongType {
} else if hasWrongType {
return "", "", "", errors.New("edge stack with config do not match the environment type")
}
@ -153,8 +150,7 @@ func (handler *Handler) storeManifestFromGitRepository(tx dataservices.DataStore
repositoryPassword = repositoryConfig.Authentication.Password
}
err = handler.GitService.CloneRepository(projectPath, repositoryConfig.URL, repositoryConfig.ReferenceName, repositoryUsername, repositoryPassword, repositoryConfig.TLSSkipVerify)
if err != nil {
if err := handler.GitService.CloneRepository(projectPath, repositoryConfig.URL, repositoryConfig.ReferenceName, repositoryUsername, repositoryPassword, repositoryConfig.TLSSkipVerify); err != nil {
return "", "", "", err
}

10
api/http/handler/edgestacks/edgestack_create_string.go
View file

@ -76,8 +76,7 @@ func (payload *edgeStackFromStringPayload) Validate(r *http.Request) error {
// @router /edge_stacks/create/string [post]
func (handler *Handler) createEdgeStackFromFileContent(r *http.Request, tx dataservices.DataStoreTx, dryrun bool) (*portainer.EdgeStack, error) {
var payload edgeStackFromStringPayload
err := request.DecodeAndValidateJSONPayload(r, &payload)
if err != nil {
if err := request.DecodeAndValidateJSONPayload(r, &payload); err != nil {
return nil, err
}
@ -96,11 +95,9 @@ func (handler *Handler) createEdgeStackFromFileContent(r *http.Request, tx datas
}
func (handler *Handler) storeFileContent(tx dataservices.DataStoreTx, stackFolder string, deploymentType portainer.EdgeStackDeploymentType, relatedEndpointIds []portainer.EndpointID, fileContent []byte) (composePath, manifestPath, projectPath string, err error) {
hasWrongType, err := hasWrongEnvironmentType(tx.Endpoint(), relatedEndpointIds, deploymentType)
if err != nil {
if hasWrongType, err := hasWrongEnvironmentType(tx.Endpoint(), relatedEndpointIds, deploymentType); err != nil {
return "", "", "", fmt.Errorf("unable to check for existence of non fitting environments: %w", err)
}
if hasWrongType {
} else if hasWrongType {
return "", "", "", errors.New("edge stack with config do not match the environment type")
}
@ -124,7 +121,6 @@ func (handler *Handler) storeFileContent(tx dataservices.DataStoreTx, stackFolde
}
return "", manifestPath, projectPath, nil
}
errMessage := fmt.Sprintf("invalid deployment type: %d", deploymentType)

31
api/http/handler/edgestacks/edgestack_create_test.go
View file

@ -8,6 +8,7 @@ import (
"testing"
portainer "github.com/portainer/portainer/api"
"github.com/stretchr/testify/require"
"github.com/segmentio/encoding/json"
)
@ -28,9 +29,7 @@ func TestCreateAndInspect(t *testing.T) {
}
err := handler.DataStore.EdgeGroup().Create(&edgeGroup)
if err != nil {
t.Fatal(err)
}
require.NoError(t, err)
endpointRelation := portainer.EndpointRelation{
EndpointID: endpoint.ID,
@ -38,9 +37,7 @@ func TestCreateAndInspect(t *testing.T) {
}
err = handler.DataStore.EndpointRelation().Create(&endpointRelation)
if err != nil {
t.Fatal(err)
}
require.NoError(t, err)
payload := edgeStackFromStringPayload{
Name: "test-stack",
@ -50,16 +47,14 @@ func TestCreateAndInspect(t *testing.T) {
}
jsonPayload, err := json.Marshal(payload)
if err != nil {
t.Fatal("JSON marshal error:", err)
}
require.NoError(t, err)
r := bytes.NewBuffer(jsonPayload)
// Create EdgeStack
req, err := http.NewRequest(http.MethodPost, "/edge_stacks/create/string", r)
if err != nil {
t.Fatal("request error:", err)
}
require.NoError(t, err)
req.Header.Add("x-api-key", rawAPIKey)
rec := httptest.NewRecorder()
handler.ServeHTTP(rec, req)
@ -70,15 +65,11 @@ func TestCreateAndInspect(t *testing.T) {
data := portainer.EdgeStack{}
err = json.NewDecoder(rec.Body).Decode(&data)
if err != nil {
t.Fatal("error decoding response:", err)
}
require.NoError(t, err)
// Inspect
req, err = http.NewRequest(http.MethodGet, fmt.Sprintf("/edge_stacks/%d", data.ID), nil)
if err != nil {
t.Fatal("request error:", err)
}
require.NoError(t, err)
req.Header.Add("x-api-key", rawAPIKey)
rec = httptest.NewRecorder()
@ -90,9 +81,7 @@ func TestCreateAndInspect(t *testing.T) {
data = portainer.EdgeStack{}
err = json.NewDecoder(rec.Body).Decode(&data)
if err != nil {
t.Fatal("error decoding response:", err)
}
require.NoError(t, err)
if payload.Name != data.Name {
t.Fatalf("expected EdgeStack Name %s, found %s", payload.Name, data.Name)

5
api/http/handler/edgestacks/edgestack_delete.go
View file

@ -30,10 +30,9 @@ func (handler *Handler) edgeStackDelete(w http.ResponseWriter, r *http.Request)
return httperror.BadRequest("Invalid edge stack identifier route variable", err)
}
err = handler.DataStore.UpdateTx(func(tx dataservices.DataStoreTx) error {
if err := handler.DataStore.UpdateTx(func(tx dataservices.DataStoreTx) error {
return handler.deleteEdgeStack(tx, portainer.EdgeStackID(edgeStackID))
})
if err != nil {
}); err != nil {
var httpErr *httperror.HandlerError
if errors.As(err, &httpErr) {
return httpErr

33
api/http/handler/edgestacks/edgestack_delete_test.go
View file

@ -8,9 +8,10 @@ import (
"testing"
portainer "github.com/portainer/portainer/api"
"github.com/stretchr/testify/assert"
"github.com/segmentio/encoding/json"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// Delete
@ -23,9 +24,7 @@ func TestDeleteAndInspect(t *testing.T) {
// Inspect
req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("/edge_stacks/%d", edgeStack.ID), nil)
if err != nil {
t.Fatal("request error:", err)
}
require.NoError(t, err)
req.Header.Add("x-api-key", rawAPIKey)
rec := httptest.NewRecorder()
@ -37,9 +36,7 @@ func TestDeleteAndInspect(t *testing.T) {
data := portainer.EdgeStack{}
err = json.NewDecoder(rec.Body).Decode(&data)
if err != nil {
t.Fatal("error decoding response:", err)
}
require.NoError(t, err)
if data.ID != edgeStack.ID {
t.Fatalf("expected EdgeStackID %d, found %d", int(edgeStack.ID), data.ID)
@ -47,9 +44,7 @@ func TestDeleteAndInspect(t *testing.T) {
// Delete
req, err = http.NewRequest(http.MethodDelete, fmt.Sprintf("/edge_stacks/%d", edgeStack.ID), nil)
if err != nil {
t.Fatal("request error:", err)
}
require.NoError(t, err)
req.Header.Add("x-api-key", rawAPIKey)
rec = httptest.NewRecorder()
@ -61,9 +56,7 @@ func TestDeleteAndInspect(t *testing.T) {
// Inspect
req, err = http.NewRequest(http.MethodGet, fmt.Sprintf("/edge_stacks/%d", edgeStack.ID), nil)
if err != nil {
t.Fatal("request error:", err)
}
require.NoError(t, err)
req.Header.Add("x-api-key", rawAPIKey)
rec = httptest.NewRecorder()
@ -117,15 +110,12 @@ func TestDeleteEdgeStack_RemoveProjectFolder(t *testing.T) {
}
var buf bytes.Buffer
if err := json.NewEncoder(&buf).Encode(payload); err != nil {
t.Fatal("error encoding payload:", err)
}
err := json.NewEncoder(&buf).Encode(payload)
require.NoError(t, err)
// Create
req, err := http.NewRequest(http.MethodPost, "/edge_stacks/create/string", &buf)
if err != nil {
t.Fatal("request error:", err)
}
require.NoError(t, err)
req.Header.Add("x-api-key", rawAPIKey)
rec := httptest.NewRecorder()
@ -138,9 +128,8 @@ func TestDeleteEdgeStack_RemoveProjectFolder(t *testing.T) {
assert.DirExists(t, handler.FileService.GetEdgeStackProjectPath("1"))
// Delete
if req, err = http.NewRequest(http.MethodDelete, "/edge_stacks/1", nil); err != nil {
t.Fatal("request error:", err)
}
req, err = http.NewRequest(http.MethodDelete, "/edge_stacks/1", nil)
require.NoError(t, err)
req.Header.Add("x-api-key", rawAPIKey)
rec = httptest.NewRecorder()

31
api/http/handler/edgestacks/edgestack_inspect.go
View file

@ -4,6 +4,7 @@ import (
"net/http"
portainer "github.com/portainer/portainer/api"
"github.com/portainer/portainer/api/dataservices"
httperror "github.com/portainer/portainer/pkg/libhttp/error"
"github.com/portainer/portainer/pkg/libhttp/request"
"github.com/portainer/portainer/pkg/libhttp/response"
@ -33,5 +34,35 @@ func (handler *Handler) edgeStackInspect(w http.ResponseWriter, r *http.Request)
return handlerDBErr(err, "Unable to find an edge stack with the specified identifier inside the database")
}
if err := fillEdgeStackStatus(handler.DataStore, edgeStack); err != nil {
return handlerDBErr(err, "Unable to retrieve edge stack status from the database")
}
return response.JSON(w, edgeStack)
}
func fillEdgeStackStatus(tx dataservices.DataStoreTx, edgeStack *portainer.EdgeStack) error {
status, err := tx.EdgeStackStatus().ReadAll(edgeStack.ID)
if err != nil {
return err
}
edgeStack.Status = make(map[portainer.EndpointID]portainer.EdgeStackStatus, len(status))
emptyStatus := make([]portainer.EdgeStackDeploymentStatus, 0)
for _, s := range status {
if s.Status == nil {
s.Status = emptyStatus
}
edgeStack.Status[s.EndpointID] = portainer.EdgeStackStatus{
Status: s.Status,
EndpointID: s.EndpointID,
DeploymentInfo: s.DeploymentInfo,
ReadyRePullImage: s.ReadyRePullImage,
}
}
return nil
}

6
api/http/handler/edgestacks/edgestack_list.go
View file

@ -25,5 +25,11 @@ func (handler *Handler) edgeStackList(w http.ResponseWriter, r *http.Request) *h
return httperror.InternalServerError("Unable to retrieve edge stacks from the database", err)
}
for i := range edgeStacks {
if err := fillEdgeStackStatus(handler.DataStore, &edgeStacks[i]); err != nil {
return handlerDBErr(err, "Unable to retrieve edge stack status from the database")
}
}
return response.JSON(w, edgeStacks)
}

59
api/http/handler/edgestacks/edgestack_status_update.go
View file

@ -9,11 +9,10 @@ import (
"time"
portainer "github.com/portainer/portainer/api"
"github.com/portainer/portainer/api/dataservices"
httperror "github.com/portainer/portainer/pkg/libhttp/error"
"github.com/portainer/portainer/pkg/libhttp/request"
"github.com/portainer/portainer/pkg/libhttp/response"
"github.com/rs/zerolog/log"
)
type updateStatusPayload struct {
@ -78,12 +77,25 @@ func (handler *Handler) edgeStackStatusUpdate(w http.ResponseWriter, r *http.Req
return httperror.Forbidden("Permission denied to access environment", fmt.Errorf("unauthorized edge endpoint operation: %w. Environment name: %s", err, endpoint.Name))
}
updateFn := func(stack *portainer.EdgeStack) (*portainer.EdgeStack, error) {
return handler.updateEdgeStackStatus(stack, stack.ID, payload)
}
var stack *portainer.EdgeStack
stack, err := handler.stackCoordinator.UpdateStatus(r, portainer.EdgeStackID(stackID), updateFn)
if err != nil {
if err := handler.DataStore.UpdateTx(func(tx dataservices.DataStoreTx) error {
var err error
stack, err = tx.EdgeStack().EdgeStack(portainer.EdgeStackID(stackID))
if err != nil {
if dataservices.IsErrObjectNotFound(err) {
return nil
}
return httperror.InternalServerError("Unable to retrieve Edge stack from the database", err)
}
if err := handler.updateEdgeStackStatus(tx, stack, stack.ID, payload); err != nil {
return httperror.InternalServerError("Unable to update Edge stack status", err)
}
return nil
}); err != nil {
var httpErr *httperror.HandlerError
if errors.As(err, &httpErr) {
return httpErr
@ -96,43 +108,34 @@ func (handler *Handler) edgeStackStatusUpdate(w http.ResponseWriter, r *http.Req
return nil
}
if err := fillEdgeStackStatus(handler.DataStore, stack); err != nil {
return handlerDBErr(err, "Unable to retrieve edge stack status from the database")
}
return response.JSON(w, stack)
}
func (handler *Handler) updateEdgeStackStatus(stack *portainer.EdgeStack, stackID portainer.EdgeStackID, payload updateStatusPayload) (*portainer.EdgeStack, error) {
func (handler *Handler) updateEdgeStackStatus(tx dataservices.DataStoreTx, stack *portainer.EdgeStack, stackID portainer.EdgeStackID, payload updateStatusPayload) error {
if payload.Version > 0 && payload.Version < stack.Version {
return stack, nil
return nil
}
status := *payload.Status
log.Debug().
Int("stackID", int(stackID)).
Int("status", int(status)).
Msg("Updating stack status")
deploymentStatus := portainer.EdgeStackDeploymentStatus{
Type: status,
Error: payload.Error,
Time: payload.Time,
}
updateEnvStatus(payload.EndpointID, stack, deploymentStatus)
return stack, nil
}
func updateEnvStatus(environmentId portainer.EndpointID, stack *portainer.EdgeStack, deploymentStatus portainer.EdgeStackDeploymentStatus) {
if deploymentStatus.Type == portainer.EdgeStackStatusRemoved {
delete(stack.Status, environmentId)
return
return tx.EdgeStackStatus().Delete(stackID, payload.EndpointID)
}
environmentStatus, ok := stack.Status[environmentId]
if !ok {
environmentStatus = portainer.EdgeStackStatus{
EndpointID: environmentId,
environmentStatus, err := tx.EdgeStackStatus().Read(stackID, payload.EndpointID)
if err != nil {
environmentStatus = &portainer.EdgeStackStatusForEnv{
EndpointID: payload.EndpointID,
Status: []portainer.EdgeStackDeploymentStatus{},
}
}
@ -143,5 +146,5 @@ func updateEnvStatus(environmentId portainer.EndpointID, stack *portainer.EdgeSt
environmentStatus.Status = append(environmentStatus.Status, deploymentStatus)
}
stack.Status[environmentId] = environmentStatus
return tx.EdgeStackStatus().Update(stackID, payload.EndpointID, environmentStatus)
}

155
api/http/handler/edgestacks/edgestack_status_update_coordinator.go
View file

@ -1,155 +0,0 @@
package edgestacks
import (
"errors"
"fmt"
"net/http"
portainer "github.com/portainer/portainer/api"
"github.com/portainer/portainer/api/dataservices"
"github.com/rs/zerolog/log"
)
type statusRequest struct {
respCh chan statusResponse
stackID portainer.EdgeStackID
updateFn statusUpdateFn
}
type statusResponse struct {
Stack *portainer.EdgeStack
Error error
}
type statusUpdateFn func(*portainer.EdgeStack) (*portainer.EdgeStack, error)
type EdgeStackStatusUpdateCoordinator struct {
updateCh chan statusRequest
dataStore dataservices.DataStore
}
var errAnotherStackUpdateInProgress = errors.New("another stack update is in progress")
func NewEdgeStackStatusUpdateCoordinator(dataStore dataservices.DataStore) *EdgeStackStatusUpdateCoordinator {
return &EdgeStackStatusUpdateCoordinator{
updateCh: make(chan statusRequest),
dataStore: dataStore,
}
}
func (c *EdgeStackStatusUpdateCoordinator) Start() {
for {
c.loop()
}
}
func (c *EdgeStackStatusUpdateCoordinator) loop() {
u := <-c.updateCh
respChs := []chan statusResponse{u.respCh}
var stack *portainer.EdgeStack
err := c.dataStore.UpdateTx(func(tx dataservices.DataStoreTx) error {
// 1. Load the edge stack
var err error
stack, err = loadEdgeStack(tx, u.stackID)
if err != nil {
return err
}
// Return early when the agent tries to update the status on a deleted stack
if stack == nil {
return nil
}
// 2. Mutate the edge stack opportunistically until there are no more pending updates
for {
stack, err = u.updateFn(stack)
if err != nil {
return err
}
if m, ok := c.getNextUpdate(stack.ID); ok {
u = m
} else {
break
}
respChs = append(respChs, u.respCh)
}
// 3. Save the changes back to the database
if err := tx.EdgeStack().UpdateEdgeStack(stack.ID, stack); err != nil {
return handlerDBErr(fmt.Errorf("unable to update Edge stack: %w.", err), "Unable to persist the stack changes inside the database")
}
return nil
})
// 4. Send back the responses
for _, ch := range respChs {
ch <- statusResponse{Stack: stack, Error: err}
}
}
func loadEdgeStack(tx dataservices.DataStoreTx, stackID portainer.EdgeStackID) (*portainer.EdgeStack, error) {
stack, err := tx.EdgeStack().EdgeStack(stackID)
if err != nil {
if dataservices.IsErrObjectNotFound(err) {
// Skip the error when the agent tries to update the status on a deleted stack
log.Debug().
Err(err).
Int("stackID", int(stackID)).
Msg("Unable to find a stack inside the database, skipping error")
return nil, nil
}
return nil, fmt.Errorf("unable to retrieve Edge stack from the database: %w.", err)
}
return stack, nil
}
func (c *EdgeStackStatusUpdateCoordinator) getNextUpdate(stackID portainer.EdgeStackID) (statusRequest, bool) {
for {
select {
case u := <-c.updateCh:
// Discard the update and let the agent retry
if u.stackID != stackID {
u.respCh <- statusResponse{Error: errAnotherStackUpdateInProgress}
continue
}
return u, true
default:
return statusRequest{}, false
}
}
}
func (c *EdgeStackStatusUpdateCoordinator) UpdateStatus(r *http.Request, stackID portainer.EdgeStackID, updateFn statusUpdateFn) (*portainer.EdgeStack, error) {
respCh := make(chan statusResponse)
defer close(respCh)
msg := statusRequest{
respCh: respCh,
stackID: stackID,
updateFn: updateFn,
}
select {
case c.updateCh <- msg:
r := <-respCh
return r.Stack, r.Error
case <-r.Context().Done():
return nil, r.Context().Err()
}
}

31
api/http/handler/edgestacks/edgestack_status_update_test.go
View file

@ -10,6 +10,7 @@ import (
portainer "github.com/portainer/portainer/api"
"github.com/segmentio/encoding/json"
"github.com/stretchr/testify/require"
)
// Update Status
@ -28,15 +29,11 @@ func TestUpdateStatusAndInspect(t *testing.T) {
}
jsonPayload, err := json.Marshal(payload)
if err != nil {
t.Fatal("request error:", err)
}
require.NoError(t, err)
r := bytes.NewBuffer(jsonPayload)
req, err := http.NewRequest(http.MethodPut, fmt.Sprintf("/edge_stacks/%d/status", edgeStack.ID), r)
if err != nil {
t.Fatal("request error:", err)
}
require.NoError(t, err)
req.Header.Set(portainer.PortainerAgentEdgeIDHeader, endpoint.EdgeID)
rec := httptest.NewRecorder()
@ -48,9 +45,7 @@ func TestUpdateStatusAndInspect(t *testing.T) {
// Get updated edge stack
req, err = http.NewRequest(http.MethodGet, fmt.Sprintf("/edge_stacks/%d", edgeStack.ID), nil)
if err != nil {
t.Fatal("request error:", err)
}
require.NoError(t, err)
req.Header.Add("x-api-key", rawAPIKey)
rec = httptest.NewRecorder()
@ -62,14 +57,10 @@ func TestUpdateStatusAndInspect(t *testing.T) {
updatedStack := portainer.EdgeStack{}
err = json.NewDecoder(rec.Body).Decode(&updatedStack)
if err != nil {
t.Fatal("error decoding response:", err)
}
require.NoError(t, err)
endpointStatus, ok := updatedStack.Status[payload.EndpointID]
if !ok {
t.Fatal("Missing status")
}
require.True(t, ok)
lastStatus := endpointStatus.Status[len(endpointStatus.Status)-1]
@ -84,8 +75,8 @@ func TestUpdateStatusAndInspect(t *testing.T) {
if endpointStatus.EndpointID != payload.EndpointID {
t.Fatalf("expected EndpointID %d, found %d", payload.EndpointID, endpointStatus.EndpointID)
}
}
func TestUpdateStatusWithInvalidPayload(t *testing.T) {
handler, _ := setupHandler(t)
@ -136,15 +127,11 @@ func TestUpdateStatusWithInvalidPayload(t *testing.T) {
for _, tc := range cases {
t.Run(tc.Name, func(t *testing.T) {
jsonPayload, err := json.Marshal(tc.Payload)
if err != nil {
t.Fatal("request error:", err)
}
require.NoError(t, err)
r := bytes.NewBuffer(jsonPayload)
req, err := http.NewRequest(http.MethodPut, fmt.Sprintf("/edge_stacks/%d/status", edgeStack.ID), r)
if err != nil {
t.Fatal("request error:", err)
}
require.NoError(t, err)
req.Header.Set(portainer.PortainerAgentEdgeIDHeader, endpoint.EdgeID)
rec := httptest.NewRecorder()

42
api/http/handler/edgestacks/edgestack_test.go
View file

@ -17,6 +17,7 @@ import (
"github.com/portainer/portainer/api/jwt"
"github.com/pkg/errors"
"github.com/stretchr/testify/require"
)
// Helpers
@ -51,27 +52,21 @@ func setupHandler(t *testing.T) (*Handler, string) {
t.Fatal(err)
}
coord := NewEdgeStackStatusUpdateCoordinator(store)
go coord.Start()
handler := NewHandler(
security.NewRequestBouncer(store, jwtService, apiKeyService),
store,
edgestacks.NewService(store),
coord,
)
handler.FileService = fs
settings, err := handler.DataStore.Settings().Settings()
if err != nil {
t.Fatal(err)
}
require.NoError(t, err)
settings.EnableEdgeComputeFeatures = true
if err := handler.DataStore.Settings().UpdateSettings(settings); err != nil {
t.Fatal(err)
}
err = handler.DataStore.Settings().UpdateSettings(settings)
require.NoError(t, err)
handler.GitService = testhelpers.NewGitService(errors.New("Clone error"), "git-service-id")
@ -90,9 +85,8 @@ func createEndpointWithId(t *testing.T, store dataservices.DataStore, endpointID
LastCheckInDate: time.Now().Unix(),
}
if err := store.Endpoint().Create(&endpoint); err != nil {
t.Fatal(err)
}
err := store.Endpoint().Create(&endpoint)
require.NoError(t, err)
return endpoint
}
@ -113,15 +107,13 @@ func createEdgeStack(t *testing.T, store dataservices.DataStore, endpointID port
PartialMatch: false,
}
if err := store.EdgeGroup().Create(&edgeGroup); err != nil {
t.Fatal(err)
}
err := store.EdgeGroup().Create(&edgeGroup)
require.NoError(t, err)
edgeStackID := portainer.EdgeStackID(14)
edgeStack := portainer.EdgeStack{
ID: edgeStackID,
Name: "test-edge-stack-" + strconv.Itoa(int(edgeStackID)),
Status: map[portainer.EndpointID]portainer.EdgeStackStatus{},
CreationDate: time.Now().Unix(),
EdgeGroups: []portainer.EdgeGroupID{edgeGroup.ID},
ProjectPath: "/project/path",
@ -138,13 +130,11 @@ func createEdgeStack(t *testing.T, store dataservices.DataStore, endpointID port
},
}
if err := store.EdgeStack().Create(edgeStack.ID, &edgeStack); err != nil {
t.Fatal(err)
}
err = store.EdgeStack().Create(edgeStack.ID, &edgeStack)
require.NoError(t, err)
if err := store.EndpointRelation().Create(&endpointRelation); err != nil {
t.Fatal(err)
}
err = store.EndpointRelation().Create(&endpointRelation)
require.NoError(t, err)
return edgeStack
}
@ -155,8 +145,8 @@ func createEdgeGroup(t *testing.T, store dataservices.DataStore) portainer.EdgeG
Name: "EdgeGroup 1",
}
if err := store.EdgeGroup().Create(&edgeGroup); err != nil {
t.Fatal(err)
}
err := store.EdgeGroup().Create(&edgeGroup)
require.NoError(t, err)
return edgeGroup
}

6
api/http/handler/edgestacks/edgestack_update.go
View file

@ -74,6 +74,10 @@ func (handler *Handler) edgeStackUpdate(w http.ResponseWriter, r *http.Request)
return httperror.InternalServerError("Unexpected error", err)
}
if err := fillEdgeStackStatus(handler.DataStore, stack); err != nil {
return handlerDBErr(err, "Unable to retrieve edge stack status from the database")
}
return response.JSON(w, stack)
}
@ -120,7 +124,7 @@ func (handler *Handler) updateEdgeStack(tx dataservices.DataStoreTx, stackID por
stack.EdgeGroups = groupsIds
if payload.UpdateVersion {
if err := handler.updateStackVersion(stack, payload.DeploymentType, []byte(payload.StackFileContent), "", relatedEndpointIds); err != nil {
if err := handler.updateStackVersion(tx, stack, payload.DeploymentType, []byte(payload.StackFileContent), "", relatedEndpointIds); err != nil {
return nil, httperror.InternalServerError("Unable to update stack version", err)
}
}

40
api/http/handler/edgestacks/edgestack_update_test.go
View file

@ -25,9 +25,8 @@ func TestUpdateAndInspect(t *testing.T) {
endpointID := portainer.EndpointID(6)
newEndpoint := createEndpointWithId(t, handler.DataStore, endpointID)
if err := handler.DataStore.Endpoint().Create(&newEndpoint); err != nil {
t.Fatal(err)
}
err := handler.DataStore.Endpoint().Create(&newEndpoint)
require.NoError(t, err)
endpointRelation := portainer.EndpointRelation{
EndpointID: endpointID,
@ -36,9 +35,8 @@ func TestUpdateAndInspect(t *testing.T) {
},
}
if err := handler.DataStore.EndpointRelation().Create(&endpointRelation); err != nil {
t.Fatal(err)
}
err = handler.DataStore.EndpointRelation().Create(&endpointRelation)
require.NoError(t, err)
newEdgeGroup := portainer.EdgeGroup{
ID: 2,
@ -49,9 +47,8 @@ func TestUpdateAndInspect(t *testing.T) {
PartialMatch: false,
}
if err := handler.DataStore.EdgeGroup().Create(&newEdgeGroup); err != nil {
t.Fatal(err)
}
err = handler.DataStore.EdgeGroup().Create(&newEdgeGroup)
require.NoError(t, err)
payload := updateEdgeStackPayload{
StackFileContent: "update-test",
@ -61,15 +58,11 @@ func TestUpdateAndInspect(t *testing.T) {
}
jsonPayload, err := json.Marshal(payload)
if err != nil {
t.Fatal("request error:", err)
}
require.NoError(t, err)
r := bytes.NewBuffer(jsonPayload)
req, err := http.NewRequest(http.MethodPut, fmt.Sprintf("/edge_stacks/%d", edgeStack.ID), r)
if err != nil {
t.Fatal("request error:", err)
}
require.NoError(t, err)
req.Header.Add("x-api-key", rawAPIKey)
rec := httptest.NewRecorder()
@ -81,9 +74,7 @@ func TestUpdateAndInspect(t *testing.T) {
// Get updated edge stack
req, err = http.NewRequest(http.MethodGet, fmt.Sprintf("/edge_stacks/%d", edgeStack.ID), nil)
if err != nil {
t.Fatal("request error:", err)
}
require.NoError(t, err)
req.Header.Add("x-api-key", rawAPIKey)
rec = httptest.NewRecorder()
@ -94,9 +85,8 @@ func TestUpdateAndInspect(t *testing.T) {
}
updatedStack := portainer.EdgeStack{}
if err := json.NewDecoder(rec.Body).Decode(&updatedStack); err != nil {
t.Fatal("error decoding response:", err)
}
err = json.NewDecoder(rec.Body).Decode(&updatedStack)
require.NoError(t, err)
if payload.UpdateVersion && updatedStack.Version != edgeStack.Version+1 {
t.Fatalf("expected EdgeStack version %d, found %d", edgeStack.Version+1, updatedStack.Version+1)
@ -226,15 +216,11 @@ func TestUpdateWithInvalidPayload(t *testing.T) {
for _, tc := range cases {
t.Run(tc.Name, func(t *testing.T) {
jsonPayload, err := json.Marshal(tc.Payload)
if err != nil {
t.Fatal("request error:", err)
}
require.NoError(t, err)
r := bytes.NewBuffer(jsonPayload)
req, err := http.NewRequest(http.MethodPut, fmt.Sprintf("/edge_stacks/%d", edgeStack.ID), r)
if err != nil {
t.Fatal("request error:", err)
}
require.NoError(t, err)
req.Header.Add("x-api-key", rawAPIKey)
rec := httptest.NewRecorder()

4
api/http/handler/edgestacks/handler.go
View file

@ -22,17 +22,15 @@ type Handler struct {
GitService portainer.GitService
edgeStacksService *edgestackservice.Service
KubernetesDeployer portainer.KubernetesDeployer
stackCoordinator *EdgeStackStatusUpdateCoordinator
}
// NewHandler creates a handler to manage environment(endpoint) group operations.
func NewHandler(bouncer security.BouncerService, dataStore dataservices.DataStore, edgeStacksService *edgestackservice.Service, stackCoordinator *EdgeStackStatusUpdateCoordinator) *Handler {
func NewHandler(bouncer security.BouncerService, dataStore dataservices.DataStore, edgeStacksService *edgestackservice.Service) *Handler {
h := &Handler{
Router: mux.NewRouter(),
requestBouncer: bouncer,
DataStore: dataStore,
edgeStacksService: edgeStacksService,
stackCoordinator: stackCoordinator,
}
h.Handle("/edge_stacks/create/{method}",

11
api/http/handler/edgestacks/utils_update_stack_version.go
View file

@ -5,15 +5,18 @@ import (
"strconv"
portainer "github.com/portainer/portainer/api"
"github.com/portainer/portainer/api/dataservices"
"github.com/portainer/portainer/api/filesystem"
edgestackutils "github.com/portainer/portainer/api/internal/edge/edgestacks"
"github.com/rs/zerolog/log"
)
func (handler *Handler) updateStackVersion(stack *portainer.EdgeStack, deploymentType portainer.EdgeStackDeploymentType, config []byte, oldGitHash string, relatedEnvironmentsIDs []portainer.EndpointID) error {
stack.Version = stack.Version + 1
stack.Status = edgestackutils.NewStatus(stack.Status, relatedEnvironmentsIDs)
func (handler *Handler) updateStackVersion(tx dataservices.DataStoreTx, stack *portainer.EdgeStack, deploymentType portainer.EdgeStackDeploymentType, config []byte, oldGitHash string, relatedEnvironmentsIDs []portainer.EndpointID) error {
stack.Version++
if err := tx.EdgeStackStatus().Clear(stack.ID, relatedEnvironmentsIDs); err != nil {
return err
}
return handler.storeStackFile(stack, deploymentType, config)
}

7
api/http/handler/endpointedge/endpointedge_status_inspect_test.go
View file

@ -287,11 +287,8 @@ func TestEdgeStackStatus(t *testing.T) {
edgeStackID := portainer.EdgeStackID(17)
edgeStack := portainer.EdgeStack{
ID: edgeStackID,
Name: "test-edge-stack-17",
Status: map[portainer.EndpointID]portainer.EdgeStackStatus{
endpointID: {},
},
ID: edgeStackID,
Name: "test-edge-stack-17",
CreationDate: time.Now().Unix(),
EdgeGroups: []portainer.EdgeGroupID{1, 2},
ProjectPath: "/project/path",

11
api/http/handler/endpoints/endpoint_delete.go
View file

@ -214,14 +214,9 @@ func (handler *Handler) deleteEndpoint(tx dataservices.DataStoreTx, endpointID p
log.Warn().Err(err).Msg("Unable to retrieve edge stacks from the database")
}
for idx := range edgeStacks {
edgeStack := &edgeStacks[idx]
if _, ok := edgeStack.Status[endpoint.ID]; ok {
delete(edgeStack.Status, endpoint.ID)
if err := tx.EdgeStack().UpdateEdgeStack(edgeStack.ID, edgeStack); err != nil {
log.Warn().Err(err).Msg("Unable to update edge stack")
}
for _, edgeStack := range edgeStacks {
if err := tx.EdgeStackStatus().Delete(edgeStack.ID, endpoint.ID); err != nil {
log.Warn().Err(err).Msg("Unable to delete edge stack status")
}
}

17
api/http/handler/endpoints/filter.go
View file

@ -247,19 +247,17 @@ func (handler *Handler) filterEndpointsByQuery(
return filteredEndpoints, totalAvailableEndpoints, nil
}
func endpointStatusInStackMatchesFilter(edgeStackStatus map[portainer.EndpointID]portainer.EdgeStackStatus, envId portainer.EndpointID, statusFilter portainer.EdgeStackStatusType) bool {
status, ok := edgeStackStatus[envId]
func endpointStatusInStackMatchesFilter(stackStatus *portainer.EdgeStackStatusForEnv, envId portainer.EndpointID, statusFilter portainer.EdgeStackStatusType) bool {
// consider that if the env has no status in the stack it is in Pending state
if statusFilter == portainer.EdgeStackStatusPending {
return !ok || len(status.Status) == 0
return stackStatus == nil || len(stackStatus.Status) == 0
}
if !ok {
if stackStatus == nil {
return false
}
return slices.ContainsFunc(status.Status, func(s portainer.EdgeStackDeploymentStatus) bool {
return slices.ContainsFunc(stackStatus.Status, func(s portainer.EdgeStackDeploymentStatus) bool {
return s.Type == statusFilter
})
}
@ -291,7 +289,12 @@ func filterEndpointsByEdgeStack(endpoints []portainer.Endpoint, edgeStackId port
if statusFilter != nil {
n := 0
for _, envId := range envIds {
if endpointStatusInStackMatchesFilter(stack.Status, envId, *statusFilter) {
edgeStackStatus, err := datastore.EdgeStackStatus().Read(edgeStackId, envId)
if err != nil {
return nil, errors.WithMessagef(err, "Unable to retrieve edge stack status for environment %d", envId)
}
if endpointStatusInStackMatchesFilter(edgeStackStatus, envId, *statusFilter) {
envIds[n] = envId
n++
}

5
api/http/server.go
View file

@ -161,10 +161,7 @@ func (server *Server) Start() error {
edgeJobsHandler.FileService = server.FileService
edgeJobsHandler.ReverseTunnelService = server.ReverseTunnelService
edgeStackCoordinator := edgestacks.NewEdgeStackStatusUpdateCoordinator(server.DataStore)
go edgeStackCoordinator.Start()
var edgeStacksHandler = edgestacks.NewHandler(requestBouncer, server.DataStore, server.EdgeStacksService, edgeStackCoordinator)
var edgeStacksHandler = edgestacks.NewHandler(requestBouncer, server.DataStore, server.EdgeStacksService)
edgeStacksHandler.FileService = server.FileService
edgeStacksHandler.GitService = server.GitService
edgeStacksHandler.KubernetesDeployer = server.KubernetesDeployer