Merge pull request #16398 from smarterclayton/disablecache

Automatic merge from submit-queue (batch tested with PRs 16546, 16398, 16157)

Backport upstream changes to watch cache enablement

Disables the watch cache for most resources by default, except those accessed by many clients. This has been shown to have minor impacts on the production workload.

Fixes #16112

contrib/completions/bash/openshift

@@ -32808,6 +32808,8 @@ _openshift_start_kubernetes_apiserver()
     local_nonpersistent_flags+=("--contention-profiling")
     flags+=("--cors-allowed-origins=")
     local_nonpersistent_flags+=("--cors-allowed-origins=")
+    flags+=("--default-watch-cache-size=")
+    local_nonpersistent_flags+=("--default-watch-cache-size=")
     flags+=("--delete-collection-workers=")
     local_nonpersistent_flags+=("--delete-collection-workers=")
     flags+=("--deserialization-cache-size=")

contrib/completions/zsh/openshift

@@ -32957,6 +32957,8 @@ _openshift_start_kubernetes_apiserver()
     local_nonpersistent_flags+=("--contention-profiling")
     flags+=("--cors-allowed-origins=")
     local_nonpersistent_flags+=("--cors-allowed-origins=")
+    flags+=("--default-watch-cache-size=")
+    local_nonpersistent_flags+=("--default-watch-cache-size=")
     flags+=("--delete-collection-workers=")
     local_nonpersistent_flags+=("--delete-collection-workers=")
     flags+=("--deserialization-cache-size=")

pkg/cmd/server/kubernetes/master/master_config.go

@@ -77,8 +77,6 @@ import (
 	"github.com/openshift/origin/pkg/version"
 )
 
-const DefaultWatchCacheSize = 1000
-
 // request paths that match this regular expression will be treated as long running
 // and not subjected to the default server timeout.
 const originLongRunningEndpointsRE = "(/|^)(buildconfigs/.*/instantiatebinary|imagestreamimports)$"
