Skip to content

Commit

Permalink
Merge pull request kubernetes#109972 from wojtek-t/clean_shutdown_1
Browse files Browse the repository at this point in the history 
Clean shutdown of couple controllers in integration tests
  • Loading branch information
@k8s-ci-robot
k8s-ci-robot authored May 11, 2022
2 parents b91c526 + 1fcc5d9 commit 564b204
Show file tree
Hide file tree
Showing 4 changed files with 93 additions and 121 deletions.
93 changes: 28 additions & 65 deletions test/integration/deployment/deployment_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -54,11 +54,8 @@ func TestNewDeployment(t *testing.T) {
}

// Start informer and controllers
stopCh := make(chan struct{})
defer close(stopCh)
informers.Start(stopCh)
go rm.Run(context.TODO(), 5)
go dc.Run(context.TODO(), 5)
stopControllers := runControllersAndInformers(t, rm, dc, informers)
defer stopControllers()

// Wait for the Deployment to be updated to revision 1
if err := tester.waitForDeploymentRevisionAndImage("1", fakeImage); err != nil {
Expand Down Expand Up @@ -121,11 +118,8 @@ func TestDeploymentRollingUpdate(t *testing.T) {
defer framework.DeleteTestingNamespace(ns, s, t)

// Start informer and controllers
stopCh := make(chan struct{})
defer close(stopCh)
informers.Start(stopCh)
go rm.Run(context.TODO(), 5)
go dc.Run(context.TODO(), 5)
stopControllers := runControllersAndInformers(t, rm, dc, informers)
defer stopControllers()

replicas := int32(20)
tester := &deploymentTester{t: t, c: c, deployment: newDeployment(name, ns.Name, replicas)}
Expand Down Expand Up @@ -264,11 +258,8 @@ func TestPausedDeployment(t *testing.T) {
}

// Start informer and controllers
stopCh := make(chan struct{})
defer close(stopCh)
informers.Start(stopCh)
go rm.Run(context.TODO(), 5)
go dc.Run(context.TODO(), 5)
stopControllers := runControllersAndInformers(t, rm, dc, informers)
defer stopControllers()

// Verify that the paused deployment won't create new replica set.
if err := tester.expectNoNewReplicaSet(); err != nil {
Expand Down Expand Up @@ -365,11 +356,8 @@ func TestScalePausedDeployment(t *testing.T) {
}

// Start informer and controllers
stopCh := make(chan struct{})
defer close(stopCh)
informers.Start(stopCh)
go rm.Run(context.TODO(), 5)
go dc.Run(context.TODO(), 5)
stopControllers := runControllersAndInformers(t, rm, dc, informers)
defer stopControllers()

// Wait for the Deployment to be updated to revision 1
if err := tester.waitForDeploymentRevisionAndImage("1", fakeImage); err != nil {
Expand Down Expand Up @@ -446,11 +434,8 @@ func TestDeploymentHashCollision(t *testing.T) {
}

// Start informer and controllers
stopCh := make(chan struct{})
defer close(stopCh)
informers.Start(stopCh)
go rm.Run(context.TODO(), 5)
go dc.Run(context.TODO(), 5)
stopControllers := runControllersAndInformers(t, rm, dc, informers)
defer stopControllers()

// Wait for the Deployment to be updated to revision 1
if err := tester.waitForDeploymentRevisionAndImage("1", fakeImage); err != nil {
Expand Down Expand Up @@ -549,11 +534,8 @@ func TestFailedDeployment(t *testing.T) {
}

// Start informer and controllers
stopCh := make(chan struct{})
defer close(stopCh)
informers.Start(stopCh)
go rm.Run(context.TODO(), 5)
go dc.Run(context.TODO(), 5)
stopControllers := runControllersAndInformers(t, rm, dc, informers)
defer stopControllers()

if err = tester.waitForDeploymentUpdatedReplicasGTE(replicas); err != nil {
t.Fatal(err)
Expand Down Expand Up @@ -590,12 +572,10 @@ func TestOverlappingDeployments(t *testing.T) {
{t: t, c: c, deployment: newDeployment(firstDeploymentName, ns.Name, replicas)},
{t: t, c: c, deployment: newDeployment(secondDeploymentName, ns.Name, replicas)},
}

// Start informer and controllers
stopCh := make(chan struct{})
defer close(stopCh)
informers.Start(stopCh)
go rm.Run(context.TODO(), 5)
go dc.Run(context.TODO(), 5)
stopControllers := runControllersAndInformers(t, rm, dc, informers)
defer stopControllers()

// Create 2 deployments with overlapping selectors
var err error
Expand Down Expand Up @@ -665,11 +645,9 @@ func TestScaledRolloutDeployment(t *testing.T) {
ns := framework.CreateTestingNamespace(name, s, t)
defer framework.DeleteTestingNamespace(ns, s, t)

stopCh := make(chan struct{})
defer close(stopCh)
informers.Start(stopCh)
go rm.Run(context.TODO(), 5)
go dc.Run(context.TODO(), 5)
// Start informer and controllers
stopControllers := runControllersAndInformers(t, rm, dc, informers)
defer stopControllers()

// Create a deployment with rolling update strategy, max surge = 3, and max unavailable = 2
var err error
Expand Down Expand Up @@ -868,11 +846,8 @@ func TestSpecReplicasChange(t *testing.T) {
}

// Start informer and controllers
stopCh := make(chan struct{})
defer close(stopCh)
informers.Start(stopCh)
go rm.Run(context.TODO(), 5)
go dc.Run(context.TODO(), 5)
stopControllers := runControllersAndInformers(t, rm, dc, informers)
defer stopControllers()

// Scale up/down deployment and verify its replicaset has matching .spec.replicas
if err = tester.scaleDeployment(2); err != nil {
Expand Down Expand Up @@ -926,11 +901,8 @@ func TestDeploymentAvailableCondition(t *testing.T) {
}

// Start informer and controllers
stopCh := make(chan struct{})
defer close(stopCh)
informers.Start(stopCh)
go rm.Run(context.TODO(), 5)
go dc.Run(context.TODO(), 5)
stopControllers := runControllersAndInformers(t, rm, dc, informers)
defer stopControllers()

// Wait for the deployment to be observed by the controller and has at least specified number of updated replicas
if err = tester.waitForDeploymentUpdatedReplicasGTE(replicas); err != nil {
Expand Down Expand Up @@ -1043,11 +1015,8 @@ func TestGeneralReplicaSetAdoption(t *testing.T) {
}

// Start informer and controllers
stopCh := make(chan struct{})
defer close(stopCh)
informers.Start(stopCh)
go rm.Run(context.TODO(), 5)
go dc.Run(context.TODO(), 5)
stopControllers := runControllersAndInformers(t, rm, dc, informers)
defer stopControllers()

// Wait for the Deployment to be updated to revision 1
if err := tester.waitForDeploymentRevisionAndImage("1", fakeImage); err != nil {
Expand Down Expand Up @@ -1135,11 +1104,8 @@ func TestDeploymentScaleSubresource(t *testing.T) {
}

// Start informer and controllers
stopCh := make(chan struct{})
defer close(stopCh)
informers.Start(stopCh)
go rm.Run(context.TODO(), 5)
go dc.Run(context.TODO(), 5)
stopControllers := runControllersAndInformers(t, rm, dc, informers)
defer stopControllers()

// Wait for the Deployment to be updated to revision 1
if err := tester.waitForDeploymentRevisionAndImage("1", fakeImage); err != nil {
Expand Down Expand Up @@ -1179,11 +1145,8 @@ func TestReplicaSetOrphaningAndAdoptionWhenLabelsChange(t *testing.T) {
}

// Start informer and controllers
stopCh := make(chan struct{})
defer close(stopCh)
informers.Start(stopCh)
go rm.Run(context.TODO(), 5)
go dc.Run(context.TODO(), 5)
stopControllers := runControllersAndInformers(t, rm, dc, informers)
defer stopControllers()

// Wait for the Deployment to be updated to revision 1
if err := tester.waitForDeploymentRevisionAndImage("1", fakeImage); err != nil {
Expand Down
9 changes: 9 additions & 0 deletions test/integration/deployment/util.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -146,6 +146,15 @@ func dcSimpleSetup(t *testing.T) (*httptest.Server, framework.CloseFunc, clients
return s, closeFn, clientSet
}

// runControllersAndInformers runs RS and deployment controllers and informers
func runControllersAndInformers(t *testing.T, rm *replicaset.ReplicaSetController, dc *deployment.DeploymentController, informers informers.SharedInformerFactory) func() {
ctx, cancelFn := context.WithCancel(context.Background())
informers.Start(ctx.Done())
go rm.Run(ctx, 5)
go dc.Run(ctx, 5)
return cancelFn
}

// addPodConditionReady sets given pod status to ready at given time
func addPodConditionReady(pod *v1.Pod, time metav1.Time) {
pod.Status = v1.PodStatus{
Expand Down
58 changes: 29 additions & 29 deletions test/integration/replicaset/replicaset_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -152,12 +152,12 @@ func rmSimpleSetup(t *testing.T) (*httptest.Server, framework.CloseFunc, clients
}

// Run RS controller and informers
func runControllerAndInformers(t *testing.T, rm *replicaset.ReplicaSetController, informers informers.SharedInformerFactory, podNum int) chan struct{} {
stopCh := make(chan struct{})
informers.Start(stopCh)
func runControllerAndInformers(t *testing.T, rm *replicaset.ReplicaSetController, informers informers.SharedInformerFactory, podNum int) func() {
ctx, cancelFn := context.WithCancel(context.Background())
informers.Start(ctx.Done())
waitToObservePods(t, informers.Core().V1().Pods().Informer(), podNum)
go rm.Run(context.TODO(), 5)
return stopCh
go rm.Run(ctx, 5)
return cancelFn
}

// wait for the podInformer to observe the pods. Call this function before
Expand Down Expand Up @@ -443,8 +443,8 @@ func TestAdoption(t *testing.T) {
t.Fatalf("Failed to create Pod: %v", err)
}

stopCh := runControllerAndInformers(t, rm, informers, 1)
defer close(stopCh)
stopControllers := runControllerAndInformers(t, rm, informers, 1)
defer stopControllers()
if err := wait.PollImmediate(interval, timeout, func() (bool, error) {
updatedPod, err := podClient.Get(context.TODO(), pod.Name, metav1.GetOptions{})
if err != nil {
Expand Down Expand Up @@ -498,8 +498,8 @@ func TestSpecReplicasChange(t *testing.T) {
defer closeFn()
ns := framework.CreateTestingNamespace("test-spec-replicas-change", s, t)
defer framework.DeleteTestingNamespace(ns, s, t)
stopCh := runControllerAndInformers(t, rm, informers, 0)
defer close(stopCh)
stopControllers := runControllerAndInformers(t, rm, informers, 0)
defer stopControllers()

rs := newRS("rs", ns.Name, 2)
rss, _ := createRSsPods(t, c, []*apps.ReplicaSet{rs}, []*v1.Pod{})
Expand Down Expand Up @@ -540,8 +540,8 @@ func TestDeletingAndFailedPods(t *testing.T) {
defer closeFn()
ns := framework.CreateTestingNamespace("test-deleting-and-failed-pods", s, t)
defer framework.DeleteTestingNamespace(ns, s, t)
stopCh := runControllerAndInformers(t, rm, informers, 0)
defer close(stopCh)
stopControllers := runControllerAndInformers(t, rm, informers, 0)
defer stopControllers()

rs := newRS("rs", ns.Name, 2)
rss, _ := createRSsPods(t, c, []*apps.ReplicaSet{rs}, []*v1.Pod{})
Expand Down Expand Up @@ -643,8 +643,8 @@ func TestPodDeletionCost(t *testing.T) {
defer closeFn()
ns := framework.CreateTestingNamespace(tc.name, s, t)
defer framework.DeleteTestingNamespace(ns, s, t)
stopCh := runControllerAndInformers(t, rm, informers, 0)
defer close(stopCh)
stopControllers := runControllerAndInformers(t, rm, informers, 0)
defer stopControllers()

rs := newRS("rs", ns.Name, 2)
rss, _ := createRSsPods(t, c, []*apps.ReplicaSet{rs}, []*v1.Pod{})
Expand Down Expand Up @@ -702,8 +702,8 @@ func TestOverlappingRSs(t *testing.T) {
defer closeFn()
ns := framework.CreateTestingNamespace("test-overlapping-rss", s, t)
defer framework.DeleteTestingNamespace(ns, s, t)
stopCh := runControllerAndInformers(t, rm, informers, 0)
defer close(stopCh)
stopControllers := runControllerAndInformers(t, rm, informers, 0)
defer stopControllers()

// Create 2 RSs with identical selectors
for i := 0; i < 2; i++ {
Expand Down Expand Up @@ -737,8 +737,8 @@ func TestPodOrphaningAndAdoptionWhenLabelsChange(t *testing.T) {
defer closeFn()
ns := framework.CreateTestingNamespace("test-pod-orphaning-and-adoption-when-labels-change", s, t)
defer framework.DeleteTestingNamespace(ns, s, t)
stopCh := runControllerAndInformers(t, rm, informers, 0)
defer close(stopCh)
stopControllers := runControllerAndInformers(t, rm, informers, 0)
defer stopControllers()

rs := newRS("rs", ns.Name, 1)
rss, _ := createRSsPods(t, c, []*apps.ReplicaSet{rs}, []*v1.Pod{})
Expand Down Expand Up @@ -814,8 +814,8 @@ func TestGeneralPodAdoption(t *testing.T) {
defer closeFn()
ns := framework.CreateTestingNamespace("test-general-pod-adoption", s, t)
defer framework.DeleteTestingNamespace(ns, s, t)
stopCh := runControllerAndInformers(t, rm, informers, 0)
defer close(stopCh)
stopControllers := runControllerAndInformers(t, rm, informers, 0)
defer stopControllers()

rs := newRS("rs", ns.Name, 1)
rss, _ := createRSsPods(t, c, []*apps.ReplicaSet{rs}, []*v1.Pod{})
Expand Down Expand Up @@ -846,8 +846,8 @@ func TestReadyAndAvailableReplicas(t *testing.T) {
defer closeFn()
ns := framework.CreateTestingNamespace("test-ready-and-available-replicas", s, t)
defer framework.DeleteTestingNamespace(ns, s, t)
stopCh := runControllerAndInformers(t, rm, informers, 0)
defer close(stopCh)
stopControllers := runControllerAndInformers(t, rm, informers, 0)
defer stopControllers()

rs := newRS("rs", ns.Name, 3)
rs.Spec.MinReadySeconds = 3600
Expand Down Expand Up @@ -898,8 +898,8 @@ func TestRSScaleSubresource(t *testing.T) {
defer closeFn()
ns := framework.CreateTestingNamespace("test-rs-scale-subresource", s, t)
defer framework.DeleteTestingNamespace(ns, s, t)
stopCh := runControllerAndInformers(t, rm, informers, 0)
defer close(stopCh)
stopControllers := runControllerAndInformers(t, rm, informers, 0)
defer stopControllers()

rs := newRS("rs", ns.Name, 1)
rss, _ := createRSsPods(t, c, []*apps.ReplicaSet{rs}, []*v1.Pod{})
Expand Down Expand Up @@ -928,8 +928,8 @@ func TestExtraPodsAdoptionAndDeletion(t *testing.T) {
}
rss, _ := createRSsPods(t, c, []*apps.ReplicaSet{rs}, podList)
rs = rss[0]
stopCh := runControllerAndInformers(t, rm, informers, 3)
defer close(stopCh)
stopControllers := runControllerAndInformers(t, rm, informers, 3)
defer stopControllers()
waitRSStable(t, c, rs)

// Verify the extra pod is deleted eventually by determining whether number of
Expand All @@ -949,8 +949,8 @@ func TestFullyLabeledReplicas(t *testing.T) {
defer closeFn()
ns := framework.CreateTestingNamespace("test-fully-labeled-replicas", s, t)
defer framework.DeleteTestingNamespace(ns, s, t)
stopCh := runControllerAndInformers(t, rm, informers, 0)
defer close(stopCh)
stopControllers := runControllerAndInformers(t, rm, informers, 0)
defer stopControllers()

extraLabelMap := map[string]string{"foo": "bar", "extraKey": "extraValue"}
rs := newRS("rs", ns.Name, 2)
Expand Down Expand Up @@ -992,8 +992,8 @@ func TestReplicaSetsAppsV1DefaultGCPolicy(t *testing.T) {
defer closeFn()
ns := framework.CreateTestingNamespace("test-default-gc-v1", s, t)
defer framework.DeleteTestingNamespace(ns, s, t)
stopCh := runControllerAndInformers(t, rm, informers, 0)
defer close(stopCh)
stopControllers := runControllerAndInformers(t, rm, informers, 0)
defer stopControllers()

rs := newRS("rs", ns.Name, 2)
fakeFinalizer := "kube.io/dummy-finalizer"
Expand Down
Loading

0 comments on commit 564b204

Please sign in to comment.