Add stress test to repeatedly restart Pods with PVCs in parallel

Change-Id: I499571cc86b1058d0e16d79e5e998d1dedfd9a4a

test/e2e/framework/pod/create.go

@@ -177,7 +177,7 @@ func MakeSecPod(podConfig *Config) (*v1.Pod, error) {
 	if len(podConfig.Command) == 0 {
 		podConfig.Command = "trap exit TERM; while true; do sleep 1; done"
 	}
-	podName := "security-context-" + string(uuid.NewUUID())
+	podName := "pod-" + string(uuid.NewUUID())
 	if podConfig.FsGroup == nil {
 		podConfig.FsGroup = func(i int64) *int64 {
 			return &i

test/e2e/storage/csi_volumes.go

@@ -31,29 +31,13 @@ var csiTestDrivers = []func() testsuites.TestDriver{
 	// Don't run tests with mock driver (drivers.InitMockCSIDriver), it does not provide persistent storage.
 }
 
-// List of testSuites to be executed in below loop
-var csiTestSuites = []func() testsuites.TestSuite{
-	testsuites.InitEphemeralTestSuite,
-	testsuites.InitVolumesTestSuite,
-	testsuites.InitVolumeIOTestSuite,
-	testsuites.InitVolumeModeTestSuite,
-	testsuites.InitSubPathTestSuite,
-	testsuites.InitProvisioningTestSuite,
-	testsuites.InitSnapshottableTestSuite,
-	testsuites.InitMultiVolumeTestSuite,
-	testsuites.InitDisruptiveTestSuite,
-	testsuites.InitVolumeExpandTestSuite,
-	testsuites.InitVolumeLimitsTestSuite,
-	testsuites.InitTopologyTestSuite,
-}
-
 // This executes testSuites for csi volumes.
 var _ = utils.SIGDescribe("CSI Volumes", func() {
 	for _, initDriver := range csiTestDrivers {
 		curDriver := initDriver()
 
 		ginkgo.Context(testsuites.GetDriverNameWithFeatureTags(curDriver), func() {
-			testsuites.DefineTestSuite(curDriver, csiTestSuites)
+			testsuites.DefineTestSuite(curDriver, testsuites.CSISuites)
 		})
 	}
 })

test/e2e/storage/external/external.go

@@ -130,21 +130,6 @@ type driverDefinition struct {
 	ClientNodeName string
 }
 
-// List of testSuites to be executed for each external driver.
-var csiTestSuites = []func() testsuites.TestSuite{
-	testsuites.InitEphemeralTestSuite,
-	testsuites.InitMultiVolumeTestSuite,
-	testsuites.InitProvisioningTestSuite,
-	testsuites.InitSnapshottableTestSuite,
-	testsuites.InitSubPathTestSuite,
-	testsuites.InitVolumeIOTestSuite,
-	testsuites.InitVolumeModeTestSuite,
-	testsuites.InitVolumesTestSuite,
-	testsuites.InitVolumeExpandTestSuite,
-	testsuites.InitDisruptiveTestSuite,
-	testsuites.InitVolumeLimitsTestSuite,
-}
-
 func init() {
 	e2econfig.Flags.Var(testDriverParameter{}, "storage.testdriver", "name of a .yaml or .json file that defines a driver for storage testing, can be used more than once")
 }
@@ -182,7 +167,7 @@ func AddDriverDefinition(filename string) error {
 
 	description := "External Storage " + testsuites.GetDriverNameWithFeatureTags(driver)
 	ginkgo.Describe(description, func() {
-		testsuites.DefineTestSuite(driver, csiTestSuites)
+		testsuites.DefineTestSuite(driver, testsuites.CSISuites)
 	})
 
 	return nil

test/e2e/storage/in_tree_volumes.go

@@ -48,27 +48,13 @@ var testDrivers = []func() testsuites.TestDriver{
 	drivers.InitLocalDriverWithVolumeType(utils.LocalVolumeGCELocalSSD),
 }
 
-// List of testSuites to be executed in below loop
-var testSuites = []func() testsuites.TestSuite{
-	testsuites.InitVolumesTestSuite,
-	testsuites.InitVolumeIOTestSuite,
-	testsuites.InitVolumeModeTestSuite,
-	testsuites.InitSubPathTestSuite,
-	testsuites.InitProvisioningTestSuite,
-	testsuites.InitMultiVolumeTestSuite,
-	testsuites.InitVolumeExpandTestSuite,
-	testsuites.InitDisruptiveTestSuite,
-	testsuites.InitVolumeLimitsTestSuite,
-	testsuites.InitTopologyTestSuite,
-}
-
 // This executes testSuites for in-tree volumes.
 var _ = utils.SIGDescribe("In-tree Volumes", func() {
 	for _, initDriver := range testDrivers {
 		curDriver := initDriver()
 
 		ginkgo.Context(testsuites.GetDriverNameWithFeatureTags(curDriver), func() {
-			testsuites.DefineTestSuite(curDriver, testSuites)
+			testsuites.DefineTestSuite(curDriver, testsuites.BaseSuites)
 		})
 	}
 })

test/e2e/storage/testsuites/BUILD

@@ -10,6 +10,7 @@ go_library(
         "multivolume.go",
         "provisioning.go",
         "snapshottable.go",
+        "stress.go",
         "subpath.go",
         "testdriver.go",
         "topology.go",

test/e2e/storage/testsuites/base.go

@@ -60,6 +60,27 @@ func init() {
 
 type opCounts map[string]int64
 
+// BaseSuites is a list of storage test suites that work for in-tree and CSI drivers
+var BaseSuites = []func() TestSuite{
+	InitVolumesTestSuite,
+	InitVolumeIOTestSuite,
+	InitVolumeModeTestSuite,
+	InitSubPathTestSuite,
+	InitProvisioningTestSuite,
+	InitMultiVolumeTestSuite,
+	InitVolumeExpandTestSuite,
+	InitDisruptiveTestSuite,
+	InitVolumeLimitsTestSuite,
+	InitTopologyTestSuite,
+	InitStressTestSuite,
+}
+
+// CSISuites is a list of storage test suites that work only for CSI drivers
+var CSISuites = append(BaseSuites,
+	InitEphemeralTestSuite,
+	InitSnapshottableTestSuite,
+)
+
 // TestSuite represents an interface for a set of tests which works with TestDriver
 type TestSuite interface {
 	// GetTestSuiteInfo returns the TestSuiteInfo for this TestSuite

test/e2e/storage/testsuites/stress.go

@@ -0,0 +1,197 @@
+/*
+Copyright 2020 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.
+*/
+
+// This suite tests volumes under stress conditions
+
+package testsuites
+
+import (
+	"context"
+	"sync"
+
+	"github.com/onsi/ginkgo"
+
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	errors "k8s.io/apimachinery/pkg/util/errors"
+	clientset "k8s.io/client-go/kubernetes"
+	"k8s.io/kubernetes/test/e2e/framework"
+	e2epod "k8s.io/kubernetes/test/e2e/framework/pod"
+	e2epv "k8s.io/kubernetes/test/e2e/framework/pv"
+	e2eskipper "k8s.io/kubernetes/test/e2e/framework/skipper"
+	"k8s.io/kubernetes/test/e2e/storage/testpatterns"
+)
+
+type stressTestSuite struct {
+	tsInfo TestSuiteInfo
+}
+
+type stressTest struct {
+	config        *PerTestConfig
+	driverCleanup func()
+
+	intreeOps   opCounts
+	migratedOps opCounts
+
+	resources []*VolumeResource
+	pods      []*v1.Pod
+	// stop and wait for any async routines
+	wg      sync.WaitGroup
+	stopChs []chan struct{}
+}
+
+var _ TestSuite = &stressTestSuite{}
+
+// InitStressTestSuite returns stressTestSuite that implements TestSuite interface
+func InitStressTestSuite() TestSuite {
+	return &stressTestSuite{
+		tsInfo: TestSuiteInfo{
+			Name:       "stress",
+			FeatureTag: "[Feature: VolumeStress]",
+			TestPatterns: []testpatterns.TestPattern{
+				testpatterns.DefaultFsDynamicPV,
+				testpatterns.BlockVolModeDynamicPV,
+			},
+		},
+	}
+}
+
+func (t *stressTestSuite) GetTestSuiteInfo() TestSuiteInfo {
+	return t.tsInfo
+}
+
+func (t *stressTestSuite) SkipRedundantSuite(driver TestDriver, pattern testpatterns.TestPattern) {
+}
+
+func (t *stressTestSuite) DefineTests(driver TestDriver, pattern testpatterns.TestPattern) {
+	var (
+		dInfo = driver.GetDriverInfo()
+		cs    clientset.Interface
+	)
+
+	ginkgo.BeforeEach(func() {
+		// Check preconditions.
+		ok := false
+		_, ok = driver.(DynamicPVTestDriver)
+		if !ok {
+			e2eskipper.Skipf("Driver %s doesn't support %v -- skipping", dInfo.Name, pattern.VolType)
+		}
+
+		if !driver.GetDriverInfo().Capabilities[CapBlock] && pattern.VolMode == v1.PersistentVolumeBlock {
+			e2eskipper.Skipf("Driver %q does not support block volume mode - skipping", driver.GetDriverInfo().Name)
+		}
+	})
+
+	// This intentionally comes after checking the preconditions because it
+	// registers its own BeforeEach which creates the namespace. Beware that it
+	// also registers an AfterEach which renders f unusable. Any code using
+	// f must run inside an It or Context callback.
+	f := framework.NewDefaultFramework("stress")
+
+	init := func() *stressTest {
+		cs = f.ClientSet
+		l := &stressTest{}
+
+		// Now do the more expensive test initialization.
+		l.config, l.driverCleanup = driver.PrepareTest(f)
+		l.intreeOps, l.migratedOps = getMigrationVolumeOpCounts(f.ClientSet, dInfo.InTreePluginName)
+		l.resources = []*VolumeResource{}
+		l.pods = []*v1.Pod{}
+		l.stopChs = []chan struct{}{}
+
+		return l
+	}
+
+	cleanup := func(l *stressTest) {
+		var errs []error
+
+		framework.Logf("Stopping and waiting for all test routines to finish")
+		for _, stopCh := range l.stopChs {
+			close(stopCh)
+		}
+		l.wg.Wait()
+
+		for _, pod := range l.pods {
+			framework.Logf("Deleting pod %v", pod.Name)
+			err := e2epod.DeletePodWithWait(cs, pod)
+			errs = append(errs, err)
+		}
+
+		for _, resource := range l.resources {
+			errs = append(errs, resource.CleanupResource())
+		}
+
+		errs = append(errs, tryFunc(l.driverCleanup))
+		framework.ExpectNoError(errors.NewAggregate(errs), "while cleaning up resource")
+		validateMigrationVolumeOpCounts(f.ClientSet, dInfo.InTreePluginName, l.intreeOps, l.migratedOps)
+	}
+
+	ginkgo.It("multiple pods should access different volumes repeatedly [Slow] [Serial]", func() {
+		const (
+			numPods = 10
+			// number of times each Pod should start
+			numPodStarts = 10
+		)
+
+		var err error
+
+		l := init()
+		defer func() {
+			cleanup(l)
+		}()
+
+		for i := 0; i < numPods; i++ {
+			framework.Logf("Creating resources for pod %v/%v", i, numPods)
+			r := CreateVolumeResource(driver, l.config, pattern, t.GetTestSuiteInfo().SupportedSizeRange)
+			l.resources = append(l.resources, r)
+			l.pods = append(l.pods, e2epod.MakeSecPod(f.Namespace.Name,
+				[]*v1.PersistentVolumeClaim{r.Pvc},
+				nil, false, "", false, false, e2epv.SELinuxLabel, nil))
+			l.stopChs = append(l.stopChs, make(chan struct{}))
+		}
+
+		// Restart pod repeatedly
+		for i := 0; i < numPods; i++ {
+			podIndex := i
+			l.wg.Add(1)
+			go func() {
+				defer ginkgo.GinkgoRecover()
+				defer l.wg.Done()
+				for j := 0; j < numPodStarts; j++ {
+					select {
+					case <-l.stopChs[podIndex]:
+						return
+					default:
+						pod := l.pods[podIndex]
+						framework.Logf("Pod %v, Iteration %v/%v", podIndex, j, numPodStarts)
+						_, err = cs.CoreV1().Pods(pod.Namespace).Create(context.TODO(), pod, metav1.CreateOptions{})
+						framework.ExpectNoError(err)
+
+						err = e2epod.WaitForPodRunningInNamespace(cs, pod)
+						framework.ExpectNoError(err)
+
+						// TODO: write data per pod and validate it everytime
+
+						err = e2epod.DeletePodWithWait(f.ClientSet, pod)
+						framework.ExpectNoError(err)
+					}
+				}
+			}()
+		}
+
+		l.wg.Wait()
+	})
+}