Skip to content

Commit

Permalink
Fix check for CRD watch priming
Browse files Browse the repository at this point in the history
  • Loading branch information
mbohlool committed Jun 14, 2018
1 parent 1d10287 commit 602ddef
Show file tree
Hide file tree
Showing 4 changed files with 72 additions and 66 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -226,7 +226,7 @@ func testSimpleCRUD(t *testing.T, ns string, noxuDefinition *apiextensionsv1beta
t.Fatal(err)
}
if e, a := createdObjectMeta.GetUID(), deletedObjectMeta.GetUID(); e != a {
t.Errorf("expected %v, got %v", e, a)
t.Errorf("expected equal UID for (expected) %v, and (actual) %v", createdNoxuInstance, watchEvent.Object)
}

case <-time.After(5 * time.Second):
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -361,7 +361,7 @@ func TestValidationSchema(t *testing.T) {
},
Required: []string{"spec"},
}
noxuDefinition, err = testserver.CreateNewCustomResourceDefinition(noxuDefinition, apiExtensionClient, dynamicClient)
_, err = testserver.CreateNewCustomResourceDefinition(noxuDefinition, apiExtensionClient, dynamicClient)
if err != nil {
t.Fatalf("unable to created crd %v: %v", noxuDefinition.Name, err)
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@ go_library(
"//vendor/k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset:go_default_library",
"//vendor/k8s.io/apiextensions-apiserver/pkg/cmd/server:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/api/meta:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@ import (
apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
"k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
"k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime/schema"
Expand Down Expand Up @@ -191,6 +190,35 @@ func NewCurletInstance(namespace, name string) *unstructured.Unstructured {
}
}

func servedVersions(crd *apiextensionsv1beta1.CustomResourceDefinition) []string {
if len(crd.Spec.Versions) == 0 {
return []string{crd.Spec.Version}
}
var versions []string
for _, v := range crd.Spec.Versions {
if v.Served {
versions = append(versions, v.Name)
}
}
return versions
}

func existsInDiscovery(crd *apiextensionsv1beta1.CustomResourceDefinition, apiExtensionsClient clientset.Interface, version string) (bool, error) {
groupResource, err := apiExtensionsClient.Discovery().ServerResourcesForGroupVersion(crd.Spec.Group + "/" + version)
if err != nil {
if errors.IsNotFound(err) {
return false, nil
}
return false, err
}
for _, g := range groupResource.APIResources {
if g.Name == crd.Spec.Names.Plural {
return true, nil
}
}
return false, nil
}

// CreateNewCustomResourceDefinitionWatchUnsafe creates the CRD and makes sure
// the apiextension apiserver has installed the CRD. But it's not safe to watch
// the created CR. Please call CreateNewCustomResourceDefinition if you need to
Expand All @@ -201,19 +229,15 @@ func CreateNewCustomResourceDefinitionWatchUnsafe(crd *apiextensionsv1beta1.Cust
return nil, err
}

// wait until the resource appears in discovery
err = wait.PollImmediate(500*time.Millisecond, 30*time.Second, func() (bool, error) {
resourceList, err := apiExtensionsClient.Discovery().ServerResourcesForGroupVersion(crd.Spec.Group + "/" + crd.Spec.Version)
// wait until all resources appears in discovery
for _, version := range servedVersions(crd) {
err := wait.PollImmediate(500*time.Millisecond, 30*time.Second, func() (bool, error) {
return existsInDiscovery(crd, apiExtensionsClient, version)
})
if err != nil {
return false, nil
}
for _, resource := range resourceList.APIResources {
if resource.Name == crd.Spec.Names.Plural {
return true, nil
}
return nil, err
}
return false, nil
})
}

return crd, err
}
Expand All @@ -234,48 +258,35 @@ func CreateNewCustomResourceDefinition(crd *apiextensionsv1beta1.CustomResourceD
// For this test, we'll actually cycle, "list/watch/create/delete" until we get an RV from list that observes the create and not an error.
// This way all the tests that are checking for watches don't have to worry about RV too old problems because crazy things *could* happen
// before like the created RV could be too old to watch.
var primingErr error
wait.PollImmediate(500*time.Millisecond, 30*time.Second, func() (bool, error) {
primingErr = checkForWatchCachePrimed(crd, dynamicClientSet)
if primingErr == nil {
return true, nil
for _, version := range servedVersions(crd) {
err := wait.PollImmediate(500*time.Millisecond, 30*time.Second, func() (bool, error) {
return isWatchCachePrimed(crd, dynamicClientSet, version)
})
if err != nil {
return nil, err
}
return false, nil
})
if primingErr != nil {
return nil, primingErr
}

return crd, nil
}

func checkForWatchCachePrimed(crd *apiextensionsv1beta1.CustomResourceDefinition, dynamicClientSet dynamic.Interface) error {
// isWatchCachePrimed returns true if the watch is primed for an specified version of CRD watch
func isWatchCachePrimed(crd *apiextensionsv1beta1.CustomResourceDefinition, dynamicClientSet dynamic.Interface, version string) (bool, error) {
ns := ""
if crd.Spec.Scope != apiextensionsv1beta1.ClusterScoped {
ns = "aval"
}

gvr := schema.GroupVersionResource{Group: crd.Spec.Group, Version: crd.Spec.Version, Resource: crd.Spec.Names.Plural}
gvr := schema.GroupVersionResource{Group: crd.Spec.Group, Version: version, Resource: crd.Spec.Names.Plural}
var resourceClient dynamic.ResourceInterface
if crd.Spec.Scope != apiextensionsv1beta1.ClusterScoped {
resourceClient = dynamicClientSet.Resource(gvr).Namespace(ns)
} else {
resourceClient = dynamicClientSet.Resource(gvr)
}

initialList, err := resourceClient.List(metav1.ListOptions{})
if err != nil {
return err
}
initialListListMeta, err := meta.ListAccessor(initialList)
if err != nil {
return err
}

instanceName := "setup-instance"
instance := &unstructured.Unstructured{
Object: map[string]interface{}{
"apiVersion": crd.Spec.Group + "/" + crd.Spec.Version,
"apiVersion": crd.Spec.Group + "/" + version,
"kind": crd.Spec.Names.Kind,
"metadata": map[string]interface{}{
"namespace": ns,
Expand All @@ -289,53 +300,49 @@ func checkForWatchCachePrimed(crd *apiextensionsv1beta1.CustomResourceDefinition
"spec": map[string]interface{}{},
},
}
if _, err := resourceClient.Create(instance); err != nil {
return err
createdInstance, err := resourceClient.Create(instance)
if err != nil {
return false, err
}
err = resourceClient.Delete(createdInstance.GetName(), nil)
if err != nil {
return false, err
}
// we created something, clean it up
defer func() {
resourceClient.Delete(instanceName, nil)
}()

noxuWatch, err := resourceClient.Watch(metav1.ListOptions{ResourceVersion: initialListListMeta.GetResourceVersion()})
noxuWatch, err := resourceClient.Watch(metav1.ListOptions{ResourceVersion: createdInstance.GetResourceVersion()})
if err != nil {
return err
return false, err
}
defer noxuWatch.Stop()

select {
case watchEvent := <-noxuWatch.ResultChan():
if watch.Added == watchEvent.Type {
return nil
if watch.Error == watchEvent.Type {
return false, nil
}
return fmt.Errorf("expected add, but got %#v", watchEvent)

if watch.Deleted != watchEvent.Type {
return false, fmt.Errorf("expected DELETE, but got %#v", watchEvent)
}
return true, nil
case <-time.After(5 * time.Second):
return fmt.Errorf("gave up waiting for watch event")
return false, fmt.Errorf("gave up waiting for watch event")
}
}

func DeleteCustomResourceDefinition(crd *apiextensionsv1beta1.CustomResourceDefinition, apiExtensionsClient clientset.Interface) error {
if err := apiExtensionsClient.Apiextensions().CustomResourceDefinitions().Delete(crd.Name, nil); err != nil {
return err
}
err := wait.PollImmediate(500*time.Millisecond, 30*time.Second, func() (bool, error) {
groupResource, err := apiExtensionsClient.Discovery().ServerResourcesForGroupVersion(crd.Spec.Group + "/" + crd.Spec.Version)
for _, version := range servedVersions(crd) {
err := wait.PollImmediate(500*time.Millisecond, 30*time.Second, func() (bool, error) {
exists, err := existsInDiscovery(crd, apiExtensionsClient, version)
return !exists, err
})
if err != nil {
if errors.IsNotFound(err) {
return true, nil

}
return false, err
}
for _, g := range groupResource.APIResources {
if g.Name == crd.Spec.Names.Plural {
return false, nil
}
return err
}
return true, nil
})
return err
}
return nil
}

func CreateNewScaleClient(crd *apiextensionsv1beta1.CustomResourceDefinition, config *rest.Config) (scale.ScalesGetter, error) {
Expand Down

0 comments on commit 602ddef

Please sign in to comment.