Merge pull request #176 from winsopc/ib-k8s-nad-add-support

NAD: add-only support without pod restart

Author: almaslennikov

deployment/ib-kubernetes.yaml

@@ -15,7 +15,7 @@ rules:
     verbs: ["get", "list", "patch", "watch"]
   - apiGroups: ["k8s.cni.cncf.io"]
     resources: ["*"]
-    verbs: ["get"]
+    verbs: ["get", "list", "watch"]
   # Leader election permissions
   - apiGroups: ["coordination.k8s.io"]
     resources: ["leases"]

pkg/daemon/daemon.go

@@ -55,11 +55,15 @@ type Daemon interface {
 
 type daemon struct {
 	config            config.DaemonConfig
-	watcher           watcher.Watcher
+	podWatcher        watcher.Watcher
+	nadWatcher        watcher.Watcher // NAD watcher for network definition changes
 	kubeClient        k8sClient.Client
 	guidPool          guid.Pool
 	smClient          plugins.SubnetManagerClient
 	guidPodNetworkMap map[string]string // allocated guid mapped to the pod and network
+
+	// NAD add-only cache
+	nadCache map[string]*v1.NetworkAttachmentDefinition // network ID -> NAD
 }
 
 // Temporary struct used to proceed pods' networks
@@ -109,6 +113,7 @@ func NewDaemon() (Daemon, error) {
 	}
 
 	podEventHandler := resEvenHandler.NewPodEventHandler()
+	nadEventHandler := resEvenHandler.NewNADEventHandler()
 	client, err := k8sClient.NewK8sClient()
 	if err != nil {
 		return nil, err
@@ -151,13 +156,16 @@ func NewDaemon() (Daemon, error) {
 	}
 
 	podWatcher := watcher.NewWatcher(podEventHandler, client)
+	nadWatcher := watcher.NewWatcher(nadEventHandler, client)
 	return &daemon{
 		config:            daemonConfig,
-		watcher:           podWatcher,
+		podWatcher:        podWatcher,
+		nadWatcher:        nadWatcher,
 		kubeClient:        client,
 		guidPool:          guidPool,
 		smClient:          smClient,
 		guidPodNetworkMap: make(map[string]string),
+		nadCache:          make(map[string]*v1.NetworkAttachmentDefinition),
 	}, nil
 }
 
@@ -289,11 +297,14 @@ func (d *daemon) runLeaderLogic() {
 
 	go wait.Until(d.AddPeriodicUpdate, time.Duration(d.config.PeriodicUpdate)*time.Second, stopPeriodicsChan)
 	go wait.Until(d.DeletePeriodicUpdate, time.Duration(d.config.PeriodicUpdate)*time.Second, stopPeriodicsChan)
+	go wait.Until(d.ProcessNADChanges, time.Duration(d.config.PeriodicUpdate)*time.Second, stopPeriodicsChan)
 	defer close(stopPeriodicsChan)
 
-	// Run Watcher in background, calling watcherStopFunc() will stop the watcher
-	watcherStopFunc := d.watcher.RunBackground()
-	defer watcherStopFunc()
+	// Run both watchers in background
+	podWatcherStopFunc := d.podWatcher.RunBackground()
+	nadWatcherStopFunc := d.nadWatcher.RunBackground()
+	defer podWatcherStopFunc()
+	defer nadWatcherStopFunc()
 
 	// Run until interrupted by os signals
 	sigChan := make(chan os.Signal, 1)
@@ -303,26 +314,16 @@ func (d *daemon) runLeaderLogic() {
 }
 
 // If network identified by networkID is IbSriov return network name and spec
-//
-//nolint:nilerr
 func (d *daemon) getIbSriovNetwork(networkID string) (string, *utils.IbSriovCniSpec, error) {
-	networkNamespace, networkName, err := utils.ParseNetworkID(networkID)
+	_, networkName, err := utils.ParseNetworkID(networkID)
 	if err != nil {
 		return "", nil, fmt.Errorf("failed to parse network id %s with error: %v", networkID, err)
 	}
 
-	// Try to get net-attach-def in backoff loop
-	var netAttInfo *v1.NetworkAttachmentDefinition
-	if err = wait.ExponentialBackoff(backoffValues, func() (bool, error) {
-		netAttInfo, err = d.kubeClient.GetNetworkAttachmentDefinition(networkNamespace, networkName)
-		if err != nil {
-			log.Warn().Msgf("failed to get networkName attachment %s with error %v",
-				networkName, err)
-			return false, nil
-		}
-		return true, nil
-	}); err != nil {
-		return "", nil, fmt.Errorf("failed to get networkName attachment %s", networkName)
+	// Try to get net-attach-def from cache first, then fallback to API
+	netAttInfo, err := d.getCachedNAD(networkID)
+	if err != nil {
+		return "", nil, fmt.Errorf("failed to get network attachment %s: %v", networkName, err)
 	}
 	log.Debug().Msgf("networkName attachment %v", netAttInfo)
 
@@ -551,7 +552,7 @@ func (d *daemon) updatePodNetworkAnnotation(pi *podNetworkInfo, removedList *[]n
 //nolint:nilerr
 func (d *daemon) AddPeriodicUpdate() {
 	log.Info().Msgf("running periodic add update")
-	addMap, _ := d.watcher.GetHandler().GetResults()
+	addMap, _ := d.podWatcher.GetHandler().GetResults()
 	addMap.Lock()
 	defer addMap.Unlock()
 	// Contains ALL pods' networks
@@ -569,12 +570,10 @@ func (d *daemon) AddPeriodicUpdate() {
 		if len(pods) == 0 {
 			continue
 		}
-
-		log.Info().Msgf("processing network networkID %s", networkID)
 		networkName, ibCniSpec, err := d.getIbSriovNetwork(networkID)
 		if err != nil {
-			addMap.UnSafeRemove(networkID)
-			log.Error().Msgf("droping network: %v", err)
+			// Do not drop the network; keep for next periodic run when NAD becomes available
+			log.Warn().Msgf("NAD not ready for network %s: %v (will retry)", networkID, err)
 			continue
 		}
 
@@ -678,7 +677,7 @@ func getAllPodGUIDsForNetwork(pod *kapi.Pod, networkName string) ([]net.Hardware
 //nolint:nilerr
 func (d *daemon) DeletePeriodicUpdate() {
 	log.Info().Msg("running delete periodic update")
-	_, deleteMap := d.watcher.GetHandler().GetResults()
+	_, deleteMap := d.podWatcher.GetHandler().GetResults()
 	deleteMap.Lock()
 	defer deleteMap.Unlock()
 	for networkID, podsInterface := range deleteMap.Items {
@@ -753,6 +752,64 @@ func (d *daemon) DeletePeriodicUpdate() {
 	log.Info().Msg("delete periodic update finished")
 }
 
+// ProcessNADChanges processes NAD add events
+func (d *daemon) ProcessNADChanges() {
+	log.Debug().Msg("Processing NAD changes...")
+
+	nadHandler := d.nadWatcher.GetHandler().(*resEvenHandler.NADEventHandler)
+	addedNADs, _ := nadHandler.GetResults()
+
+	// Process NAD add events only
+	addedNADs.Lock()
+	for networkID, nad := range addedNADs.Items {
+		nadObj := nad.(*v1.NetworkAttachmentDefinition)
+
+		// Add-only: cache the NAD; ignore updates/deletes
+		d.nadCache[networkID] = nadObj
+
+		log.Info().Msgf("Successfully processed NAD event: %s", networkID)
+
+		// Remove processed item
+		addedNADs.UnSafeRemove(networkID)
+	}
+	addedNADs.Unlock()
+
+	log.Debug().Msg("NAD changes processing completed")
+}
+
+// getCachedNAD retrieves NAD from cache, falling back to API if not cached
+func (d *daemon) getCachedNAD(networkID string) (*v1.NetworkAttachmentDefinition, error) {
+	// First check cache
+	if nad, exists := d.nadCache[networkID]; exists {
+		return nad, nil
+	}
+
+	// Fall back to API call (existing behavior)
+	networkNamespace, networkName, err := utils.ParseNetworkID(networkID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse network id %s with error: %v", networkID, err)
+	}
+
+	var netAttInfo *v1.NetworkAttachmentDefinition
+	if err = wait.ExponentialBackoff(backoffValues, func() (bool, error) {
+		var getErr error
+		netAttInfo, getErr = d.kubeClient.GetNetworkAttachmentDefinition(networkNamespace, networkName)
+		if getErr != nil {
+			log.Warn().Msgf("failed to get network attachment %s with error %v", networkName, getErr)
+			// keep retrying until backoff exhausted
+			return false, nil
+		}
+		return true, nil
+	}); err != nil {
+		return nil, fmt.Errorf("failed to get network attachment %s", networkName)
+	}
+
+	// Cache the result
+	d.nadCache[networkID] = netAttInfo
+
+	return netAttInfo, nil
+}
+
 // initPool check the guids that are already allocated by the running pods
 func (d *daemon) initPool() error {
 	log.Info().Msg("Initializing GUID pool.")

pkg/k8s-client/client.go

@@ -40,6 +40,7 @@ type Client interface {
 	GetNetworkAttachmentDefinition(namespace, name string) (*netapi.NetworkAttachmentDefinition, error)
 	GetRestClient() rest.Interface
 	GetCoordinationV1() coordv1client.CoordinationV1Interface
+	GetNetClient() netclient.K8sCniCncfIoV1Interface
 }
 
 type client struct {
@@ -120,3 +121,8 @@ func (c *client) GetRestClient() rest.Interface {
 func (c *client) GetCoordinationV1() coordv1client.CoordinationV1Interface {
 	return c.clientset.CoordinationV1()
 }
+
+// GetNetClient returns the network attachment definition client for NAD watcher
+func (c *client) GetNetClient() netclient.K8sCniCncfIoV1Interface {
+	return c.netClient
+}

pkg/k8s-client/mocks/Client.go

@@ -0,0 +1,111 @@
+// Copyright 2025 NVIDIA CORPORATION & AFFILIATES
+//
+// 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.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package handler
+
+import (
+	"encoding/json"
+	"fmt"
+	"sync"
+
+	v1 "github.com/k8snetworkplumbingwg/network-attachment-definition-client/pkg/apis/k8s.cni.cncf.io/v1"
+	"github.com/rs/zerolog/log"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	"github.com/Mellanox/ib-kubernetes/pkg/utils"
+)
+
+type NADEventHandler struct {
+	addedNADs *utils.SynchronizedMap // Maps network ID to NAD for added NADs
+	nadCache  sync.Map               // Cache of current NADs by namespace/name
+}
+
+// NewNADEventHandler creates a new NAD event handler
+func NewNADEventHandler() ResourceEventHandler {
+	return &NADEventHandler{
+		addedNADs: utils.NewSynchronizedMap(),
+		nadCache:  sync.Map{},
+	}
+}
+
+// GetResourceObject returns the NAD object type for the watcher
+func (n *NADEventHandler) GetResourceObject() runtime.Object {
+	return &v1.NetworkAttachmentDefinition{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "NetworkAttachmentDefinition",
+			APIVersion: "k8s.cni.cncf.io/v1",
+		},
+	}
+}
+
+// OnAdd handles NAD creation events
+func (n *NADEventHandler) OnAdd(obj interface{}, _ bool) {
+	nad := obj.(*v1.NetworkAttachmentDefinition)
+	log.Info().Msgf("NAD add event: namespace %s name %s", nad.Namespace, nad.Name)
+
+	// Only handle InfiniBand SR-IOV networks
+	if !n.isInfiniBandNetwork(nad) {
+		log.Debug().Msgf("NAD %s/%s is not an InfiniBand network", nad.Namespace, nad.Name)
+		return
+	}
+
+	networkID := fmt.Sprintf("%s_%s", nad.Namespace, nad.Name)
+
+	// Cache the NAD for future reference
+	n.nadCache.Store(networkID, nad)
+
+	// Add to processing queue
+	n.addedNADs.Set(networkID, nad)
+
+	log.Info().Msgf("Successfully processed NAD add event: %s", networkID)
+}
+
+// OnUpdate is a no-op for add-only support
+func (n *NADEventHandler) OnUpdate(oldObj, newObj interface{}) {}
+
+// OnDelete is a no-op for add-only support
+func (n *NADEventHandler) OnDelete(obj interface{}) {}
+
+// GetResults returns the results maps for processing by the daemon
+func (n *NADEventHandler) GetResults() (*utils.SynchronizedMap, *utils.SynchronizedMap) {
+	return n.addedNADs, nil
+}
+
+// GetNADFromCache retrieves a cached NAD by network ID
+func (n *NADEventHandler) GetNADFromCache(networkID string) (*v1.NetworkAttachmentDefinition, bool) {
+	if nad, ok := n.nadCache.Load(networkID); ok {
+		return nad.(*v1.NetworkAttachmentDefinition), true
+	}
+	return nil, false
+}
+
+// isInfiniBandNetwork checks if the NAD is for InfiniBand SR-IOV
+func (n *NADEventHandler) isInfiniBandNetwork(nad *v1.NetworkAttachmentDefinition) bool {
+	// Parse the network configuration
+	var networkConfig map[string]interface{}
+	if err := json.Unmarshal([]byte(nad.Spec.Config), &networkConfig); err != nil {
+		log.Error().Msgf("Failed to parse NAD config for %s/%s: %v", nad.Namespace, nad.Name, err)
+		return false
+	}
+
+	// Check if this is an ib-sriov network
+	if cniType, ok := networkConfig["type"]; ok {
+		return cniType == utils.InfiniBandSriovCni
+	}
+
+	return false
+}