Allow kube-apiserver to test the status of kms-plugin.

staging/src/k8s.io/apiserver/go.mod

@@ -24,6 +24,7 @@ require (
 	github.com/go-openapi/swag v0.17.2 // indirect
 	github.com/gogo/protobuf v0.0.0-20171007142547-342cbe0a0415
 	github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b // indirect
+	github.com/google/go-cmp v0.3.0
 	github.com/google/gofuzz v1.0.0
 	github.com/googleapis/gnostic v0.0.0-20170729233727-0c5108395e2d
 	github.com/gorilla/websocket v1.4.0 // indirect

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

@@ -16,7 +16,9 @@ limitations under the License.
 
 package v1
 
-import metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 

staging/src/k8s.io/apiserver/pkg/server/options/BUILD

@@ -57,6 +57,7 @@ go_library(
         "//staging/src/k8s.io/apiserver/pkg/registry/generic/registry:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/server:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/server/healthz:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/server/options/encryptionconfig:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/server/resourceconfig:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/server/storage:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage/storagebackend:go_default_library",

staging/src/k8s.io/apiserver/pkg/server/options/encryptionconfig/BUILD

@@ -17,6 +17,7 @@ go_library(
         "//staging/src/k8s.io/apimachinery/pkg/runtime/serializer:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/apis/config:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/apis/config/v1:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/server/healthz:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage/value:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/aes:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope:go_default_library",
@@ -35,6 +36,7 @@ go_test(
         "//staging/src/k8s.io/apiserver/pkg/apis/config:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage/value:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope:go_default_library",
+        "//vendor/github.com/google/go-cmp/cmp:go_default_library",
     ],
 )
 

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

@@ -23,7 +23,9 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"net/http"
 	"os"
+	"sync"
 	"time"
 
 	"k8s.io/apimachinery/pkg/runtime"
@@ -31,6 +33,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/serializer"
 	apiserverconfig "k8s.io/apiserver/pkg/apis/config"
 	apiserverconfigv1 "k8s.io/apiserver/pkg/apis/config/v1"
+	"k8s.io/apiserver/pkg/server/healthz"
 	"k8s.io/apiserver/pkg/storage/value"
 	aestransformer "k8s.io/apiserver/pkg/storage/value/encrypt/aes"
 	"k8s.io/apiserver/pkg/storage/value/encrypt/envelope"
@@ -44,8 +47,113 @@ const (
 	secretboxTransformerPrefixV1 = "k8s:enc:secretbox:v1:"
 	kmsTransformerPrefixV1       = "k8s:enc:kms:v1:"
 	kmsPluginConnectionTimeout   = 3 * time.Second
+	kmsPluginHealthzTTL          = 3 * time.Second
 )
 
+type kmsPluginHealthzResponse struct {
+	err      error
+	received time.Time
+}
+
+type kmsPluginProbe struct {
+	name string
+	envelope.Service
+	lastResponse *kmsPluginHealthzResponse
+	l            *sync.Mutex
+}
+
+func (h *kmsPluginProbe) toHealthzCheck(idx int) healthz.HealthzChecker {
+	return healthz.NamedCheck(fmt.Sprintf("kms-provider-%d", idx), func(r *http.Request) error {
+		return h.Check()
+	})
+}
+
+// GetKMSPluginHealthzCheckers extracts KMSPluginProbes from the EncryptionConfig.
+func GetKMSPluginHealthzCheckers(filepath string) ([]healthz.HealthzChecker, error) {
+	f, err := os.Open(filepath)
+	if err != nil {
+		return nil, fmt.Errorf("error opening encryption provider configuration file %q: %v", filepath, err)
+	}
+	defer f.Close()
+	var result []healthz.HealthzChecker
+	probes, err := getKMSPluginProbes(f)
+	if err != nil {
+		return nil, err
+	}
+
+	for i, p := range probes {
+		probe := p
+		result = append(result, probe.toHealthzCheck(i))
+	}
+	return result, nil
+}
+
+func getKMSPluginProbes(reader io.Reader) ([]*kmsPluginProbe, error) {
+	var result []*kmsPluginProbe
+
+	configFileContents, err := ioutil.ReadAll(reader)
+	if err != nil {
+		return result, fmt.Errorf("could not read content of encryption provider configuration: %v", err)
+	}
+
+	config, err := loadConfig(configFileContents)
+	if err != nil {
+		return result, fmt.Errorf("error while parsing encrypiton provider configuration: %v", err)
+	}
+
+	for _, r := range config.Resources {
+		for _, p := range r.Providers {
+			if p.KMS != nil {
+				timeout := kmsPluginConnectionTimeout
+				if p.KMS.Timeout != nil {
+					if p.KMS.Timeout.Duration <= 0 {
+						return nil, fmt.Errorf("could not configure KMS-Plugin's probe %q, timeout should be a positive value", p.KMS.Name)
+					}
+					timeout = p.KMS.Timeout.Duration
+				}
+
+				s, err := envelope.NewGRPCService(p.KMS.Endpoint, timeout)
+				if err != nil {
+					return nil, fmt.Errorf("could not configure KMS-Plugin's probe %q, error: %v", p.KMS.Name, err)
+				}
+
+				result = append(result, &kmsPluginProbe{
+					name:         p.KMS.Name,
+					Service:      s,
+					l:            &sync.Mutex{},
+					lastResponse: &kmsPluginHealthzResponse{},
+				})
+			}
+		}
+	}
+
+	return result, nil
+}
+
+// Check encrypts and decrypts test data against KMS-Plugin's gRPC endpoint.
+func (h *kmsPluginProbe) Check() error {
+	h.l.Lock()
+	defer h.l.Unlock()
+
+	if (time.Now().Sub(h.lastResponse.received)) < kmsPluginHealthzTTL {
+		return h.lastResponse.err
+	}
+
+	p, err := h.Service.Encrypt([]byte("ping"))
+	if err != nil {
+		h.lastResponse = &kmsPluginHealthzResponse{err: err, received: time.Now()}
+		return fmt.Errorf("failed to perform encrypt section of the healthz check for KMS Provider %s, error: %v", h.name, err)
+	}
+
+	if _, err := h.Service.Decrypt(p); err != nil {
+		h.lastResponse = &kmsPluginHealthzResponse{err: err, received: time.Now()}
+		return fmt.Errorf("failed to perform decrypt section of the healthz check for KMS Provider %s, error: %v", h.name, err)
+	}
+
+	h.lastResponse = &kmsPluginHealthzResponse{err: nil, received: time.Now()}
+	return nil
+}
+
 // GetTransformerOverrides returns the transformer overrides by reading and parsing the encryption provider configuration file
 func GetTransformerOverrides(filepath string) (map[schema.GroupResource]value.Transformer, error) {
 	f, err := os.Open(filepath)

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

@@ -24,6 +24,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/google/go-cmp/cmp"
+
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/diff"
 	apiserverconfig "k8s.io/apiserver/pkg/apis/config"
@@ -507,3 +509,99 @@ resources:
 		})
 	}
 }
