Skip to content
Permalink
Browse files

Merge pull request #79439 from roycaihw/automated-cherry-pick-of-#791…

…14-upstream-release-1.15

Automated cherry pick of #79114: crd-handler: level-trigger storage recreation and fix a race
  • Loading branch information...
k8s-ci-robot committed Jul 1, 2019
2 parents 162b46b + 262ebe8 commit 1bb90b5835bbbf87febc26f711f6144394bea513
Showing with 57 additions and 7 deletions.
  1. +57 −7 staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/customresource_handler.go
@@ -170,6 +170,7 @@ func NewCustomResourceDefinitionHandler(
requestTimeout: requestTimeout,
}
crdInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
AddFunc: ret.createCustomResourceDefinition,
UpdateFunc: ret.updateCustomResourceDefinition,
DeleteFunc: func(obj interface{}) {
ret.removeDeadStorage()
@@ -242,11 +243,19 @@ func (r *crdHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {

terminating := apiextensions.IsCRDConditionTrue(crd, apiextensions.Terminating)

crdInfo, err := r.getOrCreateServingInfoFor(crd)
crdInfo, err := r.getOrCreateServingInfoFor(crd.UID, crd.Name)
if apierrors.IsNotFound(err) {
r.delegate.ServeHTTP(w, req)
return
}
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
if !hasServedCRDVersion(crdInfo.spec, requestInfo.APIVersion) {
r.delegate.ServeHTTP(w, req)
return
}

verb := strings.ToUpper(requestInfo.Verb)
resource := requestInfo.Resource
@@ -356,6 +365,16 @@ func (r *crdHandler) serveScale(w http.ResponseWriter, req *http.Request, reques
}
}

// createCustomResourceDefinition removes potentially stale storage so it gets re-created
func (r *crdHandler) createCustomResourceDefinition(obj interface{}) {
crd := obj.(*apiextensions.CustomResourceDefinition)
r.customStorageLock.Lock()
defer r.customStorageLock.Unlock()
// this could happen if the create event is merged from create-update events
r.removeStorage_locked(crd.UID)
}

// updateCustomResourceDefinition removes potentially stale storage so it gets re-created
func (r *crdHandler) updateCustomResourceDefinition(oldObj, newObj interface{}) {
oldCRD := oldObj.(*apiextensions.CustomResourceDefinition)
newCRD := newObj.(*apiextensions.CustomResourceDefinition)
@@ -376,6 +395,10 @@ func (r *crdHandler) updateCustomResourceDefinition(oldObj, newObj interface{})
}
}

if oldCRD.UID != newCRD.UID {
r.removeStorage_locked(oldCRD.UID)
}

storageMap := r.customStorage.Load().(crdStorageMap)
oldInfo, found := storageMap[newCRD.UID]
if !found {
@@ -386,15 +409,22 @@ func (r *crdHandler) updateCustomResourceDefinition(oldObj, newObj interface{})
return
}

klog.V(4).Infof("Updating customresourcedefinition %s", oldCRD.Name)
klog.V(4).Infof("Updating customresourcedefinition %s", newCRD.Name)
r.removeStorage_locked(newCRD.UID)
}

if oldInfo, ok := storageMap[types.UID(oldCRD.UID)]; ok {
// removeStorage_locked removes the cached storage with the given uid as key from the storage map. This function
// updates r.customStorage with the cleaned-up storageMap and tears down the old storage.
// NOTE: Caller MUST hold r.customStorageLock to write r.customStorage thread-safely.
func (r *crdHandler) removeStorage_locked(uid types.UID) {
storageMap := r.customStorage.Load().(crdStorageMap)
if oldInfo, ok := storageMap[uid]; ok {
// Copy because we cannot write to storageMap without a race
// as it is used without locking elsewhere.
storageMap2 := storageMap.clone()

// Remove from the CRD info map and store the map
delete(storageMap2, types.UID(oldCRD.UID))
delete(storageMap2, uid)
r.customStorage.Store(storageMap2)

// Tear down the old storage
@@ -465,22 +495,32 @@ func (r *crdHandler) tearDown(oldInfo *crdInfo) {
// GetCustomResourceListerCollectionDeleter returns the ListerCollectionDeleter of
// the given crd.
func (r *crdHandler) GetCustomResourceListerCollectionDeleter(crd *apiextensions.CustomResourceDefinition) (finalizer.ListerCollectionDeleter, error) {
info, err := r.getOrCreateServingInfoFor(crd)
info, err := r.getOrCreateServingInfoFor(crd.UID, crd.Name)
if err != nil {
return nil, err
}
return info.storages[info.storageVersion].CustomResource, nil
}

func (r *crdHandler) getOrCreateServingInfoFor(crd *apiextensions.CustomResourceDefinition) (*crdInfo, error) {
// getOrCreateServingInfoFor gets the CRD serving info for the given CRD UID if the key exists in the storage map.
// Otherwise the function fetches the up-to-date CRD using the given CRD name and creates CRD serving info.
func (r *crdHandler) getOrCreateServingInfoFor(uid types.UID, name string) (*crdInfo, error) {
storageMap := r.customStorage.Load().(crdStorageMap)
if ret, ok := storageMap[crd.UID]; ok {
if ret, ok := storageMap[uid]; ok {
return ret, nil
}

r.customStorageLock.Lock()
defer r.customStorageLock.Unlock()

// Get the up-to-date CRD when we have the lock, to avoid racing with updateCustomResourceDefinition.
// If updateCustomResourceDefinition sees an update and happens later, the storage will be deleted and
// we will re-create the updated storage on demand. If updateCustomResourceDefinition happens before,
// we make sure that we observe the same up-to-date CRD.
crd, err := r.crdLister.Get(name)
if err != nil {
return nil, err
}
storageMap = r.customStorage.Load().(crdStorageMap)
if ret, ok := storageMap[crd.UID]; ok {
return ret, nil
@@ -1060,3 +1100,13 @@ func (v *unstructuredSchemaCoercer) apply(u *unstructured.Unstructured) error {

return nil
}

// hasServedCRDVersion returns true if the given version is in the list of CRD's versions and the Served flag is set.
func hasServedCRDVersion(spec *apiextensions.CustomResourceDefinitionSpec, version string) bool {
for _, v := range spec.Versions {
if v.Name == version {
return v.Served
}
}
return false
}

0 comments on commit 1bb90b5

Please sign in to comment.
You can’t perform that action at this time.