refactor: introduce NodeProvider interface for unified node management (#512)

* refactor: introduce NodeProvider interface for unified node management

* fix: do not check for deployment type in node-funder and improve logs

* refactor: enhance restart command to utilize NodeProvider (#516)

* refactor: enhance restart command to utilize NodeProvider

* refactor: add deployment-type flag to restart command

* docs: update restart command docs in README

* refactor: simplify restart command by removing unused cluster name and namespace checks

* fix: adjust node name handling for Ingress and IngressRoute nodes

* fix: change Service Port type to interface{} for better handling of string and number values (temp)

* lint: remove unnecessary blank line

* fix: update node discovery

* feat: smoke file sizes (#505)

* feat: fize-sizes param for smoke

* ci: file-sizes in configs

* ci: file-sizes in staging

* refactor: comments

* refactor: backward compatibility for ContentSize

* fix: comments

* fix(load): use content-size for load check (#522)

---------

Co-authored-by: Ljubiša Gačević <[email protected]>

* refactor: add new config file and split nodes setup for create cluster (#487)

* chore: add new config file and split nodes setup for create cluster

* chore: update config to use latest docker image

* fix: add image flag to restart cmd

* refactor: add node groups support for filtering nodes

* refactor: enhance Client structure to include NodeGroupName and update filtering logic for beekeeper deployments

* refactor(node): improve filterClientsByNodeGroups method

---------

Co-authored-by: Ljubisa Gacevic <[email protected]>
Co-authored-by: nugaon <[email protected]>
Co-authored-by: Ljubiša Gačević <[email protected]>

* fix: minor changes

* fix: use node-grups for filtering

---------

Co-authored-by: Akrem Chabchoub <[email protected]>
Co-authored-by: nugaon <[email protected]>

Author: 3 people

README.md

@@ -539,7 +539,7 @@ beekeeper node-operator --geth-url="http://geth-swap.default.testnet.internal" -
 
 ### restart
 
-Command **restart** restarts bee nodes in Kubernetes, with optional targeting by namespace, label selectors, and node groups.
+Command **restart** restarts bee nodes in Kubernetes using a unified node management interface, with optional targeting by cluster name, namespace, label selectors, and node groups.
 
 It has following flags:
 
@@ -550,6 +550,7 @@ It has following flags:
 --image string            Container image to use when restarting pods (defaults to current image if not set).
 --node-groups strings     Comma-separated list of node groups to target for restarts (applies to all groups if not set).
 --timeout duration        Operation timeout (e.g., 5s, 10m, 1.5h). (default 5m0s)
+--deployment-type string  Indicates how the cluster was deployed: 'beekeeper' or 'helm'. (default "beekeeper")
 ```
 
 example:

cmd/beekeeper/cmd/cmd.go

@@ -12,10 +12,12 @@ import (
 	"strings"
 	"time"
 
+	"github.com/ethersphere/beekeeper/pkg/bee"
 	"github.com/ethersphere/beekeeper/pkg/config"
 	"github.com/ethersphere/beekeeper/pkg/httpx"
 	"github.com/ethersphere/beekeeper/pkg/k8s"
 	"github.com/ethersphere/beekeeper/pkg/logging"
+	"github.com/ethersphere/beekeeper/pkg/node"
 	"github.com/ethersphere/beekeeper/pkg/scheduler"
 	"github.com/ethersphere/beekeeper/pkg/swap"
 	"github.com/go-git/go-billy/v5/memfs"
@@ -433,6 +435,58 @@ func (c *command) executePeriodically(ctx context.Context, action func(ctx conte
 	return ctx.Err()
 }
 
+func (c *command) createNodeClient(ctx context.Context, useDeploymentType bool) (*node.Client, error) {
+	namespace := c.globalConfig.GetString(optionNameNamespace)
+	clusterName := c.globalConfig.GetString(optionNameClusterName)
+
+	if clusterName == "" && namespace == "" {
+		return nil, errors.New("either cluster name or namespace must be provided")
+	}
+
+	if c.globalConfig.IsSet(optionNameNamespace) && namespace == "" {
+		return nil, errors.New("namespace cannot be empty if set")
+	}
+
+	if namespace == "" && useDeploymentType && !isValidDeploymentType(c.globalConfig.GetString(optionNameDeploymentType)) {
+		return nil, errors.New("unsupported deployment type: must be 'beekeeper' or 'helm'")
+	}
+
+	if useDeploymentType {
+		c.log.Infof("using deployment type %s", c.globalConfig.GetString(optionNameDeploymentType))
+	}
+
+	var beeClients map[string]*bee.Client
+
+	if clusterName != "" {
+		cluster, err := c.setupCluster(ctx, clusterName, false)
+		if err != nil {
+			return nil, fmt.Errorf("setting up cluster %s: %w", clusterName, err)
+		}
+
+		beeClients, err = cluster.NodesClients(ctx)
+		if err != nil {
+			return nil, fmt.Errorf("failed to retrieve node clients: %w", err)
+		}
+
+		namespace = cluster.Namespace()
+	}
+
+	nodeClient := node.New(&node.ClientConfig{
+		Log:            c.log,
+		HTTPClient:     c.httpClient,
+		K8sClient:      c.k8sClient,
+		BeeClients:     beeClients,
+		Namespace:      namespace,
+		LabelSelector:  c.globalConfig.GetString(optionNameLabelSelector),
+		DeploymentType: node.DeploymentType(c.globalConfig.GetString(optionNameDeploymentType)),
+		InCluster:      c.globalConfig.GetBool(optionNameInCluster),
+		UseNamespace:   c.globalConfig.IsSet(optionNameNamespace),
+		NodeGroups:     c.globalConfig.GetStringSlice(optionNameNodeGroups),
+	})
+
+	return nodeClient, nil
+}
+
 func (c *command) setSwapClient() (err error) {
 	if c.globalConfig.IsSet(optionNameGethURL) {
 		gethUrl, err := url.Parse(c.globalConfig.GetString(optionNameGethURL))

cmd/beekeeper/cmd/node_funder.go

@@ -73,40 +73,33 @@ Requires --wallet-key for the funding account and --geth-url for blockchain acce
 					})
 				}
 
-				namespace := c.globalConfig.GetString(optionNameNamespace)
-				if namespace != "" {
-					label := c.globalConfig.GetString(optionNameLabelSelector)
-					funderClient := nodefunder.NewClient(c.k8sClient, c.globalConfig.GetBool(optionNameInCluster), label, c.log)
+				nodeClient, err := c.createNodeClient(ctx, false)
+				if err != nil {
+					return fmt.Errorf("creating node client: %w", err)
+				}
 
-					cfg.Namespace = namespace
+				if c.globalConfig.IsSet(optionNameNamespace) {
+					cfg.Namespace = nodeClient.Namespace()
 					return c.executePeriodically(ctx, func(ctx context.Context) error {
+						funderClient := nodefunder.NewClient(nodeClient, c.log)
 						return funder.Fund(ctx, cfg, funderClient, nil, logOpt)
 					})
 				}
 
-				clusterName := c.globalConfig.GetString(optionNameClusterName)
-				if clusterName != "" {
-					cluster, err := c.setupCluster(ctx, clusterName, false)
+				if c.globalConfig.IsSet(optionNameClusterName) {
+					nodes, err := nodeClient.GetNodes(ctx)
 					if err != nil {
-						return fmt.Errorf("setting up cluster %s: %w", clusterName, err)
+						return fmt.Errorf("failed to retrieve nodes: %w", err)
 					}
 
-					clients, err := cluster.NodesClients(ctx)
-					if err != nil {
-						return fmt.Errorf("failed to retrieve node clients: %w", err)
-					}
-
-					for _, node := range clients {
-						c.log.Debugf("adding node address %s", node.Name())
-						addr, err := node.Addresses(ctx)
+					for _, node := range nodes {
+						addr, err := node.Client().Node.Addresses(ctx)
 						if err != nil {
 							return fmt.Errorf("error fetching addresses for node %s: %w", node.Name(), err)
 						}
 						cfg.Addresses = append(cfg.Addresses, addr.Ethereum)
 					}
 
-					c.log.Infof("funding %d nodes in cluster %s", len(cfg.Addresses), clusterName)
-
 					return c.executePeriodically(ctx, func(ctx context.Context) error {
 						return funder.Fund(ctx, cfg, nil, nil, logOpt)
 					})
@@ -130,6 +123,7 @@ Requires --wallet-key for the funding account and --geth-url for blockchain acce
 	cmd.Flags().Float64(optionNameMinNative, 0, "Minimum amount of chain native coins (xDAI) nodes should have.")
 	cmd.Flags().Float64(optionNameMinSwarm, 0, "Minimum amount of swarm tokens (xBZZ) nodes should have.")
 	cmd.Flags().String(optionNameLabelSelector, nodeFunderLabelSelector, "Kubernetes label selector for filtering resources within the specified namespace. Use an empty string to select all resources.")
+	cmd.Flags().StringSlice(optionNameNodeGroups, nil, "List of node groups to target for node-funder (applies to all groups if not set). Only used with --cluster-name.")
 	cmd.Flags().Duration(optionNameTimeout, 5*time.Minute, "Operation timeout (e.g., 5s, 10m, 1.5h).")
 	cmd.Flags().Duration(optionNamePeriodicCheck, 0*time.Minute, "Periodic execution check interval.")
 

cmd/beekeeper/cmd/nuke.go

@@ -2,11 +2,9 @@ package cmd
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"time"
 
-	"github.com/ethersphere/beekeeper/pkg/bee"
 	"github.com/ethersphere/beekeeper/pkg/node"
 	"github.com/ethersphere/beekeeper/pkg/nuker"
 	"github.com/spf13/cobra"
@@ -28,59 +26,24 @@ func (c *command) initNukeCmd() (err error) {
 		This command provides StatefulSet update and rollback procedures to maintain cluster stability during the nuke process, ensuring safe and coordinated resets of node state.`,
 		RunE: func(cmd *cobra.Command, args []string) (err error) {
 			return c.withTimeoutHandler(cmd, func(ctx context.Context) error {
-				namespace := c.globalConfig.GetString(optionNameNamespace)
-				clusterName := c.globalConfig.GetString(optionNameClusterName)
-
-				if clusterName == "" && namespace == "" {
-					return errors.New("either cluster name or namespace must be provided")
-				}
-
-				if !isValidDeploymentType(c.globalConfig.GetString(optionNameDeploymentType)) {
-					return errors.New("unsupported deployment type: must be 'beekeeper' or 'helm'")
-				}
-
-				var beeClients map[string]*bee.Client
-
-				if clusterName != "" {
-					cluster, err := c.setupCluster(ctx, clusterName, false)
-					if err != nil {
-						return fmt.Errorf("setting up cluster %s: %w", clusterName, err)
-					}
-
-					beeClients, err = cluster.NodesClients(ctx)
-					if err != nil {
-						return fmt.Errorf("failed to retrieve node clients: %w", err)
-					}
-
-					namespace = cluster.Namespace()
+				if !c.globalConfig.GetBool(optionNameEnableK8S) {
+					return fmt.Errorf("kubernetes support must be enabled for nuke command")
 				}
 
-				nodeClient := node.New(&node.ClientConfig{
-					Log:            c.log,
-					HTTPClient:     c.httpClient,
-					K8sClient:      c.k8sClient,
-					BeeClients:     beeClients,
-					Namespace:      namespace,
-					LabelSelector:  c.globalConfig.GetString(optionNameLabelSelector),
-					DeploymentType: node.DeploymentType(c.globalConfig.GetString(optionNameDeploymentType)),
-					InCluster:      c.globalConfig.GetBool(optionNameInCluster),
-					UseNamespace:   c.globalConfig.IsSet(optionNameNamespace),
-				})
-
-				nodes, err := nodeClient.GetNodes(ctx)
+				nodeClient, err := c.createNodeClient(ctx, true)
 				if err != nil {
-					return fmt.Errorf("getting nodes: %w", err)
+					return fmt.Errorf("creating node client: %w", err)
 				}
 
 				nukerClient := nuker.New(&nuker.ClientConfig{
 					Log:                   c.log,
 					K8sClient:             c.k8sClient,
-					Nodes:                 nodes,
+					NodeProvider:          nodeClient,
 					UseRandomNeighborhood: c.globalConfig.GetBool(optionNameUseRandomNeighboorhood),
 				})
 
-				if err := nukerClient.Run(ctx, namespace, c.globalConfig.GetStringSlice(optionNameRestartArgs)); err != nil {
-					return fmt.Errorf("updating Bee cluster %s: %w", clusterName, err)
+				if err := nukerClient.Run(ctx, c.globalConfig.GetStringSlice(optionNameRestartArgs)); err != nil {
+					return fmt.Errorf("running nuke command: %w", err)
 				}
 
 				return nil
@@ -92,6 +55,7 @@ func (c *command) initNukeCmd() (err error) {
 	cmd.Flags().String(optionNameClusterName, "", "Target Beekeeper cluster name.")
 	cmd.Flags().StringP(optionNameNamespace, "n", "", "Kubernetes namespace (overrides cluster name).")
 	cmd.Flags().String(optionNameLabelSelector, beeLabelSelector, "Kubernetes label selector for filtering resources when namespace is set (use empty string for all).")
+	cmd.Flags().StringSlice(optionNameNodeGroups, nil, "List of node groups to target for nuke (applies to all groups if not set). Only used with --cluster-name.")
 	cmd.Flags().Duration(optionNameTimeout, 30*time.Minute, "Timeout")
 	cmd.Flags().StringSlice(optionNameRestartArgs, []string{"bee", "start", "--config=.bee.yaml"}, "Command to run in the Bee cluster, e.g. 'db,nuke,--config=.bee.yaml'")
 	cmd.Flags().Bool(optionNameUseRandomNeighboorhood, false, "Use random neighborhood for Bee nodes (default: false)")

cmd/beekeeper/cmd/operator.go

@@ -3,6 +3,7 @@ package cmd
 import (
 	"context"
 	"errors"
+	"fmt"
 	"time"
 
 	"github.com/ethersphere/beekeeper/pkg/funder/operator"
@@ -39,6 +40,10 @@ but this can be customized with --label-selector. Runs indefinitely until manual
 Requires --namespace, --wallet-key, and --geth-url for operation.`,
 		RunE: func(cmd *cobra.Command, args []string) (err error) {
 			return c.withTimeoutHandler(cmd, func(ctx context.Context) error {
+				if !c.globalConfig.GetBool(optionNameEnableK8S) {
+					return fmt.Errorf("kubernetes support must be enabled for nuke command")
+				}
+
 				namespace := c.globalConfig.GetString(optionNameNamespace)
 				if namespace == "" {
 					return errors.New("namespace not provided")

cmd/beekeeper/cmd/restart.go

@@ -2,7 +2,6 @@ package cmd
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"time"
 
@@ -12,11 +11,12 @@ import (
 
 func (c *command) initRestartCmd() (err error) {
 	const (
-		optionNameLabelSelector = "label-selector"
-		optionNameNamespace     = "namespace"
-		optionNameImage         = "image"
-		optionNameNodeGroups    = "node-groups"
-		optionNameTimeout       = "timeout"
+		optionNameLabelSelector  = "label-selector"
+		optionNameNamespace      = "namespace"
+		optionNameImage          = "image"
+		optionNameNodeGroups     = "node-groups"
+		optionNameTimeout        = "timeout"
+		optionNameDeploymentType = "deployment-type"
 	)
 
 	cmd := &cobra.Command{
@@ -41,58 +41,35 @@ This command is useful for:
 
 Requires either --cluster-name or --namespace to be specified.`,
 		RunE: func(cmd *cobra.Command, args []string) (err error) {
-			ctx, cancel := context.WithTimeout(cmd.Context(), c.globalConfig.GetDuration(optionNameTimeout))
-			defer cancel()
-
-			clusterName := c.globalConfig.GetString(optionNameClusterName)
-			namespace := c.globalConfig.GetString(optionNameNamespace)
-
-			if clusterName == "" && namespace == "" {
-				return errors.New("either cluster name or namespace must be provided")
-			}
-
-			restartClient := restart.NewClient(c.k8sClient, c.log)
-
-			if clusterName != "" {
-				clusterConfig, ok := c.config.Clusters[clusterName]
-				if !ok {
-					return fmt.Errorf("cluster config %s not defined", clusterName)
+			return c.withTimeoutHandler(cmd, func(ctx context.Context) error {
+				if !c.globalConfig.GetBool(optionNameEnableK8S) {
+					return fmt.Errorf("kubernetes support must be enabled for nuke command")
 				}
 
-				cluster, err := c.setupCluster(ctx, clusterName, false)
+				nodeClient, err := c.createNodeClient(ctx, true)
 				if err != nil {
-					return fmt.Errorf("setting up cluster %s: %w", clusterName, err)
+					return fmt.Errorf("creating node client: %w", err)
 				}
 
-				c.log.Infof("restarting cluster %s", clusterName)
+				restartClient := restart.NewClient(nodeClient, c.k8sClient, c.log)
 
-				if err := restartClient.RestartCluster(ctx,
-					cluster,
-					clusterConfig.GetNamespace(),
-					c.globalConfig.GetString(optionNameImage),
-					c.globalConfig.GetStringSlice(optionNameNodeGroups),
-				); err != nil {
-					return fmt.Errorf("restarting cluster %s: %w", clusterName, err)
+				if err := restartClient.Restart(ctx, c.globalConfig.GetString(optionNameImage)); err != nil {
+					return fmt.Errorf("restarting pods: %w", err)
 				}
 
 				return nil
-			}
-
-			if err := restartClient.RestartPods(ctx, namespace, c.globalConfig.GetString(optionNameLabelSelector)); err != nil {
-				return fmt.Errorf("restarting pods in namespace %s: %w", namespace, err)
-			}
-
-			return nil
+			})
 		},
 		PreRunE: c.preRunE,
 	}
 
 	cmd.Flags().String(optionNameClusterName, "", "Kubernetes cluster to operate on (overrides namespace and label selector).")
 	cmd.Flags().StringP(optionNameNamespace, "n", "", "Namespace to delete pods from (only used if cluster name is not set).")
-	cmd.Flags().String(optionNameLabelSelector, "", "Label selector for resources in the namespace (only used with namespace).")
+	cmd.Flags().String(optionNameLabelSelector, beeLabelSelector, "Label selector for resources in the namespace (only used with namespace).")
 	cmd.Flags().String(optionNameImage, "", "Container image to use when restarting pods (defaults to current image if not set).")
-	cmd.Flags().StringSlice(optionNameNodeGroups, nil, "List of node groups to target for restarts (applies to all groups if not set).")
+	cmd.Flags().StringSlice(optionNameNodeGroups, nil, "List of node groups to target for restarts (applies to all groups if not set). Only used with --cluster-name.")
 	cmd.Flags().Duration(optionNameTimeout, 5*time.Minute, "Operation timeout (e.g., 5s, 10m, 1.5h).")
+	cmd.Flags().String(optionNameDeploymentType, "beekeeper", "Indicates how the cluster was deployed: 'beekeeper' or 'helm'.")
 
 	c.root.AddCommand(cmd)