+
+func TestKMSPluginHealthz(t *testing.T) {
+	service, err := envelope.NewGRPCService("unix:///tmp/testprovider.sock", kmsPluginConnectionTimeout)
+	if err != nil {
+		t.Fatalf("Could not initialize envelopeService, error: %v", err)
+	}
+
+	testCases := []struct {
+		desc    string
+		config  string
+		want    []*kmsPluginProbe
+		wantErr bool
+	}{
+		{
+			desc: "Install Healthz",
+			config: `kind: EncryptionConfiguration
+apiVersion: apiserver.config.k8s.io/v1
+resources:
+  - resources:
+    - secrets
+    providers:
+    - kms:
+        name: foo
+        endpoint: unix:///tmp/testprovider.sock
+        timeout:   15s
+`,
+			want: []*kmsPluginProbe{
+				{
+					name:    "foo",
+					Service: service,
+				},
+			},
+		},
+		{
+			desc: "Install multiple healthz",
+			config: `kind: EncryptionConfiguration
+apiVersion: apiserver.config.k8s.io/v1
+resources:
+  - resources:
+    - secrets
+    providers:
+    - kms:
+        name: foo
+        endpoint: unix:///tmp/testprovider.sock
+        timeout:   15s
+    - kms:
+        name: bar
+        endpoint: unix:///tmp/testprovider.sock
+        timeout:   15s
+`,
+			want: []*kmsPluginProbe{
+				{
+					name:    "foo",
+					Service: service,
+				},
+				{
+					name:    "bar",
+					Service: service,
+				},
+			},
+		},
+		{
+			desc: "No KMS Providers",
+			config: `kind: EncryptionConfiguration
+apiVersion: apiserver.config.k8s.io/v1
+resources:
+  - resources:
+    - secrets
+    providers:
+    - aesgcm:
+        keys:
+        - name: key1
+          secret: c2VjcmV0IGlzIHNlY3VyZQ==
+`,
+		},
+	}
+
+	for _, tt := range testCases {
+		t.Run(tt.desc, func(t *testing.T) {
+			got, err := getKMSPluginProbes(strings.NewReader(tt.config))
+			if err != nil && !tt.wantErr {
+				t.Fatalf("got %v, want nil for error", err)
+			}
+
+			if d := cmp.Diff(tt.want, got, cmp.Comparer(serviceComparer)); d != "" {
+				t.Fatalf("HealthzConfig mismatch (-want +got):\n%s", d)
+			}
+		})
+	}
+}
+
+// As long as got and want contain envelope.Service we will return true.
+// If got has an envelope.Service and want does note (or vice versa) this will return false.
+func serviceComparer(_, _ envelope.Service) bool {
+	return true
+}

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

