Clean up kms test
Signed-off-by: Rita Zhang <rita.z.zhang@gmail.com> Kubernetes-commit: 906f0607efc5e12b3108e2b12cdac119fdba7909
This commit is contained in:
parent
c7cbc5a59c
commit
bef9a879e0
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
Loading…
Reference in New Issue