Repository: kubernetes/kubernetes
File: pkg/scheduler/framework/plugins/dynamicresources/dra_manager.go (Go)

Code before:
	// - Pod A and B get scheduled as above.
	// - PreBind for pod A gets called first, then fails with a temporary API error.
	//   It removes the updated claim from the assume cache because of that.
	// - PreBind for pod B gets called next and succeeds with adding the
	//   allocation and its own reservedFor entry.
	// - The assume cache is now not reflecting that the claim is allocated,
	//   which could lead to reusing the same resource for some other claim.
	//
	// A sync.Map is used because in practice sharing of a claim between
	// pods is expected to be rare compared to per-pod claim, so we end up
	// hitting the "multiple goroutines read, write, and overwrite entries
	// for disjoint sets of keys" case that sync.Map is optimized for.
	inFlightAllocations *sync.Map
	allocatedDevices    *allocatedDevices
	logger              klog.Logger
}

func (c *claimTracker) ClaimHasPendingAllocation(claimUID types.UID) bool {
	_, found := c.inFlightAllocations.Load(claimUID)
	return found
}

func (c *claimTracker) SignalClaimPendingAllocation(claimUID types.UID, allocatedClaim *resourceapi.ResourceClaim) error {
	c.inFlightAllocations.Store(claimUID, allocatedClaim)
	// This is the same verbosity as the corresponding log in the assume cache.
	c.logger.V(5).Info("Added in-flight claim", "claim", klog.KObj(allocatedClaim), "uid", claimUID, "version", allocatedClaim.ResourceVersion)
	// There's no reason to return an error in this implementation, but the error is helpful for other implementations.
	// For example, implementations that have to deal with fake claims might want to return an error if the allocation
	// is for an invalid claim.
	return nil
}

func (c *claimTracker) RemoveClaimPendingAllocation(claimUID types.UID) (deleted bool) {
	claim, found := c.inFlightAllocations.LoadAndDelete(claimUID)
	// The assume cache doesn't log this, but maybe it should.
	if found {
		claim := claim.(*resourceapi.ResourceClaim)
		c.logger.V(5).Info("Removed in-flight claim", "claim", klog.KObj(claim), "uid", claimUID, "version", claim.ResourceVersion)
	} else {
		c.logger.V(5).Info("Redundant remove of in-flight claim, not found", "uid", claimUID)
	}
	return found
}

func (c *claimTracker) Get(namespace, claimName string) (*resourceapi.ResourceClaim, error) {
	obj, err := c.cache.Get(namespace + "/" + claimName)
	if err != nil {
		return nil, err
	}
	claim, ok := obj.(*resourceapi.ResourceClaim)
	if !ok {

Review comment (suggestion):
Let's first merge this fix PR without the logging changes and then do those in https://github.com/kubernetes/kubernetes/pull/136268. That'll make backporting easier (there's one small conflict between the two PRs).

The next push will drop this.


Code after:
	//   It removes the updated claim from the assume cache because of that.
	// - PreBind for pod B gets called next and succeeds with adding the
	//   allocation and its own reservedFor entry.
	// - The assume cache is now not reflecting that the claim is allocated,
	//   which could lead to reusing the same resource for some other claim.
	//
	// A sync.Map is used because in practice sharing of a claim between
	// pods is expected to be rare compared to per-pod claim, so we end up
	// hitting the "multiple goroutines read, write, and overwrite entries
	// for disjoint sets of keys" case that sync.Map is optimized for.
	inFlightAllocations *sync.Map
	allocatedDevices    *allocatedDevices
	logger              klog.Logger
}

func (c *claimTracker) ClaimHasPendingAllocation(claimUID types.UID) bool {
	_, found := c.inFlightAllocations.Load(claimUID)
	return found
}

func (c *claimTracker) SignalClaimPendingAllocation(claimUID types.UID, allocatedClaim *resourceapi.ResourceClaim) error {
	c.inFlightAllocations.Store(claimUID, allocatedClaim)
	// There's no reason to return an error in this implementation, but the error is helpful for other implementations.
	// For example, implementations that have to deal with fake claims might want to return an error if the allocation
	// is for an invalid claim.
	return nil
}

func (c *claimTracker) RemoveClaimPendingAllocation(claimUID types.UID) (deleted bool) {
	_, found := c.inFlightAllocations.LoadAndDelete(claimUID)
	return found
}

func (c *claimTracker) Get(namespace, claimName string) (*resourceapi.ResourceClaim, error) {
	obj, err := c.cache.Get(namespace + "/" + claimName)
	if err != nil {
		return nil, err
	}
	claim, ok := obj.(*resourceapi.ResourceClaim)
	if !ok {
		return nil, fmt.Errorf("unexpected object type %T for assumed object %s/%s", obj, namespace, claimName)
	}
	return claim, nil
}

func (c *claimTracker) List() ([]*resourceapi.ResourceClaim, error) {
	var result []*resourceapi.ResourceClaim
	// Probably not worth adding an index for?
	objs := c.cache.List(nil)
	for _, obj := range objs {
		claim, ok := obj.(*resourceapi.ResourceClaim)