@@ -31,6 +31,7 @@ import (
 	genericregistry "k8s.io/apiserver/pkg/registry/generic/registry"
 	"k8s.io/apiserver/pkg/server"
 	"k8s.io/apiserver/pkg/server/healthz"
+	"k8s.io/apiserver/pkg/server/options/encryptionconfig"
 	serverstorage "k8s.io/apiserver/pkg/server/storage"
 	"k8s.io/apiserver/pkg/storage/storagebackend"
 	storagefactory "k8s.io/apiserver/pkg/storage/storagebackend/factory"
@@ -204,6 +205,16 @@ func (s *EtcdOptions) addEtcdHealthEndpoint(c *server.Config) error {
 	c.HealthzChecks = append(c.HealthzChecks, healthz.NamedCheck("etcd", func(r *http.Request) error {
 		return healthCheck()
 	}))
+
+	if s.EncryptionProviderConfigFilepath != "" {
+		kmsPluginHealthzChecks, err := encryptionconfig.GetKMSPluginHealthzCheckers(s.EncryptionProviderConfigFilepath)
+		if err != nil {
+			return err
+		}
+
+		c.HealthzChecks = append(c.HealthzChecks, kmsPluginHealthzChecks...)
+	}
+
 	return nil
 }
 

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

@@ -23,6 +23,7 @@ import (
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
+	"k8s.io/apiserver/pkg/server"
 	"k8s.io/apiserver/pkg/storage/storagebackend"
 )
 
@@ -194,3 +195,48 @@ func TestParseWatchCacheSizes(t *testing.T) {
 		})
 	}
 }
+
+func TestKMSHealthzEndpoint(t *testing.T) {
+	testCases := []struct {
+		name                 string
+		encryptionConfigPath string
+		wantChecks           []string
+	}{
+		{
+			name:                 "single kms-provider, expect single kms healthz check",
+			encryptionConfigPath: "testdata/encryption-configs/single-kms-provider.yaml",
+			wantChecks:           []string{"etcd", "kms-provider-0"},
+		},
+		{
+			name:                 "two kms-providers, expect two kms healthz checks",
+			encryptionConfigPath: "testdata/encryption-configs/multiple-kms-providers.yaml",
+			wantChecks:           []string{"etcd", "kms-provider-0", "kms-provider-1"},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			serverConfig := &server.Config{}
+			etcdOptions := &EtcdOptions{
+				EncryptionProviderConfigFilepath: tc.encryptionConfigPath,
+			}
+			if err := etcdOptions.addEtcdHealthEndpoint(serverConfig); err != nil {
+				t.Fatalf("Failed to add healthz error: %v", err)
+			}
+
+			for _, n := range tc.wantChecks {
+				found := false
+				for _, h := range serverConfig.HealthzChecks {
+					if n == h.Name() {
+						found = true
+						break
+					}
+				}
+				if !found {
+					t.Errorf("Missing HealthzChecker %s", n)
+				}
+				found = false
+			}
+		})
+	}
+}

staging/src/k8s.io/apiserver/pkg/server/options/testdata/encryption-configs/multiple-kms-providers.yaml

@@ -0,0 +1,14 @@
+kind: EncryptionConfiguration
+apiVersion: apiserver.config.k8s.io/v1
+resources:
+  - resources:
+      - secrets
+    providers:
+      - kms:
+          name: kms-provider-1
+          cachesize: 1000
+          endpoint: unix:///@provider1.sock
+      - kms:
+          name: kms-provider-2
+          cachesize: 1000
+          endpoint: unix:///@provider2.sock

staging/src/k8s.io/apiserver/pkg/server/options/testdata/encryption-configs/single-kms-provider.yaml

@@ -0,0 +1,10 @@
+kind: EncryptionConfiguration
+apiVersion: apiserver.config.k8s.io/v1
+resources:
+  - resources:
+      - secrets
+    providers:
+      - kms:
+          name: kms-provider-1
+          cachesize: 1000
+          endpoint: unix:///@kms-provider.sock

test/integration/master/BUILD

