Add the ability to override CAPI group via env variable and discover API version.

This change adds detection for an environment variable to specify the group for the clusterapi resources. If the environment
variable `CAPI_GROUP` is specified, then it will
be used instead of the default.
This also decouples the API group from the version and let the latter to be discovered dynamically.

Author: elmiko

cluster-autoscaler/cloudprovider/clusterapi/clusterapi_controller.go

@@ -19,12 +19,14 @@ package clusterapi
 import (
 	"context"
 	"fmt"
+	"os"
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/dynamic/dynamicinformer"
 	"k8s.io/client-go/informers"
@@ -38,7 +40,9 @@ import (
 const (
 	machineProviderIDIndex = "machineProviderIDIndex"
 	nodeProviderIDIndex    = "nodeProviderIDIndex"
-	defaultMachineAPI      = "v1alpha2.cluster.x-k8s.io"
+	defaultCAPIGroup       = "cluster.x-k8s.io"
+	// CAPIGroupEnvVar contains the environment variable name which allows overriding defaultCAPIGroup.
+	CAPIGroupEnvVar = "CAPI_GROUP"
 )
 
 // machineController watches for Nodes, Machines, MachineSets and
@@ -271,37 +275,58 @@ func (c *machineController) machinesInMachineSet(machineSet *MachineSet) ([]*Mac
 	return result, nil
 }
 
+// getCAPIGroup returns a string that specifies the group for the API.
+// It will return either the value from the
+// CAPI_GROUP environment variable, or the default value i.e cluster.x-k8s.io.
+func getCAPIGroup() string {
+	g := os.Getenv(CAPIGroupEnvVar)
+	if g == "" {
+		g = defaultCAPIGroup
+	}
+	klog.V(4).Infof("Using API Group %q", g)
+	return g
+}
+
 // newMachineController constructs a controller that watches Nodes,
 // Machines and MachineSet as they are added, updated and deleted on
 // the cluster.
 func newMachineController(
 	dynamicclient dynamic.Interface,
 	kubeclient kubeclient.Interface,
+	discoveryclient discovery.DiscoveryInterface,
 ) (*machineController, error) {
 	kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeclient, 0)
 	informerFactory := dynamicinformer.NewFilteredDynamicSharedInformerFactory(dynamicclient, 0, metav1.NamespaceAll, nil)
 
-	// TODO(alberto): let environment variable to override defaultMachineAPI
-	machineDeploymentResource, _ := schema.ParseResourceArg(fmt.Sprintf("machinedeployments.%v", defaultMachineAPI))
+	CAPIGroup := getCAPIGroup()
+	CAPIVersion, err := getAPIGroupPreferredVersion(discoveryclient, CAPIGroup)
+	if err != nil {
+		panic("CAPIVersion")
+	}
+	klog.Infof("Using version %q for API group %q", CAPIVersion, CAPIGroup)
+
+	machineDeploymentResource, _ := schema.ParseResourceArg(fmt.Sprintf("machinedeployments.%v.%v", CAPIVersion, CAPIGroup))
+	if machineDeploymentResource == nil {
+		panic("MachineDeployment")
+	}
 
-	machineSetResource, _ := schema.ParseResourceArg(fmt.Sprintf("machinesets.%v", defaultMachineAPI))
+	machineSetResource, _ := schema.ParseResourceArg(fmt.Sprintf("machinesets.%v.%v", CAPIVersion, CAPIGroup))
 	if machineSetResource == nil {
 		panic("MachineSetResource")
 	}
 
-	machineResource, _ := schema.ParseResourceArg(fmt.Sprintf("machines.%v", defaultMachineAPI))
+	machineResource, _ := schema.ParseResourceArg(fmt.Sprintf("machines.%v.%v", CAPIVersion, CAPIGroup))
 	if machineResource == nil {
 		panic("machineResource")
 	}
+
 	machineInformer := informerFactory.ForResource(*machineResource)
 	machineSetInformer := informerFactory.ForResource(*machineSetResource)
-	var machineDeploymentInformer informers.GenericInformer
-
-	machineDeploymentInformer = informerFactory.ForResource(*machineDeploymentResource)
-	machineDeploymentInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{})
+	machineDeploymentInformer := informerFactory.ForResource(*machineDeploymentResource)
 
 	machineInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{})
 	machineSetInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{})
+	machineDeploymentInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{})
 
 	nodeInformer := kubeInformerFactory.Core().V1().Nodes().Informer()
 	nodeInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{})
@@ -332,6 +357,21 @@ func newMachineController(
 	}, nil
 }
 
