kubernetes
/
apiserver
mirror of https://github.com/kubernetes/apiserver.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #116881 from ritazh/kms-test-fix 

kmsv2: test cleanup

Kubernetes-commit: 66c78653f1c5bca4f9992c971c886043ab5f86da
This commit is contained in:
Kubernetes Publisher 2023-04-11 19:16:31 -07:00
parent c7cbc5a59c bef9a879e0
commit 90c5f14b70
6 changed files with 61 additions and 164 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
go.mod
View File

@ -44,7 +44,7 @@ require (
gopkg.in/square/go-jose.v2 v2.6.0
k8s.io/api v0.0.0-20230327181730-ec59454f24f5
k8s.io/apimachinery v0.0.0-20230411235754-c1741f1470ae
k8s.io/client-go v0.0.0-20230412000719-20bcfef42d0d
k8s.io/client-go v0.0.0-20230412040258-8005e0d28ba2
k8s.io/component-base v0.0.0-20230327183930-4fb97becef37
k8s.io/klog/v2 v2.90.1
k8s.io/kms v0.0.0-20230315071547-f5c193c64781
@ -126,7 +126,7 @@ require (
replace (
k8s.io/api => k8s.io/api v0.0.0-20230327181730-ec59454f24f5
k8s.io/apimachinery => k8s.io/apimachinery v0.0.0-20230411235754-c1741f1470ae
k8s.io/client-go => k8s.io/client-go v0.0.0-20230412000719-20bcfef42d0d
k8s.io/client-go => k8s.io/client-go v0.0.0-20230412040258-8005e0d28ba2
k8s.io/component-base => k8s.io/component-base v0.0.0-20230327183930-4fb97becef37
k8s.io/kms => k8s.io/kms v0.0.0-20230315071547-f5c193c64781
)

4
go.sum
View File

@ -882,8 +882,8 @@ k8s.io/api v0.0.0-20230327181730-ec59454f24f5 h1:w1YnXoFKIeL9JZfIYRn9CRdkX8/kW10
k8s.io/api v0.0.0-20230327181730-ec59454f24f5/go.mod h1:RAYThv5ELmDNk2Kt+PWiwWtAI1TZyap1JK1QStRzwyw=
k8s.io/apimachinery v0.0.0-20230411235754-c1741f1470ae h1:DAZ4vnjauD5+IAaf5khBE2B2OtQpmjUbqrtUVN76DYI=
k8s.io/apimachinery v0.0.0-20230411235754-c1741f1470ae/go.mod h1:5ikh59fK3AJ287GUvpUsryoMFtH9zj/ARfWCo3AyXTM=
k8s.io/client-go v0.0.0-20230412000719-20bcfef42d0d h1:YKMu9zz1p+fgHvMvOgsblzwed0i4no1KWCud9XrCyWQ=
k8s.io/client-go v0.0.0-20230412000719-20bcfef42d0d/go.mod h1:OxG0CHMCzsv5/3LcNDNXMggbUbsrNzwa/4wfzm234x0=
k8s.io/client-go v0.0.0-20230412040258-8005e0d28ba2 h1:Mxfg52BtCwS9aQ0uqJSBtpQrDvFdbTZve06ttHHOqRk=
k8s.io/client-go v0.0.0-20230412040258-8005e0d28ba2/go.mod h1:OxG0CHMCzsv5/3LcNDNXMggbUbsrNzwa/4wfzm234x0=
k8s.io/component-base v0.0.0-20230327183930-4fb97becef37 h1:zhFwQvCptlnYZrPc5cZZUVPMoU8ao2kX+6dwRhF2r2s=
k8s.io/component-base v0.0.0-20230327183930-4fb97becef37/go.mod h1:ldAHwJrM01rErecGRS2RZfis66nxmBLNz9jSHBOg3VI=
k8s.io/klog/v2 v2.90.1 h1:m4bYOKall2MmOiRaR1J+We67Do7vm9KiQVlT96lnHUw=

76
pkg/storage/value/encrypt/envelope/grpc_service_unix_test.go
View File

@ -66,14 +66,7 @@ func TestKMSPluginLateStart(t *testing.T) {
defer destroyService(service)
time.Sleep(callTimeout / 2)
f, err := mock.NewBase64Plugin(s.path)
if err != nil {
t.Fatalf("failed to start test KMS provider server, error: %v", err)
}
if err := f.Start(); err != nil {
t.Fatalf("Failed to start kms-plugin, err: %v", err)
}
defer f.CleanUp()
_ = mock.NewBase64Plugin(t, s.path)
data := []byte("test data")
_, err = service.Encrypt(data)
@ -158,16 +151,8 @@ func TestTimeouts(t *testing.T) {
// Simulating delayed start of kms-plugin, kube-apiserver is up before the plugin, if requested by the testcase.
time.Sleep(tt.pluginDelay)
f, err := mock.NewBase64Plugin(socketName.path)
if err != nil {
t.Errorf("failed to construct test KMS provider server, error: %v", err)
return
}
if err := f.Start(); err != nil {
t.Errorf("Failed to start test KMS provider server, error: %v", err)
return
}
defer f.CleanUp()
_ = mock.NewBase64Plugin(t, socketName.path)
kmsPluginWG.Done()
// Keeping plugin up to process requests.
testCompletedWG.Wait()
@ -206,13 +191,7 @@ func TestIntermittentConnectionLoss(t *testing.T) {
encryptErr error
)
// Start KMS Plugin
f, err := mock.NewBase64Plugin(endpoint.path)
if err != nil {
t.Fatalf("failed to start test KMS provider server, error: %v", err)
}
if err := f.Start(); err != nil {
t.Fatalf("Failed to start kms-plugin, err: %v", err)
}
f := mock.NewBase64Plugin(t, endpoint.path)
ctx := testContext(t)
@ -228,10 +207,10 @@ func TestIntermittentConnectionLoss(t *testing.T) {
t.Fatalf("failed when execute encrypt, error: %v", err)
}
t.Log("Connected to KMSPlugin")
f.CleanUp()
// Stop KMS Plugin - simulating connection loss
t.Log("KMS Plugin is stopping")
f.CleanUp()
time.Sleep(2 * time.Second)
wg1.Add(1)
@ -250,14 +229,7 @@ func TestIntermittentConnectionLoss(t *testing.T) {
wg1.Wait()
time.Sleep(blackOut)
// Start KMS Plugin
f, err = mock.NewBase64Plugin(endpoint.path)
if err != nil {
t.Fatalf("failed to start test KMS provider server, error: %v", err)
}
if err := f.Start(); err != nil {
t.Fatalf("Failed to start kms-plugin, err: %v", err)
}
defer f.CleanUp()
_ = mock.NewBase64Plugin(t, endpoint.path)
t.Log("Restarted KMS Plugin")
wg2.Wait()
@ -274,15 +246,8 @@ func TestUnsupportedVersion(t *testing.T) {
wantErr := fmt.Errorf(versionErrorf, ver, kmsapiVersion)
endpoint := newEndpoint()
f, err := mock.NewBase64Plugin(endpoint.path)
if err != nil {
t.Fatalf("failed to start test KMS provider server, error: %ver", err)
}
f := mock.NewBase64Plugin(t, endpoint.path)
f.SetVersion(ver)
if err := f.Start(); err != nil {
t.Fatalf("Failed to start kms-plugin, err: %v", err)
}
defer f.CleanUp()
ctx := testContext(t)
@ -318,14 +283,7 @@ func TestGRPCService(t *testing.T) {
t.Parallel()
// Start a test gRPC server.
endpoint := newEndpoint()
f, err := mock.NewBase64Plugin(endpoint.path)
if err != nil {
t.Fatalf("failed to construct test KMS provider server, error: %v", err)
}
if err := f.Start(); err != nil {
t.Fatalf("Failed to start kms-plugin, err: %v", err)
}
defer f.CleanUp()
_ = mock.NewBase64Plugin(t, endpoint.path)
ctx := testContext(t)
@ -359,14 +317,7 @@ func TestGRPCServiceConcurrentAccess(t *testing.T) {
t.Parallel()
// Start a test gRPC server.
endpoint := newEndpoint()
f, err := mock.NewBase64Plugin(endpoint.path)
if err != nil {
t.Fatalf("failed to start test KMS provider server, error: %v", err)
}
if err := f.Start(); err != nil {
t.Fatalf("Failed to start kms-plugin, err: %v", err)
}
defer f.CleanUp()
_ = mock.NewBase64Plugin(t, endpoint.path)
ctx := testContext(t)
@ -416,14 +367,7 @@ func destroyService(service Service) {
func TestInvalidConfiguration(t *testing.T) {
t.Parallel()
// Start a test gRPC server.
f, err := mock.NewBase64Plugin(newEndpoint().path)
if err != nil {
t.Fatalf("failed to start test KMS provider server, error: %v", err)
}
if err := f.Start(); err != nil {
t.Fatalf("Failed to start kms-plugin, err: %v", err)
}
defer f.CleanUp()
_ = mock.NewBase64Plugin(t, newEndpoint().path)
ctx := testContext(t)

72
pkg/storage/value/encrypt/envelope/kmsv2/grpc_service_unix_test.go
View File

@ -69,14 +69,7 @@ func TestKMSPluginLateStart(t *testing.T) {
defer destroyService(service)
time.Sleep(callTimeout / 2)
f, err := mock.NewBase64Plugin(s.path)
if err != nil {
t.Fatalf("failed to start test KMS provider server, error: %v", err)
}
if err := f.Start(); err != nil {
t.Fatalf("Failed to start kms-plugin, err: %v", err)
}
defer f.CleanUp()
_ = mock.NewBase64Plugin(t, s.path)
data := []byte("test data")
uid := string(uuid.NewUUID())
@ -162,16 +155,7 @@ func TestTimeouts(t *testing.T) {
// Simulating delayed start of kms-plugin, kube-apiserver is up before the plugin, if requested by the testcase.
time.Sleep(tt.pluginDelay)
f, err := mock.NewBase64Plugin(socketName.path)
if err != nil {
t.Errorf("failed to construct test KMS provider server, error: %v", err)
return
}
if err := f.Start(); err != nil {
t.Errorf("Failed to start test KMS provider server, error: %v", err)
return
}
defer f.CleanUp()
_ = mock.NewBase64Plugin(t, socketName.path)
kmsPluginWG.Done()
// Keeping plugin up to process requests.
testCompletedWG.Wait()
@ -211,13 +195,7 @@ func TestIntermittentConnectionLoss(t *testing.T) {
encryptErr error
)
// Start KMS Plugin
f, err := mock.NewBase64Plugin(endpoint.path)
if err != nil {
t.Fatalf("failed to start test KMS provider server, error: %v", err)
}
if err := f.Start(); err != nil {
t.Fatalf("Failed to start kms-plugin, err: %v", err)
}
f := mock.NewBase64Plugin(t, endpoint.path)
ctx := testContext(t)
@ -255,14 +233,7 @@ func TestIntermittentConnectionLoss(t *testing.T) {
wg1.Wait()
time.Sleep(blackOut)
// Start KMS Plugin
f, err = mock.NewBase64Plugin(endpoint.path)
if err != nil {
t.Fatalf("failed to start test KMS provider server, error: %v", err)
}
if err := f.Start(); err != nil {
t.Fatalf("Failed to start kms-plugin, err: %v", err)
}
defer f.CleanUp()
_ = mock.NewBase64Plugin(t, endpoint.path)
t.Log("Restarted KMS Plugin")
wg2.Wait()
@ -277,14 +248,7 @@ func TestGRPCService(t *testing.T) {
t.Parallel()
// Start a test gRPC server.
endpoint := newEndpoint()
f, err := mock.NewBase64Plugin(endpoint.path)
if err != nil {
t.Fatalf("failed to construct test KMS provider server, error: %v", err)
}
if err := f.Start(); err != nil {
t.Fatalf("Failed to start kms-plugin, err: %v", err)
}
defer f.CleanUp()
_ = mock.NewBase64Plugin(t, endpoint.path)
ctx := testContext(t)
@ -320,14 +284,7 @@ func TestGRPCServiceConcurrentAccess(t *testing.T) {
t.Parallel()
// Start a test gRPC server.
endpoint := newEndpoint()
f, err := mock.NewBase64Plugin(endpoint.path)
if err != nil {
t.Fatalf("failed to start test KMS provider server, error: %v", err)
}
if err := f.Start(); err != nil {
t.Fatalf("Failed to start kms-plugin, err: %v", err)
}
defer f.CleanUp()
_ = mock.NewBase64Plugin(t, endpoint.path)
ctx := testContext(t)
@ -379,14 +336,7 @@ func destroyService(service kmsservice.Service) {
func TestInvalidConfiguration(t *testing.T) {
t.Parallel()
// Start a test gRPC server.
f, err := mock.NewBase64Plugin(newEndpoint().path)
if err != nil {
t.Fatalf("failed to start test KMS provider server, error: %v", err)
}
if err := f.Start(); err != nil {
t.Fatalf("Failed to start kms-plugin, err: %v", err)
}
defer f.CleanUp()
_ = mock.NewBase64Plugin(t, newEndpoint().path)
ctx := testContext(t)
@ -410,13 +360,7 @@ func TestInvalidConfiguration(t *testing.T) {
func TestKMSOperationsMetric(t *testing.T) {
endpoint := newEndpoint()
f, err := mock.NewBase64Plugin(endpoint.path)
if err != nil {
t.Fatalf("failed to start test KMS provider server, error: %v", err)
}
if err := f.Start(); err != nil {
t.Fatalf("Failed to start kms-plugin, err: %v", err)
}
_ = mock.NewBase64Plugin(t, endpoint.path)
ctx := testContext(t)

32
pkg/storage/value/encrypt/envelope/testing/v1beta1/kms_plugin_mock.go
View File

@ -25,9 +25,8 @@ import (
"fmt"
"net"
"os"
"runtime"
"strings"
"sync"
"testing"
"time"
"google.golang.org/grpc"
@ -60,7 +59,7 @@ type Base64Plugin struct {
}
// NewBase64Plugin is a constructor for Base64Plugin.
func NewBase64Plugin(socketPath string) (*Base64Plugin, error) {
func NewBase64Plugin(t *testing.T, socketPath string) *Base64Plugin {
server := grpc.NewServer()
result := &Base64Plugin{
grpcServer: server,
@ -70,19 +69,26 @@ func NewBase64Plugin(socketPath string) (*Base64Plugin, error) {
}
kmsapi.RegisterKeyManagementServiceServer(server, result)
return result, nil
if err := result.start(); err != nil {
t.Fatalf("failed to start KMS plugin, err: %v", err)
}
t.Cleanup(result.CleanUp)
if err := waitForBase64PluginToBeUp(result); err != nil {
t.Fatalf("failed to start KMS plugin: err: %v", err)
}
return result
}
// WaitForBase64PluginToBeUp waits until the plugin is ready to serve requests.
func WaitForBase64PluginToBeUp(plugin *Base64Plugin) error {
// waitForBase64PluginToBeUp waits until the plugin is ready to serve requests.
func waitForBase64PluginToBeUp(plugin *Base64Plugin) error {
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 {
return fmt.Errorf("failed to start kms-plugin, error: %v", gRPCErr)
if pollErr != nil {
return fmt.Errorf("failed to start KMS plugin, gRPC error: %v, poll error: %v", gRPCErr, pollErr)
}
return nil
@ -98,8 +104,8 @@ func (s *Base64Plugin) SetVersion(ver string) {
s.ver = ver
}
// Start starts plugin's gRPC service.
func (s *Base64Plugin) Start() error {
// start starts plugin's gRPC service.
func (s *Base64Plugin) start() error {
var err error
s.listener, err = net.Listen(unixProtocol, s.socketPath)
if err != nil {
@ -114,10 +120,8 @@ func (s *Base64Plugin) Start() error {
// CleanUp stops gRPC server and the underlying listener.
func (s *Base64Plugin) CleanUp() {
s.grpcServer.Stop()
s.listener.Close()
if !strings.HasPrefix(s.socketPath, "@") || runtime.GOOS != "linux" {
os.Remove(s.socketPath)
}
_ = s.listener.Close()
_ = os.Remove(s.socketPath)
}
// EnterFailedState places the plugin into failed state.

37
pkg/storage/value/encrypt/envelope/testing/v2/kms_plugin_mock.go
View File

@ -25,9 +25,8 @@ import (
"fmt"
"net"
"os"
"runtime"
"strings"
"sync"
"testing"
"time"
"google.golang.org/grpc"
@ -61,7 +60,7 @@ type Base64Plugin struct {
}
// NewBase64Plugin is a constructor for Base64Plugin.
func NewBase64Plugin(socketPath string) (*Base64Plugin, error) {
func NewBase64Plugin(t *testing.T, socketPath string) *Base64Plugin {
server := grpc.NewServer()
result := &Base64Plugin{
grpcServer: server,
@ -72,11 +71,19 @@ func NewBase64Plugin(socketPath string) (*Base64Plugin, error) {
}
kmsapi.RegisterKeyManagementServiceServer(server, result)
return result, nil
if err := result.start(); err != nil {
t.Fatalf("failed to start KMS plugin, err: %v", err)
}
t.Cleanup(result.CleanUp)
if err := waitForBase64PluginToBeUp(result); err != nil {
t.Fatalf("failed to start KMS plugin: err: %v", err)
}
return result
}
// WaitForBase64PluginToBeUp waits until the plugin is ready to serve requests.
func WaitForBase64PluginToBeUp(plugin *Base64Plugin) error {
// waitForBase64PluginToBeUp waits until the plugin is ready to serve requests.
func waitForBase64PluginToBeUp(plugin *Base64Plugin) error {
var gRPCErr error
var resp *kmsapi.StatusResponse
pollErr := wait.PollImmediate(1*time.Second, wait.ForeverTestTimeout, func() (bool, error) {
@ -84,14 +91,14 @@ func WaitForBase64PluginToBeUp(plugin *Base64Plugin) error {
return gRPCErr == nil && resp.Healthz == "ok", nil
})
if pollErr == wait.ErrWaitTimeout {
return fmt.Errorf("failed to start kms-plugin, error: %v", gRPCErr)
if pollErr != nil {
return fmt.Errorf("failed to start kms-plugin, gRPC error: %v, poll error: %v", gRPCErr, pollErr)
}
return nil
}
// WaitForBase64PluginToBeUpdated waits until the plugin updates keyID.
// waitForBase64PluginToBeUpdated waits until the plugin updates keyID.
func WaitForBase64PluginToBeUpdated(plugin *Base64Plugin) error {
var gRPCErr error
var resp *kmsapi.StatusResponse
@ -103,7 +110,7 @@ func WaitForBase64PluginToBeUpdated(plugin *Base64Plugin) error {
})
if updatePollErr != nil {
return fmt.Errorf("failed to update keyID for kmsv2-plugin, error: %w", gRPCErr)
return fmt.Errorf("failed to update keyID for kmsv2-plugin, gRPC error: %w, updatePoll error: %w", gRPCErr, updatePollErr)
}
return nil
@ -119,8 +126,8 @@ func (s *Base64Plugin) SetVersion(ver string) {
s.ver = ver
}
// Start starts plugin's gRPC service.
func (s *Base64Plugin) Start() error {
// start starts plugin's gRPC service.
func (s *Base64Plugin) start() error {
var err error
s.listener, err = net.Listen(unixProtocol, s.socketPath)
if err != nil {
@ -135,10 +142,8 @@ func (s *Base64Plugin) Start() error {
// CleanUp stops gRPC server and the underlying listener.
func (s *Base64Plugin) CleanUp() {
s.grpcServer.Stop()
s.listener.Close()
if !strings.HasPrefix(s.socketPath, "@") || runtime.GOOS != "linux" {
os.Remove(s.socketPath)
}
_ = s.listener.Close()
_ = os.Remove(s.socketPath)
}
// EnterFailedState places the plugin into failed state.