Merge pull request kubernetes#84387 from immutableT/interceptor

Move the common logic of checking for kms-plugin's version into gRPC client interceptor.

Author: k8s-ci-robot

staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/grpc_service.go

@@ -61,29 +61,31 @@ func NewGRPCService(endpoint string, callTimeout time.Duration) (Service, error)
 		return nil, err
 	}
 
-	connection, err := grpc.Dial(addr, grpc.WithInsecure(), grpc.WithDefaultCallOptions(grpc.WaitForReady(true)), grpc.WithContextDialer(
-		func(context.Context, string) (net.Conn, error) {
-			// Ignoring addr and timeout arguments:
-			// addr - comes from the closure
-			c, err := net.DialUnix(unixProtocol, nil, &net.UnixAddr{Name: addr})
-			if err != nil {
-				klog.Errorf("failed to create connection to unix socket: %s, error: %v", addr, err)
-			} else {
-				klog.V(4).Infof("Successfully dialed Unix socket %v", addr)
-			}
-			return c, err
-		}))
+	s := &gRPCService{callTimeout: callTimeout}
+	s.connection, err = grpc.Dial(
+		addr,
+		grpc.WithInsecure(),
+		grpc.WithUnaryInterceptor(s.interceptor),
+		grpc.WithDefaultCallOptions(grpc.WaitForReady(true)),
+		grpc.WithContextDialer(
+			func(context.Context, string) (net.Conn, error) {
+				// Ignoring addr and timeout arguments:
+				// addr - comes from the closure
+				c, err := net.DialUnix(unixProtocol, nil, &net.UnixAddr{Name: addr})
+				if err != nil {
+					klog.Errorf("failed to create connection to unix socket: %s, error: %v", addr, err)
+				} else {
+					klog.V(4).Infof("Successfully dialed Unix socket %v", addr)
+				}
+				return c, err
+			}))
 
 	if err != nil {
 		return nil, fmt.Errorf("failed to create connection to %s, error: %v", endpoint, err)
 	}
 
-	kmsClient := kmsapi.NewKeyManagementServiceClient(connection)
-	return &gRPCService{
-		kmsClient:   kmsClient,
-		connection:  connection,
-		callTimeout: callTimeout,
-	}, nil
+	s.kmsClient = kmsapi.NewKeyManagementServiceClient(s.connection)
+	return s, nil
 }
 
 // Parse the endpoint to extract schema, host or path.
@@ -139,10 +141,6 @@ func (g *gRPCService) Decrypt(cipher []byte) ([]byte, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), g.callTimeout)
 	defer cancel()
 
-	if err := g.checkAPIVersion(ctx); err != nil {
-		return nil, err
-	}
-
 	request := &kmsapi.DecryptRequest{Cipher: cipher, Version: kmsapiVersion}
 	response, err := g.kmsClient.Decrypt(ctx, request)
 	if err != nil {
@@ -155,9 +153,6 @@ func (g *gRPCService) Decrypt(cipher []byte) ([]byte, error) {
 func (g *gRPCService) Encrypt(plain []byte) ([]byte, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), g.callTimeout)
 	defer cancel()
-	if err := g.checkAPIVersion(ctx); err != nil {
-		return nil, err
-	}
 
 	request := &kmsapi.EncryptRequest{Plain: plain, Version: kmsapiVersion}
 	response, err := g.kmsClient.Encrypt(ctx, request)
@@ -166,3 +161,21 @@ func (g *gRPCService) Encrypt(plain []byte) ([]byte, error) {
 	}
 	return response.Cipher, nil
 }
+
+func (g *gRPCService) interceptor(
+	ctx context.Context,
+	method string,
+	req interface{},
+	reply interface{},
+	cc *grpc.ClientConn,
+	invoker grpc.UnaryInvoker,
+	opts ...grpc.CallOption,
+) error {
+	if !kmsapi.IsVersionCheckMethod(method) {
+		if err := g.checkAPIVersion(ctx); err != nil {
+			return err
+		}
+	}
+
+	return invoker(ctx, method, req, reply, cc, opts...)
+}

staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/v1beta1/BUILD

@@ -2,7 +2,10 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library")
 
 go_library(
     name = "go_default_library",
-    srcs = ["service.pb.go"],
+    srcs = [
+        "service.pb.go",
+        "v1beta1.go",
+    ],
     importmap = "k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/v1beta1",
     importpath = "k8s.io/apiserver/pkg/storage/value/encrypt/envelope/v1beta1",
     visibility = ["//visibility:public"],

staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/v1beta1/v1beta1.go

@@ -0,0 +1,23 @@
+/*
+Copyright 2019 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 v1beta1 contains definition of kms-plugin's gRPC service.
+package v1beta1
+
+// IsVersionCheckMethod determines whether the supplied method is a version check against kms-plugin.
+func IsVersionCheckMethod(method string) bool {
+	return method == "/v1beta1.KeyManagementService/Version"
+}