Update kube-proxy daemonset image tag during KCP upgrade

controlplane/kubeadm/controllers/kubeadm_control_plane_controller.go

@@ -251,7 +251,8 @@ func (r *KubeadmControlPlaneReconciler) reconcile(ctx context.Context, cluster *
 		return r.upgradeControlPlane(ctx, cluster, kcp, ownedMachines, requireUpgrade)
 	}
 
-	// If we've made it this far, we we can assume that all ownedMachines are up to date
+	// If we've made it this far, we can assume that all ownedMachines are up to date
+
 	numMachines := len(ownedMachines)
 	desiredReplicas := int(*kcp.Spec.Replicas)
 
@@ -272,6 +273,16 @@ func (r *KubeadmControlPlaneReconciler) reconcile(ctx context.Context, cluster *
 		return r.scaleDownControlPlane(ctx, cluster, kcp, ownedMachines)
 	}
 
+	// Update kube-proxy image name
+	workloadCluster, err := r.managementCluster.GetWorkloadCluster(ctx, util.ObjectKey(cluster))
+	if err != nil {
+		// Just log and continue, control-plane cluster may be uninitialized
+		logger.Error(err, "failed to get remote client for workload cluster")
+	} else if err := workloadCluster.UpdateKubeProxyImageInfo(ctx, kcp); err != nil {
+		logger.Error(err, "failed to update kube-proxy image name in kube-proxy daemonset")
+		return ctrl.Result{}, err
+	}
+
 	return ctrl.Result{}, nil
 }
 

controlplane/kubeadm/controllers/kubeadm_control_plane_controller_test.go

@@ -540,7 +540,7 @@ func TestReconcileInitializeControlPlane(t *testing.T) {
 		recorder: record.NewFakeRecorder(32),
 		managementCluster: &fakeManagementCluster{
 			Management: &internal.Management{Client: fakeClient},
-			Workload:   fakeWorkloadCluster{},
+			Workload:   fakeWorkloadCluster{Workload: &internal.Workload{Client: fakeClient}, Status: internal.ClusterStatus{}},
 		},
 	}
 

controlplane/kubeadm/internal/cluster_test.go

