Merge pull request #48502 from ironcladlou/memcachediscovery

Automatic merge from submit-queue (batch tested with PRs 47700, 48464, 48502)

Add a refreshing discovery client

Introduce a discovery client (implementing `CachedDiscoveryInterface`) which caches discovery information in memory and which can be actively refreshed by the user.

This implementation fetches from discovery upon refresh and could later be improved to maintain updates from a watch.

Extracted from https://github.com/kubernetes/kubernetes/pull/47665 and https://github.com/kubernetes/kubernetes/pull/46000 to help reduce the scope of https://github.com/kubernetes/kubernetes/pull/48065.

```release-note
NONE
```

staging/src/k8s.io/client-go/discovery/cached/BUILD

@@ -0,0 +1,36 @@
+package(default_visibility = ["//visibility:public"])
+
+licenses(["notice"])
+
+load(
+    "@io_bazel_rules_go//go:def.bzl",
+    "go_library",
+    "go_test",
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["memcache_test.go"],
+    library = ":go_default_library",
+    tags = ["automanaged"],
+    deps = [
+        "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//vendor/k8s.io/client-go/discovery/fake:go_default_library",
+    ],
+)
+
+go_library(
+    name = "go_default_library",
+    srcs = ["memcache.go"],
+    tags = ["automanaged"],
+    deps = [
+        "//vendor/github.com/emicklei/go-restful-swagger12:go_default_library",
+        "//vendor/github.com/googleapis/gnostic/OpenAPIv2:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/version:go_default_library",
+        "//vendor/k8s.io/client-go/discovery:go_default_library",
+        "//vendor/k8s.io/client-go/rest:go_default_library",
+    ],
+)

staging/src/k8s.io/client-go/discovery/cached/memcache.go

