Merge pull request kubernetes-sigs#6684 from prometherion/refactor/docker-node

🌱 patching Docker-based nodes provider ID using client-runtime

Author: k8s-ci-robot

test/infrastructure/docker/controllers/alias.go

@@ -24,6 +24,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 
+	"sigs.k8s.io/cluster-api/controllers/remote"
 	"sigs.k8s.io/cluster-api/test/infrastructure/container"
 	dockercontrollers "sigs.k8s.io/cluster-api/test/infrastructure/docker/internal/controllers"
 )
@@ -35,13 +36,15 @@ import (
 type DockerMachineReconciler struct {
 	Client           client.Client
 	ContainerRuntime container.Runtime
+	Tracker          *remote.ClusterCacheTracker
 }
 
 // SetupWithManager sets up the reconciler with the Manager.
 func (r *DockerMachineReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, options controller.Options) error {
 	return (&dockercontrollers.DockerMachineReconciler{
 		Client:           r.Client,
 		ContainerRuntime: r.ContainerRuntime,
+		Tracker:          r.Tracker,
 	}).SetupWithManager(ctx, mgr, options)
 }
 

test/infrastructure/docker/exp/controllers/alias.go

@@ -25,6 +25,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 
+	"sigs.k8s.io/cluster-api/controllers/remote"
 	"sigs.k8s.io/cluster-api/test/infrastructure/container"
 	dockermachinepoolcontrollers "sigs.k8s.io/cluster-api/test/infrastructure/docker/exp/internal/controllers"
 )
@@ -34,6 +35,7 @@ type DockerMachinePoolReconciler struct {
 	Client           client.Client
 	Scheme           *runtime.Scheme
 	ContainerRuntime container.Runtime
+	Tracker          *remote.ClusterCacheTracker
 }
 
 // SetupWithManager will add watches for this controller.
@@ -42,5 +44,6 @@ func (r *DockerMachinePoolReconciler) SetupWithManager(ctx context.Context, mgr
 		Client:           r.Client,
 		Scheme:           r.Scheme,
 		ContainerRuntime: r.ContainerRuntime,
+		Tracker:          r.Tracker,
 	}).SetupWithManager(ctx, mgr, options)
 }

test/infrastructure/docker/exp/internal/controllers/dockermachinepool_controller.go

@@ -34,6 +34,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
+	"sigs.k8s.io/cluster-api/controllers/remote"
 	expv1 "sigs.k8s.io/cluster-api/exp/api/v1beta1"
 	utilexp "sigs.k8s.io/cluster-api/exp/util"
 	"sigs.k8s.io/cluster-api/test/infrastructure/container"
@@ -49,6 +50,7 @@ type DockerMachinePoolReconciler struct {
 	Client           client.Client
 	Scheme           *runtime.Scheme
 	ContainerRuntime container.Runtime
+	Tracker          *remote.ClusterCacheTracker
 }
 
 // +kubebuilder:rbac:groups=infrastructure.cluster.x-k8s.io,resources=dockermachinepools,verbs=get;list;watch;create;update;patch;delete
@@ -186,7 +188,11 @@ func (r *DockerMachinePoolReconciler) reconcileNormal(ctx context.Context, clust
 	}
 
 	// Reconcile machines and updates Status.Instances
-	res, err := pool.ReconcileMachines(ctx)
+	remoteClient, err := r.Tracker.GetClient(ctx, client.ObjectKeyFromObject(cluster))
+	if err != nil {
+		return ctrl.Result{}, errors.Wrap(err, "failed to generate workload cluster client")
+	}
+	res, err := pool.ReconcileMachines(ctx, remoteClient)
 	if err != nil {
 		return res, err
 	}

test/infrastructure/docker/exp/internal/docker/nodepool.go