@@ -23,6 +23,7 @@ import (
 	"strings"
 	"testing"
 
+	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -225,6 +226,7 @@ type fakeClient struct {
 	get       map[string]interface{}
 	getErr    error
 	createErr error
+	patchErr  error
 }
 
 func (f *fakeClient) Get(_ context.Context, key client.ObjectKey, obj runtime.Object) error {
@@ -239,6 +241,8 @@ func (f *fakeClient) Get(_ context.Context, key client.ObjectKey, obj runtime.Ob
 		l.DeepCopyInto(obj.(*rbacv1.RoleBinding))
 	case *rbacv1.Role:
 		l.DeepCopyInto(obj.(*rbacv1.Role))
+	case *appsv1.DaemonSet:
+		l.DeepCopyInto(obj.(*appsv1.DaemonSet))
 	case nil:
 		return apierrors.NewNotFound(schema.GroupResource{}, key.Name)
 	default:
@@ -266,6 +270,13 @@ func (f *fakeClient) Create(_ context.Context, _ runtime.Object, _ ...client.Cre
 	return nil
 }
 
+func (f *fakeClient) Patch(_ context.Context, _ runtime.Object, _ client.Patch, _ ...client.PatchOption) error {
+	if f.patchErr != nil {
+		return f.patchErr
+	}
+	return nil
+}
+
 func TestManagementCluster_healthCheck_NoError(t *testing.T) {
 	tests := []struct {
 		name             string

controlplane/kubeadm/internal/workload_cluster.go

@@ -29,18 +29,25 @@ import (
 
 	"github.com/blang/semver"
 	"github.com/pkg/errors"
+	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1alpha3"
+	controlplanev1 "sigs.k8s.io/cluster-api/controlplane/kubeadm/api/v1alpha3"
 	"sigs.k8s.io/cluster-api/controlplane/kubeadm/internal/etcd"
 	etcdutil "sigs.k8s.io/cluster-api/controlplane/kubeadm/internal/etcd/util"
 	"sigs.k8s.io/cluster-api/util"
 	"sigs.k8s.io/cluster-api/util/certs"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
 )
 
+const (
+	kubeProxyDaemonSetName = "kube-proxy"
+)
+
 var (
 	ErrControlPlaneMinNodes = errors.New("cluster has fewer than 2 control plane nodes; removing an etcd member is not supported")
 )
@@ -64,6 +71,7 @@ type WorkloadCluster interface {
 	UpdateKubernetesVersionInKubeadmConfigMap(ctx context.Context, version string) error
 	UpdateEtcdVersionInKubeadmConfigMap(ctx context.Context, imageRepository, imageTag string) error
 	UpdateKubeletConfigMap(ctx context.Context, version semver.Version) error
+	UpdateKubeProxyImageInfo(ctx context.Context, kcp *controlplanev1.KubeadmControlPlane) error
 	RemoveEtcdMemberForMachine(ctx context.Context, machine *clusterv1.Machine) error
 	RemoveMachineFromKubeadmConfigMap(ctx context.Context, machine *clusterv1.Machine) error
 }
@@ -563,3 +571,30 @@ func firstNodeNotMatchingName(name string, nodes []corev1.Node) *corev1.Node {
 	}
 	return nil
 }
+
+// UpdateKubeProxyImageInfo updates kube-proxy image in the kube-proxy DaemonSet.
+func (w *Workload) UpdateKubeProxyImageInfo(ctx context.Context, kcp *controlplanev1.KubeadmControlPlane) error {
+	ds := &appsv1.DaemonSet{}
+
+	if err := w.Client.Get(ctx, ctrlclient.ObjectKey{Name: kubeProxyDaemonSetName, Namespace: metav1.NamespaceSystem}, ds); err != nil {
+		if apierrors.IsNotFound(err) {
+			// if kube-proxy is missing, return without errors
+			return nil
+		}
+		return errors.Wrapf(err, "failed to determine if %s daemonset already exists", kubeProxyDaemonSetName)
+	}
+
+	newImageName, err := util.ModifyImageTag(ds.Spec.Template.Spec.Containers[0].Image, kcp.Spec.Version)
+	if err != nil {
+		return err
+	}
+
+	if len(ds.Spec.Template.Spec.Containers) > 0 && ds.Spec.Template.Spec.Containers[0].Image != newImageName {
+		patch := client.MergeFrom(ds.DeepCopy())
+		ds.Spec.Template.Spec.Containers[0].Image = newImageName
+		if err := w.Client.Patch(ctx, ds, patch); err != nil {
+			return errors.Wrap(err, "error patching kube-proxy DaemonSet")
+		}
+	}
+	return nil
+}

controlplane/kubeadm/internal/workload_cluster_test.go

@@ -22,9 +22,12 @@ import (
 	"testing"
 
 	"github.com/blang/semver"
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"sigs.k8s.io/cluster-api/controlplane/kubeadm/api/v1alpha3"
 	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -104,3 +107,76 @@ func TestCluster_ReconcileKubeletRBACBinding_Error(t *testing.T) {
 		})
 	}
 }
+
+func TestUpdateKubeProxyImageInfo(t *testing.T) {
+	ds := &appsv1.DaemonSet{
+		Spec: appsv1.DaemonSetSpec{
+			Template: corev1.PodTemplateSpec{
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{{Image: "k8s.gcr.io/kube-proxy:v1.16.3"}},
+				},
+			},
+		},
+	}
+
+	dsImageInDigestFormat := ds.DeepCopy()
+	dsImageInDigestFormat.Spec.Template.Spec.Containers[0].Image = "k8s.gcr.io/kube-proxy@sha256:47bfd"
+
+	dsImageEmpty := ds.DeepCopy()
+	dsImageEmpty.Spec.Template.Spec.Containers[0].Image = ""
+
+	tests := []struct {
+		name      string
+		ds        *appsv1.DaemonSet
+		expectErr bool
+		clientGet map[string]interface{}
+		patchErr  error
+	}{
+		{
+			name:      "succeeds if patch correctly",
+			ds:        ds,
+			expectErr: false,
+			clientGet: map[string]interface{}{
+				"kube-system/" + kubeProxyDaemonSetName: ds,
+			},
+		},
+		{
+			name:      "returns error if image in kube-proxy ds was in digest format",
+			ds:        dsImageInDigestFormat,
+			expectErr: true,
+			clientGet: map[string]interface{}{
+				"kube-system/" + kubeProxyDaemonSetName: dsImageInDigestFormat,
+			},
+		},
+		{
+			name:      "returns error if image in kube-proxy ds was in wrong format",
+			ds:        ds,
+			expectErr: true,
+			clientGet: map[string]interface{}{
+				"kube-system/" + kubeProxyDaemonSetName: dsImageEmpty,
+			},
+		},
+	}
+
+	ctx := context.Background()
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+
+			fakeClient := &fakeClient{
+				get: tt.clientGet,
+			}
+			c := &Workload{
+				Client: fakeClient,
+			}
+			err := c.UpdateKubeProxyImageInfo(ctx, &v1alpha3.KubeadmControlPlane{Spec: v1alpha3.KubeadmControlPlaneSpec{Version: "1.16.3"}})
+			if err != nil && !tt.expectErr {
+				t.Fatalf("expected no error, got %s", err)
+			}
+			if err == nil && tt.expectErr {
+				t.Fatal("expected error but got none")
+			}
+
+		})
+	}
+}

