Merge pull request #51765 from mitake/etcd3-compaction

cmd/kube-apiserver/app/options/options_test.go

@@ -114,11 +114,12 @@ func TestAddFlags(t *testing.T) {
 				ServerList: nil,
 				Prefix:     "/registry",
 				DeserializationCacheSize: 0,
-				Copier:   kapi.Scheme,
-				Quorum:   false,
-				KeyFile:  "/var/run/kubernetes/etcd.key",
-				CAFile:   "/var/run/kubernetes/etcdca.crt",
-				CertFile: "/var/run/kubernetes/etcdce.crt",
+				Copier:             kapi.Scheme,
+				Quorum:             false,
+				KeyFile:            "/var/run/kubernetes/etcd.key",
+				CAFile:             "/var/run/kubernetes/etcdca.crt",
+				CertFile:           "/var/run/kubernetes/etcdce.crt",
+				CompactionInterval: storagebackend.DefaultCompactInterval,
 			},
 			DefaultStorageMediaType: "application/vnd.kubernetes.protobuf",
 			DeleteCollectionWorkers: 1,

staging/src/k8s.io/apiserver/pkg/server/options/etcd.go

@@ -148,6 +148,10 @@ func (s *EtcdOptions) AddFlags(fs *pflag.FlagSet) {
 
 	fs.StringVar(&s.EncryptionProviderConfigFilepath, "experimental-encryption-provider-config", s.EncryptionProviderConfigFilepath,
 		"The file containing configuration for encryption providers to be used for storing secrets in etcd")
+
+	fs.DurationVar(&s.StorageConfig.CompactionInterval, "etcd-compaction-interval", s.StorageConfig.CompactionInterval,
+		"The interval of compaction requests. If 0, the compaction request from apiserver is disabled.")
+
 }
 
 func (s *EtcdOptions) ApplyTo(c *server.Config) error {

staging/src/k8s.io/apiserver/pkg/storage/etcd3/compact.go

@@ -27,8 +27,7 @@ import (
 )
 
 const (
-	compactInterval = 5 * time.Minute
-	compactRevKey   = "compact_rev_key"
+	compactRevKey = "compact_rev_key"
 )
 
 var (
@@ -44,7 +43,7 @@ func init() {
 // By default, we save the most recent 10 minutes data and compact versions > 10minutes ago.
 // It should be enough for slow watchers and to tolerate burst.
 // TODO: We might keep a longer history (12h) in the future once storage API can take advantage of past version of keys.
-func StartCompactor(ctx context.Context, client *clientv3.Client) {
+func StartCompactor(ctx context.Context, client *clientv3.Client, compactInterval time.Duration) {
 	endpointsMapMu.Lock()
 	defer endpointsMapMu.Unlock()
 
@@ -60,7 +59,9 @@ func StartCompactor(ctx context.Context, client *clientv3.Client) {
 		endpointsMap[ep] = struct{}{}
 	}
 
-	go compactor(ctx, client, compactInterval)
+	if compactInterval != 0 {
+		go compactor(ctx, client, compactInterval)
+	}
 }
 
 // compactor periodically compacts historical versions of keys in etcd.

staging/src/k8s.io/apiserver/pkg/storage/storagebackend/config.go

@@ -17,6 +17,8 @@ limitations under the License.
 package storagebackend
 
 import (
+	"time"
+
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apiserver/pkg/storage/value"
 )
@@ -25,6 +27,8 @@ const (
 	StorageTypeUnset = ""
 	StorageTypeETCD2 = "etcd2"
 	StorageTypeETCD3 = "etcd3"
+
+	DefaultCompactInterval = 5 * time.Minute
 )
 
 // Config is configuration for creating a storage backend.
@@ -55,6 +59,10 @@ type Config struct {
 	Copier runtime.ObjectCopier
 	// Transformer allows the value to be transformed prior to persisting into etcd.
 	Transformer value.Transformer
+
+	// CompactionInterval is an interval of requesting compaction from apiserver.
+	// If the value is 0, no compaction will be issued.
+	CompactionInterval time.Duration
 }
 
 func NewDefaultConfig(prefix string, copier runtime.ObjectCopier, codec runtime.Codec) *Config {
@@ -63,7 +71,8 @@ func NewDefaultConfig(prefix string, copier runtime.ObjectCopier, codec runtime.
 		// Default cache size to 0 - if unset, its size will be set based on target
 		// memory usage.
 		DeserializationCacheSize: 0,
-		Copier: copier,
-		Codec:  codec,
+		Copier:             copier,
+		Codec:              codec,
+		CompactionInterval: DefaultCompactInterval,
 	}
 }

staging/src/k8s.io/apiserver/pkg/storage/storagebackend/factory/etcd3.go

@@ -51,7 +51,7 @@ func newETCD3Storage(c storagebackend.Config) (storage.Interface, DestroyFunc, e
 		return nil, nil, err
 	}
 	ctx, cancel := context.WithCancel(context.Background())
-	etcd3.StartCompactor(ctx, client)
+	etcd3.StartCompactor(ctx, client, c.CompactionInterval)
 	destroyFunc := func() {
 		cancel()
 		client.Close()