@@ -79,7 +79,7 @@ func NewNodePool(ctx context.Context, c client.Client, cluster *clusterv1.Cluste
 // currently the nodepool supports only a recreate strategy for replacing old nodes with new ones
 // (all existing machines are killed before new ones are created).
 // TODO: consider if to support a Rollout strategy (a more progressive node replacement).
-func (np *NodePool) ReconcileMachines(ctx context.Context) (ctrl.Result, error) {
+func (np *NodePool) ReconcileMachines(ctx context.Context, remoteClient client.Client) (ctrl.Result, error) {
 	desiredReplicas := int(*np.machinePool.Spec.Replicas)
 
 	// Delete all the machines in excess (outdated machines or machines exceeding desired replica count).
@@ -140,7 +140,7 @@ func (np *NodePool) ReconcileMachines(ctx context.Context) (ctrl.Result, error)
 	result := ctrl.Result{}
 	for i := range np.machines {
 		machine := np.machines[i]
-		if res, err := np.reconcileMachine(ctx, machine); err != nil || !res.IsZero() {
+		if res, err := np.reconcileMachine(ctx, machine, remoteClient); err != nil || !res.IsZero() {
 			if err != nil {
 				return ctrl.Result{}, errors.Wrap(err, "failed to reconcile machine")
 			}
@@ -240,7 +240,7 @@ func (np *NodePool) refresh(ctx context.Context) error {
 }
 
 // reconcileMachine will build and provision a docker machine and update the docker machine pool status for that instance.
-func (np *NodePool) reconcileMachine(ctx context.Context, machine *docker.Machine) (ctrl.Result, error) {
+func (np *NodePool) reconcileMachine(ctx context.Context, machine *docker.Machine, remoteClient client.Client) (ctrl.Result, error) {
 	log := ctrl.LoggerFrom(ctx)
 
 	var machineStatus infraexpv1.DockerMachinePoolInstanceStatus
@@ -334,7 +334,7 @@ func (np *NodePool) reconcileMachine(ctx context.Context, machine *docker.Machin
 		// Usually a cloud provider will do this, but there is no docker-cloud provider.
 		// Requeue if there is an error, as this is likely momentary load balancer
 		// state changes during control plane provisioning.
-		if err := externalMachine.SetNodeProviderID(ctx); err != nil {
+		if err = externalMachine.SetNodeProviderID(ctx, remoteClient); err != nil {
 			log.V(4).Info("transient error setting the provider id")
 			return ctrl.Result{Requeue: true}, nil //nolint:nilerr
 		}

test/infrastructure/docker/internal/controllers/dockermachine_controller.go

@@ -35,6 +35,7 @@ import (
 
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
 	bootstrapv1 "sigs.k8s.io/cluster-api/bootstrap/kubeadm/api/v1beta1"
+	"sigs.k8s.io/cluster-api/controllers/remote"
 	"sigs.k8s.io/cluster-api/test/infrastructure/container"
 	infrav1 "sigs.k8s.io/cluster-api/test/infrastructure/docker/api/v1beta1"
 	"sigs.k8s.io/cluster-api/test/infrastructure/docker/internal/docker"
@@ -49,6 +50,7 @@ import (
 type DockerMachineReconciler struct {
 	client.Client
 	ContainerRuntime container.Runtime
+	Tracker          *remote.ClusterCacheTracker
 }
 
 // +kubebuilder:rbac:groups=infrastructure.cluster.x-k8s.io,resources=dockermachines,verbs=get;list;watch;create;update;patch;delete
@@ -310,7 +312,11 @@ func (r *DockerMachineReconciler) reconcileNormal(ctx context.Context, cluster *
 	// Usually a cloud provider will do this, but there is no docker-cloud provider.
 	// Requeue if there is an error, as this is likely momentary load balancer
 	// state changes during control plane provisioning.
-	if err := externalMachine.SetNodeProviderID(ctx); err != nil {
+	remoteClient, err := r.Tracker.GetClient(ctx, client.ObjectKeyFromObject(cluster))
+	if err != nil {
+		return ctrl.Result{}, errors.Wrap(err, "failed to generate workload cluster client")
+	}
+	if err := externalMachine.SetNodeProviderID(ctx, remoteClient); err != nil {
 		if errors.As(err, &docker.ContainerNotRunningError{}) {
 			return ctrl.Result{}, errors.Wrap(err, "failed to patch the Kubernetes node with the machine providerID")
 		}

test/infrastructure/docker/internal/docker/machine.go

@@ -28,8 +28,11 @@ import (
 
 	"github.com/go-logr/logr"
 	"github.com/pkg/errors"
+	corev1 "k8s.io/api/core/v1"
+	apimachinerytypes "k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/kind/pkg/apis/config/v1alpha4"
 	"sigs.k8s.io/kind/pkg/cluster/constants"
 
@@ -42,6 +45,7 @@ import (
 	"sigs.k8s.io/cluster-api/test/infrastructure/docker/internal/provisioning/cloudinit"
 	"sigs.k8s.io/cluster-api/test/infrastructure/docker/internal/provisioning/ignition"
 	clusterapicontainer "sigs.k8s.io/cluster-api/util/container"
+	"sigs.k8s.io/cluster-api/util/patch"
 )
 
 const (
@@ -381,7 +385,7 @@ func (m *Machine) CheckForBootstrapSuccess(ctx context.Context) error {
 }
 
 // SetNodeProviderID sets the docker provider ID for the kubernetes node.
-func (m *Machine) SetNodeProviderID(ctx context.Context) error {
+func (m *Machine) SetNodeProviderID(ctx context.Context, c client.Client) error {
 	log := ctrl.LoggerFrom(ctx)
 
 	kubectlNode, err := m.getKubectlNode(ctx)
@@ -392,20 +396,21 @@ func (m *Machine) SetNodeProviderID(ctx context.Context) error {
 		return errors.Wrapf(ContainerNotRunningError{Name: kubectlNode.Name}, "unable to set NodeProviderID")
 	}
 
+	node := &corev1.Node{}
+	if err = c.Get(ctx, apimachinerytypes.NamespacedName{Name: m.ContainerName()}, node); err != nil {
+		return errors.Wrap(err, "failed to retrieve node")
+	}
+
 	log.Info("Setting Kubernetes node providerID")
-	patch := fmt.Sprintf(`{"spec": {"providerID": %q}}`, m.ProviderID())
-	cmd := kubectlNode.Commander.Command(
-		"kubectl",
-		"--kubeconfig", "/etc/kubernetes/kubelet.conf",
-		"patch",
-		"node", m.ContainerName(),
-		"--patch", patch,
-	)
-	lines, err := cmd.RunLoggingOutputOnFail(ctx)
+
+	patchHelper, err := patch.NewHelper(node, c)
 	if err != nil {
-		for _, line := range lines {
-			log.Info(line)
-		}
+		return err
+	}
+
+	node.Spec.ProviderID = m.ProviderID()
+
+	if err = patchHelper.Patch(ctx, node); err != nil {
 		return errors.Wrap(err, "failed update providerID")
 	}
 

test/infrastructure/docker/internal/docker/types/node.go

@@ -18,8 +18,6 @@ limitations under the License.
 package types
 
 import (
-	"bufio"
-	"bytes"
 	"context"
 	"io"
 	"path/filepath"
@@ -164,22 +162,6 @@ type ContainerCmd struct {
 	stderr   io.Writer
 }
 
-// RunLoggingOutputOnFail runs the cmd, logging error output if Run returns an error.
-func (c *ContainerCmd) RunLoggingOutputOnFail(ctx context.Context) ([]string, error) {
-	var buff bytes.Buffer
-	c.SetStdout(&buff)
-	c.SetStderr(&buff)
-	err := c.Run(ctx)
-	out := make([]string, 0)
-	if err != nil {
-		scanner := bufio.NewScanner(&buff)
-		for scanner.Scan() {
-			out = append(out, scanner.Text())
-		}
-	}
-	return out, errors.WithStack(err)
-}
-
 // Run will run a configured ContainerCmd inside a container instance.
 func (c *ContainerCmd) Run(ctx context.Context) error {
 	containerRuntime, err := container.RuntimeFrom(ctx)

test/infrastructure/docker/main.go

@@ -185,9 +185,33 @@ func setupReconcilers(ctx context.Context, mgr ctrl.Manager) {
 		os.Exit(1)
 	}
 
+	log := ctrl.Log.WithName("remote").WithName("ClusterCacheTracker")
+	tracker, err := remote.NewClusterCacheTracker(
+		mgr,
+		remote.ClusterCacheTrackerOptions{
+			Log:     &log,
+			Indexes: remote.DefaultIndexes,
+		},
+	)
+	if err != nil {
+		setupLog.Error(err, "unable to create cluster cache tracker")
+		os.Exit(1)
+	}
+
+	if err := (&remote.ClusterCacheReconciler{
+		Client:  mgr.GetClient(),
+		Tracker: tracker,
+	}).SetupWithManager(ctx, mgr, controller.Options{
+		MaxConcurrentReconciles: concurrency,
+	}); err != nil {
+		setupLog.Error(err, "unable to create controller", "controller", "ClusterCacheReconciler")
+		os.Exit(1)
+	}
+
 	if err := (&controllers.DockerMachineReconciler{
 		Client:           mgr.GetClient(),
 		ContainerRuntime: runtimeClient,
+		Tracker:          tracker,
 	}).SetupWithManager(ctx, mgr, controller.Options{
 		MaxConcurrentReconciles: concurrency,
 	}); err != nil {
@@ -207,9 +231,8 @@ func setupReconcilers(ctx context.Context, mgr ctrl.Manager) {
 		if err := (&expcontrollers.DockerMachinePoolReconciler{
 			Client:           mgr.GetClient(),
 			ContainerRuntime: runtimeClient,
-		}).SetupWithManager(ctx, mgr, controller.Options{
-			MaxConcurrentReconciles: concurrency,
-		}); err != nil {
+			Tracker:          tracker,
+		}).SetupWithManager(ctx, mgr, controller.Options{MaxConcurrentReconciles: concurrency}); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "DockerMachinePool")
 			os.Exit(1)
 		}