Merge pull request #619 from freehan/unit-test

Add more unit tests for transaction syncer

pkg/neg/syncers/syncer.go

@@ -28,6 +28,10 @@ import (
 	"k8s.io/client-go/tools/record"
 )
 
+type syncerCore interface {
+	sync() error
+}
+
 // syncer is a NEG syncer skeleton.
 // It handles state transitions and backoff retry operations.
 type syncer struct {
@@ -36,7 +40,7 @@ type syncer struct {
 	negName string
 
 	// NEG sync function
-	syncFunc func() error
+	core syncerCore
 
 	// event recording
 	serviceLister cache.Indexer
@@ -53,11 +57,11 @@ type syncer struct {
 	backoff backoffHandler
 }
 
-func newSyncer(negSyncerKey NegSyncerKey, networkEndpointGroupName string, serviceLister cache.Indexer, recorder record.EventRecorder) *syncer {
+func newSyncer(negSyncerKey NegSyncerKey, networkEndpointGroupName string, serviceLister cache.Indexer, recorder record.EventRecorder, core syncerCore) *syncer {
 	return &syncer{
 		NegSyncerKey:  negSyncerKey,
 		negName:       networkEndpointGroupName,
-		syncFunc:      func() error { return nil },
+		core:          core,
 		serviceLister: serviceLister,
 		recorder:      recorder,
 		stopped:       true,
@@ -81,7 +85,7 @@ func (s *syncer) Start() error {
 		for {
 			// equivalent to never retry
 			retryCh := make(<-chan time.Time)
-			err := s.syncFunc()
+			err := s.core.sync()
 			if err != nil {
 				delay, retryErr := s.backoff.NextRetryDelay()
 				retryMesg := ""
@@ -158,7 +162,3 @@ func (s *syncer) IsShuttingDown() bool {
 	defer s.stateLock.Unlock()
 	return s.shuttingDown
 }
-
-func (s *syncer) SetSyncFunc(syncFunc func() error) {
-	s.syncFunc = syncFunc
-}

pkg/neg/syncers/syncer_test.go

@@ -27,11 +27,12 @@ import (
 	"k8s.io/client-go/tools/record"
 	backendconfigclient "k8s.io/ingress-gce/pkg/backendconfig/client/clientset/versioned/fake"
 	"k8s.io/ingress-gce/pkg/context"
+	negtypes "k8s.io/ingress-gce/pkg/neg/types"
 	"k8s.io/ingress-gce/pkg/utils"
 )
 
 type syncerTester struct {
-	syncer
+	syncer negtypes.NegSyncer
 	// keep track of the number of syncs
 	syncCount int
 	// syncError is true, then sync function return error
@@ -53,7 +54,7 @@ func (t *syncerTester) sync() error {
 	return nil
 }
 
-func newTestNegSyncer() *syncerTester {
+func newSyncerTester() *syncerTester {
 	testNegName := "test-neg-name"
 	kubeClient := fake.NewSimpleClientset()
 	backendConfigClient := backendconfigclient.NewSimpleClientset()
@@ -73,94 +74,96 @@ func newTestNegSyncer() *syncerTester {
 		TargetPort: "80",
 	}
 
-	s := &syncerTester{
-		syncer: *newSyncer(
-			negSyncerKey,
-			testNegName,
-			context.ServiceInformer.GetIndexer(),
-			record.NewFakeRecorder(100),
-		),
+	st := &syncerTester{
 		syncCount: 0,
 		blockSync: false,
 		syncError: false,
 		ch:        make(chan interface{}),
 	}
-	s.SetSyncFunc(s.sync)
-	return s
+
+	s := newSyncer(
+		negSyncerKey,
+		testNegName,
+		context.ServiceInformer.GetIndexer(),
+		record.NewFakeRecorder(100),
+		st,
+	)
+	st.syncer = s
+	return st
 }
 
 func TestStartAndStopNoopSyncer(t *testing.T) {
-	syncer := newTestNegSyncer()
-	if !syncer.IsStopped() {
+	syncerTester := newSyncerTester()
+	if !syncerTester.syncer.IsStopped() {
 		t.Fatalf("Syncer is not stopped after creation.")
 	}
-	if syncer.IsShuttingDown() {
+	if syncerTester.syncer.IsShuttingDown() {
 		t.Fatalf("Syncer is shutting down after creation.")
 	}
 
-	if err := syncer.Start(); err != nil {
+	if err := syncerTester.syncer.Start(); err != nil {
 		t.Fatalf("Failed to start syncer: %v", err)
 	}
-	if syncer.IsStopped() {
+	if syncerTester.syncer.IsStopped() {
 		t.Fatalf("Syncer is stopped after Start.")
 	}
-	if syncer.IsShuttingDown() {
+	if syncerTester.syncer.IsShuttingDown() {
 		t.Fatalf("Syncer is shutting down after Start.")
 	}
 
 	// blocks sync function
-	syncer.blockSync = true
-	syncer.Stop()
-	if !syncer.IsShuttingDown() {
+	syncerTester.blockSync = true
+	syncerTester.syncer.Stop()
+	if !syncerTester.syncer.IsShuttingDown() {
 		// assume syncer needs 5 second for sync
 		t.Fatalf("Syncer is not shutting down after Start.")
 	}
 
-	if !syncer.IsStopped() {
+	if !syncerTester.syncer.IsStopped() {
 		t.Fatalf("Syncer is not stopped after Stop.")
 	}
 
 	// unblock sync function
-	syncer.ch <- struct{}{}
+	syncerTester.ch <- struct{}{}
 	if err := wait.PollImmediate(time.Second, 3*time.Second, func() (bool, error) {
-		return !syncer.IsShuttingDown() && syncer.IsStopped(), nil
+		return !syncerTester.syncer.IsShuttingDown() && syncerTester.syncer.IsStopped(), nil
 	}); err != nil {
 		t.Fatalf("Syncer failed to shutdown: %v", err)
 	}
 
-	if err := syncer.Start(); err != nil {
+	if err := syncerTester.syncer.Start(); err != nil {
 		t.Fatalf("Failed to restart syncer: %v", err)
 	}
-	if syncer.IsStopped() {
+	if syncerTester.syncer.IsStopped() {
 		t.Fatalf("Syncer is stopped after restart.")
 	}
-	if syncer.IsShuttingDown() {
+	if syncerTester.syncer.IsShuttingDown() {
 		t.Fatalf("Syncer is shutting down after restart.")
 	}
 
-	syncer.Stop()
-	if !syncer.IsStopped() {
+	syncerTester.syncer.Stop()
+	if !syncerTester.syncer.IsStopped() {
 		t.Fatalf("Syncer is not stopped after Stop.")
 	}
 }
 
 func TestRetryOnSyncError(t *testing.T) {
 	maxRetry := 3
-	syncer := newTestNegSyncer()
-	syncer.syncError = true
-	if err := syncer.Start(); err != nil {
+	syncerTester := newSyncerTester()
+	syncerTester.syncError = true
+	if err := syncerTester.syncer.Start(); err != nil {
 		t.Fatalf("Failed to start syncer: %v", err)
 	}
-	syncer.backoff = NewExponentialBackendOffHandler(maxRetry, 0, 0)
+	syncerTester.syncer.(*syncer).backoff = NewExponentialBackendOffHandler(maxRetry, 0, 0)
 
 	if err := wait.PollImmediate(time.Second, 5*time.Second, func() (bool, error) {
 		// In 5 seconds, syncer should be able to retry 3 times.
-		return syncer.syncCount == maxRetry+1, nil
+		return syncerTester.syncCount == maxRetry+1, nil
 	}); err != nil {
 		t.Errorf("Syncer failed to retry and record error: %v", err)
 	}
 
-	if syncer.syncCount != maxRetry+1 {
-		t.Errorf("Expect sync count to be %v, but got %v", maxRetry+1, syncer.syncCount)
+	if syncerTester.syncCount != maxRetry+1 {
+		t.Errorf("Expect sync count to be %v, but got %v", maxRetry+1, syncerTester.syncCount)
 	}
 }

pkg/neg/syncers/transaction.go

@@ -61,21 +61,23 @@ type transactionSyncer struct {
 }
 
 func NewTransactionSyncer(negSyncerKey NegSyncerKey, networkEndpointGroupName string, recorder record.EventRecorder, cloud negtypes.NetworkEndpointGroupCloud, zoneGetter negtypes.ZoneGetter, serviceLister cache.Indexer, endpointLister cache.Indexer) negtypes.NegSyncer {
-	syncer := newSyncer(negSyncerKey, networkEndpointGroupName, serviceLister, recorder)
+	// TransactionSyncer implements the syncer core
 	ts := &transactionSyncer{
 		NegSyncerKey:   negSyncerKey,
 		negName:        networkEndpointGroupName,
-		syncer:         syncer,
 		needInit:       true,
 		transactions:   NewTransactionTable(),
 		serviceLister:  serviceLister,
 		endpointLister: endpointLister,
 		recorder:       recorder,
 		cloud:          cloud,
 		zoneGetter:     zoneGetter,
-		retry:          NewDelayRetryHandler(func() { syncer.Sync() }, NewExponentialBackendOffHandler(maxRetries, minRetryDelay, maxRetryDelay)),
 	}
-	syncer.SetSyncFunc(ts.sync)
+	// Syncer implements life cycle logic
+	syncer := newSyncer(negSyncerKey, networkEndpointGroupName, serviceLister, recorder, ts)
+	// transactionSyncer needs syncer interface for internals
+	ts.syncer = syncer
+	ts.retry = NewDelayRetryHandler(func() { syncer.Sync() }, NewExponentialBackendOffHandler(maxRetries, minRetryDelay, maxRetryDelay))
 	return syncer
 }
 
@@ -285,7 +287,7 @@ func (s *transactionSyncer) commitTransaction(err error, networkEndpointMap map[
 	for encodedEndpoint := range networkEndpointMap {
 		entry, ok := s.transactions.Get(encodedEndpoint)
 		if !ok {
-			glog.Errorf("Endpoint %q was not found in the transaction table.")
+			glog.Errorf("Endpoint %q was not found in the transaction table.", encodedEndpoint)
 			needSync = true
 			continue
 		}