Merge pull request #1501 from zqzten/kubernetes_version

feat: introduce Kubernetes version to status of KubeFedCluster

.github/workflows/build-and-test.yml

@@ -11,7 +11,7 @@ jobs:
 
       - uses: actions/setup-go@v2
         with:
-          go-version: '^1.16.6'
+          go-version: '~1.16'
 
       - name: Run tests
         run: |

.github/workflows/test-and-push.yml

@@ -16,7 +16,7 @@ jobs:
 
       - uses: actions/setup-go@v2
         with:
-          go-version: '^1.16.6'
+          go-version: '~1.16'
 
       - name: Run tests
         run: DOWNLOAD_BINARIES=y bash -x ./scripts/pre-commit.sh

charts/kubefed/charts/controllermanager/crds/crds.yaml

@@ -457,6 +457,8 @@ spec:
     kind: KubeFedCluster
     listKind: KubeFedClusterList
     plural: kubefedclusters
+    shortNames:
+    - kfc
     singular: kubefedcluster
   scope: Namespaced
   versions:
@@ -467,6 +469,9 @@ spec:
     - jsonPath: .status.conditions[?(@.type=='Ready')].status
       name: ready
       type: string
+    - jsonPath: .status.kubernetesVersion
+      name: kubernetes-version
+      type: string
     name: v1beta1
     schema:
       openAPIV3Schema:
@@ -559,6 +564,10 @@ spec:
                   - type
                   type: object
                 type: array
+              kubernetesVersion:
+                description: KubernetesVersion is the Kubernetes git version of the
+                  cluster.
+                type: string
               region:
                 description: Region is the name of the region in which all of the
                   nodes in the cluster exist.  e.g. 'us-east1'.

cmd/controller-manager/app/leaderelection/leaderelection.go