@@ -135,57 +135,68 @@ go_library(
         "//test/integration/framework:go_default_library",
         "//vendor/github.com/coreos/etcd/clientv3:go_default_library",
         "//vendor/github.com/prometheus/client_golang/prometheus:go_default_library",
+        "//vendor/k8s.io/klog:go_default_library",
         "//vendor/sigs.k8s.io/yaml:go_default_library",
     ] + select({
         "@io_bazel_rules_go//go/platform:android": [
             "//staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/v1beta1:go_default_library",
             "//vendor/google.golang.org/grpc:go_default_library",
-            "//vendor/k8s.io/klog:go_default_library",
+            "//vendor/google.golang.org/grpc/codes:go_default_library",
+            "//vendor/google.golang.org/grpc/status:go_default_library",
         ],
         "@io_bazel_rules_go//go/platform:darwin": [
             "//staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/v1beta1:go_default_library",
             "//vendor/google.golang.org/grpc:go_default_library",
-            "//vendor/k8s.io/klog:go_default_library",
+            "//vendor/google.golang.org/grpc/codes:go_default_library",
+            "//vendor/google.golang.org/grpc/status:go_default_library",
         ],
         "@io_bazel_rules_go//go/platform:dragonfly": [
             "//staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/v1beta1:go_default_library",
             "//vendor/google.golang.org/grpc:go_default_library",
-            "//vendor/k8s.io/klog:go_default_library",
+            "//vendor/google.golang.org/grpc/codes:go_default_library",
+            "//vendor/google.golang.org/grpc/status:go_default_library",
         ],
         "@io_bazel_rules_go//go/platform:freebsd": [
             "//staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/v1beta1:go_default_library",
             "//vendor/google.golang.org/grpc:go_default_library",
-            "//vendor/k8s.io/klog:go_default_library",
+            "//vendor/google.golang.org/grpc/codes:go_default_library",
+            "//vendor/google.golang.org/grpc/status:go_default_library",
         ],
         "@io_bazel_rules_go//go/platform:linux": [
             "//staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/v1beta1:go_default_library",
             "//vendor/google.golang.org/grpc:go_default_library",
-            "//vendor/k8s.io/klog:go_default_library",
+            "//vendor/google.golang.org/grpc/codes:go_default_library",
+            "//vendor/google.golang.org/grpc/status:go_default_library",
         ],
         "@io_bazel_rules_go//go/platform:nacl": [
             "//staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/v1beta1:go_default_library",
             "//vendor/google.golang.org/grpc:go_default_library",
-            "//vendor/k8s.io/klog:go_default_library",
+            "//vendor/google.golang.org/grpc/codes:go_default_library",
+            "//vendor/google.golang.org/grpc/status:go_default_library",
         ],
         "@io_bazel_rules_go//go/platform:netbsd": [
             "//staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/v1beta1:go_default_library",
             "//vendor/google.golang.org/grpc:go_default_library",
-            "//vendor/k8s.io/klog:go_default_library",
+            "//vendor/google.golang.org/grpc/codes:go_default_library",
+            "//vendor/google.golang.org/grpc/status:go_default_library",
         ],
         "@io_bazel_rules_go//go/platform:openbsd": [
             "//staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/v1beta1:go_default_library",
             "//vendor/google.golang.org/grpc:go_default_library",
-            "//vendor/k8s.io/klog:go_default_library",
+            "//vendor/google.golang.org/grpc/codes:go_default_library",
+            "//vendor/google.golang.org/grpc/status:go_default_library",
         ],
         "@io_bazel_rules_go//go/platform:plan9": [
             "//staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/v1beta1:go_default_library",
             "//vendor/google.golang.org/grpc:go_default_library",
-            "//vendor/k8s.io/klog:go_default_library",
+            "//vendor/google.golang.org/grpc/codes:go_default_library",
+            "//vendor/google.golang.org/grpc/status:go_default_library",
         ],
         "@io_bazel_rules_go//go/platform:solaris": [
             "//staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/v1beta1:go_default_library",
             "//vendor/google.golang.org/grpc:go_default_library",
-            "//vendor/k8s.io/klog:go_default_library",
+            "//vendor/google.golang.org/grpc/codes:go_default_library",
+            "//vendor/google.golang.org/grpc/status:go_default_library",
         ],
         "//conditions:default": [],
     }),

test/integration/master/kms_plugin_mock.go