+func getAPIGroupPreferredVersion(client discovery.DiscoveryInterface, APIGroup string) (string, error) {
+	groupList, err := client.ServerGroups()
+	if err != nil {
+		return "", fmt.Errorf("failed to get ServerGroups: %v", err)
+	}
+
+	for _, group := range groupList.Groups {
+		if group.Name == APIGroup {
+			return group.PreferredVersion.Version, nil
+		}
+	}
+
+	return "", fmt.Errorf("failed to find API group %q", APIGroup)
+}
+
 func (c *machineController) machineSetProviderIDs(machineSet *MachineSet) ([]string, error) {
 	machines, err := c.machinesInMachineSet(machineSet)
 	if err != nil {

cluster-autoscaler/cloudprovider/clusterapi/clusterapi_controller_test.go

@@ -18,6 +18,7 @@ package clusterapi
 
 import (
 	"fmt"
+	"os"
 	"path"
 	"reflect"
 	"sort"
@@ -29,8 +30,10 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
+	fakediscovery "k8s.io/client-go/discovery/fake"
 	fakedynamic "k8s.io/client-go/dynamic/fake"
 	fakekube "k8s.io/client-go/kubernetes/fake"
+	clientgotesting "k8s.io/client-go/testing"
 	"k8s.io/utils/pointer"
 )
 
@@ -53,6 +56,8 @@ type testSpec struct {
 	rootIsMachineDeployment bool
 }
 
+const customCAPIGroup = "custom.x-k8s.io"
+
 func mustCreateTestController(t *testing.T, testConfigs ...*testConfig) (*machineController, testControllerShutdownFunc) {
 	t.Helper()
 
@@ -76,7 +81,19 @@ func mustCreateTestController(t *testing.T, testConfigs ...*testConfig) (*machin
 
 	kubeclientSet := fakekube.NewSimpleClientset(nodeObjects...)
 	dynamicClientset := fakedynamic.NewSimpleDynamicClient(runtime.NewScheme(), machineObjects...)
-	controller, err := newMachineController(dynamicClientset, kubeclientSet)
+	discoveryClient := &fakediscovery.FakeDiscovery{
+		Fake: &clientgotesting.Fake{
+			Resources: []*v1.APIResourceList{
+				{
+					GroupVersion: fmt.Sprintf("%s/v1beta1", customCAPIGroup),
+				},
+				{
+					GroupVersion: fmt.Sprintf("%s/v1alpha3", defaultCAPIGroup),
+				},
+			},
+		},
+	}
+	controller, err := newMachineController(dynamicClientset, kubeclientSet, discoveryClient)
 	if err != nil {
 		t.Fatal("failed to create test controller")
 	}
@@ -136,7 +153,7 @@ func createTestConfigs(specs ...testSpec) []*testConfig {
 
 		config.machineSet = &MachineSet{
 			TypeMeta: v1.TypeMeta{
-				APIVersion: "cluster.x-k8s.io/v1alpha2",
+				APIVersion: fmt.Sprintf("%s/v1alpha3", defaultCAPIGroup),
 				Kind:       "MachineSet",
 			},
 			ObjectMeta: v1.ObjectMeta{
@@ -152,7 +169,7 @@ func createTestConfigs(specs ...testSpec) []*testConfig {
 		} else {
 			config.machineDeployment = &MachineDeployment{
 				TypeMeta: v1.TypeMeta{
-					APIVersion: "cluster.x-k8s.io/v1alpha2",
+					APIVersion: fmt.Sprintf("%s/v1alpha3", defaultCAPIGroup),
 					Kind:       "MachineDeployment",
 				},
 				ObjectMeta: v1.ObjectMeta{
@@ -211,7 +228,7 @@ func makeLinkedNodeAndMachine(i int, namespace string, owner v1.OwnerReference)
 
 	machine := &Machine{
 		TypeMeta: v1.TypeMeta{
-			APIVersion: "cluster.x-k8s.io/v1alpha2",
+			APIVersion: fmt.Sprintf("%s/v1alpha3", defaultCAPIGroup),
 			Kind:       "Machine",
 		},
 		ObjectMeta: v1.ObjectMeta{
@@ -991,3 +1008,121 @@ func TestControllerMachineSetNodeNamesUsingStatusNodeRefName(t *testing.T) {
 		}
 	}
 }
+
+func TestControllerGetAPIVersionGroup(t *testing.T) {
+	expected := "mygroup"
+	if err := os.Setenv(CAPIGroupEnvVar, expected); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	observed := getCAPIGroup()
+	if observed != expected {
+		t.Fatalf("Wrong Version Group detected, expected %q, got %q", expected, observed)
+	}
+
+	expected = defaultCAPIGroup
+	if err := os.Setenv(CAPIGroupEnvVar, ""); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	observed = getCAPIGroup()
+	if observed != expected {
+		t.Fatalf("Wrong Version Group detected, expected %q, got %q", expected, observed)
+	}
+}
+
+func TestControllerGetAPIVersionGroupWithMachineDeployments(t *testing.T) {
+	testConfig := createMachineDeploymentTestConfig(testNamespace, 1, map[string]string{
+		nodeGroupMinSizeAnnotationKey: "1",
+		nodeGroupMaxSizeAnnotationKey: "1",
+	})
+	if err := os.Setenv(CAPIGroupEnvVar, customCAPIGroup); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	testConfig.machineDeployment.TypeMeta.APIVersion = fmt.Sprintf("%s/v1beta1", customCAPIGroup)
+	testConfig.machineSet.TypeMeta.APIVersion = fmt.Sprintf("%s/v1beta1", customCAPIGroup)
+	for _, machine := range testConfig.machines {
+		machine.TypeMeta.APIVersion = fmt.Sprintf("%s/v1beta1", customCAPIGroup)
+	}
+	controller, stop := mustCreateTestController(t, testConfig)
+	defer stop()
+
+	machineDeployments, err := controller.listMachineDeployments(testNamespace, labels.Everything())
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if l := len(machineDeployments); l != 1 {
+		t.Fatalf("Incorrect number of MachineDeployments, expected 1, got %d", l)
+	}
+
+	machineSets, err := controller.listMachineSets(testNamespace, labels.Everything())
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if l := len(machineSets); l != 1 {
+		t.Fatalf("Incorrect number of MachineSets, expected 1, got %d", l)
+	}
+
+	machines, err := controller.listMachines(testNamespace, labels.Everything())
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if l := len(machines); l != 1 {
+		t.Fatalf("Incorrect number of Machines, expected 1, got %d", l)
+	}
+
+	if err := os.Unsetenv(CAPIGroupEnvVar); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
+func TestGetAPIGroupPreferredVersion(t *testing.T) {
+	testCases := []struct {
+		description      string
+		APIGroup         string
+		preferredVersion string
+		error            bool
+	}{
+		{
+			description:      "find version for default API group",
+			APIGroup:         defaultCAPIGroup,
+			preferredVersion: "v1alpha3",
+			error:            false,
+		},
+		{
+			description:      "find version for another API group",
+			APIGroup:         customCAPIGroup,
+			preferredVersion: "v1beta1",
+			error:            false,
+		},
+		{
+			description:      "API group does not exist",
+			APIGroup:         "does.not.exist",
+			preferredVersion: "",
+			error:            true,
+		},
+	}
+
+	discoveryClient := &fakediscovery.FakeDiscovery{
+		Fake: &clientgotesting.Fake{
+			Resources: []*v1.APIResourceList{
+				{
+					GroupVersion: fmt.Sprintf("%s/v1beta1", customCAPIGroup),
+				},
+				{
+					GroupVersion: fmt.Sprintf("%s/v1alpha3", defaultCAPIGroup),
+				},
+			},
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.description, func(t *testing.T) {
+			version, err := getAPIGroupPreferredVersion(discoveryClient, tc.APIGroup)
+			if (err != nil) != tc.error {
+				t.Errorf("expected to have error: %t. Had an error: %t", tc.error, err != nil)
+			}
+			if version != tc.preferredVersion {
+				t.Errorf("expected %v, got: %v", tc.preferredVersion, version)
+			}
+		})
+	}
+}

cluster-autoscaler/cloudprovider/clusterapi/clusterapi_provider.go

@@ -24,6 +24,7 @@ import (
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
 	"k8s.io/autoscaler/cluster-autoscaler/config"
 	"k8s.io/autoscaler/cluster-autoscaler/utils/errors"
+	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/clientcmd"
@@ -146,12 +147,17 @@ func BuildClusterAPI(opts config.AutoscalingOptions, do cloudprovider.NodeGroupD
 		klog.Fatalf("could not generate dynamic client for config")
 	}
 
-	kubeclient, err := kubernetes.NewForConfig(externalConfig)
+	kubeClient, err := kubernetes.NewForConfig(externalConfig)
 	if err != nil {
 		klog.Fatalf("create kube clientset failed: %v", err)
 	}
 
-	controller, err := newMachineController(dc, kubeclient)
+	discoveryClient, err := discovery.NewDiscoveryClientForConfig(externalConfig)
+	if err != nil {
+		klog.Fatalf("create discovery client failed: %v", err)
+	}
+
+	controller, err := newMachineController(dc, kubeClient, discoveryClient)
 	if err != nil {
 		klog.Fatal(err)
 	}