topology-updater:compute pod set fingerprint

Add an option to compute the fingerprint of the current pod set on each
node.

Report this new fingerprint using an annotation in NRT object.

cmd/nfd-topology-updater/main.go

@@ -143,6 +143,7 @@ func initFlags(flagset *flag.FlagSet) (*topology.Args, *resourcemonitor.Args) {
 		"Pod Resource Socket path to use.")
 	flagset.StringVar(&args.ConfigFile, "config", "/etc/kubernetes/node-feature-discovery/nfd-topology-updater.conf",
 		"Config file to use.")
+	flagset.BoolVar(&resourcemonitorArgs.PodSetFingerprint, "pods-fingerprint", false, "If enable, compute and report the pod set fingerprint")
 
 	klog.InitFlags(flagset)
 

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/google/go-cmp v0.5.9
 	github.com/jaypipes/ghw v0.8.1-0.20210827132705-c7224150a17e
 	github.com/k8stopologyawareschedwg/noderesourcetopology-api v0.0.13
+	github.com/k8stopologyawareschedwg/podfingerprint v0.1.1
 	github.com/klauspost/cpuid/v2 v2.2.3
 	github.com/onsi/ginkgo/v2 v2.4.0
 	github.com/onsi/gomega v1.23.0
@@ -50,6 +51,7 @@ require (
 	github.com/Microsoft/go-winio v0.4.17 // indirect
 	github.com/Microsoft/hcsshim v0.8.22 // indirect
 	github.com/NYTimes/gziphandler v1.1.1 // indirect
+	github.com/OneOfOne/xxhash v1.2.8 // indirect
 	github.com/StackExchange/wmi v0.0.0-20190523213315-cbe66965904d // indirect
 	github.com/antlr/antlr4/runtime/Go/antlr v1.4.10 // indirect
 	github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e // indirect

go.sum

@@ -86,6 +86,8 @@ github.com/NYTimes/gziphandler v0.0.0-20170623195520-56545f4a5d46/go.mod h1:3wb0
 github.com/NYTimes/gziphandler v1.1.1 h1:ZUDjpQae29j0ryrS0u/B8HZfJBtBQHjqw2rQ2cqUQ3I=
 github.com/NYTimes/gziphandler v1.1.1/go.mod h1:n/CVRwUEOgIxrgPvAQhUUr9oeUtvrhMomdKFjzJNB0c=
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
+github.com/OneOfOne/xxhash v1.2.8 h1:31czK/TI9sNkxIKfaUfGlU47BAxQ0ztGgd9vPyqimf8=
+github.com/OneOfOne/xxhash v1.2.8/go.mod h1:eZbhyaAYD41SGSSsnmcpxVoRiQ/MPUTjUdIIOT9Um7Q=
 github.com/PuerkitoBio/purell v1.1.1/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbtSwDGJws/X0=
 github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578/go.mod h1:uGdkoq3SwY9Y+13GIhn11/XLaGBb4BfwItxLd5jeuXE=
 github.com/StackExchange/wmi v0.0.0-20190523213315-cbe66965904d h1:G0m3OIz70MZUWq3EgK3CesDbo8upS2Vm9/P3FtgI+Jk=
@@ -422,6 +424,8 @@ github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7V
 github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM=
 github.com/k8stopologyawareschedwg/noderesourcetopology-api v0.0.13 h1:Y1RjPskyGMkVtNL8lq75bEdjqgq8gi+JJ1oWaz/mIJE=
 github.com/k8stopologyawareschedwg/noderesourcetopology-api v0.0.13/go.mod h1:AkACMQGiTgCt0lQw3m7TTU8PLH9lYKNK5e9DqFf5VuM=
+github.com/k8stopologyawareschedwg/podfingerprint v0.1.1 h1:uNEj+avp3yJkJMvkmk6iosibVauSo+owEKV2JyuKNsQ=
+github.com/k8stopologyawareschedwg/podfingerprint v0.1.1/go.mod h1:C23pM15t06dXg/OihGlqBvnYzLr+MXDXJ7zMfbNAyXI=
 github.com/karrick/godirwalk v1.17.0 h1:b4kY7nqDdioR/6qnbHQyDvmA17u5G1cZ6J+CZXwSWoI=
 github.com/karrick/godirwalk v1.17.0/go.mod h1:j4mkqPuvaLI8mp1DroR3P6ad7cyYd4c1qeJ3RV7ULlk=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=

pkg/nfd-topology-updater/nfd-topology-updater.go

@@ -114,7 +114,7 @@ func (w *nfdTopologyUpdater) Run() error {
 
 	var resScan resourcemonitor.ResourcesScanner
 
-	resScan, err = resourcemonitor.NewPodResourcesScanner(w.resourcemonitorArgs.Namespace, podResClient, w.apihelper)
+	resScan, err = resourcemonitor.NewPodResourcesScanner(w.resourcemonitorArgs.Namespace, podResClient, w.apihelper, w.resourcemonitorArgs.PodSetFingerprint)
 	if err != nil {
 		return fmt.Errorf("failed to initialize ResourceMonitor instance: %w", err)
 	}
@@ -138,7 +138,7 @@ func (w *nfdTopologyUpdater) Run() error {
 		select {
 		case <-crTrigger.C:
 			klog.Infof("Scanning")
-			podResources, err := resScan.Scan()
+			podResources, annotations, err := resScan.Scan()
 			utils.KlogDump(1, "podResources are", "  ", podResources)
 			if err != nil {
 				klog.Warningf("Scan failed: %v", err)
@@ -147,7 +147,7 @@ func (w *nfdTopologyUpdater) Run() error {
 			zones = resAggr.Aggregate(podResources)
 			utils.KlogDump(1, "After aggregating resources identified zones are", "  ", zones)
 			if !w.args.NoPublish {
-				if err = w.updateNodeResourceTopology(zones); err != nil {
+				if err = w.updateNodeResourceTopology(zones, annotations); err != nil {
 					return err
 				}
 			}
@@ -172,7 +172,7 @@ func (w *nfdTopologyUpdater) Stop() {
 	}
 }
 
-func (w *nfdTopologyUpdater) updateNodeResourceTopology(zoneInfo v1alpha1.ZoneList) error {
+func (w *nfdTopologyUpdater) updateNodeResourceTopology(zoneInfo v1alpha1.ZoneList, annotations map[string]string) error {
 	cli, err := w.apihelper.GetTopologyClient()
 	if err != nil {
 		return err
@@ -182,7 +182,8 @@ func (w *nfdTopologyUpdater) updateNodeResourceTopology(zoneInfo v1alpha1.ZoneLi
 	if errors.IsNotFound(err) {
 		nrtNew := v1alpha1.NodeResourceTopology{
 			ObjectMeta: metav1.ObjectMeta{
-				Name: w.nodeInfo.nodeName,
+				Name:        w.nodeInfo.nodeName,
+				Annotations: annotations,
 			},
 			Zones:            zoneInfo,
 			TopologyPolicies: []string{w.nodeInfo.tmPolicy},
@@ -199,6 +200,7 @@ func (w *nfdTopologyUpdater) updateNodeResourceTopology(zoneInfo v1alpha1.ZoneLi
 
 	nrtMutated := nrt.DeepCopy()
 	nrtMutated.Zones = zoneInfo
+	nrtMutated.Annotations = mergeAnnotations(nrt.Annotations, annotations)
 
 	nrtUpdated, err := cli.TopologyV1alpha1().NodeResourceTopologies().Update(context.TODO(), nrtMutated, metav1.UpdateOptions{})
 	if err != nil {
@@ -231,3 +233,13 @@ func (w *nfdTopologyUpdater) configure() error {
 	klog.Infof("configuration file %q parsed:\n %v", w.configFilePath, w.config)
 	return nil
 }
+
+func mergeAnnotations(annotations ...map[string]string) map[string]string {
+	ret := make(map[string]string)
+	for _, annotation := range annotations {
+		for k, v := range annotation {
+			ret[k] = v
+		}
+	}
+	return ret
+}

pkg/resourcemonitor/podresourcesscanner.go

@@ -27,20 +27,24 @@ import (
 	podresourcesapi "k8s.io/kubelet/pkg/apis/podresources/v1"
 
 	"sigs.k8s.io/node-feature-discovery/pkg/apihelper"
+
+	"github.com/k8stopologyawareschedwg/podfingerprint"
 )
 
 type PodResourcesScanner struct {
 	namespace         string
 	podResourceClient podresourcesapi.PodResourcesListerClient
 	apihelper         apihelper.APIHelpers
+	podFingerprint    bool
 }
 
 // NewPodResourcesScanner creates a new ResourcesScanner instance
-func NewPodResourcesScanner(namespace string, podResourceClient podresourcesapi.PodResourcesListerClient, kubeApihelper apihelper.APIHelpers) (ResourcesScanner, error) {
+func NewPodResourcesScanner(namespace string, podResourceClient podresourcesapi.PodResourcesListerClient, kubeApihelper apihelper.APIHelpers, podFingerprint bool) (ResourcesScanner, error) {
 	resourcemonitorInstance := &PodResourcesScanner{
 		namespace:         namespace,
 		podResourceClient: podResourceClient,
 		apihelper:         kubeApihelper,
+		podFingerprint:    podFingerprint,
 	}
 	if resourcemonitorInstance.namespace != "*" {
 		klog.Infof("watching namespace %q", resourcemonitorInstance.namespace)
@@ -113,14 +117,14 @@ func hasIntegralCPUs(pod *corev1.Pod, container *corev1.Container) bool {
 }
 
 // Scan gathers all the PodResources from the system, using the podresources API client.
-func (resMon *PodResourcesScanner) Scan() ([]PodResources, error) {
+func (resMon *PodResourcesScanner) Scan() ([]PodResources, map[string]string, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultPodResourcesTimeout)
 	defer cancel()
 
 	// Pod Resource API client
 	resp, err := resMon.podResourceClient.List(ctx, &podresourcesapi.ListPodResourcesRequest{})
 	if err != nil {
-		return nil, fmt.Errorf("can't receive response: %v.Get(_) = _, %w", resMon.podResourceClient, err)
+		return nil, nil, fmt.Errorf("can't receive response: %v.Get(_) = _, %w", resMon.podResourceClient, err)
 	}
 
 	var podResData []PodResources
@@ -130,7 +134,7 @@ func (resMon *PodResourcesScanner) Scan() ([]PodResources, error) {
 		hasDevice := hasDevice(podResource)
 		isWatchable, isIntegralGuaranteed, err := resMon.isWatchable(podResource.GetNamespace(), podResource.GetName(), hasDevice)
 		if err != nil {
-			return nil, fmt.Errorf("checking if pod in a namespace is watchable, namespace:%v, pod name %v: %v", podResource.GetNamespace(), podResource.GetName(), err)
+			return nil, nil, fmt.Errorf("checking if pod in a namespace is watchable, namespace:%v, pod name %v: %v", podResource.GetNamespace(), podResource.GetName(), err)
 		}
 		if !isWatchable {
 			continue
@@ -198,7 +202,22 @@ func (resMon *PodResourcesScanner) Scan() ([]PodResources, error) {
 
 	}
 
-	return podResData, nil
+	if resMon.podFingerprint && len(podResData) > 0 {
+		var status podfingerprint.Status
+		podFingerprintSign, err := computePodFingerprint(podResData, &status)
+		if err != nil {
+			klog.Errorf("podFingerprint: Unable to compute fingerprint %v", err)
+			return podResData, nil, nil
+		}
+		klog.Infof("podFingerprint: " + status.Repr())
+
+		annotations := map[string]string{
+			podfingerprint.Annotation: podFingerprintSign,
+		}
+		return podResData, annotations, nil
+	}
+
+	return podResData, nil, nil
 }
 
 func hasDevice(podResource *podresourcesapi.PodResources) bool {
@@ -225,3 +244,14 @@ func getNumaNodeIds(topologyInfo *podresourcesapi.TopologyInfo) []int {
 
 	return topology
 }
+
+func computePodFingerprint(podResources []PodResources, status *podfingerprint.Status) (string, error) {
+	fingerprint := podfingerprint.NewTracingFingerprint(len(podResources), status)
+	for _, podResource := range podResources {
+		err := fingerprint.Add(podResource.Namespace, podResource.Name)
+		if err != nil {
+			return "", err
+		}
+	}
+	return fingerprint.Sign(), nil
+}