@@ -147,7 +145,7 @@ func BuildKubeAPIserverOptions(masterConfig configapi.MasterConfig) (*kapiserver
 	server.Etcd.StorageConfig.KeyFile = masterConfig.EtcdClientInfo.ClientCert.KeyFile
 	server.Etcd.StorageConfig.CertFile = masterConfig.EtcdClientInfo.ClientCert.CertFile
 	server.Etcd.StorageConfig.CAFile = masterConfig.EtcdClientInfo.CA
-	server.Etcd.DefaultWatchCacheSize = DefaultWatchCacheSize
+	server.Etcd.DefaultWatchCacheSize = 0
 
 	server.GenericServerRunOptions.CorsAllowedOriginList = masterConfig.CORSAllowedOrigins
 	server.GenericServerRunOptions.MaxRequestsInFlight = masterConfig.ServingInfo.MaxRequestsInFlight
@@ -438,6 +436,20 @@ func buildKubeApiserverConfig(
 		return originLongRunningRequestRE.MatchString(r.URL.Path) || kubeLongRunningFunc(r, requestInfo)
 	}
 
+	if apiserverOptions.Etcd.EnableWatchCache {
+		glog.V(2).Infof("Initializing cache sizes based on %dMB limit", apiserverOptions.GenericServerRunOptions.TargetRAMMB)
+		sizes := cachesize.NewHeuristicWatchCacheSizes(apiserverOptions.GenericServerRunOptions.TargetRAMMB)
+		if userSpecified, err := genericoptions.ParseWatchCacheSizes(apiserverOptions.Etcd.WatchCacheSizes); err == nil {
+			for resource, size := range userSpecified {
+				sizes[resource] = size
+			}
+		}
+		apiserverOptions.Etcd.WatchCacheSizes, err = genericoptions.WriteWatchCacheSizes(sizes)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	if err := apiserverOptions.Etcd.ApplyWithStorageFactoryTo(storageFactory, genericConfig); err != nil {
 		return nil, err
 	}
@@ -523,12 +535,6 @@ func buildKubeApiserverConfig(
 		EnableCoreControllers: true,
 	}
 
-	if apiserverOptions.Etcd.EnableWatchCache {
-		// TODO(rebase): upstream also does the following:
-		// cachesize.InitializeWatchCacheSizes(s.GenericServerRunOptions.TargetRAMMB)
-		cachesize.SetWatchCacheSizes(apiserverOptions.GenericServerRunOptions.WatchCacheSizes)
-	}
-
 	if kubeApiserverConfig.EnableCoreControllers {
 		ttl := masterConfig.KubernetesMasterConfig.MasterEndpointReconcileTTL
 		interval := ttl * 2 / 3

pkg/cmd/server/kubernetes/master/master_test.go

@@ -25,8 +25,7 @@ func TestNewMasterLeasesHasCorrectTTL(t *testing.T) {
 	}
 
 	restOptions := generic.RESTOptions{StorageConfig: etcdStorage, Decorator: generic.UndecoratedStorage, DeleteCollectionWorkers: 1}
-	watchCacheDisabled := 0
-	storageInterface, _ := restOptions.Decorator(kapi.Scheme, restOptions.StorageConfig, &watchCacheDisabled, nil, "masterleases", nil, nil, nil, nil)
+	storageInterface, _ := restOptions.Decorator(kapi.Scheme, restOptions.StorageConfig, nil, "masterleases", nil, nil, nil, nil)
 	defer server.Terminate(t)
 
 	masterLeases := newMasterLeases(storageInterface, 15)

pkg/security/registry/securitycontextconstraints/etcd/etcd.go

@@ -6,7 +6,6 @@ import (
 	"k8s.io/apiserver/pkg/registry/generic/registry"
 	"k8s.io/apiserver/pkg/registry/rest"
 	"k8s.io/kubernetes/pkg/api"
-	"k8s.io/kubernetes/pkg/registry/cachesize"
 
 	securityapi "github.com/openshift/origin/pkg/security/apis/security"
 	"github.com/openshift/origin/pkg/security/registry/securitycontextconstraints"
@@ -30,7 +29,6 @@ func NewREST(optsGetter generic.RESTOptionsGetter) *REST {
 		},
 		PredicateFunc:            securitycontextconstraints.Matcher,
 		DefaultQualifiedResource: securityapi.Resource("securitycontextconstraints"),
-		WatchCacheSize:           cachesize.GetWatchCacheSizeByResource("securitycontextconstraints"),
 
 		CreateStrategy:      securitycontextconstraints.Strategy,
 		UpdateStrategy:      securitycontextconstraints.Strategy,

pkg/util/restoptions/configgetter.go

@@ -1,22 +1,16 @@
 package restoptions
 
 import (
-	"fmt"
-	"strconv"
-	"strings"
 	"sync"
 
-	"k8s.io/apimachinery/pkg/runtime"
+	"github.com/golang/glog"
+
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	kerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apiserver/pkg/registry/generic"
 	"k8s.io/apiserver/pkg/registry/generic/registry"
+	"k8s.io/apiserver/pkg/server/options"
 	serverstorage "k8s.io/apiserver/pkg/server/storage"
-	"k8s.io/apiserver/pkg/storage"
-	"k8s.io/apiserver/pkg/storage/storagebackend"
-	"k8s.io/apiserver/pkg/storage/storagebackend/factory"
 
-	"github.com/golang/glog"
 	configapi "github.com/openshift/origin/pkg/cmd/server/api"
 	kubernetes "github.com/openshift/origin/pkg/cmd/server/kubernetes/master"
 )
@@ -41,7 +35,6 @@ type configRESTOptionsGetter struct {
 }
 
 // NewConfigGetter returns a restoptions.Getter implemented using information from the provided master config.
-// By default, the etcd watch cache is enabled with a size of 1000 per resource type.
 // TODO: this class should either not need to know about configapi.MasterConfig, or not be in pkg/util
 func NewConfigGetter(masterOptions configapi.MasterConfig, defaultResourceConfig *serverstorage.ResourceConfig, resourcePrefixOverrides map[schema.GroupResource]string, enforcedStorageVersions map[schema.GroupResource]schema.GroupVersion, quorumResources map[schema.GroupResource]struct{}) (Getter, error) {
 	apiserverOptions, err := kubernetes.BuildKubeAPIserverOptions(masterOptions)
@@ -55,27 +48,24 @@ func NewConfigGetter(masterOptions configapi.MasterConfig, defaultResourceConfig
 	storageFactory.DefaultResourcePrefixes = resourcePrefixOverrides
 	storageFactory.StorageConfig.Prefix = masterOptions.EtcdStorageConfig.OpenShiftStoragePrefix
 
-	// TODO: refactor vendor/k8s.io/kubernetes/pkg/registry/cachesize to remove our custom cache size code
-	errs := []error{}
-	cacheSizes := map[schema.GroupResource]int{}
-	for _, c := range apiserverOptions.GenericServerRunOptions.WatchCacheSizes {
-		tokens := strings.Split(c, "#")
-		if len(tokens) != 2 {
-			errs = append(errs, fmt.Errorf("invalid watch cache size value '%s', expecting <resource>#<size> format (e.g. builds#100)", c))
-			continue
+	// perform watch cache heuristic like upstream
+	if apiserverOptions.Etcd.EnableWatchCache {
+		glog.V(2).Infof("Initializing cache sizes based on %dMB limit", apiserverOptions.GenericServerRunOptions.TargetRAMMB)
+		sizes := newHeuristicWatchCacheSizes(apiserverOptions.GenericServerRunOptions.TargetRAMMB)
+		if userSpecified, err := options.ParseWatchCacheSizes(apiserverOptions.Etcd.WatchCacheSizes); err == nil {
+			for resource, size := range userSpecified {
+				sizes[resource] = size
+			}
 		}
-
-		resource := schema.ParseGroupResource(tokens[0])
-
-		size, err := strconv.Atoi(tokens[1])
+		apiserverOptions.Etcd.WatchCacheSizes, err = options.WriteWatchCacheSizes(sizes)
 		if err != nil {
-			errs = append(errs, fmt.Errorf("invalid watch cache size value '%s': %v", c, err))
-			continue
+			return nil, err
 		}
-		cacheSizes[resource] = size
 	}
-	if len(errs) > 0 {
-		return nil, kerrors.NewAggregate(errs)
+
+	cacheSizes, err := options.ParseWatchCacheSizes(apiserverOptions.Etcd.WatchCacheSizes)
+	if err != nil {
+		return nil, err
 	}
 
 	return &configRESTOptionsGetter{
@@ -108,41 +98,14 @@ func (g *configRESTOptionsGetter) GetRESTOptions(resource schema.GroupResource)
 		config.Quorum = true
 	}
 
-	configuredCacheSize, specified := g.cacheSizes[resource]
-	if !specified || configuredCacheSize < 0 {
-		configuredCacheSize = g.defaultCacheSize
-	}
-	storageWithCacher := registry.StorageWithCacher(configuredCacheSize)
-
-	decorator := func(
-		copier runtime.ObjectCopier,
-		storageConfig *storagebackend.Config,
-		requestedSize *int,
-		objectType runtime.Object,
-		resourcePrefix string,
-		keyFunc func(obj runtime.Object) (string, error),
-		newListFn func() runtime.Object,
-		getAttrsFunc storage.AttrFunc,
-		triggerFn storage.TriggerPublisherFunc,
-	) (storage.Interface, factory.DestroyFunc) {
-		// use the origin default cache size, not the one in registry.StorageWithCacher
-		capacity := &configuredCacheSize
-		if requestedSize != nil {
-			capacity = requestedSize
-		}
-
-		if *capacity == 0 || !g.cacheEnabled {
-			glog.V(5).Infof("using uncached watch storage for %s (quorum=%t)", resource.String(), storageConfig.Quorum)
-			return generic.UndecoratedStorage(copier, storageConfig, capacity, objectType, resourcePrefix, keyFunc, newListFn, getAttrsFunc, triggerFn)
-		}
-
-		glog.V(5).Infof("using watch cache storage (capacity=%v, quorum=%t) for %s %#v", *capacity, storageConfig.Quorum, resource.String(), storageConfig)
-		return storageWithCacher(copier, storageConfig, capacity, objectType, resourcePrefix, keyFunc, newListFn, getAttrsFunc, triggerFn)
+	cacheSize, ok := g.cacheSizes[resource]
+	if !ok {
+		cacheSize = g.defaultCacheSize
 	}
 
 	resourceOptions := generic.RESTOptions{
 		StorageConfig:           config,
-		Decorator:               decorator,
+		Decorator:               registry.StorageWithCacher(cacheSize),
 		DeleteCollectionWorkers: g.deleteCollectionWorkers,
 		EnableGarbageCollection: g.enableGarbageCollection,
 		ResourcePrefix:          g.storageFactory.ResourcePrefix(resource),
@@ -151,3 +114,25 @@ func (g *configRESTOptionsGetter) GetRESTOptions(resource schema.GroupResource)
 
 	return resourceOptions, nil
 }
+
+// newHeuristicWatchCacheSizes returns a map of suggested watch cache sizes based on total
+// memory. It reuses the upstream heuristic and adds OpenShift specific resources.
+func newHeuristicWatchCacheSizes(expectedRAMCapacityMB int) map[schema.GroupResource]int {
+	// TODO: Revisit this heuristic, copied from upstream
+	clusterSize := expectedRAMCapacityMB / 60
+
+	// default enable watch caches for resources that will have a high number of clients accessing it
+	// and where the write rate may be significant
+	watchCacheSizes := make(map[schema.GroupResource]int)
+	watchCacheSizes[schema.GroupResource{Group: "network.openshift.io", Resource: "hostsubnets"}] = maxInt(5*clusterSize, 100)
+	watchCacheSizes[schema.GroupResource{Group: "network.openshift.io", Resource: "netnamespaces"}] = maxInt(5*clusterSize, 100)
+	watchCacheSizes[schema.GroupResource{Group: "network.openshift.io", Resource: "egressnetworkpolicies"}] = maxInt(10*clusterSize, 100)
+	return watchCacheSizes
+}
+
+func maxInt(a, b int) int {
+	if a > b {
+		return a
+	}
+	return b
+}

test/integration/watch_cache_test.go

@@ -15,7 +15,6 @@ import (
 	coreclient "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/core/internalversion"
 
 	configapi "github.com/openshift/origin/pkg/cmd/server/api"
-	serverkube "github.com/openshift/origin/pkg/cmd/server/kubernetes/master"
 	testutil "github.com/openshift/origin/test/util"
 	testserver "github.com/openshift/origin/test/util/server"
 )
@@ -124,9 +123,13 @@ func TestDefaultWatchCacheSize(t *testing.T) {
 	etcdOptions := apiserveroptions.NewEtcdOptions(&storagebackend.Config{})
 	kubeDefaultCacheSize := etcdOptions.DefaultWatchCacheSize
 	if kubeDefaultCacheSize != 100 {
-		t.Fatalf("upstream DefaultWatchCacheSize changed from 100 to %q", kubeDefaultCacheSize)
+		t.Fatalf("upstream DefaultWatchCacheSize changed to %d", kubeDefaultCacheSize)
 	}
-	testWatchCacheWithConfig(t, master, serverkube.DefaultWatchCacheSize, kubeDefaultCacheSize)
+	if master.KubernetesMasterConfig.APIServerArguments == nil {
+		master.KubernetesMasterConfig.APIServerArguments = configapi.ExtendedArguments{}
+	}
+	master.KubernetesMasterConfig.APIServerArguments["watch-cache-sizes"] = []string{"namespaces#100"}
+	testWatchCacheWithConfig(t, master, 100, 0)
 }
 
 func TestWatchCacheSizeWithFlag(t *testing.T) {
@@ -140,5 +143,5 @@ func TestWatchCacheSizeWithFlag(t *testing.T) {
 	}
 	master.KubernetesMasterConfig.APIServerArguments["watch-cache-sizes"] = []string{"namespaces#2000"}
 
-	testWatchCacheWithConfig(t, master, 2000, serverkube.DefaultWatchCacheSize)
+	testWatchCacheWithConfig(t, master, 2000, 0)
 }