Merge pull request #224 from openshift-kni/revert-informer-mode-4.14

[release-4.14][manual] revert #223

apis/config/types.go

@@ -176,14 +176,6 @@ const (
 	CacheResyncOnlyExclusiveResources CacheResyncMethod = "OnlyExclusiveResources"
 )
 
-// CacheInformerMode is a "string" type
-type CacheInformerMode string
-
-const (
-	CacheInformerShared    CacheInformerMode = "Shared"
-	CacheInformerDedicated CacheInformerMode = "Dedicated"
-)
-
 // NodeResourceTopologyCache define configuration details for the NodeResourceTopology cache.
 type NodeResourceTopologyCache struct {
 	// ForeignPodsDetect sets how foreign pods should be handled.
@@ -200,11 +192,6 @@ type NodeResourceTopologyCache struct {
 	// Has no effect if caching is disabled (CacheResyncPeriod is zero) or if DiscardReservedNodes
 	// is enabled. "Autodetect" is the default, reads hint from NRT objects. Fallback is "All".
 	ResyncMethod *CacheResyncMethod
-	// InformerMode controls the channel the cache uses to get updates about pods.
-	// "Shared" uses the default settings; "Dedicated" creates a specific subscription which is
-	// guaranteed to best suit the cache needs, at cost of one extra connection.
-	// If unspecified, default is "Dedicated"
-	InformerMode *CacheInformerMode
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

apis/config/v1/defaults.go

@@ -89,8 +89,6 @@ var (
 
 	defaultResyncMethod = CacheResyncAutodetect
 
-	defaultInformerMode = CacheInformerDedicated
-
 	// Defaults for NetworkOverhead
 	// DefaultWeightsName contains the default costs to be used by networkAware plugins
 	DefaultWeightsName = "UserDefined"
@@ -202,9 +200,6 @@ func SetDefaults_NodeResourceTopologyMatchArgs(obj *NodeResourceTopologyMatchArg
 	if obj.Cache.ResyncMethod == nil {
 		obj.Cache.ResyncMethod = &defaultResyncMethod
 	}
-	if obj.Cache.InformerMode == nil {
-		obj.Cache.InformerMode = &defaultInformerMode
-	}
 }
 
 // SetDefaults_PreemptionTolerationArgs reuses SetDefaults_DefaultPreemptionArgs

apis/config/v1/defaults_test.go

@@ -205,7 +205,6 @@ func TestSchedulingDefaults(t *testing.T) {
 				Cache: &NodeResourceTopologyCache{
 					ForeignPodsDetect: &defaultForeignPodsDetect,
 					ResyncMethod:      &defaultResyncMethod,
-					InformerMode:      &defaultInformerMode,
 				},
 			},
 		},

apis/config/v1/types.go

@@ -174,14 +174,6 @@ const (
 	CacheResyncOnlyExclusiveResources CacheResyncMethod = "OnlyExclusiveResources"
 )
 
-// CacheInformerMode is a "string" type
-type CacheInformerMode string
-
-const (
-	CacheInformerShared    CacheInformerMode = "Shared"
-	CacheInformerDedicated CacheInformerMode = "Dedicated"
-)
-
 // NodeResourceTopologyCache define configuration details for the NodeResourceTopology cache.
 type NodeResourceTopologyCache struct {
 	// ForeignPodsDetect sets how foreign pods should be handled.
@@ -198,11 +190,6 @@ type NodeResourceTopologyCache struct {
 	// Has no effect if caching is disabled (CacheResyncPeriod is zero) or if DiscardReservedNodes
 	// is enabled. "Autodetect" is the default, reads hint from NRT objects. Fallback is "All".
 	ResyncMethod *CacheResyncMethod `json:"resyncMethod,omitempty"`
-	// InformerMode controls the channel the cache uses to get updates about pods.
-	// "Shared" uses the default settings; "Dedicated" creates a specific subscription which is
-	// guaranteed to best suit the cache needs, at cost of one extra connection.
-	// If unspecified, default is "Dedicated"
-	InformerMode *CacheInformerMode `json:"informerMode,omitempty"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

apis/config/v1beta3/defaults.go

@@ -89,8 +89,6 @@ var (
 
 	defaultResyncMethod = CacheResyncAutodetect
 
-	defaultInformerMode = CacheInformerDedicated
-
 	// Defaults for NetworkOverhead
 	// DefaultWeightsName contains the default costs to be used by networkAware plugins
 	DefaultWeightsName = "UserDefined"
@@ -202,9 +200,6 @@ func SetDefaults_NodeResourceTopologyMatchArgs(obj *NodeResourceTopologyMatchArg
 	if obj.Cache.ResyncMethod == nil {
 		obj.Cache.ResyncMethod = &defaultResyncMethod
 	}
-	if obj.Cache.InformerMode == nil {
-		obj.Cache.InformerMode = &defaultInformerMode
-	}
 }
 
 // SetDefaults_PreemptionTolerationArgs reuses SetDefaults_DefaultPreemptionArgs

apis/config/v1beta3/defaults_test.go

@@ -205,7 +205,6 @@ func TestSchedulingDefaults(t *testing.T) {
 				Cache: &NodeResourceTopologyCache{
 					ForeignPodsDetect: &defaultForeignPodsDetect,
 					ResyncMethod:      &defaultResyncMethod,
-					InformerMode:      &defaultInformerMode,
 				},
 			},
 		},

apis/config/v1beta3/types.go

@@ -174,14 +174,6 @@ const (
 	CacheResyncOnlyExclusiveResources CacheResyncMethod = "OnlyExclusiveResources"
 )
 
-// CacheInformerMode is a "string" type
-type CacheInformerMode string
-
-const (
-	CacheInformerShared    CacheInformerMode = "Shared"
-	CacheInformerDedicated CacheInformerMode = "Dedicated"
-)
-
 // NodeResourceTopologyCache define configuration details for the NodeResourceTopology cache.
 type NodeResourceTopologyCache struct {
 	// ForeignPodsDetect sets how foreign pods should be handled.
@@ -198,11 +190,6 @@ type NodeResourceTopologyCache struct {
 	// Has no effect if caching is disabled (CacheResyncPeriod is zero) or if DiscardReservedNodes
 	// is enabled. "Autodetect" is the default, reads hint from NRT objects. Fallback is "All".
 	ResyncMethod *CacheResyncMethod `json:"resyncMethod,omitempty"`
-	// InformerMode controls the channel the cache uses to get updates about pods.
-	// "Shared" uses the default settings; "Dedicated" creates a specific subscription which is
-	// guaranteed to best suit the cache needs, at cost of one extra connection.
-	// If unspecified, default is "Dedicated"
-	InformerMode *CacheInformerMode `json:"informerMode,omitempty"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

cmd/noderesourcetopology-plugin/main.go

@@ -36,6 +36,7 @@ import (
 	_ "sigs.k8s.io/scheduler-plugins/apis/config/scheme"
 
 	knifeatures "sigs.k8s.io/scheduler-plugins/pkg-kni/features"
+	kniinformer "sigs.k8s.io/scheduler-plugins/pkg-kni/podinformer"
 
 	"github.com/k8stopologyawareschedwg/podfingerprint"
 )
@@ -65,6 +66,8 @@ func main() {
 
 	rand.Seed(time.Now().UnixNano())
 
+	kniinformer.Setup()
+
 	// Register custom plugins to the scheduler framework.
 	// Later they can consist of scheduler profile(s) and hence
 	// used by various kinds of workloads.

pkg-kni/podinformer/podinformer.go

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2023 Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package podinformer
+
+import (
+	"context"
+	"os"
+	"strconv"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	coreinformers "k8s.io/client-go/informers/core/v1"
+	podlisterv1 "k8s.io/client-go/listers/core/v1"
+	"k8s.io/client-go/tools/cache"
+	k8scache "k8s.io/client-go/tools/cache"
+	"k8s.io/klog/v2"
+	"k8s.io/kubernetes/pkg/scheduler/framework"
+)
+
+const (
+	nrtInformerEnvVar string = "NRT_ENABLE_INFORMER"
+)
+
+var (
+	enabled bool
+)
+
+func IsEnabled() bool {
+	return enabled
+}
+
+func Setup() {
+	hasNRTInf, ok := os.LookupEnv(nrtInformerEnvVar)
+	if !ok || hasNRTInf == "" {
+		klog.InfoS("NRT specific informer disabled", "variableFound", ok, "valueGiven", hasNRTInf != "")
+		return
+	}
+	val, err := strconv.ParseBool(hasNRTInf)
+	if err != nil {
+		klog.Error(err, "NRT specific informer disabled")
+		return
+	}
+	klog.InfoS("NRT specific informer status", "value", val)
+	enabled = val
+}
+
+func FromHandle(handle framework.Handle) (k8scache.SharedIndexInformer, podlisterv1.PodLister) {
+	if !IsEnabled() {
+		podHandle := handle.SharedInformerFactory().Core().V1().Pods() // shortcut
+		return podHandle.Informer(), podHandle.Lister()
+	}
+
+	podInformer := coreinformers.NewFilteredPodInformer(handle.ClientSet(), metav1.NamespaceAll, 0, cache.Indexers{}, nil)
+	podLister := podlisterv1.NewPodLister(podInformer.GetIndexer())
+
+	klog.V(5).InfoS("Start custom pod informer")
+	ctx := context.Background()
+	go podInformer.Run(ctx.Done())
+
+	klog.V(5).InfoS("Syncing custom pod informer")
+	cache.WaitForCacheSync(ctx.Done(), podInformer.HasSynced)
+	klog.V(5).InfoS("Synced custom pod informer")
+
+	return podInformer, podLister
+}
+
+func IsPodRelevantForState(pod *corev1.Pod) bool {
+	if pod == nil {
+		return false // should never happen
+	}
+	if IsEnabled() {
+		return true // consider all pods including ones in terminal phase
+	}
+	return pod.Status.Phase == corev1.PodRunning // we are interested only about nodes which consume resources
+}

pkg/noderesourcetopology/cache/overreserve.go

@@ -27,16 +27,15 @@ import (
 	"github.com/k8stopologyawareschedwg/podfingerprint"
 
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	podlisterv1 "k8s.io/client-go/listers/core/v1"
+	k8scache "k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
+	"k8s.io/kubernetes/pkg/scheduler/framework"
 
 	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
 
 	apiconfig "sigs.k8s.io/scheduler-plugins/apis/config"
-	"sigs.k8s.io/scheduler-plugins/pkg/noderesourcetopology/podprovider"
-	"sigs.k8s.io/scheduler-plugins/pkg/noderesourcetopology/resourcerequests"
 	"sigs.k8s.io/scheduler-plugins/pkg/noderesourcetopology/stringify"
 )
 
@@ -51,10 +50,9 @@ type OverReserve struct {
 	nodesWithForeignPods   counter
 	podLister              podlisterv1.PodLister
 	resyncMethod           apiconfig.CacheResyncMethod
-	isPodRelevant          podprovider.PodFilterFunc
 }
 
-func NewOverReserve(cfg *apiconfig.NodeResourceTopologyCache, client ctrlclient.Client, podLister podlisterv1.PodLister, isPodRelevant podprovider.PodFilterFunc) (*OverReserve, error) {
+func NewOverReserve(cfg *apiconfig.NodeResourceTopologyCache, client ctrlclient.Client, podLister podlisterv1.PodLister) (*OverReserve, error) {
 	if client == nil || podLister == nil {
 		return nil, fmt.Errorf("nrtcache: received nil references")
 	}
@@ -76,7 +74,6 @@ func NewOverReserve(cfg *apiconfig.NodeResourceTopologyCache, client ctrlclient.
 		nodesWithForeignPods:   newCounter(),
 		podLister:              podLister,
 		resyncMethod:           resyncMethod,
-		isPodRelevant:          isPodRelevant,
 	}
 	return obj, nil
 }
@@ -202,7 +199,7 @@ func (ov *OverReserve) Resync() {
 	}
 
 	// node -> pod identifier (namespace, name)
-	nodeToObjsMap, err := makeNodeToPodDataMap(ov.podLister, ov.isPodRelevant, logID)
+	nodeToObjsMap, err := makeNodeToPodDataMap(ov.podLister, logID)
 	if err != nil {
 		klog.ErrorS(err, "cannot find the mapping between running pods and nodes")
 		return
@@ -270,32 +267,16 @@ func (ov *OverReserve) FlushNodes(logID string, nrts ...*topologyv1alpha2.NodeRe
 	}
 }
 
+func InformerFromHandle(handle framework.Handle) (k8scache.SharedIndexInformer, podlisterv1.PodLister) {
+	podHandle := handle.SharedInformerFactory().Core().V1().Pods() // shortcut
+	return podHandle.Informer(), podHandle.Lister()
+}
+
 // to be used only in tests
 func (ov *OverReserve) Store() *nrtStore {
 	return ov.nrts
 }
 
-func makeNodeToPodDataMap(podLister podlisterv1.PodLister, isPodRelevant podprovider.PodFilterFunc, logID string) (map[string][]podData, error) {
-	nodeToObjsMap := make(map[string][]podData)
-	pods, err := podLister.List(labels.Everything())
-	if err != nil {
-		return nodeToObjsMap, err
-	}
-	for _, pod := range pods {
-		if !isPodRelevant(pod, logID) {
-			continue
-		}
-		nodeObjs := nodeToObjsMap[pod.Spec.NodeName]
-		nodeObjs = append(nodeObjs, podData{
-			Namespace:             pod.Namespace,
-			Name:                  pod.Name,
-			HasExclusiveResources: resourcerequests.AreExclusiveForPod(pod),
-		})
-		nodeToObjsMap[pod.Spec.NodeName] = nodeObjs
-	}
-	return nodeToObjsMap, nil
-}
-
 func logIDFromTime() string {
 	return fmt.Sprintf("resync%v", time.Now().UnixMilli())
 }