@@ -0,0 +1,185 @@
+/*
+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 cached
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+
+	"github.com/emicklei/go-restful-swagger12"
+	"github.com/googleapis/gnostic/OpenAPIv2"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/version"
+	"k8s.io/client-go/discovery"
+	restclient "k8s.io/client-go/rest"
+)
+
+// memCacheClient can Invalidate() to stay up-to-date with discovery
+// information.
+//
+// TODO: Switch to a watch interface. Right now it will poll anytime
+// Invalidate() is called.
+type memCacheClient struct {
+	delegate discovery.DiscoveryInterface
+
+	lock                   sync.RWMutex
+	groupToServerResources map[string]*metav1.APIResourceList
+	groupList              *metav1.APIGroupList
+	cacheValid             bool
+}
+
+var (
+	ErrCacheEmpty    = errors.New("the cache has not been filled yet")
+	ErrCacheNotFound = errors.New("not found")
+)
+
+var _ discovery.CachedDiscoveryInterface = &memCacheClient{}
+
+// ServerResourcesForGroupVersion returns the supported resources for a group and version.
+func (d *memCacheClient) ServerResourcesForGroupVersion(groupVersion string) (*metav1.APIResourceList, error) {
+	d.lock.RLock()
+	defer d.lock.RUnlock()
+	if !d.cacheValid {
+		return nil, ErrCacheEmpty
+	}
+	cachedVal, ok := d.groupToServerResources[groupVersion]
+	if !ok {
+		return nil, ErrCacheNotFound
+	}
+	return cachedVal, nil
+}
+
+// ServerResources returns the supported resources for all groups and versions.
+func (d *memCacheClient) ServerResources() ([]*metav1.APIResourceList, error) {
+	apiGroups, err := d.ServerGroups()
+	if err != nil {
+		return nil, err
+	}
+	groupVersions := metav1.ExtractGroupVersions(apiGroups)
+	result := []*metav1.APIResourceList{}
+	for _, groupVersion := range groupVersions {
+		resources, err := d.ServerResourcesForGroupVersion(groupVersion)
+		if err != nil {
+			return nil, err
+		}
+		result = append(result, resources)
+	}
+	return result, nil
+}
+
+func (d *memCacheClient) ServerGroups() (*metav1.APIGroupList, error) {
+	d.lock.RLock()
+	defer d.lock.RUnlock()
+	if d.groupList == nil {
+		return nil, ErrCacheEmpty
+	}
+	return d.groupList, nil
+}
+
+func (d *memCacheClient) RESTClient() restclient.Interface {
+	return d.delegate.RESTClient()
+}
+
+// TODO: Should this also be cached? The results seem more likely to be
+// inconsistent with ServerGroups and ServerResources given the requirement to
+// actively Invalidate.
+func (d *memCacheClient) ServerPreferredResources() ([]*metav1.APIResourceList, error) {
+	return d.delegate.ServerPreferredResources()
+}
+
+// TODO: Should this also be cached? The results seem more likely to be
+// inconsistent with ServerGroups and ServerResources given the requirement to
+// actively Invalidate.
+func (d *memCacheClient) ServerPreferredNamespacedResources() ([]*metav1.APIResourceList, error) {
+	return d.delegate.ServerPreferredNamespacedResources()
+}
+
+func (d *memCacheClient) ServerVersion() (*version.Info, error) {
+	return d.delegate.ServerVersion()
+}
+
+func (d *memCacheClient) SwaggerSchema(version schema.GroupVersion) (*swagger.ApiDeclaration, error) {
+	return d.delegate.SwaggerSchema(version)
+}
+
+func (d *memCacheClient) OpenAPISchema() (*openapi_v2.Document, error) {
+	return d.delegate.OpenAPISchema()
+}
+
+func (d *memCacheClient) Fresh() bool {
+	d.lock.RLock()
+	defer d.lock.RUnlock()
+	// Fresh is supposed to tell the caller whether or not to retry if the cache
+	// fails to find something. The idea here is that Invalidate will be called
+	// periodically and therefore we'll always be returning the latest data. (And
+	// in the future we can watch and stay even more up-to-date.) So we only
+	// return false if the cache has never been filled.
+	return d.cacheValid
+}
+
+// Invalidate refreshes the cache, blocking calls until the cache has been
+// refreshed. It would be trivial to make a version that does this in the
+// background while continuing to respond to requests if needed.
+func (d *memCacheClient) Invalidate() {
+	d.lock.Lock()
+	defer d.lock.Unlock()
+
+	// TODO: Could this multiplicative set of calls be replaced by a single call
+	// to ServerResources? If it's possible for more than one resulting
+	// APIResourceList to have the same GroupVersion, the lists would need merged.
+	gl, err := d.delegate.ServerGroups()
+	if err != nil || len(gl.Groups) == 0 {
+		utilruntime.HandleError(fmt.Errorf("couldn't get current server API group list; will keep using cached value. (%v)", err))
+		return
+	}
+
+	rl := map[string]*metav1.APIResourceList{}
+	for _, g := range gl.Groups {
+		for _, v := range g.Versions {
+			r, err := d.delegate.ServerResourcesForGroupVersion(v.GroupVersion)
+			if err != nil || len(r.APIResources) == 0 {
+				utilruntime.HandleError(fmt.Errorf("couldn't get resource list for %v: %v", v.GroupVersion, err))
+				if cur, ok := d.groupToServerResources[v.GroupVersion]; ok {
+					// retain the existing list, if we had it.
+					r = cur
+				} else {
+					continue
+				}
+			}
+			rl[v.GroupVersion] = r
+		}
+	}
+
+	d.groupToServerResources, d.groupList = rl, gl
+	d.cacheValid = true
+}
+
+// NewMemCacheClient creates a new CachedDiscoveryInterface which caches
+// discovery information in memory and will stay up-to-date if Invalidate is
+// called with regularity.
+//
+// NOTE: The client will NOT resort to live lookups on cache misses.
+func NewMemCacheClient(delegate discovery.DiscoveryInterface) discovery.CachedDiscoveryInterface {
+	return &memCacheClient{
+		delegate:               delegate,
+		groupToServerResources: map[string]*metav1.APIResourceList{},
+	}
+}

staging/src/k8s.io/client-go/discovery/cached/memcache_test.go

@@ -0,0 +1,132 @@
+/*
+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 cached
+
+import (
+	"errors"
+	"reflect"
+	"sync"
+	"testing"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/discovery/fake"
+)
+
+type fakeDiscovery struct {
+	*fake.FakeDiscovery
+
+	lock        sync.Mutex
+	groupList   *metav1.APIGroupList
+	resourceMap map[string]*metav1.APIResourceList
+}
+
+func (c *fakeDiscovery) ServerResourcesForGroupVersion(groupVersion string) (*metav1.APIResourceList, error) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	if rl, ok := c.resourceMap[groupVersion]; ok {
+		return rl, nil
+	}
+	return nil, errors.New("doesn't exist")
+}
+
+func (c *fakeDiscovery) ServerGroups() (*metav1.APIGroupList, error) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	if c.groupList == nil {
+		return nil, errors.New("doesn't exist")
+	}
+	return c.groupList, nil
+}
+
+func TestClient(t *testing.T) {
+	fake := &fakeDiscovery{
+		groupList: &metav1.APIGroupList{
+			Groups: []metav1.APIGroup{{
+				Name: "astronomy",
+				Versions: []metav1.GroupVersionForDiscovery{{
+					GroupVersion: "astronomy/v8beta1",
+					Version:      "v8beta1",
+				}},
+			}},
+		},
+		resourceMap: map[string]*metav1.APIResourceList{
+			"astronomy/v8beta1": {
+				GroupVersion: "astronomy/v8beta1",
+				APIResources: []metav1.APIResource{{
+					Name:         "dwarfplanets",
+					SingularName: "dwarfplanet",
+					Namespaced:   true,
+					Kind:         "DwarfPlanet",
+					ShortNames:   []string{"dp"},
+				}},
+			},
+		},
+	}
+
+	c := NewMemCacheClient(fake)
+	g, err := c.ServerGroups()
+	if err == nil {
+		t.Errorf("Unexpected non-error.")
+	}
+	if c.Fresh() {
+		t.Errorf("Expected not fresh.")
+	}
+
+	c.Invalidate()
+	if !c.Fresh() {
+		t.Errorf("Expected fresh.")
+	}
+
+	g, err = c.ServerGroups()
+	if err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+	if e, a := fake.groupList, g; !reflect.DeepEqual(e, a) {
+		t.Errorf("Expected %#v, got %#v", e, a)
+	}
+	r, err := c.ServerResourcesForGroupVersion("astronomy/v8beta1")
+	if err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+	if e, a := fake.resourceMap["astronomy/v8beta1"], r; !reflect.DeepEqual(e, a) {
+		t.Errorf("Expected %#v, got %#v", e, a)
+	}
+
+	fake.lock.Lock()
+	fake.resourceMap = map[string]*metav1.APIResourceList{
+		"astronomy/v8beta1": {
+			GroupVersion: "astronomy/v8beta1",
+			APIResources: []metav1.APIResource{{
+				Name:         "stars",
+				SingularName: "star",
+				Namespaced:   true,
+				Kind:         "Star",
+				ShortNames:   []string{"s"},
+			}},
+		},
+	}
+	fake.lock.Unlock()
+
+	c.Invalidate()
+	r, err = c.ServerResourcesForGroupVersion("astronomy/v8beta1")
+	if err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+	if e, a := fake.resourceMap["astronomy/v8beta1"], r; !reflect.DeepEqual(e, a) {
+		t.Errorf("Expected %#v, got %#v", e, a)
+	}
+}

staging/src/k8s.io/client-go/discovery/discovery_client.go

@@ -55,7 +55,11 @@ type DiscoveryInterface interface {
 // CachedDiscoveryInterface is a DiscoveryInterface with cache invalidation and freshness.
 type CachedDiscoveryInterface interface {
 	DiscoveryInterface
-	// Fresh returns true if no cached data was used that had been retrieved before the instantiation.
+	// Fresh is supposed to tell the caller whether or not to retry if the cache
+	// fails to find something (false = retry, true = no need to retry).
+	//
+	// TODO: this needs to be revisited, this interface can't be locked properly
+	// and doesn't make a lot of sense.
 	Fresh() bool
 	// Invalidate enforces that no cached data is used in the future that is older than the current time.
 	Invalidate()