refactor: use cached client tracker in the provider

Instead of creating workload cluster Kubernetes client each time and
closing it after that, use CAPI standard class to cache the client.

Signed-off-by: Artem Chernyshev <[email protected]>

Author: Unix4ever

controllers/configs.go

@@ -17,10 +17,8 @@ import (
 	talosconfig "github.com/talos-systems/talos/pkg/machinery/client/config"
 	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/client-go/util/connrotation"
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -43,41 +41,6 @@ func newDialer() *connrotation.Dialer {
 	return connrotation.NewDialer((&net.Dialer{Timeout: 30 * time.Second, KeepAlive: 30 * time.Second}).DialContext)
 }
 
-// kubeconfigForCluster will fetch a kubeconfig secret based on cluster name/namespace,
-// use it to create a clientset, and return it.
-func (r *TalosControlPlaneReconciler) kubeconfigForCluster(ctx context.Context, cluster client.ObjectKey) (*kubernetesClient, error) {
-	kubeconfigSecret := &corev1.Secret{}
-
-	err := r.Client.Get(ctx,
-		types.NamespacedName{
-			Namespace: cluster.Namespace,
-			Name:      cluster.Name + "-kubeconfig",
-		},
-		kubeconfigSecret,
-	)
-	if err != nil {
-		return nil, err
-	}
-
-	config, err := clientcmd.RESTConfigFromKubeConfig(kubeconfigSecret.Data["value"])
-	if err != nil {
-		return nil, err
-	}
-
-	dialer := newDialer()
-	config.Dial = dialer.DialContext
-
-	clientset, err := kubernetes.NewForConfig(config)
-	if err != nil {
-		return nil, err
-	}
-
-	return &kubernetesClient{
-		Clientset: clientset,
-		dialer:    dialer,
-	}, nil
-}
-
 // talosconfigForMachine will generate a talosconfig that uses *all* found addresses as the endpoints.
 func (r *TalosControlPlaneReconciler) talosconfigForMachines(ctx context.Context, tcp *controlplanev1.TalosControlPlane, machines ...clusterv1.Machine) (*talosclient.Client, error) {
 	if len(machines) == 0 {
@@ -132,7 +95,7 @@ func (r *TalosControlPlaneReconciler) talosconfigFromWorkloadCluster(ctx context
 		return nil, fmt.Errorf("at least one machine should be provided")
 	}
 
-	clientset, err := r.kubeconfigForCluster(ctx, cluster)
+	c, err := r.Tracker.GetClient(ctx, cluster)
 	if err != nil {
 		return nil, err
 	}
@@ -146,8 +109,10 @@ func (r *TalosControlPlaneReconciler) talosconfigFromWorkloadCluster(ctx context
 			return nil, fmt.Errorf("%q machine does not have a nodeRef", machine.Name)
 		}
 
+		var node v1.Node
+
 		// grab all addresses as endpoints
-		node, err := clientset.CoreV1().Nodes().Get(ctx, machine.Status.NodeRef.Name, metav1.GetOptions{})
+		err := c.Get(ctx, types.NamespacedName{Name: machine.Status.NodeRef.Name, Namespace: machine.Status.NodeRef.Namespace}, &node)
 		if err != nil {
 			return nil, err
 		}

controllers/etcd.go

@@ -13,18 +13,10 @@ import (
 	"github.com/talos-systems/talos/pkg/machinery/api/machine"
 	talosclient "github.com/talos-systems/talos/pkg/machinery/client"
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
-	"sigs.k8s.io/cluster-api/util"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 func (r *TalosControlPlaneReconciler) etcdHealthcheck(ctx context.Context, tcp *controlplanev1.TalosControlPlane, cluster *clusterv1.Cluster, ownedMachines []clusterv1.Machine) error {
-	kubeclient, err := r.kubeconfigForCluster(ctx, util.ObjectKey(cluster))
-	if err != nil {
-		return err
-	}
-
-	defer kubeclient.Close() //nolint:errcheck
-
 	machines := []clusterv1.Machine{}
 
 	for _, machine := range ownedMachines {

controllers/taloscontrolplane_controller.go

@@ -24,16 +24,19 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	corev1 "k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/selection"
+	"k8s.io/apimachinery/pkg/types"
 	kerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apiserver/pkg/storage/names"
 	"k8s.io/utils/pointer"
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
 	"sigs.k8s.io/cluster-api/controllers/external"
+	"sigs.k8s.io/cluster-api/controllers/remote"
 	"sigs.k8s.io/cluster-api/util"
 	"sigs.k8s.io/cluster-api/util/annotations"
 	"sigs.k8s.io/cluster-api/util/conditions"
@@ -66,6 +69,7 @@ type TalosControlPlaneReconciler struct {
 	APIReader client.Reader
 	Log       logr.Logger
 	Scheme    *runtime.Scheme
+	Tracker   *remote.ClusterCacheTracker
 }
 
 func (r *TalosControlPlaneReconciler) SetupWithManager(mgr ctrl.Manager, options controller.Options) error {
@@ -309,19 +313,21 @@ func (r *TalosControlPlaneReconciler) scaleDownControlPlane(ctx context.Context,
 
 	r.Log.Info("Found control plane machines", "machines", len(machines))
 
-	kubeclient, err := r.kubeconfigForCluster(ctx, cluster)
+	client, err := r.Tracker.GetClient(ctx, cluster)
 	if err != nil {
 		return ctrl.Result{RequeueAfter: 20 * time.Second}, err
 	}
 
-	defer kubeclient.Close() //nolint:errcheck
-
 	deleteMachine := machines[0]
 	for _, machine := range machines {
 		if !machine.ObjectMeta.DeletionTimestamp.IsZero() {
 			r.Log.Info("machine is in process of deletion", "machine", machine.Name)
 
-			node, err := kubeclient.CoreV1().Nodes().Get(ctx, machine.Status.NodeRef.Name, metav1.GetOptions{})
+			var node v1.Node
+
+			name := types.NamespacedName{Name: machine.Status.NodeRef.Name, Namespace: machine.Status.NodeRef.Namespace}
+
+			err := client.Get(ctx, name, &node)
 			if err != nil {
 				// It's possible for the node to already be deleted in the workload cluster, so we just
 				// requeue if that's that case instead of throwing a scary error.
@@ -333,7 +339,7 @@ func (r *TalosControlPlaneReconciler) scaleDownControlPlane(ctx context.Context,
 
 			r.Log.Info("Deleting node", "machine", machine.Name, "node", node.Name)
 
-			err = kubeclient.CoreV1().Nodes().Delete(ctx, node.Name, metav1.DeleteOptions{})
+			err = client.Delete(ctx, &node)
 			if err != nil {
 				return ctrl.Result{RequeueAfter: 20 * time.Second}, err
 			}
@@ -408,7 +414,11 @@ func (r *TalosControlPlaneReconciler) scaleDownControlPlane(ctx context.Context,
 
 	r.Log.Info("deleting node", "machine", deleteMachine.Name, "node", node.Name)
 
-	err = kubeclient.CoreV1().Nodes().Delete(ctx, node.Name, metav1.DeleteOptions{})
+	n := &v1.Node{}
+	n.Name = node.Name
+	n.Namespace = node.Namespace
+
+	err = client.Delete(ctx, n)
 	if err != nil {
 		return ctrl.Result{RequeueAfter: 20 * time.Second}, err
 	}
@@ -657,23 +667,23 @@ func (r *TalosControlPlaneReconciler) updateStatus(ctx context.Context, tcp *con
 		return nil
 	}
 
-	kubeclient, err := r.kubeconfigForCluster(ctx, util.ObjectKey(cluster))
+	c, err := r.Tracker.GetClient(ctx, util.ObjectKey(cluster))
 	if err != nil {
 		r.Log.Info("failed to get kubeconfig for the cluster", "error", err)
 
 		return nil
 	}
 
-	defer kubeclient.Close() //nolint:errcheck
-
 	nodeSelector := labels.NewSelector()
 	req, err := labels.NewRequirement(constants.LabelNodeRoleMaster, selection.Exists, []string{})
 	if err != nil {
 		return err
 	}
 
-	nodes, err := kubeclient.CoreV1().Nodes().List(ctx, metav1.ListOptions{
-		LabelSelector: nodeSelector.Add(*req).String(),
+	var nodes v1.NodeList
+
+	err = c.List(ctx, &nodes, &client.ListOptions{
+		LabelSelector: nodeSelector.Add(*req),
 	})
 
 	if err != nil {

main.go

@@ -17,6 +17,7 @@ limitations under the License.
 package main
 
 import (
+	"context"
 	"flag"
 	"math/rand"
 	"os"
@@ -29,7 +30,9 @@ import (
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
+	"sigs.k8s.io/cluster-api/controllers/remote"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 	// +kubebuilder:scaffold:imports
@@ -76,10 +79,31 @@ func main() {
 		os.Exit(1)
 	}
 
+	// Set up a ClusterCacheTracker to provide to controllers
+	// requiring a connection to a remote cluster
+	tracker, err := remote.NewClusterCacheTracker(mgr, remote.ClusterCacheTrackerOptions{
+		Indexes:               remote.DefaultIndexes,
+		ClientUncachedObjects: []client.Object{},
+	})
+	if err != nil {
+		setupLog.Error(err, "unable to create cluster cache tracker")
+		os.Exit(1)
+	}
+
+	if err := (&remote.ClusterCacheReconciler{
+		Client:  mgr.GetClient(),
+		Log:     ctrl.Log.WithName("remote").WithName("ClusterCacheReconciler"),
+		Tracker: tracker,
+	}).SetupWithManager(context.Background(), mgr, controller.Options{MaxConcurrentReconciles: 10}); err != nil {
+		setupLog.Error(err, "unable to create controller", "controller", "ClusterCacheReconciler")
+		os.Exit(1)
+	}
+
 	if err = (&controllers.TalosControlPlaneReconciler{
 		Client:    mgr.GetClient(),
 		APIReader: mgr.GetAPIReader(),
 		Log:       ctrl.Log.WithName("controllers").WithName("TalosControlPlane"),
+		Tracker:   tracker,
 		Scheme:    mgr.GetScheme(),
 	}).SetupWithManager(mgr, controller.Options{MaxConcurrentReconciles: 10}); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "TalosControlPlane")