[YUNIKORN-2631] Support canonical labels for queue/applicationId in Admission Controller

.golangci.yml

@@ -35,6 +35,8 @@ linters-settings:
     local-prefixes: github.com/apache/yunikorn
   govet:
     check-shadowing: true
+  goconst:
+    min-occurrences: 5
   funlen:
     lines: 120
     statements: 80

pkg/admission/admission_controller.go

@@ -322,7 +322,7 @@ func (c *AdmissionController) processPodUpdate(req *admissionv1.AdmissionRequest
 
 func (c *AdmissionController) shouldProcessAdmissionReview(namespace string, labels map[string]string) bool {
 	if c.shouldProcessNamespace(namespace) &&
-		(labels[constants.LabelApplicationID] != "" || labels[constants.SparkLabelAppID] != "" || c.shouldLabelNamespace(namespace)) {
+		(labels[constants.CanonicalLabelApplicationID] != "" || labels[constants.LabelApplicationID] != "" || labels[constants.SparkLabelAppID] != "" || c.shouldLabelNamespace(namespace)) {
 		return true
 	}
 

pkg/admission/admission_controller_test.go

@@ -81,9 +81,11 @@ func TestUpdateLabels(t *testing.T) {
 	assert.Equal(t, patch[0].Op, "add")
 	assert.Equal(t, patch[0].Path, "/metadata/labels")
 	if updatedMap, ok := patch[0].Value.(map[string]string); ok {
-		assert.Equal(t, len(updatedMap), 3)
+		assert.Equal(t, len(updatedMap), 5)
 		assert.Equal(t, updatedMap["random"], "random")
+		assert.Equal(t, updatedMap[constants.CanonicalLabelQueueName], "root.default")
 		assert.Equal(t, updatedMap[constants.LabelQueueName], "root.default")
+		assert.Equal(t, strings.HasPrefix(updatedMap[constants.CanonicalLabelApplicationID], constants.AutoGenAppPrefix), true)
 		assert.Equal(t, strings.HasPrefix(updatedMap[constants.LabelApplicationID], constants.AutoGenAppPrefix), true)
 	} else {
 		t.Fatal("patch info content is not as expected")
@@ -104,8 +106,8 @@ func TestUpdateLabels(t *testing.T) {
 			UID:             "7f5fd6c5d5",
 			ResourceVersion: "10654",
 			Labels: map[string]string{
-				"random":                     "random",
-				constants.LabelApplicationID: "app-0001",
+				"random":                              "random",
+				constants.CanonicalLabelApplicationID: "app-0001",
 			},
 		},
 		Spec:   v1.PodSpec{},
@@ -117,9 +119,11 @@ func TestUpdateLabels(t *testing.T) {
 	assert.Equal(t, patch[0].Op, "add")
 	assert.Equal(t, patch[0].Path, "/metadata/labels")
 	if updatedMap, ok := patch[0].Value.(map[string]string); ok {
-		assert.Equal(t, len(updatedMap), 3)
+		assert.Equal(t, len(updatedMap), 5)
 		assert.Equal(t, updatedMap["random"], "random")
+		assert.Equal(t, updatedMap[constants.CanonicalLabelQueueName], "root.default")
 		assert.Equal(t, updatedMap[constants.LabelQueueName], "root.default")
+		assert.Equal(t, updatedMap[constants.CanonicalLabelApplicationID], "app-0001")
 		assert.Equal(t, updatedMap[constants.LabelApplicationID], "app-0001")
 	} else {
 		t.Fatal("patch info content is not as expected")
@@ -140,8 +144,8 @@ func TestUpdateLabels(t *testing.T) {
 			UID:             "7f5fd6c5d5",
 			ResourceVersion: "10654",
 			Labels: map[string]string{
-				"random":                 "random",
-				constants.LabelQueueName: "root.abc",
+				"random":                          "random",
+				constants.CanonicalLabelQueueName: "root.abc",
 			},
 		},
 		Spec:   v1.PodSpec{},
@@ -154,9 +158,11 @@ func TestUpdateLabels(t *testing.T) {
 	assert.Equal(t, patch[0].Op, "add")
 	assert.Equal(t, patch[0].Path, "/metadata/labels")
 	if updatedMap, ok := patch[0].Value.(map[string]string); ok {
-		assert.Equal(t, len(updatedMap), 3)
+		assert.Equal(t, len(updatedMap), 5)
 		assert.Equal(t, updatedMap["random"], "random")
+		assert.Equal(t, updatedMap[constants.CanonicalLabelQueueName], "root.abc")
 		assert.Equal(t, updatedMap[constants.LabelQueueName], "root.abc")
+		assert.Equal(t, strings.HasPrefix(updatedMap[constants.CanonicalLabelApplicationID], constants.AutoGenAppPrefix), true)
 		assert.Equal(t, strings.HasPrefix(updatedMap[constants.LabelApplicationID], constants.AutoGenAppPrefix), true)
 	} else {
 		t.Fatal("patch info content is not as expected")
@@ -186,8 +192,10 @@ func TestUpdateLabels(t *testing.T) {
 	assert.Equal(t, patch[0].Op, "add")
 	assert.Equal(t, patch[0].Path, "/metadata/labels")
 	if updatedMap, ok := patch[0].Value.(map[string]string); ok {
-		assert.Equal(t, len(updatedMap), 2)
+		assert.Equal(t, len(updatedMap), 4)
+		assert.Equal(t, updatedMap[constants.CanonicalLabelQueueName], "root.default")
 		assert.Equal(t, updatedMap[constants.LabelQueueName], "root.default")
+		assert.Equal(t, strings.HasPrefix(updatedMap[constants.CanonicalLabelApplicationID], constants.AutoGenAppPrefix), true)
 		assert.Equal(t, strings.HasPrefix(updatedMap[constants.LabelApplicationID], constants.AutoGenAppPrefix), true)
 	} else {
 		t.Fatal("patch info content is not as expected")
@@ -214,8 +222,10 @@ func TestUpdateLabels(t *testing.T) {
 	assert.Equal(t, patch[0].Op, "add")
 	assert.Equal(t, patch[0].Path, "/metadata/labels")
 	if updatedMap, ok := patch[0].Value.(map[string]string); ok {
-		assert.Equal(t, len(updatedMap), 2)
+		assert.Equal(t, len(updatedMap), 4)
+		assert.Equal(t, updatedMap[constants.CanonicalLabelQueueName], "root.default")
 		assert.Equal(t, updatedMap[constants.LabelQueueName], "root.default")
+		assert.Equal(t, strings.HasPrefix(updatedMap[constants.CanonicalLabelApplicationID], constants.AutoGenAppPrefix), true)
 		assert.Equal(t, strings.HasPrefix(updatedMap[constants.LabelApplicationID], constants.AutoGenAppPrefix), true)
 	} else {
 		t.Fatal("patch info content is not as expected")
@@ -240,8 +250,10 @@ func TestUpdateLabels(t *testing.T) {
 	assert.Equal(t, patch[0].Op, "add")
 	assert.Equal(t, patch[0].Path, "/metadata/labels")
 	if updatedMap, ok := patch[0].Value.(map[string]string); ok {
-		assert.Equal(t, len(updatedMap), 2)
+		assert.Equal(t, len(updatedMap), 4)
+		assert.Equal(t, updatedMap[constants.CanonicalLabelQueueName], "root.default")
 		assert.Equal(t, updatedMap[constants.LabelQueueName], "root.default")
+		assert.Equal(t, strings.HasPrefix(updatedMap[constants.CanonicalLabelApplicationID], constants.AutoGenAppPrefix), true)
 		assert.Equal(t, strings.HasPrefix(updatedMap[constants.LabelApplicationID], constants.AutoGenAppPrefix), true)
 	} else {
 		t.Fatal("patch info content is not as expected")

pkg/admission/conf/am_conf.go

@@ -117,8 +117,13 @@ func NewAdmissionControllerConf(configMaps []*v1.ConfigMap) *AdmissionController
 	return acc
 }
 
-func (acc *AdmissionControllerConf) RegisterHandlers(configMaps informersv1.ConfigMapInformer) {
-	configMaps.Informer().AddEventHandler(&configMapUpdateHandler{conf: acc})
+func (acc *AdmissionControllerConf) RegisterHandlers(configMaps informersv1.ConfigMapInformer) error {
+	_, err := configMaps.Informer().AddEventHandler(&configMapUpdateHandler{conf: acc})
+	if err != nil {
+		return fmt.Errorf("failed to create register handlers: %w", err)
+	}
+
+	return nil
 }
 
 func (acc *AdmissionControllerConf) GetNamespace() string {

pkg/admission/namespace_cache.go

@@ -19,6 +19,8 @@
 package admission
 
 import (
+	"fmt"
+
 	v1 "k8s.io/api/core/v1"
 	informersv1 "k8s.io/client-go/informers/core/v1"
 	"k8s.io/client-go/tools/cache"
@@ -53,14 +55,17 @@ type nsFlags struct {
 }
 
 // NewNamespaceCache creates a new cache and registers the handler for the cache with the Informer.
-func NewNamespaceCache(namespaces informersv1.NamespaceInformer) *NamespaceCache {
+func NewNamespaceCache(namespaces informersv1.NamespaceInformer) (*NamespaceCache, error) {
 	nsc := &NamespaceCache{
 		nameSpaces: make(map[string]nsFlags),
 	}
 	if namespaces != nil {
-		namespaces.Informer().AddEventHandler(&namespaceUpdateHandler{cache: nsc})
+		_, err := namespaces.Informer().AddEventHandler(&namespaceUpdateHandler{cache: nsc})
+		if err != nil {
+			return nil, fmt.Errorf("failed to create namespace cache: %w", err)
+		}
 	}
-	return nsc
+	return nsc, nil
 }
 
 // enableYuniKorn returns the value for the enableYuniKorn flag (tri-state UNSET, TRUE or FALSE) for the namespace.

pkg/admission/namespace_cache_test.go

@@ -35,7 +35,8 @@ import (
 const testNS = "test-ns"
 
 func TestFlags(t *testing.T) {
-	cache := NewNamespaceCache(nil)
+	cache, nsErr := NewNamespaceCache(nil)
+	assert.NilError(t, nsErr)
 	cache.nameSpaces["notset"] = nsFlags{
 		enableYuniKorn: UNSET,
 		generateAppID:  UNSET,
@@ -69,7 +70,8 @@ func TestNamespaceHandlers(t *testing.T) {
 	kubeClient := client.NewKubeClientMock(false)
 
 	informers := NewInformers(kubeClient, "default")
-	cache := NewNamespaceCache(informers.Namespace)
+	cache, nsErr := NewNamespaceCache(informers.Namespace)
+	assert.NilError(t, nsErr)
 	informers.Start()
 	defer informers.Stop()
 

pkg/admission/priority_class_cache.go

@@ -19,6 +19,8 @@
 package admission
 
 import (
+	"fmt"
+
 	schedulingv1 "k8s.io/api/scheduling/v1"
 	informersv1 "k8s.io/client-go/informers/scheduling/v1"
 	"k8s.io/client-go/tools/cache"
@@ -36,14 +38,17 @@ type PriorityClassCache struct {
 }
 
 // NewPriorityClassCache creates a new cache and registers the handler for the cache with the Informer.
-func NewPriorityClassCache(priorityClasses informersv1.PriorityClassInformer) *PriorityClassCache {
+func NewPriorityClassCache(priorityClasses informersv1.PriorityClassInformer) (*PriorityClassCache, error) {
 	pcc := &PriorityClassCache{
 		priorityClasses: make(map[string]bool),
 	}
 	if priorityClasses != nil {
-		priorityClasses.Informer().AddEventHandler(&priorityClassUpdateHandler{cache: pcc})
+		_, err := priorityClasses.Informer().AddEventHandler(&priorityClassUpdateHandler{cache: pcc})
+		if err != nil {
+			return nil, fmt.Errorf("failed to create a new cache and register the handler: %w", err)
+		}
 	}
-	return pcc
+	return pcc, nil
 }
 
 // isPreemptSelfAllowed returns the preemption value. Only returns false if configured.

pkg/admission/priority_class_cache_test.go

@@ -35,7 +35,8 @@ import (
 const testPC = "test-pc"
 
 func TestIsPreemptSelfAllowed(t *testing.T) {
-	cache := NewPriorityClassCache(nil)
+	cache, pcErr := NewPriorityClassCache(nil)
+	assert.NilError(t, pcErr)
 	cache.priorityClasses["yes"] = true
 	cache.priorityClasses["no"] = false
 
@@ -49,7 +50,8 @@ func TestPriorityClassHandlers(t *testing.T) {
 	kubeClient := client.NewKubeClientMock(false)
 
 	informers := NewInformers(kubeClient, "default")
-	cache := NewPriorityClassCache(informers.PriorityClass)
+	cache, pcErr := NewPriorityClassCache(informers.PriorityClass)
+	assert.NilError(t, pcErr)
 	informers.Start()
 	defer informers.Stop()
 

pkg/admission/util.go

@@ -35,30 +35,43 @@ func updatePodLabel(pod *v1.Pod, namespace string, generateUniqueAppIds bool, de
 		result[k] = v
 	}
 
+	canonicalAppID := utils.GetPodLabelValue(pod, constants.CanonicalLabelApplicationID)
 	sparkAppID := utils.GetPodLabelValue(pod, constants.SparkLabelAppID)
 	labelAppID := utils.GetPodLabelValue(pod, constants.LabelApplicationID)
 	annotationAppID := utils.GetPodAnnotationValue(pod, constants.AnnotationApplicationID)
-	if sparkAppID == "" && labelAppID == "" && annotationAppID == "" {
+	if canonicalAppID == "" && sparkAppID == "" && labelAppID == "" && annotationAppID == "" {
 		// if app id not exist, generate one
 		// for each namespace, we group unnamed pods to one single app - if GenerateUniqueAppId is not set
 		// if GenerateUniqueAppId:
 		//		application ID convention: ${NAMESPACE}-${POD_UID}
 		// else
 		// 		application ID convention: ${AUTO_GEN_PREFIX}-${NAMESPACE}-${AUTO_GEN_SUFFIX}
 		generatedID := utils.GenerateApplicationID(namespace, generateUniqueAppIds, string(pod.UID))
+
+		result[constants.CanonicalLabelApplicationID] = generatedID
+		// Deprecated: After 1.7.0, admission controller will only add canonical label if application ID was not set
 		result[constants.LabelApplicationID] = generatedID
+	} else if canonicalAppID != "" {
+		// Deprecated: Added in 1.6.0 for backward compatibility, in case the prior shim version can't handle canonical label
+		result[constants.LabelApplicationID] = canonicalAppID
 	}
 
+	canonicalQueueName := utils.GetPodLabelValue(pod, constants.CanonicalLabelQueueName)
 	labelQueueName := utils.GetPodLabelValue(pod, constants.LabelQueueName)
 	annotationQueueName := utils.GetPodAnnotationValue(pod, constants.AnnotationQueueName)
-	if labelQueueName == "" && annotationQueueName == "" {
+	if canonicalQueueName == "" && labelQueueName == "" && annotationQueueName == "" {
 		// if queueName not exist, generate one
 		// if defaultQueueName is "", skip adding default queue name to the pod labels
 		if defaultQueueName != "" {
 			// for undefined configuration, am_conf will add 'root.default' to retain existing behavior
 			// if a custom name is configured for default queue, it will be used instead of root.default
+			result[constants.CanonicalLabelQueueName] = defaultQueueName
+			// Deprecated: After 1.7.0, admission controller will only add canonical label if queue was not set
 			result[constants.LabelQueueName] = defaultQueueName
 		}
+	} else if canonicalQueueName != "" {
+		// Deprecated: Added in 1.6.0 for backward compatibility, in case the prior shim version can't handle canonical label
+		result[constants.LabelQueueName] = canonicalQueueName
 	}
 
 	return result