@@ -36,8 +36,9 @@ import (
 
 func NewKubeFedLeaderElector(opts *options.Options, fnStartControllers func(*options.Options, <-chan struct{})) (*leaderelection.LeaderElector, error) {
 	const component = "kubefed-controller-manager"
-	restclient.AddUserAgent(opts.Config.KubeConfig, "kubefed-leader-election")
-	leaderElectionClient := kubeclient.NewForConfigOrDie(opts.Config.KubeConfig)
+	kubeConfig := restclient.CopyConfig(opts.Config.KubeConfig)
+	restclient.AddUserAgent(kubeConfig, "kubefed-leader-election")
+	leaderElectionClient := kubeclient.NewForConfigOrDie(kubeConfig)
 
 	hostname, err := os.Hostname()
 	if err != nil {

docs/cluster-registration.md

@@ -34,12 +34,14 @@ Check the status of the joined clusters by using the following command.
 ```bash
 kubectl -n kube-federation-system get kubefedclusters
 
-NAME       READY   AGE
-cluster1   True    1m
-cluster2   True    1m
+NAME       AGE   READY   KUBERNETES-VERSION
+cluster1   1m    True    v1.21.2
+cluster2   1m    True    v1.22.0
 
 ```
 
+The Kubernetes version is checked periodically along with the cluster health check so that it would be automatically updated within the cluster health check period after a Kubernetes upgrade/downgrade of the cluster.
+
 # Joining kind clusters on MacOS
 
 A Kubernetes cluster deployed with [kind](https://sigs.k8s.io/kind) on Docker

pkg/apis/core/v1beta1/kubefedcluster_types.go

@@ -70,6 +70,9 @@ type LocalSecretReference struct {
 type KubeFedClusterStatus struct {
 	// Conditions is an array of current cluster conditions.
 	Conditions []ClusterCondition `json:"conditions"`
+	// KubernetesVersion is the Kubernetes git version of the cluster.
+	// +optional
+	KubernetesVersion string `json:"kubernetesVersion,omitempty"`
 	// Zones are the names of availability zones in which the nodes of the cluster exist, e.g. 'us-east1-a'.
 	// +optional
 	Zones []string `json:"zones,omitempty"`
@@ -81,7 +84,8 @@ type KubeFedClusterStatus struct {
 // +kubebuilder:object:root=true
 // +kubebuilder:printcolumn:name=age,type=date,JSONPath=.metadata.creationTimestamp
 // +kubebuilder:printcolumn:name=ready,type=string,JSONPath=.status.conditions[?(@.type=='Ready')].status
-// +kubebuilder:resource:path=kubefedclusters
+// +kubebuilder:printcolumn:name=kubernetes-version,type=string,JSONPath=.status.kubernetesVersion
+// +kubebuilder:resource:path=kubefedclusters,shortName=kfc
 // +kubebuilder:subresource:status
 
 // KubeFedCluster configures KubeFed to be aware of a Kubernetes

pkg/controller/kubefedcluster/clusterclient.go

@@ -18,6 +18,7 @@ package kubefedcluster
 
 import (
 	"context"
+	"fmt"
 	"strings"
 	"time"
 
@@ -79,8 +80,8 @@ func NewClusterClientSet(c *fedv1b1.KubeFedCluster, client generic.Client, fedNa
 	return &clusterClientSet, err
 }
 
-// GetClusterHealthStatus gets the kubernetes cluster health status by requesting "/healthz"
-func (c *ClusterClient) GetClusterHealthStatus() (*fedv1b1.KubeFedClusterStatus, error) {
+// GetClusterStatus gets the kubernetes cluster's health and version status
+func (c *ClusterClient) GetClusterStatus() (*fedv1b1.KubeFedClusterStatus, error) {
 	clusterStatus := fedv1b1.KubeFedClusterStatus{}
 	currentTime := metav1.Now()
 	clusterReady := ClusterReady
@@ -141,6 +142,8 @@ func (c *ClusterClient) GetClusterHealthStatus() (*fedv1b1.KubeFedClusterStatus,
 	body, err := c.kubeClient.DiscoveryClient.RESTClient().Get().AbsPath("/healthz").Do(context.Background()).Raw()
 	if err != nil {
 		runtime.HandleError(errors.Wrapf(err, "Failed to do cluster health check for cluster %q", c.clusterName))
+		msg := fmt.Sprintf("%s: %v", ClusterNotReachableMsg, err)
+		newClusterOfflineCondition.Message = &msg
 		clusterStatus.Conditions = append(clusterStatus.Conditions, newClusterOfflineCondition)
 		metrics.RegisterKubefedClusterTotal(metrics.ClusterOffline, c.clusterName)
 	} else {
@@ -150,6 +153,13 @@ func (c *ClusterClient) GetClusterHealthStatus() (*fedv1b1.KubeFedClusterStatus,
 		} else {
 			metrics.RegisterKubefedClusterTotal(metrics.ClusterReady, c.clusterName)
 			clusterStatus.Conditions = append(clusterStatus.Conditions, newClusterReadyCondition)
+
+			version, err := c.kubeClient.DiscoveryClient.ServerVersion()
+			if err != nil {
+				runtime.HandleError(errors.Wrapf(err, "Failed to get Kubernetes version of cluster %q", c.clusterName))
+			} else {
+				clusterStatus.KubernetesVersion = version.GitVersion
+			}
 		}
 	}
 

pkg/controller/kubefedcluster/controller.go

@@ -96,7 +96,8 @@ func StartClusterController(config *util.ControllerConfig, clusterHealthCheckCon
 func newClusterController(config *util.ControllerConfig, clusterHealthCheckConfig *util.ClusterHealthCheckConfig) (*ClusterController, error) {
 	kubeConfig := restclient.CopyConfig(config.KubeConfig)
 	kubeConfig.Timeout = clusterHealthCheckConfig.Timeout
-	client := genericclient.NewForConfigOrDieWithUserAgent(kubeConfig, "cluster-controller")
+	restclient.AddUserAgent(kubeConfig, "cluster-controller")
+	client := genericclient.NewForConfigOrDie(kubeConfig)
 
 	cc := &ClusterController{
 		client:                   client,
@@ -242,7 +243,7 @@ func (cc *ClusterController) updateIndividualClusterStatus(cluster *fedv1b1.Kube
 
 	clusterClient := storedData.clusterKubeClient
 
-	currentClusterStatus, err := clusterClient.GetClusterHealthStatus()
+	currentClusterStatus, err := clusterClient.GetClusterStatus()
 	if err != nil {
 		cc.RecordError(cluster, "RetrievingClusterHealthFailed", errors.Wrap(err, "Failed to retrieve health of the cluster"))
 		klog.Errorf("Failed to retrieve health of the cluster %s: %v", cluster.Name, err)
@@ -278,7 +279,7 @@ func thresholdAdjustedClusterStatus(clusterStatus *fedv1b1.KubeFedClusterStatus,
 	if storedData.resultRun < threshold {
 		// Success/Failure is below threshold - leave the probe state unchanged.
 		probeTime := clusterStatus.Conditions[0].LastProbeTime
-		clusterStatus = storedData.clusterStatus
+		clusterStatus.Conditions = storedData.clusterStatus.Conditions
 		setProbeTime(clusterStatus, probeTime)
 	} else if clusterStatusEqual(clusterStatus, storedData.clusterStatus) {
 		// preserve the last transition time

pkg/controller/schedulingpreference/controller.go

@@ -137,7 +137,7 @@ func newSchedulingPreferenceController(config *util.ControllerConfig, scheduling
 	s.clusterDeliverer = util.NewDelayingDeliverer()
 
 	s.store, s.controller, err = util.NewGenericInformer(
-		config.KubeConfig,
+		kubeConfig,
 		config.TargetNamespace,
 		s.scheduler.ObjectType(),
 		util.NoResyncPeriod,

pkg/controller/status/controller.go

@@ -25,6 +25,7 @@ import (
 	"time"
 
 	"k8s.io/apimachinery/pkg/util/wait"
+	restclient "k8s.io/client-go/rest"
 
 	"github.com/pkg/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -104,14 +105,16 @@ func newKubeFedStatusController(controllerConfig *util.ControllerConfig, typeCon
 		return nil, errors.Errorf("Status collection is not supported for %q", federatedAPIResource.Kind)
 	}
 	userAgent := fmt.Sprintf("%s-controller", strings.ToLower(statusAPIResource.Kind))
-	client := genericclient.NewForConfigOrDieWithUserAgent(controllerConfig.KubeConfig, userAgent)
+	kubeConfig := restclient.CopyConfig(controllerConfig.KubeConfig)
+	restclient.AddUserAgent(kubeConfig, userAgent)
+	client := genericclient.NewForConfigOrDie(kubeConfig)
 
-	federatedTypeClient, err := util.NewResourceClient(controllerConfig.KubeConfig, &federatedAPIResource)
+	federatedTypeClient, err := util.NewResourceClient(kubeConfig, &federatedAPIResource)
 	if err != nil {
 		return nil, err
 	}
 
-	statusClient, err := util.NewResourceClient(controllerConfig.KubeConfig, statusAPIResource)
+	statusClient, err := util.NewResourceClient(kubeConfig, statusAPIResource)
 	if err != nil {
 		return nil, err
 	}

pkg/controller/sync/controller.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/pkg/errors"
+	restclient "k8s.io/client-go/rest"
 
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -113,9 +114,10 @@ func newKubeFedSyncController(controllerConfig *util.ControllerConfig, typeConfi
 	federatedTypeAPIResource := typeConfig.GetFederatedType()
 	userAgent := fmt.Sprintf("%s-controller", strings.ToLower(federatedTypeAPIResource.Kind))
 
-	// Initialize non-dynamic clients first to avoid polluting config
-	client := genericclient.NewForConfigOrDieWithUserAgent(controllerConfig.KubeConfig, userAgent)
-	kubeClient := kubeclient.NewForConfigOrDie(controllerConfig.KubeConfig)
+	kubeConfig := restclient.CopyConfig(controllerConfig.KubeConfig)
+	restclient.AddUserAgent(kubeConfig, userAgent)
+	client := genericclient.NewForConfigOrDie(kubeConfig)
+	kubeClient := kubeclient.NewForConfigOrDie(kubeConfig)
 
 	broadcaster := record.NewBroadcaster()
 	broadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeClient.CoreV1().Events("")})
@@ -589,7 +591,9 @@ func (s *KubeFedSyncController) deleteFromClusters(fedResource FederatedResource
 	if len(remainingClusters) > 0 {
 		fedKind := fedResource.FederatedKind()
 		fedName := fedResource.FederatedName()
-		klog.V(2).Infof("Waiting for resources managed by %s %q to be removed from the following clusters: %s", fedKind, fedName, strings.Join(remainingClusters, ", "))
+		remainingClustersStr := strings.Join(remainingClusters, ", ")
+		klog.V(2).Infof("Waiting for resources managed by %s %q to be removed from the following clusters: %s", fedKind, fedName, remainingClustersStr)
+		fedResource.RecordEvent("WaitForRemovalInCluster", "Waiting for managed resources to be removed from the following clusters: %s", remainingClustersStr)
 		return true, nil
 	}
 	err = s.ensureRemovedOrUnmanaged(fedResource)

pkg/controller/sync/dispatch/retain.go

@@ -47,52 +47,52 @@ func RetainClusterFields(targetKind string, desiredObj, clusterObj, fedObj *unst
 
 func retainServiceFields(desiredObj, clusterObj *unstructured.Unstructured) error {
 	// healthCheckNodePort is allocated by APIServer and unchangeable, so it should be retained while updating
-	healthCheckNodePort, ok, err := unstructured.NestedInt64(clusterObj.Object, "spec", "healthCheckNodePort")
+	healthCheckNodePort, ok, err := unstructured.NestedInt64(clusterObj.Object, util.SpecField, util.HealthCheckNodePortField)
 	if err != nil {
 		return errors.Wrap(err, "Error retrieving healthCheckNodePort from service")
 	}
 	if ok && healthCheckNodePort > 0 {
-		if err = unstructured.SetNestedField(desiredObj.Object, healthCheckNodePort, "spec", "healthCheckNodePort"); err != nil {
+		if err = unstructured.SetNestedField(desiredObj.Object, healthCheckNodePort, util.SpecField, util.HealthCheckNodePortField); err != nil {
 			return errors.Wrap(err, "Error setting healthCheckNodePort for service")
 		}
 	}
 
 	// ClusterIP and NodePort are allocated to Service by cluster, so retain the same if any while updating
 
 	// Retain clusterip and clusterips
-	clusterIP, ok, err := unstructured.NestedString(clusterObj.Object, "spec", "clusterIP")
+	clusterIP, ok, err := unstructured.NestedString(clusterObj.Object, util.SpecField, util.ClusterIPField)
 	if err != nil {
 		return errors.Wrap(err, "Error retrieving clusterIP from cluster service")
 	}
 	// !ok could indicate that a cluster ip was not assigned
 	if ok && clusterIP != "" {
-		err := unstructured.SetNestedField(desiredObj.Object, clusterIP, "spec", "clusterIP")
+		err := unstructured.SetNestedField(desiredObj.Object, clusterIP, util.SpecField, util.ClusterIPField)
 		if err != nil {
 			return errors.Wrap(err, "Error setting clusterIP for service")
 		}
 	}
-	clusterIPs, ok, err := unstructured.NestedStringSlice(clusterObj.Object, "spec", "clusterIPs")
+	clusterIPs, ok, err := unstructured.NestedStringSlice(clusterObj.Object, util.SpecField, util.ClusterIPsField)
 	if err != nil {
 		return errors.Wrap(err, "Error retrieving clusterIPs from cluster service")
 	}
 	// !ok could indicate that cluster ips was not assigned
 	if ok && len(clusterIPs) > 0 {
-		err := unstructured.SetNestedStringSlice(desiredObj.Object, clusterIPs, "spec", "clusterIPs")
+		err := unstructured.SetNestedStringSlice(desiredObj.Object, clusterIPs, util.SpecField, util.ClusterIPsField)
 		if err != nil {
 			return errors.Wrap(err, "Error setting clusterIPs for service")
 		}
 	}
 
 	// Retain nodeports
-	clusterPorts, ok, err := unstructured.NestedSlice(clusterObj.Object, "spec", "ports")
+	clusterPorts, ok, err := unstructured.NestedSlice(clusterObj.Object, util.SpecField, util.PortsField)
 	if err != nil {
 		return errors.Wrap(err, "Error retrieving ports from cluster service")
 	}
 	if !ok {
 		return nil
 	}
 	var desiredPorts []interface{}
-	desiredPorts, ok, err = unstructured.NestedSlice(desiredObj.Object, "spec", "ports")
+	desiredPorts, ok, err = unstructured.NestedSlice(desiredObj.Object, util.SpecField, util.PortsField)
 	if err != nil {
 		return errors.Wrap(err, "Error retrieving ports from service")
 	}
@@ -112,7 +112,7 @@ func retainServiceFields(desiredObj, clusterObj *unstructured.Unstructured) erro
 			}
 		}
 	}
-	err = unstructured.SetNestedSlice(desiredObj.Object, desiredPorts, "spec", "ports")
+	err = unstructured.SetNestedSlice(desiredObj.Object, desiredPorts, util.SpecField, util.PortsField)
 	if err != nil {
 		return errors.Wrap(err, "Error setting ports for service")
 	}

pkg/controller/util/constants.go

@@ -41,6 +41,12 @@ const (
 	StatusField   = "status"
 	MetadataField = "metadata"
 
+	// Service fields
+	HealthCheckNodePortField = "healthCheckNodePort"
+	ClusterIPField           = "clusterIP"
+	ClusterIPsField          = "clusterIPs"
+	PortsField               = "ports"
+
 	// ServiceAccount fields
 	SecretsField = "secrets"
 

pkg/schedulingtypes/plugin.go

@@ -29,6 +29,7 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
+	restclient "k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
 
@@ -60,7 +61,9 @@ type Plugin struct {
 func NewPlugin(controllerConfig *util.ControllerConfig, eventHandlers SchedulerEventHandlers, typeConfig typeconfig.Interface, nsAPIResource *metav1.APIResource) (*Plugin, error) {
 	targetAPIResource := typeConfig.GetTargetType()
 	userAgent := fmt.Sprintf("%s-replica-scheduler", strings.ToLower(targetAPIResource.Kind))
-	client := genericclient.NewForConfigOrDieWithUserAgent(controllerConfig.KubeConfig, userAgent)
+	kubeConfig := restclient.CopyConfig(controllerConfig.KubeConfig)
+	restclient.AddUserAgent(kubeConfig, userAgent)
+	client := genericclient.NewForConfigOrDie(kubeConfig)
 
 	targetInformer, err := util.NewFederatedInformer(
 		controllerConfig,
@@ -84,13 +87,13 @@ func NewPlugin(controllerConfig *util.ControllerConfig, eventHandlers SchedulerE
 	kubeFedEventHandler := eventHandlers.KubeFedEventHandler
 
 	federatedTypeAPIResource := typeConfig.GetFederatedType()
-	p.federatedTypeClient, err = util.NewResourceClient(controllerConfig.KubeConfig, &federatedTypeAPIResource)
+	p.federatedTypeClient, err = util.NewResourceClient(kubeConfig, &federatedTypeAPIResource)
 	if err != nil {
 		return nil, err
 	}
 	p.federatedStore, p.federatedController = util.NewResourceInformer(p.federatedTypeClient, targetNamespace, &federatedTypeAPIResource, kubeFedEventHandler)
 
-	p.fedNsClient, err = util.NewResourceClient(controllerConfig.KubeConfig, nsAPIResource)
+	p.fedNsClient, err = util.NewResourceClient(kubeConfig, nsAPIResource)
 	if err != nil {
 		return nil, err
 	}