Expose kms timeout value via encryption config.

staging/src/k8s.io/apiserver/pkg/apis/config/types.go

@@ -123,4 +123,7 @@ type KMSConfiguration struct {
 	CacheSize int32
 	// endpoint is the gRPC server listening address, for example "unix:///var/run/kms-provider.sock".
 	Endpoint string
+	// Timeout for gRPC calls to kms-plugin (ex. 5s). The default is 3 seconds.
+	// +optional
+	Timeout *metav1.Duration
 }

staging/src/k8s.io/apiserver/pkg/apis/config/v1/types.go

@@ -84,4 +84,7 @@ type KMSConfiguration struct {
 	CacheSize int32 `json:"cachesize,omitempty"`
 	// endpoint is the gRPC server listening address, for example "unix:///var/run/kms-provider.sock".
 	Endpoint string `json:"endpoint"`
+	// Timeout for gRPC calls to kms-plugin (ex. 5s). The default is 3 seconds.
+	// +optional
+	Timeout *metav1.Duration `json:"timeout,omitempty"`
 }

staging/src/k8s.io/apiserver/pkg/apis/config/v1/zz_generated.conversion.go

@@ -23,6 +23,7 @@ package v1
 import (
 	unsafe "unsafe"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	conversion "k8s.io/apimachinery/pkg/conversion"
 	runtime "k8s.io/apimachinery/pkg/runtime"
 	config "k8s.io/apiserver/pkg/apis/config"
@@ -180,6 +181,7 @@ func autoConvert_v1_KMSConfiguration_To_config_KMSConfiguration(in *KMSConfigura
 	out.Name = in.Name
 	out.CacheSize = in.CacheSize
 	out.Endpoint = in.Endpoint
+	out.Timeout = (*metav1.Duration)(unsafe.Pointer(in.Timeout))
 	return nil
 }
 
@@ -192,6 +194,7 @@ func autoConvert_config_KMSConfiguration_To_v1_KMSConfiguration(in *config.KMSCo
 	out.Name = in.Name
 	out.CacheSize = in.CacheSize
 	out.Endpoint = in.Endpoint
+	out.Timeout = (*metav1.Duration)(unsafe.Pointer(in.Timeout))
 	return nil
 }
 

staging/src/k8s.io/apiserver/pkg/apis/config/v1/zz_generated.deepcopy.go

@@ -21,6 +21,7 @@ limitations under the License.
 package v1
 
 import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	runtime "k8s.io/apimachinery/pkg/runtime"
 )
 
@@ -96,6 +97,11 @@ func (in *IdentityConfiguration) DeepCopy() *IdentityConfiguration {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *KMSConfiguration) DeepCopyInto(out *KMSConfiguration) {
 	*out = *in
+	if in.Timeout != nil {
+		in, out := &in.Timeout, &out.Timeout
+		*out = new(metav1.Duration)
+		**out = **in
+	}
 	return
 }
 
@@ -151,7 +157,7 @@ func (in *ProviderConfiguration) DeepCopyInto(out *ProviderConfiguration) {
 	if in.KMS != nil {
 		in, out := &in.KMS, &out.KMS
 		*out = new(KMSConfiguration)
-		**out = **in
+		(*in).DeepCopyInto(*out)
 	}
 	return
 }

staging/src/k8s.io/apiserver/pkg/apis/config/zz_generated.deepcopy.go

@@ -21,6 +21,7 @@ limitations under the License.
 package config
 
 import (
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	runtime "k8s.io/apimachinery/pkg/runtime"
 )
 
@@ -112,6 +113,11 @@ func (in *IdentityConfiguration) DeepCopy() *IdentityConfiguration {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *KMSConfiguration) DeepCopyInto(out *KMSConfiguration) {
 	*out = *in
+	if in.Timeout != nil {
+		in, out := &in.Timeout, &out.Timeout
+		*out = new(v1.Duration)
+		**out = **in
+	}
 	return
 }
 
@@ -186,7 +192,7 @@ func (in *ProviderConfiguration) DeepCopyInto(out *ProviderConfiguration) {
 	if in.KMS != nil {
 		in, out := &in.KMS, &out.KMS
 		*out = new(KMSConfiguration)
-		**out = **in
+		(*in).DeepCopyInto(*out)
 	}
 	return
 }

