Merge branch 'NVIDIA:main' into two-container-one-pod

Author: guptaNswati

.github/workflows/golang.yaml

@@ -34,7 +34,7 @@ jobs:
         go-version: ${{ inputs.golang_versions }}
 
     - name: Lint
-      uses: golangci/golangci-lint-action@v7
+      uses: golangci/golangci-lint-action@v8
       with:
         version: latest
         args: -v --timeout 5m

cmd/compute-domain-controller/computedomain.go

@@ -191,7 +191,7 @@ func (m *ComputeDomainManager) RemoveFinalizer(ctx context.Context, uid string)
 	return nil
 }
 
-// AssertWorkloadsCompletes ensures that all workloads asssociated with a ComputeDomain have completed.
+// AssertWorkloadsCompletes ensures that all workloads associated with a ComputeDomain have completed.
 //
 // TODO: We should probably also check to ensure that all ResourceClaims
 // generated from our ResourceClaimTemplate for workloads are gone. Doing

cmd/compute-domain-kubelet-plugin/checkpoint.go

@@ -3,6 +3,7 @@ package main
 import (
 	"encoding/json"
 
+	resourceapi "k8s.io/api/resource/v1beta1"
 	"k8s.io/kubernetes/pkg/kubelet/checkpointmanager/checksum"
 )
 