test/infrastructure/docker/go.sum

@@ -66,6 +66,7 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no=
+github.com/docker/distribution v2.7.1+incompatible h1:a5mlkVzth6W5A4fOsS3D2EO5BUmsJpcB+cRlLU7cSug=
 github.com/docker/distribution v2.7.1+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w=
 github.com/docker/docker v0.7.3-0.20190327010347-be7ac8be2ae0/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/docker/go-units v0.3.3/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
@@ -266,6 +267,7 @@ github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7J
 github.com/onsi/gomega v1.8.1/go.mod h1:Ho0h+IUsWyvy1OpqCwxlQ/21gkhVunqlU8fDGcoTdcA=
 github.com/onsi/gomega v1.9.0 h1:R1uwffexN6Pr340GtYRIdZmAiN4J+iw6WG4wog1DUXg=
 github.com/onsi/gomega v1.9.0/go.mod h1:Ho0h+IUsWyvy1OpqCwxlQ/21gkhVunqlU8fDGcoTdcA=
+github.com/opencontainers/go-digest v1.0.0-rc1 h1:WzifXhOVOEOuFYOJAW6aQqW0TooG2iki3E3Ii+WN7gQ=
 github.com/opencontainers/go-digest v1.0.0-rc1/go.mod h1:cMLVZDEM3+U2I4VmLI6N8jQYUd2OVphdqWwCJHrFt2s=
 github.com/pborman/uuid v1.2.0/go.mod h1:X/NO0urCmaxf9VXbdlT7C2Yzkj2IKimNn4k+gtPdI/k=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=

util/util.go

@@ -24,6 +24,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/docker/distribution/reference"
 	"github.com/pkg/errors"
 	v1 "k8s.io/api/core/v1"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
@@ -61,6 +62,27 @@ func RandomString(n int) string {
 	return string(result)
 }
 
+// ModifyImageTag takes an imageName (e.g., registry/repo:tag), and returns an image name with updated tag
+func ModifyImageTag(imageName, tagName string) (string, error) {
+	namedRef, err := reference.ParseNormalizedNamed(imageName)
+	if err != nil {
+		return "", errors.Wrap(err, "failed to parse image name")
+
+	}
+	// return error if images use digest as version instead of tag
+	if _, isCanonical := namedRef.(reference.Canonical); isCanonical {
+		return "", errors.New("image uses digest as version, cannot update tag ")
+	}
+
+	// update the image tag with tagName
+	namedTagged, err := reference.WithTag(namedRef, tagName)
+	if err != nil {
+		return "", errors.Wrap(err, "failed to update image tag")
+	}
+
+	return reference.FamiliarString(reference.TagNameOnly(namedTagged)), nil
+}
+
 // GetMachinesForCluster returns a list of machines associated with the cluster.
 func GetMachinesForCluster(ctx context.Context, c client.Client, cluster *clusterv1.Cluster) (*clusterv1.MachineList, error) {
 	var machines clusterv1.MachineList