Merge pull request #24537 from mwielgus/fix-race-hpa

Automatic merge from submit-queue

Add locks in HPA test

Fixes #23361

pkg/controller/podautoscaler/horizontal_test.go

@@ -20,6 +20,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"sync"
 	"testing"
 	"time"
 
@@ -66,10 +67,12 @@ type fakeResource struct {
 }
 
 type testCase struct {
+	sync.Mutex
 	minReplicas     int
 	maxReplicas     int
 	initialReplicas int
 	desiredReplicas int
+
 	// CPU target utilization as a percentage of the requested resources.
 	CPUTarget           int
 	CPUCurrent          int
@@ -88,6 +91,7 @@ type testCase struct {
 	resource *fakeResource
 }
 
+// Needs to be called under a lock.
 func (tc *testCase) computeCPUCurrent() {
 	if len(tc.reportedLevels) != len(tc.reportedCPURequests) || len(tc.reportedLevels) == 0 {
 		return
@@ -111,6 +115,8 @@ func (tc *testCase) prepareTestClient(t *testing.T) *fake.Clientset {
 		MatchLabels: map[string]string{"name": podNamePrefix},
 	}
 
+	tc.Lock()
+
 	tc.scaleUpdated = false
 	tc.statusUpdated = false
 	tc.eventCreated = false
@@ -126,9 +132,13 @@ func (tc *testCase) prepareTestClient(t *testing.T) *fake.Clientset {
 			kind:       "replicationcontrollers",
 		}
 	}
+	tc.Unlock()
 
 	fakeClient := &fake.Clientset{}
 	fakeClient.AddReactor("list", "horizontalpodautoscalers", func(action core.Action) (handled bool, ret runtime.Object, err error) {
+		tc.Lock()
+		defer tc.Unlock()
+
 		obj := &extensions.HorizontalPodAutoscalerList{
 			Items: []extensions.HorizontalPodAutoscaler{
 				{
@@ -154,6 +164,7 @@ func (tc *testCase) prepareTestClient(t *testing.T) *fake.Clientset {
 				},
 			},
 		}
+
 		if tc.CPUTarget > 0.0 {
 			obj.Items[0].Spec.CPUUtilization = &extensions.CPUTargetUtilization{TargetPercentage: tc.CPUTarget}
 		}
@@ -169,6 +180,9 @@ func (tc *testCase) prepareTestClient(t *testing.T) *fake.Clientset {
 	})
 
 	fakeClient.AddReactor("get", "replicationcontrollers", func(action core.Action) (handled bool, ret runtime.Object, err error) {
+		tc.Lock()
+		defer tc.Unlock()
+
 		obj := &extensions.Scale{
 			ObjectMeta: api.ObjectMeta{
 				Name:      tc.resource.name,
@@ -186,6 +200,9 @@ func (tc *testCase) prepareTestClient(t *testing.T) *fake.Clientset {
 	})
 
 	fakeClient.AddReactor("get", "deployments", func(action core.Action) (handled bool, ret runtime.Object, err error) {
+		tc.Lock()
+		defer tc.Unlock()
+
 		obj := &extensions.Scale{
 			ObjectMeta: api.ObjectMeta{
 				Name:      tc.resource.name,
@@ -203,6 +220,9 @@ func (tc *testCase) prepareTestClient(t *testing.T) *fake.Clientset {
 	})
 
 	fakeClient.AddReactor("get", "replicasets", func(action core.Action) (handled bool, ret runtime.Object, err error) {
+		tc.Lock()
+		defer tc.Unlock()
+
 		obj := &extensions.Scale{
 			ObjectMeta: api.ObjectMeta{
 				Name:      tc.resource.name,
@@ -220,6 +240,9 @@ func (tc *testCase) prepareTestClient(t *testing.T) *fake.Clientset {
 	})
 
 	fakeClient.AddReactor("list", "pods", func(action core.Action) (handled bool, ret runtime.Object, err error) {
+		tc.Lock()
+		defer tc.Unlock()
+
 		obj := &api.PodList{}
 		for i := 0; i < len(tc.reportedCPURequests); i++ {
 			podName := fmt.Sprintf("%s-%d", podNamePrefix, i)
@@ -252,6 +275,9 @@ func (tc *testCase) prepareTestClient(t *testing.T) *fake.Clientset {
 	})
 
 	fakeClient.AddProxyReactor("services", func(action core.Action) (handled bool, ret restclient.ResponseWrapper, err error) {
+		tc.Lock()
+		defer tc.Unlock()
+
 		timestamp := time.Now()
 		metrics := heapster.MetricResultList{}
 		for _, level := range tc.reportedLevels {
@@ -266,6 +292,9 @@ func (tc *testCase) prepareTestClient(t *testing.T) *fake.Clientset {
 	})
 
 	fakeClient.AddReactor("update", "replicationcontrollers", func(action core.Action) (handled bool, ret runtime.Object, err error) {
+		tc.Lock()
+		defer tc.Unlock()
+
 		obj := action.(testclient.UpdateAction).GetObject().(*extensions.Scale)
 		replicas := action.(testclient.UpdateAction).GetObject().(*extensions.Scale).Spec.Replicas
 		assert.Equal(t, tc.desiredReplicas, replicas)
@@ -274,6 +303,9 @@ func (tc *testCase) prepareTestClient(t *testing.T) *fake.Clientset {
 	})
 
 	fakeClient.AddReactor("update", "deployments", func(action core.Action) (handled bool, ret runtime.Object, err error) {
+		tc.Lock()
+		defer tc.Unlock()
+
 		obj := action.(testclient.UpdateAction).GetObject().(*extensions.Scale)
 		replicas := action.(testclient.UpdateAction).GetObject().(*extensions.Scale).Spec.Replicas
 		assert.Equal(t, tc.desiredReplicas, replicas)
@@ -282,6 +314,9 @@ func (tc *testCase) prepareTestClient(t *testing.T) *fake.Clientset {
 	})
 
 	fakeClient.AddReactor("update", "replicasets", func(action core.Action) (handled bool, ret runtime.Object, err error) {
+		tc.Lock()
+		defer tc.Unlock()
+
 		obj := action.(testclient.UpdateAction).GetObject().(*extensions.Scale)
 		replicas := action.(testclient.UpdateAction).GetObject().(*extensions.Scale).Spec.Replicas
 		assert.Equal(t, tc.desiredReplicas, replicas)
@@ -290,6 +325,9 @@ func (tc *testCase) prepareTestClient(t *testing.T) *fake.Clientset {
 	})
 
 	fakeClient.AddReactor("update", "horizontalpodautoscalers", func(action core.Action) (handled bool, ret runtime.Object, err error) {
+		tc.Lock()
+		defer tc.Unlock()
+
 		obj := action.(testclient.UpdateAction).GetObject().(*extensions.HorizontalPodAutoscaler)
 		assert.Equal(t, namespace, obj.Namespace)
 		assert.Equal(t, hpaName, obj.Name)
@@ -305,6 +343,9 @@ func (tc *testCase) prepareTestClient(t *testing.T) *fake.Clientset {
 	})
 
 	fakeClient.AddReactor("*", "events", func(action core.Action) (handled bool, ret runtime.Object, err error) {
+		tc.Lock()
+		defer tc.Unlock()
+
 		obj := action.(testclient.CreateAction).GetObject().(*api.Event)
 		if tc.verifyEvents {
 			assert.Equal(t, "SuccessfulRescale", obj.Reason)
@@ -321,6 +362,9 @@ func (tc *testCase) prepareTestClient(t *testing.T) *fake.Clientset {
 }
 
 func (tc *testCase) verifyResults(t *testing.T) {
+	tc.Lock()
+	defer tc.Unlock()
+
 	assert.Equal(t, tc.initialReplicas != tc.desiredReplicas, tc.scaleUpdated)
 	assert.True(t, tc.statusUpdated)
 	if tc.verifyEvents {
@@ -351,9 +395,13 @@ func (tc *testCase) runTest(t *testing.T) {
 	defer close(stop)
 	go hpaController.Run(stop)
 
+	tc.Lock()
 	if tc.verifyEvents {
+		tc.Unlock()
 		// We need to wait for events to be broadcasted (sleep for longer than record.sleepDuration).
 		time.Sleep(2 * time.Second)
+	} else {
+		tc.Unlock()
 	}
 	// Wait for HPA to be processed.
 	<-tc.processed