@@ -23,8 +23,11 @@ import (
 	"encoding/base64"
 	"fmt"
 	"net"
+	"sync"
 
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
 	kmsapi "k8s.io/apiserver/pkg/storage/value/encrypt/envelope/v1beta1"
 	"k8s.io/klog"
@@ -32,33 +35,32 @@ import (
 
 const (
 	kmsAPIVersion = "v1beta1"
-	sockFile      = "@kms-provider.sock"
 	unixProtocol  = "unix"
 )
 
 // base64Plugin gRPC sever for a mock KMS provider.
 // Uses base64 to simulate encrypt and decrypt.
 type base64Plugin struct {
-	grpcServer *grpc.Server
+	grpcServer         *grpc.Server
-	listener   net.Listener
+	listener           net.Listener
-
+	mu                 *sync.Mutex
-	// Allow users of the plugin to sense requests that were passed to KMS.
+	lastEncryptRequest *kmsapi.EncryptRequest
-	encryptRequest chan *kmsapi.EncryptRequest
+	inFailedState      bool
 }
 
-func newBase64Plugin() (*base64Plugin, error) {
+func newBase64Plugin(socketPath string) (*base64Plugin, error) {
-	listener, err := net.Listen(unixProtocol, sockFile)
+	listener, err := net.Listen(unixProtocol, socketPath)
 	if err != nil {
 		return nil, fmt.Errorf("failed to listen on the unix socket, error: %v", err)
 	}
-	klog.Infof("Listening on %s", sockFile)
+	klog.Infof("Listening on %s", socketPath)
 
 	server := grpc.NewServer()
 
 	result := &base64Plugin{
-		grpcServer:     server,
+		grpcServer: server,
-		listener:       listener,
+		listener:   listener,
-		encryptRequest: make(chan *kmsapi.EncryptRequest, 1),
+		mu:         &sync.Mutex{},
 	}
 
 	kmsapi.RegisterKeyManagementServiceServer(server, result)
@@ -73,6 +75,18 @@ func (s *base64Plugin) cleanUp() {
 
 var testProviderAPIVersion = kmsAPIVersion
 
+func (s *base64Plugin) enterFailedState() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.inFailedState = true
+}
+
+func (s *base64Plugin) exitFailedState() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.inFailedState = false
+}
+
 func (s *base64Plugin) Version(ctx context.Context, request *kmsapi.VersionRequest) (*kmsapi.VersionResponse, error) {
 	return &kmsapi.VersionResponse{Version: testProviderAPIVersion, RuntimeName: "testKMS", RuntimeVersion: "0.0.1"}, nil
 }
@@ -80,6 +94,12 @@ func (s *base64Plugin) Version(ctx context.Context, request *kmsapi.VersionReque
 func (s *base64Plugin) Decrypt(ctx context.Context, request *kmsapi.DecryptRequest) (*kmsapi.DecryptResponse, error) {
 	klog.Infof("Received Decrypt Request for DEK: %s", string(request.Cipher))
 
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.inFailedState {
+		return nil, status.Error(codes.FailedPrecondition, "failed precondition - key disabled")
+	}
+
 	buf := make([]byte, base64.StdEncoding.DecodedLen(len(request.Cipher)))
 	n, err := base64.StdEncoding.Decode(buf, request.Cipher)
 	if err != nil {
@@ -91,7 +111,13 @@ func (s *base64Plugin) Decrypt(ctx context.Context, request *kmsapi.DecryptReque
 
 func (s *base64Plugin) Encrypt(ctx context.Context, request *kmsapi.EncryptRequest) (*kmsapi.EncryptResponse, error) {
 	klog.Infof("Received Encrypt Request for DEK: %x", request.Plain)
-	s.encryptRequest <- request
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.lastEncryptRequest = request
+
+	if s.inFailedState {
+		return nil, status.Error(codes.FailedPrecondition, "failed precondition - key disabled")
+	}
 
 	buf := make([]byte, base64.StdEncoding.EncodedLen(len(request.Plain)))
 	base64.StdEncoding.Encode(buf, request.Plain)

test/integration/master/kms_transformation_test.go

@@ -24,52 +24,70 @@ import (
 	"crypto/aes"
 	"encoding/binary"
 	"fmt"
+
+	"net/http"
 	"strings"
 	"testing"
+	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apiserver/pkg/storage/value"
 	aestransformer "k8s.io/apiserver/pkg/storage/value/encrypt/aes"
-
 	kmsapi "k8s.io/apiserver/pkg/storage/value/encrypt/envelope/v1beta1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
 )
 
 const (
-	kmsPrefix     = "k8s:enc:kms:v1:grpc-kms-provider:"
 	dekKeySizeLen = 2
-
-	kmsConfigYAML = `
-kind: EncryptionConfiguration
-apiVersion: apiserver.config.k8s.io/v1
-resources:
-  - resources:
-    - secrets
-    providers:
-    - kms:
-       name: grpc-kms-provider
-       cachesize: 1000
-       endpoint: unix:///@kms-provider.sock
-`
 )
 
-// rawDEKKEKSecret provides operations for working with secrets transformed with Data Encryption Key(DEK) Key Encryption Kye(KEK) envelop.
+type envelope struct {
-type rawDEKKEKSecret []byte
+	providerName string
+	rawEnvelope  []byte
+	plainTextDEK []byte
+}
 
-func (r rawDEKKEKSecret) getDEKLen() int {
+func (r envelope) prefix() string {
+	return fmt.Sprintf("k8s:enc:kms:v1:%s:", r.providerName)
+}
+
+func (r envelope) prefixLen() int {
+	return len(r.prefix())
+}
+
+func (r envelope) dekLen() int {
 	// DEK's length is stored in the two bytes that follow the prefix.
-	return int(binary.BigEndian.Uint16(r[len(kmsPrefix) : len(kmsPrefix)+dekKeySizeLen]))
+	return int(binary.BigEndian.Uint16(r.rawEnvelope[r.prefixLen() : r.prefixLen()+dekKeySizeLen]))
 }
 
-func (r rawDEKKEKSecret) getDEK() []byte {
+func (r envelope) cipherTextDEK() []byte {
-	return r[len(kmsPrefix)+dekKeySizeLen : len(kmsPrefix)+dekKeySizeLen+r.getDEKLen()]
+	return r.rawEnvelope[r.prefixLen()+dekKeySizeLen : r.prefixLen()+dekKeySizeLen+r.dekLen()]
 }
 
-func (r rawDEKKEKSecret) getStartOfPayload() int {
+func (r envelope) startOfPayload(providerName string) int {
-	return len(kmsPrefix) + dekKeySizeLen + r.getDEKLen()
+	return r.prefixLen() + dekKeySizeLen + r.dekLen()
 }
 
-func (r rawDEKKEKSecret) getPayload() []byte {
+func (r envelope) cipherTextPayload() []byte {
-	return r[r.getStartOfPayload():]
+	return r.rawEnvelope[r.startOfPayload(r.providerName):]
+}
+
+func (r envelope) plainTextPayload(secretETCDPath string) ([]byte, error) {
+	block, err := aes.NewCipher(r.plainTextDEK)
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialize AES Cipher: %v", err)
+	}
+	// etcd path of the key is used as the authenticated context - need to pass it to decrypt
+	ctx := value.DefaultContext([]byte(secretETCDPath))
+	aescbcTransformer := aestransformer.NewCBCTransformer(block)
+	plainSecret, _, err := aescbcTransformer.TransformFromStorage(r.cipherTextPayload(), ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to transform from storage via AESCBC, err: %v", err)
+	}
+
+	return plainSecret, nil
 }
 
 // TestKMSProvider is an integration test between KubeAPI, ETCD and KMS Plugin
@@ -77,60 +95,77 @@ func (r rawDEKKEKSecret) getPayload() []byte {
 // 1. Raw records in ETCD that were processed by KMS Provider should be prefixed with k8s:enc:kms:v1:grpc-kms-provider-name:
 // 2. Data Encryption Key (DEK) should be generated by envelopeTransformer and passed to KMS gRPC Plugin
 // 3. KMS gRPC Plugin should encrypt the DEK with a Key Encryption Key (KEK) and pass it back to envelopeTransformer
-// 4. The payload (ex. Secret) should be encrypted via AES CBC transform
+// 4. The cipherTextPayload (ex. Secret) should be encrypted via AES CBC transform
 // 5. Prefix-EncryptedDEK-EncryptedPayload structure should be deposited to ETCD
 func TestKMSProvider(t *testing.T) {
-	pluginMock, err := newBase64Plugin()
+	encryptionConfig := `
+kind: EncryptionConfiguration
+apiVersion: apiserver.config.k8s.io/v1
+resources:
+  - resources:
+    - secrets
+    providers:
+    - kms:
+       name: kms-provider
+       cachesize: 1000
+       endpoint: unix:///@kms-provider.sock
+`
+
+	providerName := "kms-provider"
+	pluginMock, err := newBase64Plugin("@kms-provider.sock")
 	if err != nil {
 		t.Fatalf("failed to create mock of KMS Plugin: %v", err)
 	}
-	defer pluginMock.cleanUp()
-	serveErr := make(chan error, 1)
-	go func() {
-		serveErr <- pluginMock.grpcServer.Serve(pluginMock.listener)
-	}()
 
-	test, err := newTransformTest(t, kmsConfigYAML)
+	go pluginMock.grpcServer.Serve(pluginMock.listener)
+	defer pluginMock.cleanUp()
+	kmsPluginMustBeUp(t, pluginMock)
+
+	test, err := newTransformTest(t, encryptionConfig)
 	if err != nil {
-		t.Fatalf("failed to start KUBE API Server with encryptionConfig\n %s", kmsConfigYAML)
+		t.Fatalf("failed to start KUBE API Server with encryptionConfig\n %s, error: %v", encryptionConfig, err)
 	}
 	defer test.cleanUp()
 
-	// As part of newTransformTest a new secret was created, so KMS Mock should have been exercised by this point.
+	test.secret, err = test.createSecret(testSecret, testNamespace)
-	if len(serveErr) != 0 {
-		t.Fatalf("KMSPlugin failed while serving requests: %v", <-serveErr)
-	}
-
-	secretETCDPath := test.getETCDPath()
-	var rawSecretAsSeenByETCD rawDEKKEKSecret
-	rawSecretAsSeenByETCD, err = test.getRawSecretFromETCD()
 	if err != nil {
-		t.Fatalf("failed to read %s from etcd: %v", secretETCDPath, err)
+		t.Fatalf("Failed to create test secret, error: %v", err)
-	}
-
-	if !bytes.HasPrefix(rawSecretAsSeenByETCD, []byte(kmsPrefix)) {
-		t.Fatalf("expected secret to be prefixed with %s, but got %s", kmsPrefix, rawSecretAsSeenByETCD)
 	}
 
 	// Since Data Encryption Key (DEK) is randomly generated (per encryption operation), we need to ask KMS Mock for it.
-	dekPlainAsSeenByKMS, err := getDEKFromKMSPlugin(pluginMock)
+	plainTextDEK := pluginMock.lastEncryptRequest.Plain
 	if err != nil {
 		t.Fatalf("failed to get DEK from KMS: %v", err)
 	}
 
-	decryptResponse, err := pluginMock.Decrypt(context.Background(),
+	secretETCDPath := test.getETCDPath()
-		&kmsapi.DecryptRequest{Version: kmsAPIVersion, Cipher: rawSecretAsSeenByETCD.getDEK()})
+	rawEnvelope, err := test.getRawSecretFromETCD()
+	if err != nil {
+		t.Fatalf("failed to read %s from etcd: %v", secretETCDPath, err)
+	}
+	envelope := envelope{
+		providerName: providerName,
+		rawEnvelope:  rawEnvelope,
+		plainTextDEK: plainTextDEK,
+	}
+
+	wantPrefix := "k8s:enc:kms:v1:kms-provider:"
+	if !bytes.HasPrefix(rawEnvelope, []byte(wantPrefix)) {
+		t.Fatalf("expected secret to be prefixed with %s, but got %s", wantPrefix, rawEnvelope)
+	}
+
+	decryptResponse, err := pluginMock.Decrypt(context.Background(), &kmsapi.DecryptRequest{Version: kmsAPIVersion, Cipher: envelope.cipherTextDEK()})
 	if err != nil {
 		t.Fatalf("failed to decrypt DEK, %v", err)
 	}
 	dekPlainAsWouldBeSeenByETCD := decryptResponse.Plain
 
-	if !bytes.Equal(dekPlainAsSeenByKMS, dekPlainAsWouldBeSeenByETCD) {
+	if !bytes.Equal(plainTextDEK, dekPlainAsWouldBeSeenByETCD) {
-		t.Fatalf("expected dekPlainAsSeenByKMS %v to be passed to KMS Plugin, but got %s",
+		t.Fatalf("expected plainTextDEK %v to be passed to KMS Plugin, but got %s",
-			dekPlainAsSeenByKMS, dekPlainAsWouldBeSeenByETCD)
+			plainTextDEK, dekPlainAsWouldBeSeenByETCD)
 	}
 
-	plainSecret, err := decryptPayload(dekPlainAsWouldBeSeenByETCD, rawSecretAsSeenByETCD, secretETCDPath)
+	plainSecret, err := envelope.plainTextPayload(secretETCDPath)
 	if err != nil {
 		t.Fatalf("failed to transform from storage via AESCBC, err: %v", err)
 	}
@@ -144,32 +179,124 @@ func TestKMSProvider(t *testing.T) {
 	if secretVal != string(s.Data[secretKey]) {
 		t.Fatalf("expected %s from KubeAPI, but got %s", secretVal, string(s.Data[secretKey]))
 	}
-	test.printMetrics()
 }
 
-func getDEKFromKMSPlugin(pluginMock *base64Plugin) ([]byte, error) {
+func TestKMSHealthz(t *testing.T) {
-	// We expect KMS to already have seen an encryptRequest. Hence non-blocking call.
+	encryptionConfig := `
-	e, ok := <-pluginMock.encryptRequest
+kind: EncryptionConfiguration
+apiVersion: apiserver.config.k8s.io/v1
+resources:
+  - resources:
+    - secrets
+    providers:
+    - kms:
+       name: provider-1
+       endpoint: unix:///@kms-provider-1.sock
+    - kms:
+       name: provider-2
+       endpoint: unix:///@kms-provider-2.sock
+`
 
-	if !ok {
+	pluginMock1, err := newBase64Plugin("@kms-provider-1.sock")
-		return nil, fmt.Errorf("failed to sense encryptRequest from KMS Plugin Mock")
-	}
-
-	return e.Plain, nil
-}
-
-func decryptPayload(key []byte, secret rawDEKKEKSecret, secretETCDPath string) ([]byte, error) {
-	block, err := aes.NewCipher(key)
 	if err != nil {
-		return nil, fmt.Errorf("failed to initialize AES Cipher: %v", err)
+		t.Fatalf("failed to create mock of KMS Plugin #1: %v", err)
-	}
-	// etcd path of the key is used as the authenticated context - need to pass it to decrypt
-	ctx := value.DefaultContext([]byte(secretETCDPath))
-	aescbcTransformer := aestransformer.NewCBCTransformer(block)
-	plainSecret, _, err := aescbcTransformer.TransformFromStorage(secret.getPayload(), ctx)
-	if err != nil {
-		return nil, fmt.Errorf("failed to transform from storage via AESCBC, err: %v", err)
 	}
 
-	return plainSecret, nil
+	go pluginMock1.grpcServer.Serve(pluginMock1.listener)
+	defer pluginMock1.cleanUp()
+	kmsPluginMustBeUp(t, pluginMock1)
+
+	pluginMock2, err := newBase64Plugin("@kms-provider-2.sock")
+	if err != nil {
+		t.Fatalf("failed to create mock of KMS Plugin #2: %v", err)
+	}
+
+	go pluginMock2.grpcServer.Serve(pluginMock2.listener)
+	defer pluginMock2.cleanUp()
+	kmsPluginMustBeUp(t, pluginMock2)
+
+	test, err := newTransformTest(t, encryptionConfig)
+	if err != nil {
+		t.Fatalf("Failed to start kube-apiserver, error: %v", err)
+	}
+	defer test.cleanUp()
+
+	// Name of the healthz check is calculated based on a constant "kms-provider-" + position of the
+	// provider in the config.
+
+	// Stage 1 - Since all kms-plugins are guaranteed to be up, healthz checks for:
+	// healthz/kms-provider-0 and /healthz/kms-provider-1 should be OK.
+	mustBeHealthy(t, "kms-provider-0", test.kubeAPIServer.ClientConfig)
+	mustBeHealthy(t, "kms-provider-1", test.kubeAPIServer.ClientConfig)
+
+	// Stage 2 - kms-plugin for provider-1 is down. Therefore, expect the health check for provider-1
+	// to fail, but provider-2 should still be OK
+	pluginMock1.enterFailedState()
+	mustBeUnHealthy(t, "kms-provider-0", test.kubeAPIServer.ClientConfig)
+	mustBeHealthy(t, "kms-provider-1", test.kubeAPIServer.ClientConfig)
+	pluginMock1.exitFailedState()
+
+	// Stage 3 - kms-plugin for provider-1 is now up. Therefore, expect the health check for provider-1
+	// to succeed now, but provider-2 is now down.
+	// Need to sleep since health check chases responses for 3 seconds.
+	pluginMock2.enterFailedState()
+	mustBeHealthy(t, "kms-provider-0", test.kubeAPIServer.ClientConfig)
+	mustBeUnHealthy(t, "kms-provider-1", test.kubeAPIServer.ClientConfig)
+}
+
+func kmsPluginMustBeUp(t *testing.T, plugin *base64Plugin) {
+	t.Helper()
+	var gRPCErr error
+	pollErr := wait.PollImmediate(1*time.Second, wait.ForeverTestTimeout, func() (bool, error) {
+		_, gRPCErr = plugin.Encrypt(context.Background(), &kmsapi.EncryptRequest{Plain: []byte("foo")})
+		return gRPCErr == nil, nil
+	})
+
+	if pollErr == wait.ErrWaitTimeout {
+		t.Fatalf("failed to start kms-plugin, error: %v", gRPCErr)
+	}
+}
+
+func mustBeHealthy(t *testing.T, checkName string, clientConfig *rest.Config) {
+	t.Helper()
+	var restErr error
+	pollErr := wait.PollImmediate(2*time.Second, wait.ForeverTestTimeout, func() (bool, error) {
+		status, err := getHealthz(checkName, clientConfig)
+		if err != nil {
+			return false, err
+		}
+		return status == http.StatusOK, nil
+	})
+
+	if pollErr == wait.ErrWaitTimeout {
+		t.Fatalf("failed to get the expected healthz status of OK for check: %s, error: %v", restErr, checkName)
+	}
+}
+
+func mustBeUnHealthy(t *testing.T, checkName string, clientConfig *rest.Config) {
+	t.Helper()
+	var restErr error
+	pollErr := wait.PollImmediate(2*time.Second, wait.ForeverTestTimeout, func() (bool, error) {
+		status, err := getHealthz(checkName, clientConfig)
+		if err != nil {
+			return false, err
+		}
+		return status != http.StatusOK, nil
+	})
+
+	if pollErr == wait.ErrWaitTimeout {
+		t.Fatalf("failed to get the expected healthz status of !OK for check: %s, error: %v", restErr, checkName)
+	}
+}
+
+func getHealthz(checkName string, clientConfig *rest.Config) (int, error) {
+	client, err := kubernetes.NewForConfig(clientConfig)
+	if err != nil {
+		return 0, fmt.Errorf("failed to create a client: %v", err)
+	}
+
+	result := client.CoreV1().RESTClient().Get().AbsPath(fmt.Sprintf("/healthz/%v", checkName)).Do()
+	status := 0
+	result.StatusCode(&status)
+	return status, nil
 }

test/integration/master/secrets_transformation_test.go

@@ -90,6 +90,10 @@ func TestSecretsShouldBeTransformed(t *testing.T) {
 			t.Errorf("failed to setup test for envelop %s, error was %v", tt.transformerPrefix, err)
 			continue
 		}
+		test.secret, err = test.createSecret(testSecret, testNamespace)
+		if err != nil {
+			t.Fatalf("Failed to create test secret, error: %v", err)
+		}
 		test.run(tt.unSealFunc, tt.transformerPrefix)
 		test.cleanUp()
 	}

test/integration/master/transformation_testcase.go

@@ -27,6 +27,8 @@ import (
 	"strings"
 	"testing"
 
+	"k8s.io/klog"
+
 	"github.com/coreos/etcd/clientv3"
 	"github.com/prometheus/client_golang/prometheus"
 	"sigs.k8s.io/yaml"
@@ -81,6 +83,7 @@ func newTransformTest(l kubeapiservertesting.Logger, transformerConfigYAML strin
 	if e.kubeAPIServer, err = kubeapiservertesting.StartTestServer(l, nil, e.getEncryptionOptions(), e.storageConfig); err != nil {
 		return nil, fmt.Errorf("failed to start KubeAPI server: %v", err)
 	}
+	klog.Infof("Started kube-apiserver %v", e.kubeAPIServer.ClientConfig.Host)
 
 	if e.restClient, err = kubernetes.NewForConfig(e.kubeAPIServer.ClientConfig); err != nil {
 		return nil, fmt.Errorf("error while creating rest client: %v", err)
@@ -90,10 +93,6 @@ func newTransformTest(l kubeapiservertesting.Logger, transformerConfigYAML strin
 		return nil, err
 	}
 
-	if e.secret, err = e.createSecret(testSecret, e.ns.Name); err != nil {
-		return nil, err
-	}
-
 	return &e, nil
 }
 
@@ -118,7 +117,7 @@ func (e *transformTest) run(unSealSecretFunc unSealSecret, expectedEnvelopePrefi
 
 	// etcd path of the key is used as the authenticated context - need to pass it to decrypt
 	ctx := value.DefaultContext([]byte(e.getETCDPath()))
-	// Envelope header precedes the payload
+	// Envelope header precedes the cipherTextPayload
 	sealedData := response.Kvs[0].Value[len(expectedEnvelopePrefix):]
 	transformerConfig, err := e.getEncryptionConfig()
 	if err != nil {