dra scheduler: preserve allocation in assume cache

Storing a modified claim with allocation and the original resource version
was not reliable: if an update was received, it replaced the modified
claim and the resource that was reserved for the claim might have been
used for some other claim.

To fix this, the assume cache gets copied and modified slightly so that it
keeps the allocation until the process of writing it back or dropping it
is complete. This is then not a general-purpose cache anymore.

Logging got extended to diagnose this problem better. It started to occur in
E2E tests after splitting the claim update so that first the finalizer is set
and then the status, because setting the finalizer triggered an update.

pkg/scheduler/framework/plugins/dynamicresources/assume_cache.go

@@ -0,0 +1,368 @@
+/*
+Copyright 2017 The Kubernetes Authors.
+
+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 dynamicresources
+
+import (
+	"fmt"
+	"strconv"
+	"sync"
+
+	resourcev1alpha2 "k8s.io/api/resource/v1alpha2"
+	"k8s.io/klog/v2"
+
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/client-go/tools/cache"
+)
+
+// AssumeCache is a cache on top of the informer that allows for updating
+// objects outside of informer events and also restoring the informer
+// cache's version of the object.  Objects are assumed to be
+// Kubernetes API objects that implement meta.Interface
+type AssumeCache interface {
+	// Assume updates the object in-memory only
+	Assume(obj interface{}) error
+
+	// Restore the informer cache's version of the object
+	Restore(objName string)
+
+	// Get the object by name
+	Get(objName string) (interface{}, error)
+
+	// GetAPIObj gets the API object by name
+	GetAPIObj(objName string) (interface{}, error)
+
+	// List all the objects in the cache
+	List(indexObj interface{}) []interface{}
+}
+
+type errWrongType struct {
+	typeName string
+	object   interface{}
+}
+
+func (e *errWrongType) Error() string {
+	return fmt.Sprintf("could not convert object to type %v: %+v", e.typeName, e.object)
+}
+
+type errNotFound struct {
+	typeName   string
+	objectName string
+}
+
+func (e *errNotFound) Error() string {
+	return fmt.Sprintf("could not find %v %q", e.typeName, e.objectName)
+}
+
+type errObjectName struct {
+	detailedErr error
+}
+
+func (e *errObjectName) Error() string {
+	return fmt.Sprintf("failed to get object name: %v", e.detailedErr)
+}
+
+// assumeCache stores two pointers to represent a single object:
+//   - The pointer to the informer object.
+//   - The pointer to the latest object, which could be the same as
+//     the informer object, or an in-memory object.
+//
+// An informer update always overrides the latest object pointer.
+//
+// Assume() only updates the latest object pointer.
+// Restore() sets the latest object pointer back to the informer object.
+// Get/List() always returns the latest object pointer.
+type assumeCache struct {
+	// The logger that was chosen when setting up the cache.
+	// Will be used for all operations.
+	logger klog.Logger
+
+	// Synchronizes updates to store
+	rwMutex sync.RWMutex
+
+	// describes the object stored
+	description string
+
+	// Stores objInfo pointers
+	store cache.Indexer
+
+	// Index function for object
+	indexFunc cache.IndexFunc
+	indexName string
+}
+
+type objInfo struct {
+	// name of the object
+	name string
+
+	// Latest version of object could be cached-only or from informer
+	latestObj interface{}
+
+	// Latest object from informer
+	apiObj interface{}
+}
+
+func objInfoKeyFunc(obj interface{}) (string, error) {
+	objInfo, ok := obj.(*objInfo)
+	if !ok {
+		return "", &errWrongType{"objInfo", obj}
+	}
+	return objInfo.name, nil
+}
+
+func (c *assumeCache) objInfoIndexFunc(obj interface{}) ([]string, error) {
+	objInfo, ok := obj.(*objInfo)
+	if !ok {
+		return []string{""}, &errWrongType{"objInfo", obj}
+	}
+	return c.indexFunc(objInfo.latestObj)
+}
+
+// NewAssumeCache creates an assume cache for general objects.
+func NewAssumeCache(logger klog.Logger, informer cache.SharedIndexInformer, description, indexName string, indexFunc cache.IndexFunc) AssumeCache {
+	c := &assumeCache{
+		logger:      logger,
+		description: description,
+		indexFunc:   indexFunc,
+		indexName:   indexName,
+	}
+	indexers := cache.Indexers{}
+	if indexName != "" && indexFunc != nil {
+		indexers[indexName] = c.objInfoIndexFunc
+	}
+	c.store = cache.NewIndexer(objInfoKeyFunc, indexers)
+
+	// Unit tests don't use informers
+	if informer != nil {
+		informer.AddEventHandler(
+			cache.ResourceEventHandlerFuncs{
+				AddFunc:    c.add,
+				UpdateFunc: c.update,
+				DeleteFunc: c.delete,
+			},
+		)
+	}
+	return c
+}
+
+func (c *assumeCache) add(obj interface{}) {
+	if obj == nil {
+		return
+	}
+
+	name, err := cache.MetaNamespaceKeyFunc(obj)
+	if err != nil {
+		c.logger.Error(&errObjectName{err}, "Add failed")
+		return
+	}
+
+	c.rwMutex.Lock()
+	defer c.rwMutex.Unlock()
+
+	latestObj := obj
+	if objInfo, _ := c.getObjInfo(name); objInfo != nil {
+		newVersion, err := c.getObjVersion(name, obj)
+		if err != nil {
+			c.logger.Error(err, "Add failed: couldn't get object version")
+			return
+		}
+
+		storedVersion, err := c.getObjVersion(name, objInfo.latestObj)
+		if err != nil {
+			c.logger.Error(err, "Add failed: couldn't get stored object version")
+			return
+		}
+
+		// Only update object if version is newer.
+		// This is so we don't override assumed objects due to informer resync.
+		if newVersion <= storedVersion {
+			c.logger.V(10).Info("Skip adding object to assume cache because version is not newer than storedVersion", "description", c.description, "cacheKey", name, "newVersion", newVersion, "storedVersion", storedVersion)
+			return
+		}
+
+		// If we we have an assumed object, then it always has an
+		// allocation. If the new object doesn't, then we have to keep
+		// overriding allocation because the scheduler plugin is still
+		// in the process of writing out that change.  Eventually, the
+		// plugin will store the actual written claim or restore the
+		// original claim.
+		if claim := obj.(*resourcev1alpha2.ResourceClaim); claim.Status.Allocation == nil {
+			claim := claim.DeepCopy()
+			claim.Status.DriverName = objInfo.latestObj.(*resourcev1alpha2.ResourceClaim).Status.DriverName
+			claim.Status.Allocation = objInfo.latestObj.(*resourcev1alpha2.ResourceClaim).Status.Allocation
+			latestObj = claim
+		}
+	}
+
+	objInfo := &objInfo{name: name, latestObj: latestObj, apiObj: obj}
+	if err = c.store.Update(objInfo); err != nil {
+		c.logger.Info("Error occurred while updating stored object", "err", err)
+	} else {
+		c.logger.V(10).Info("Adding object to assume cache", "description", c.description, "cacheKey", name, "assumeCache", obj)
+	}
+}
+
+func (c *assumeCache) update(oldObj interface{}, newObj interface{}) {
+	c.add(newObj)
+}
+
+func (c *assumeCache) delete(obj interface{}) {
+	if obj == nil {
+		return
+	}
+
+	name, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
+	if err != nil {
+		c.logger.Error(&errObjectName{err}, "Failed to delete")
+		return
+	}
+
+	c.rwMutex.Lock()
+	defer c.rwMutex.Unlock()
+
+	objInfo := &objInfo{name: name}
+	err = c.store.Delete(objInfo)
+	if err != nil {
+		c.logger.Error(err, "Failed to delete", "description", c.description, "cacheKey", name)
+	}
+}
+
+func (c *assumeCache) getObjVersion(name string, obj interface{}) (int64, error) {
+	objAccessor, err := meta.Accessor(obj)
+	if err != nil {
+		return -1, err
+	}
+
+	objResourceVersion, err := strconv.ParseInt(objAccessor.GetResourceVersion(), 10, 64)
+	if err != nil {
+		return -1, fmt.Errorf("error parsing ResourceVersion %q for %v %q: %s", objAccessor.GetResourceVersion(), c.description, name, err)
+	}
+	return objResourceVersion, nil
+}
+
+func (c *assumeCache) getObjInfo(name string) (*objInfo, error) {
+	obj, ok, err := c.store.GetByKey(name)
+	if err != nil {
+		return nil, err
+	}
+	if !ok {
+		return nil, &errNotFound{c.description, name}
+	}
+
+	objInfo, ok := obj.(*objInfo)
+	if !ok {
+		return nil, &errWrongType{"objInfo", obj}
+	}
+	return objInfo, nil
+}
+
+func (c *assumeCache) Get(objName string) (interface{}, error) {
+	c.rwMutex.RLock()
+	defer c.rwMutex.RUnlock()
+
+	objInfo, err := c.getObjInfo(objName)
+	if err != nil {
+		return nil, err
+	}
+	return objInfo.latestObj, nil
+}
+
+func (c *assumeCache) GetAPIObj(objName string) (interface{}, error) {
+	c.rwMutex.RLock()
+	defer c.rwMutex.RUnlock()
+
+	objInfo, err := c.getObjInfo(objName)
+	if err != nil {
+		return nil, err
+	}
+	return objInfo.apiObj, nil
+}
+
+func (c *assumeCache) List(indexObj interface{}) []interface{} {
+	c.rwMutex.RLock()
+	defer c.rwMutex.RUnlock()
+
+	allObjs := []interface{}{}
+	var objs []interface{}
+	if c.indexName != "" {
+		o, err := c.store.Index(c.indexName, &objInfo{latestObj: indexObj})
+		if err != nil {
+			c.logger.Error(err, "List index error")
+			return nil
+		}
+		objs = o
+	} else {
+		objs = c.store.List()
+	}
+
+	for _, obj := range objs {
+		objInfo, ok := obj.(*objInfo)
+		if !ok {
+			c.logger.Error(&errWrongType{"objInfo", obj}, "List error")
+			continue
+		}
+		allObjs = append(allObjs, objInfo.latestObj)
+	}
+	return allObjs
+}
+
+func (c *assumeCache) Assume(obj interface{}) error {
+	name, err := cache.MetaNamespaceKeyFunc(obj)
+	if err != nil {
+		return &errObjectName{err}
+	}
+
+	c.rwMutex.Lock()
+	defer c.rwMutex.Unlock()
+
+	objInfo, err := c.getObjInfo(name)
+	if err != nil {
+		return err
+	}
+
+	newVersion, err := c.getObjVersion(name, obj)
+	if err != nil {
+		return err
+	}
+
+	storedVersion, err := c.getObjVersion(name, objInfo.latestObj)
+	if err != nil {
+		return err
+	}
+
+	if newVersion < storedVersion {
+		return fmt.Errorf("%v %q is out of sync (stored: %d, assume: %d)", c.description, name, storedVersion, newVersion)
+	}
+
+	// Only update the cached object
+	objInfo.latestObj = obj
+	c.logger.V(4).Info("Assumed object", "description", c.description, "cacheKey", name, "version", newVersion)
+	return nil
+}
+
+func (c *assumeCache) Restore(objName string) {
+	c.rwMutex.Lock()
+	defer c.rwMutex.Unlock()
+
+	objInfo, err := c.getObjInfo(objName)
+	if err != nil {
+		// This could be expected if object got deleted
+		c.logger.V(5).Info("Restore object", "description", c.description, "cacheKey", objName, "err", err)
+	} else {
+		objInfo.latestObj = objInfo.apiObj
+		c.logger.V(4).Info("Restored object", "description", c.description, "cacheKey", objName)
+	}
+}