@@ -12,14 +13,22 @@ type Checkpoint struct {
 }
 
 type CheckpointV1 struct {
-	PreparedClaims PreparedClaims `json:"preparedClaims,omitempty"`
+	PreparedClaims PreparedClaimsByUID `json:"preparedClaims,omitempty"`
+}
+
+// key: stringified claim UUID
+type PreparedClaimsByUID map[string]PreparedClaim
+
+type PreparedClaim struct {
+	Status          resourceapi.ResourceClaimStatus `json:"status,omitempty"`
+	PreparedDevices PreparedDevices                 `json:"preparedDevices,omitempty"`
 }
 
 func newCheckpoint() *Checkpoint {
 	pc := &Checkpoint{
 		Checksum: 0,
 		V1: &CheckpointV1{
-			PreparedClaims: make(PreparedClaims),
+			PreparedClaims: make(PreparedClaimsByUID),
 		},
 	}
 	return pc

cmd/compute-domain-kubelet-plugin/computedomain.go

@@ -325,7 +325,7 @@ func (m *ComputeDomainManager) GetNodeIPs(ctx context.Context, cdUID string) ([]
 	}
 
 	if cd.Status.Nodes == nil {
-		return nil, fmt.Errorf("error getting status of nodes in ComputeDomain: %w", err)
+		return nil, fmt.Errorf("no nodes set for ComputeDomain")
 	}
 
 	if len(cd.Status.Nodes) != cd.Spec.NumNodes {
@@ -437,7 +437,7 @@ func (m *ComputeDomainManager) periodicCleanup(ctx context.Context) {
 				continue
 			}
 			if err != nil {
-				klog.Errorf("error checking for existenc of directory '%s': %v", m.configFilesRoot, err)
+				klog.Errorf("error checking for existence of directory '%s': %v", m.configFilesRoot, err)
 				continue
 			}
 

cmd/compute-domain-kubelet-plugin/device_state.go

@@ -24,8 +24,8 @@ import (
 
 	resourceapi "k8s.io/api/resource/v1beta1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/dynamic-resource-allocation/kubeletplugin"
 	"k8s.io/klog/v2"
-	drapbv1 "k8s.io/kubelet/pkg/apis/dra/v1beta1"
 	"k8s.io/kubernetes/pkg/kubelet/checkpointmanager"
 	cdiapi "tags.cncf.io/container-device-interface/pkg/cdi"
 
@@ -115,33 +115,37 @@ func NewDeviceState(ctx context.Context, config *Config) (*DeviceState, error) {
 	}
 
 	for _, c := range checkpoints {
-		if c == DriverPluginCheckpointFile {
+		if c == DriverPluginCheckpointFileBasename {
 			return state, nil
 		}
 	}
 
 	checkpoint := newCheckpoint()
-	if err := state.checkpointManager.CreateCheckpoint(DriverPluginCheckpointFile, checkpoint); err != nil {
+	if err := state.checkpointManager.CreateCheckpoint(DriverPluginCheckpointFileBasename, checkpoint); err != nil {
 		return nil, fmt.Errorf("unable to sync to checkpoint: %v", err)
 	}
 
 	return state, nil
 }
 
-func (s *DeviceState) Prepare(ctx context.Context, claim *resourceapi.ResourceClaim) ([]*drapbv1.Device, error) {
+func (s *DeviceState) Prepare(ctx context.Context, claim *resourceapi.ResourceClaim) ([]kubeletplugin.Device, error) {
 	s.Lock()
 	defer s.Unlock()
 
 	claimUID := string(claim.UID)
 
 	checkpoint := newCheckpoint()
-	if err := s.checkpointManager.GetCheckpoint(DriverPluginCheckpointFile, checkpoint); err != nil {
-		return nil, fmt.Errorf("unable to sync from checkpoint: %v", err)
+	if err := s.checkpointManager.GetCheckpoint(DriverPluginCheckpointFileBasename, checkpoint); err != nil {
+		return nil, fmt.Errorf("unable to get checkpoint: %w", err)
 	}
-	preparedClaims := checkpoint.V1.PreparedClaims
 
-	if preparedClaims[claimUID] != nil {
-		return preparedClaims[claimUID].GetDevices(), nil
+	preparedClaim, exists := checkpoint.V1.PreparedClaims[claimUID]
+	if exists {
+		// Make this a noop. Associated device(s) has/ave been prepared by us.
+		// Prepare() must be idempotent, as it may be invoked more than once per
+		// claim (and actual device preparation must happen at most once).
+		klog.V(6).Infof("skip prepare: claim %v found in checkpoint", claimUID)
+		return preparedClaim.PreparedDevices.GetDevices(), nil
 	}
 
 	preparedDevices, err := s.prepareDevices(ctx, claim)
@@ -153,50 +157,65 @@ func (s *DeviceState) Prepare(ctx context.Context, claim *resourceapi.ResourceCl
 		return nil, fmt.Errorf("unable to create CDI spec file for claim: %w", err)
 	}
 
-	preparedClaims[claimUID] = preparedDevices
-	if err := s.checkpointManager.CreateCheckpoint(DriverPluginCheckpointFile, checkpoint); err != nil {
-		return nil, fmt.Errorf("unable to sync to checkpoint: %v", err)
+	// Add ResourceClaimStatus API object to node-local checkpoint: the
+	// 'unprepare' code path must use local state exclusively (ResourceClaim
+	// object might have been deleted from the API server).
+	checkpoint.V1.PreparedClaims[claimUID] = PreparedClaim{
+		Status:          claim.Status,
+		PreparedDevices: preparedDevices,
+	}
+	if err := s.checkpointManager.CreateCheckpoint(DriverPluginCheckpointFileBasename, checkpoint); err != nil {
+		return nil, fmt.Errorf("unable to create checkpoint: %w", err)
 	}
+	klog.V(6).Infof("checkpoint written for claim %v", claimUID)
 
-	return preparedClaims[claimUID].GetDevices(), nil
+	return preparedDevices.GetDevices(), nil
 }
 
-func (s *DeviceState) Unprepare(ctx context.Context, claim *resourceapi.ResourceClaim) error {
+func (s *DeviceState) Unprepare(ctx context.Context, claimRef kubeletplugin.NamespacedObject) error {
 	s.Lock()
 	defer s.Unlock()
 
-	claimUID := string(claim.UID)
-
-	if err := s.unprepareDevices(ctx, claim); err != nil {
-		return fmt.Errorf("unprepare devices failed: %w", err)
-	}
+	claimUID := string(claimRef.UID)
 
+	// Rely on local checkpoint state for ability to clean up.
 	checkpoint := newCheckpoint()
-	if err := s.checkpointManager.GetCheckpoint(DriverPluginCheckpointFile, checkpoint); err != nil {
-		return fmt.Errorf("unable to sync from checkpoint: %v", err)
+	if err := s.checkpointManager.GetCheckpoint(DriverPluginCheckpointFileBasename, checkpoint); err != nil {
+		return fmt.Errorf("unable to get checkpoint: %w", err)
 	}
-	preparedClaims := checkpoint.V1.PreparedClaims
 
-	if preparedClaims[claimUID] == nil {
+	pc, exists := checkpoint.V1.PreparedClaims[claimUID]
+	if !exists {
+		// Not an error: if this claim UID is not in the checkpoint then this
+		// device was never prepared or has already been unprepared (assume that
+		// Prepare+Checkpoint are done transactionally). Note that
+		// claimRef.String() contains namespace, name, UID.
+		klog.Infof("unprepare noop: claim not found in checkpoint data: %v", claimRef.String())
 		return nil
 	}
 
+	if err := s.unprepareDevices(ctx, &pc.Status); err != nil {
+		return fmt.Errorf("unprepare devices failed: %w", err)
+	}
+
 	err := s.cdi.DeleteClaimSpecFile(claimUID)
 	if err != nil {
 		return fmt.Errorf("unable to delete CDI spec file for claim: %w", err)
 	}
 
-	delete(preparedClaims, claimUID)
-	if err := s.checkpointManager.CreateCheckpoint(DriverPluginCheckpointFile, checkpoint); err != nil {
-		return fmt.Errorf("unable to sync to checkpoint: %v", err)
+	// Write new checkpoint reflecting that all devices for this claim have been
+	// unprepared (by virtue of removing its UID from all mappings).
+	delete(checkpoint.V1.PreparedClaims, claimUID)
+	if err := s.checkpointManager.CreateCheckpoint(DriverPluginCheckpointFileBasename, checkpoint); err != nil {
+		return fmt.Errorf("create checkpoint failed: %w", err)
 	}
 
 	return nil
 }
 
 func (s *DeviceState) prepareDevices(ctx context.Context, claim *resourceapi.ResourceClaim) (PreparedDevices, error) {
 	// Generate a mapping of each OpaqueDeviceConfigs to the Device.Results it applies to
-	configResultsMap, err := s.getConfigResultsMap(claim)
+	configResultsMap, err := s.getConfigResultsMap(&claim.Status)
 	if err != nil {
 		return nil, fmt.Errorf("error generating configResultsMap: %w", err)
 	}
@@ -254,8 +273,8 @@ func (s *DeviceState) prepareDevices(ctx context.Context, claim *resourceapi.Res
 				cdiDevices = append(cdiDevices, d)
 			}
 
-			device := &drapbv1.Device{
-				RequestNames: []string{result.Request},
+			device := kubeletplugin.Device{
+				Requests:     []string{result.Request},
 				PoolName:     result.Pool,
 				DeviceName:   result.Device,
 				CDIDeviceIDs: cdiDevices,
@@ -266,12 +285,12 @@ func (s *DeviceState) prepareDevices(ctx context.Context, claim *resourceapi.Res
 			case ComputeDomainChannelType:
 				preparedDevice.Channel = &PreparedComputeDomainChannel{
 					Info:   s.allocatable[result.Device].Channel,
-					Device: device,
+					Device: &device,
 				}
 			case ComputeDomainDaemonType:
 				preparedDevice.Daemon = &PreparedComputeDomainDaemon{
 					Info:   s.allocatable[result.Device].Daemon,
-					Device: device,
+					Device: &device,
 				}
 			}
 
@@ -283,9 +302,9 @@ func (s *DeviceState) prepareDevices(ctx context.Context, claim *resourceapi.Res
 	return preparedDevices, nil
 }
 
-func (s *DeviceState) unprepareDevices(ctx context.Context, claim *resourceapi.ResourceClaim) error {
+func (s *DeviceState) unprepareDevices(ctx context.Context, cs *resourceapi.ResourceClaimStatus) error {
 	// Generate a mapping of each OpaqueDeviceConfigs to the Device.Results it applies to
-	configResultsMap, err := s.getConfigResultsMap(claim)
+	configResultsMap, err := s.getConfigResultsMap(cs)
 	if err != nil {
 		return fmt.Errorf("error generating configResultsMap: %w", err)
 	}
@@ -407,12 +426,12 @@ func (s *DeviceState) applyComputeDomainDaemonConfig(ctx context.Context, config
 	return &configState, nil
 }
 
-func (s *DeviceState) getConfigResultsMap(claim *resourceapi.ResourceClaim) (map[runtime.Object][]*resourceapi.DeviceRequestAllocationResult, error) {
+func (s *DeviceState) getConfigResultsMap(rcs *resourceapi.ResourceClaimStatus) (map[runtime.Object][]*resourceapi.DeviceRequestAllocationResult, error) {
 	// Retrieve the full set of device configs for the driver.
 	configs, err := GetOpaqueDeviceConfigs(
 		configapi.Decoder,
 		DriverName,
-		claim.Status.Allocation.Devices.Config,
+		rcs.Allocation.Devices.Config,
 	)
 	if err != nil {
 		return nil, fmt.Errorf("error getting opaque device configs: %v", err)
@@ -433,7 +452,7 @@ func (s *DeviceState) getConfigResultsMap(claim *resourceapi.ResourceClaim) (map
 	// Look through the configs and figure out which one will be applied to
 	// each device allocation result based on their order of precedence and type.
 	configResultsMap := make(map[runtime.Object][]*resourceapi.DeviceRequestAllocationResult)
-	for _, result := range claim.Status.Allocation.Devices.Results {
+	for _, result := range rcs.Allocation.Devices.Results {
 		device, exists := s.allocatable[result.Device]
 		if !exists {
 			return nil, fmt.Errorf("requested device is not allocatable: %v", result.Device)