staging/src/k8s.io/apiserver/pkg/server/options/encryptionconfig/config.go

@@ -173,8 +173,16 @@ func GetPrefixTransformers(config *apiserverconfig.ResourceConfiguration) ([]val
 				return nil, fmt.Errorf("remote KMS provider can't use empty string as endpoint")
 			}
 
+			timeout := kmsPluginConnectionTimeout
+			if provider.KMS.Timeout != nil {
+				if provider.KMS.Timeout.Duration < 0 {
+					return nil, fmt.Errorf("could not configure KMS plugin %q, timeout should be positive value", provider.KMS.Name)
+				}
+				timeout = provider.KMS.Timeout.Duration
+			}
+
 			// Get gRPC client service with endpoint.
-			envelopeService, err := envelopeServiceFactory(provider.KMS.Endpoint, kmsPluginConnectionTimeout)
+			envelopeService, err := envelopeServiceFactory(provider.KMS.Endpoint, timeout)
 			if err != nil {
 				return nil, fmt.Errorf("could not configure KMS plugin %q, error: %v", provider.KMS.Name, err)
 			}

staging/src/k8s.io/apiserver/pkg/server/options/encryptionconfig/config_test.go

@@ -419,3 +419,91 @@ func TestEncryptionProviderConfigNoEndpointForKMS(t *testing.T) {
 		t.Fatalf("invalid configuration file (kms has no endpoint) got parsed:\n%s", incorrectConfigNoEndpointForKMS)
 	}
 }
+
+func TestKMSConfigTimeout(t *testing.T) {
+	testCases := []struct {
+		desc    string
+		config  string
+		want    time.Duration
+		wantErr string
+	}{
+		{
+			desc: "duration explicitly provided",
+			config: `kind: EncryptionConfiguration
+apiVersion: apiserver.config.k8s.io/v1
+resources:
+  - resources:
+    - secrets
+    providers:
+    - kms:
+        name: foo
+        endpoint: unix:///tmp/testprovider.sock
+        timeout:   15s
+`,
+			want: 15 * time.Second,
+		},
+		{
+			desc: "duration explicitly provided as 0",
+			config: `kind: EncryptionConfiguration
+apiVersion: apiserver.config.k8s.io/v1
+resources:
+  - resources:
+    - secrets
+    providers:
+    - kms:
+        name: foo
+        endpoint: unix:///tmp/testprovider.sock
+        timeout:   0
+`,
+			want: 0,
+		},
+		{
+			desc: "duration is not provided, default will be supplied",
+			config: `kind: EncryptionConfiguration
+apiVersion: apiserver.config.k8s.io/v1
+resources:
+  - resources:
+    - secrets
+    providers:
+    - kms:
+        name: foo
+        endpoint: unix:///tmp/testprovider.sock
+`,
+			want: kmsPluginConnectionTimeout,
+		},
+		{
+			desc: "duration is invalid (negative), error should be returned",
+			config: `kind: EncryptionConfiguration
+apiVersion: apiserver.config.k8s.io/v1
+resources:
+  - resources:
+    - secrets
+    providers:
+    - kms:
+        name: foo
+        endpoint: unix:///tmp/testprovider.sock
+        timeout:   -15s
+
+`,
+			wantErr: "timeout should be positive value",
+		},
+	}
+
+	for _, tt := range testCases {
+		t.Run(tt.desc, func(t *testing.T) {
+			// mocking envelopeServiceFactory to sense the value of the supplied timeout.
+			envelopeServiceFactory = func(endpoint string, callTimeout time.Duration) (envelope.Service, error) {
+				if callTimeout != tt.want {
+					t.Fatalf("got timeout: %v, want %v", callTimeout, tt.want)
+				}
+
+				return newMockEnvelopeService(endpoint, callTimeout)
+			}
+
+			// mocked envelopeServiceFactory is called during ParseEncryptionConfiguration.
+			if _, err := ParseEncryptionConfiguration(strings.NewReader(tt.config)); err != nil && !strings.Contains(err.Error(), tt.wantErr) {
+				t.Fatalf("unable to parse yaml\n%s\nerror: %v", tt.config, err)
+			}
+		})
+	}
+}