EE-2404: EE-1120: Chore/refactor edge

cmd/agent/main.go

@@ -13,12 +13,10 @@ import (
 	"github.com/portainer/agent"
 	"github.com/portainer/agent/crypto"
 	"github.com/portainer/agent/docker"
+	"github.com/portainer/agent/edge"
 	"github.com/portainer/agent/exec"
-	"github.com/portainer/agent/filesystem"
 	"github.com/portainer/agent/ghw"
 	"github.com/portainer/agent/http"
-	"github.com/portainer/agent/http/client"
-	"github.com/portainer/agent/internal/edge"
 	"github.com/portainer/agent/kubernetes"
 	"github.com/portainer/agent/logutils"
 	"github.com/portainer/agent/net"
@@ -179,17 +177,18 @@ func main() {
 	// !Security
 
 	// Edge
-	edgeManagerParameters := &edge.ManagerParameters{
-		Options:           options,
-		AdvertiseAddr:     advertiseAddr,
-		ClusterService:    clusterService,
-		DockerInfoService: dockerInfoService,
-		ContainerPlatform: containerPlatform,
-	}
-	edgeManager := edge.NewManager(edgeManagerParameters)
-
+	var edgeManager *edge.Manager
 	if options.EdgeMode {
-		edgeKey, err := retrieveEdgeKey(options.EdgeKey, clusterService)
+		edgeManagerParameters := &edge.ManagerParameters{
+			Options:           options,
+			AdvertiseAddr:     advertiseAddr,
+			ClusterService:    clusterService,
+			DockerInfoService: dockerInfoService,
+			ContainerPlatform: containerPlatform,
+		}
+		edgeManager = edge.NewManager(edgeManagerParameters)
+
+		edgeKey, err := edge.RetrieveEdgeKey(options.EdgeKey, clusterService)
 		if err != nil {
 			log.Printf("[ERROR] [main,edge] [message: Unable to retrieve Edge key] [error: %s]", err)
 		}
@@ -232,11 +231,11 @@ func main() {
 		ContainerPlatform:    containerPlatform,
 	}
 
-	if edgeManager.IsEdgeModeEnabled() {
+	if options.EdgeMode {
 		config.Addr = advertiseAddr
 	}
 
-	err = startAPIServer(config)
+	err = startAPIServer(config, options.EdgeMode)
 	if err != nil && !errors.Is(err, gohttp.ErrServerClosed) {
 		log.Fatalf("[ERROR] [main,http] [message: Unable to start Agent API server] [error: %s]", err)
 	}
@@ -250,14 +249,14 @@ func main() {
 	fmt.Printf("[DEBUG] [main] [message: shutting down] [signal: %s]", s)
 }
 
-func startAPIServer(config *http.APIServerConfig) error {
+func startAPIServer(config *http.APIServerConfig, edgeMode bool) error {
 	server := http.NewAPIServer(config)
 
-	if config.EdgeManager.IsEdgeModeEnabled() {
-		return server.StartUnsecured()
+	if edgeMode {
+		return server.StartUnsecured(edgeMode)
 	}
 
-	return server.StartSecured()
+	return server.StartSecured(edgeMode)
 }
 
 func parseOptions() (*agent.Options, error) {
@@ -289,71 +288,3 @@ func serveEdgeUI(edgeManager *edge.Manager, serverAddr, serverPort string) {
 		}
 	}()
 }
-
-func retrieveEdgeKey(edgeKey string, clusterService agent.ClusterService) (string, error) {
-
-	if edgeKey != "" {
-		log.Println("[INFO] [main,edge] [message: Edge key loaded from options]")
-		return edgeKey, nil
-	}
-
-	var keyRetrievalError error
-
-	edgeKey, keyRetrievalError = retrieveEdgeKeyFromFilesystem()
-	if keyRetrievalError != nil {
-		return "", keyRetrievalError
-	}
-
-	if edgeKey == "" && clusterService != nil {
-		edgeKey, keyRetrievalError = retrieveEdgeKeyFromCluster(clusterService)
-		if keyRetrievalError != nil {
-			return "", keyRetrievalError
-		}
-	}
-
-	return edgeKey, nil
-}
-
-func retrieveEdgeKeyFromFilesystem() (string, error) {
-	var edgeKey string
-
-	edgeKeyFilePath := fmt.Sprintf("%s/%s", agent.DataDirectory, agent.EdgeKeyFile)
-
-	keyFileExists, err := filesystem.FileExists(edgeKeyFilePath)
-	if err != nil {
-		return "", err
-	}
-
-	if keyFileExists {
-		filesystemKey, err := filesystem.ReadFromFile(edgeKeyFilePath)
-		if err != nil {
-			return "", err
-		}
-
-		log.Println("[INFO] [main,edge] [message: Edge key loaded from the filesystem]")
-		edgeKey = string(filesystemKey)
-	}
-
-	return edgeKey, nil
-}
-
-func retrieveEdgeKeyFromCluster(clusterService agent.ClusterService) (string, error) {
-	var edgeKey string
-
-	member := clusterService.GetMemberWithEdgeKeySet()
-	if member != nil {
-		httpCli := client.NewAPIClient()
-
-		memberAddr := fmt.Sprintf("%s:%s", member.IPAddress, member.Port)
-		memberKey, err := httpCli.GetEdgeKey(memberAddr)
-		if err != nil {
-			log.Printf("[ERROR] [main,edge,http,cluster] [message: Unable to retrieve Edge key from cluster member] [error: %s]", err)
-			return "", err
-		}
-
-		log.Println("[INFO] [main,edge] [message: Edge key loaded from cluster]")
-		edgeKey = memberKey
-	}
-
-	return edgeKey, nil
-}

internal/edge/edge.go → edge/edge.go

@@ -7,6 +7,8 @@ import (
 	"time"
 
 	"github.com/portainer/agent"
+	"github.com/portainer/agent/edge/scheduler"
+	"github.com/portainer/agent/edge/stack"
 )
 
 type (
@@ -18,10 +20,9 @@ type (
 		clusterService    agent.ClusterService
 		dockerInfoService agent.DockerInfoService
 		key               *edgeKey
-		logsManager       *logsManager
+		logsManager       *scheduler.LogsManager
 		pollService       *PollService
-		pollServiceConfig *pollServiceConfig
-		stackManager      *StackManager
+		stackManager      *stack.StackManager
 	}
 
 	// ManagerParameters represents an object used to create a Manager
@@ -48,7 +49,7 @@ func NewManager(parameters *ManagerParameters) *Manager {
 // Start starts the manager
 func (manager *Manager) Start() error {
 	if !manager.IsKeySet() {
-		return errors.New("Unable to start Edge manager without key")
+		return errors.New("unable to start Edge manager without key")
 	}
 
 	apiServerAddr := fmt.Sprintf("%s:%s", manager.advertiseAddr, manager.agentOptions.AgentServerPort)
@@ -59,24 +60,24 @@ func (manager *Manager) Start() error {
 		PollFrequency:           agent.DefaultEdgePollInterval,
 		InactivityTimeout:       manager.agentOptions.EdgeInactivityTimeout,
 		InsecurePoll:            manager.agentOptions.EdgeInsecurePoll,
-		Tunnel:                  manager.agentOptions.EdgeTunnel,
+		TunnelCapability:        manager.agentOptions.EdgeTunnel,
 		PortainerURL:            manager.key.PortainerInstanceURL,
 		EndpointID:              manager.key.EndpointID,
 		TunnelServerAddr:        manager.key.TunnelServerAddr,
 		TunnelServerFingerprint: manager.key.TunnelServerFingerprint,
 		ContainerPlatform:       manager.containerPlatform,
 	}
 
-	log.Printf("[DEBUG] [internal,edge] [api_addr: %s] [edge_id: %s] [poll_frequency: %s] [inactivity_timeout: %s] [insecure_poll: %t] [tunnel: %t]", pollServiceConfig.APIServerAddr, pollServiceConfig.EdgeID, pollServiceConfig.PollFrequency, pollServiceConfig.InactivityTimeout, pollServiceConfig.InsecurePoll, pollServiceConfig.Tunnel)
+	log.Printf("[DEBUG] [internal,edge] [api_addr: %s] [edge_id: %s] [poll_frequency: %s] [inactivity_timeout: %s] [insecure_poll: %t] [tunnel_capability: %t]", pollServiceConfig.APIServerAddr, pollServiceConfig.EdgeID, pollServiceConfig.PollFrequency, pollServiceConfig.InactivityTimeout, pollServiceConfig.InsecurePoll, manager.agentOptions.EdgeTunnel)
 
-	stackManager, err := newStackManager(manager.key.PortainerInstanceURL, manager.key.EndpointID, manager.agentOptions.EdgeID, pollServiceConfig.InsecurePoll, pollServiceConfig.Tunnel)
+	stackManager, err := stack.NewStackManager(manager.key.PortainerInstanceURL, manager.key.EndpointID, manager.agentOptions.EdgeID, pollServiceConfig.InsecurePoll)
 	if err != nil {
 		return err
 	}
 	manager.stackManager = stackManager
 
-	manager.logsManager = newLogsManager(manager.key.PortainerInstanceURL, manager.key.EndpointID, manager.agentOptions.EdgeID, pollServiceConfig.InsecurePoll, pollServiceConfig.Tunnel)
-	manager.logsManager.start()
+	manager.logsManager = scheduler.NewLogsManager(manager.key.PortainerInstanceURL, manager.key.EndpointID, manager.agentOptions.EdgeID, pollServiceConfig.InsecurePoll)
+	manager.logsManager.Start()
 
 	pollService, err := newPollService(manager.stackManager, manager.logsManager, pollServiceConfig)
 	if err != nil {
@@ -92,11 +93,6 @@ func (manager *Manager) Start() error {
 	return nil
 }
 
-// IsEdgeModeEnabled returns true if edge mode is enabled
-func (manager *Manager) IsEdgeModeEnabled() bool {
-	return manager.agentOptions.EdgeMode
-}
-
 // ResetActivityTimer resets the activity timer
 func (manager *Manager) ResetActivityTimer() {
 	manager.pollService.resetActivityTimer()
@@ -143,13 +139,13 @@ func (manager *Manager) startEdgeBackgroundProcessOnKubernetes(runtimeCheckFrequ
 					return
 				}
 
-				err = manager.stackManager.setEngineStatus(engineTypeKubernetes)
+				err = manager.stackManager.SetEngineStatus(stack.EngineTypeKubernetes)
 				if err != nil {
 					log.Printf("[ERROR] [internal,edge,runtime] [message: unable to set engine status] [error: %s]", err)
 					return
 				}
 
-				err = manager.stackManager.start()
+				err = manager.stackManager.Start()
 				if err != nil {
 					log.Printf("[ERROR] [internal,edge,runtime] [message: unable to start stack manager] [error: %s]", err)
 					return
@@ -190,22 +186,22 @@ func (manager *Manager) checkDockerRuntimeConfig() error {
 	log.Printf("[DEBUG] [internal,edge,runtime,docker] [message: Docker runtime configuration check] [engine_status: %d] [leader_node: %t]", runtimeConfiguration.DockerConfiguration.EngineStatus, agentRunsOnLeaderNode)
 
 	if !agentRunsOnSwarm || agentRunsOnLeaderNode {
-		engineStatus := engineTypeDockerStandalone
+		engineStatus := stack.EngineTypeDockerStandalone
 		if agentRunsOnSwarm {
-			engineStatus = engineTypeDockerSwarm
+			engineStatus = stack.EngineTypeDockerSwarm
 		}
 
 		err = manager.pollService.start()
 		if err != nil {
 			return err
 		}
 
-		err = manager.stackManager.setEngineStatus(engineStatus)
+		err = manager.stackManager.SetEngineStatus(engineStatus)
 		if err != nil {
 			return err
 		}
 
-		err = manager.stackManager.start()
+		err = manager.stackManager.Start()
 		if err != nil {
 			return err
 		}
@@ -216,7 +212,7 @@ func (manager *Manager) checkDockerRuntimeConfig() error {
 			return err
 		}
 
-		err = manager.stackManager.stop()
+		err = manager.stackManager.Stop()
 		if err != nil {
 			return err
 		}

internal/edge/key.go → edge/key.go

@@ -4,6 +4,7 @@ import (
 	"encoding/base64"
 	"errors"
 	"fmt"
+	"log"
 	"strings"
 
 	"github.com/portainer/agent"
@@ -19,7 +20,7 @@ type edgeKey struct {
 }
 
 // SetKey parses and associates an Edge key to the agent.
-// If the agent is running inside a Swarm cluster, it will also set the "set" flag to specify that a key is set on this agent in the cluster.
+// If the agent is running inside a cluster, it will also set the "set" flag to specify that a key is set on this agent in the cluster.
 func (manager *Manager) SetKey(key string) error {
 	edgeKey, err := parseEdgeKey(key)
 	if err != nil {
@@ -58,10 +59,7 @@ func (manager *Manager) GetKey() string {
 
 // IsKeySet returns true if an Edge key is associated to the agent
 func (manager *Manager) IsKeySet() bool {
-	if manager.key == nil {
-		return false
-	}
-	return true
+	return manager.key != nil
 }
 
 // PropagateKeyInCluster propagates the Edge key associated to the agent to all the other agents inside the cluster
@@ -94,7 +92,6 @@ func (manager *Manager) PropagateKeyInCluster() error {
 
 // parseEdgeKey decodes a base64 encoded key and extract the decoded information from the following
 // format: <portainer_instance_url>|<tunnel_server_addr>|<tunnel_server_fingerprint>|<endpoint_id>
-// <client_credentials> are expected in the user:password format
 func parseEdgeKey(key string) (*edgeKey, error) {
 	decodedKey, err := base64.RawStdEncoding.DecodeString(key)
 	if err != nil {
@@ -122,3 +119,66 @@ func encodeKey(edgeKey *edgeKey) string {
 	encodedKey := base64.RawStdEncoding.EncodeToString([]byte(keyInfo))
 	return encodedKey
 }
+
+func RetrieveEdgeKey(edgeKey string, clusterService agent.ClusterService) (string, error) {
+	if edgeKey != "" {
+		log.Println("[INFO] [main,edge] [message: Edge key loaded from options]")
+		return edgeKey, nil
+	}
+
+	var keyRetrievalError error
+
+	edgeKey, keyRetrievalError = retrieveEdgeKeyFromFilesystem()
+	if keyRetrievalError != nil {
+		return "", keyRetrievalError
+	}
+
+	if edgeKey == "" && clusterService != nil {
+		edgeKey, keyRetrievalError = retrieveEdgeKeyFromCluster(clusterService)
+		if keyRetrievalError != nil {
+			return "", keyRetrievalError
+		}
+	}
+
+	return edgeKey, nil
+}
+
+func retrieveEdgeKeyFromFilesystem() (string, error) {
+	edgeKeyFilePath := fmt.Sprintf("%s/%s", agent.DataDirectory, agent.EdgeKeyFile)
+
+	keyFileExists, err := filesystem.FileExists(edgeKeyFilePath)
+	if err != nil {
+		return "", err
+	}
+
+	if !keyFileExists {
+		return "", nil
+	}
+
+	filesystemKey, err := filesystem.ReadFromFile(edgeKeyFilePath)
+	if err != nil {
+		return "", err
+	}
+
+	log.Println("[INFO] [main,edge] [message: Edge key loaded from the filesystem]")
+	return string(filesystemKey), nil
+}
+
+func retrieveEdgeKeyFromCluster(clusterService agent.ClusterService) (string, error) {
+	member := clusterService.GetMemberWithEdgeKeySet()
+	if member == nil {
+		return "", nil
+	}
+
+	httpCli := client.NewAPIClient()
+
+	memberAddr := fmt.Sprintf("%s:%s", member.IPAddress, member.Port)
+	memberKey, err := httpCli.GetEdgeKey(memberAddr)
+	if err != nil {
+		log.Printf("[ERROR] [main,edge,http,cluster] [message: Unable to retrieve Edge key from cluster member] [error: %s]", err)
+		return "", err
+	}
+
+	log.Println("[INFO] [main,edge] [message: Edge key loaded from cluster]")
+	return memberKey, nil
+}