Replace router svcAddrUpdateLocks with new throttler package for code readability&reusability (#1047)

Rename svcAddrUpdateLock to Throttler and move it to the top-level directory as a package for code import.

router/functionHandler.go

@@ -39,6 +39,7 @@ import (
 	"github.com/fission/fission/crd"
 	executorClient "github.com/fission/fission/executor/client"
 	"github.com/fission/fission/redis"
+	"github.com/fission/fission/throttler"
 )
 
 const (
@@ -46,56 +47,63 @@ const (
 	X_FORWARDED_HOST = "X-Forwarded-Host"
 )
 
-type tsRoundTripperParams struct {
-	timeout         time.Duration
-	timeoutExponent int
-	keepAlive       time.Duration
+type (
+	functionHandler struct {
+		fmap                     *functionServiceMap
+		frmap                    *functionRecorderMap
+		trmap                    *triggerRecorderMap
+		executor                 *executorClient.Client
+		function                 *metav1.ObjectMeta
+		httpTrigger              *crd.HTTPTrigger
+		functionMetadataMap      map[string]*metav1.ObjectMeta
+		fnWeightDistributionList []FunctionWeightDistribution
+		tsRoundTripperParams     *tsRoundTripperParams
+		recorderName             string
+		isDebugEnv               bool
+		svcAddrUpdateThrottler   *throttler.Throttler
+	}
 
-	// maxRetires is the max times for RetryingRoundTripper to retry a request.
-	// Default maxRetries is 10, which means router will retry for
-	// up to 10 times and abort it if still not succeeded.
-	maxRetries int
+	tsRoundTripperParams struct {
+		timeout         time.Duration
+		timeoutExponent int
+		keepAlive       time.Duration
 
-	// svcAddrRetryCount is the max times for RetryingRoundTripper to retry with a specific service address
-	// Router sends requests to a specific service address for each function.
-	// A service address is considered as an invalid one if amount of non-network
-	// errors router received is higher than svcAddrRetryCount. In this situation,
-	// remove it from cache and try to get a new one from executor.
-	// Default svcAddrRetryCount is 5.
-	svcAddrRetryCount int
-}
+		// maxRetires is the max times for RetryingRoundTripper to retry a request.
+		// Default maxRetries is 10, which means router will retry for
+		// up to 10 times and abort it if still not succeeded.
+		maxRetries int
 
-type functionHandler struct {
-	fmap                     *functionServiceMap
-	frmap                    *functionRecorderMap
-	trmap                    *triggerRecorderMap
-	executor                 *executorClient.Client
-	function                 *metav1.ObjectMeta
-	httpTrigger              *crd.HTTPTrigger
-	functionMetadataMap      map[string]*metav1.ObjectMeta
-	fnWeightDistributionList []FunctionWeightDistribution
-	tsRoundTripperParams     *tsRoundTripperParams
-	recorderName             string
-	isDebugEnv               bool
-	svcAddrUpdateLocks       *svcAddrUpdateLocks
-}
+		// svcAddrRetryCount is the max times for RetryingRoundTripper to retry with a specific service address
+		// Router sends requests to a specific service address for each function.
+		// A service address is considered as an invalid one if amount of non-network
+		// errors router received is higher than svcAddrRetryCount. In this situation,
+		// remove it from cache and try to get a new one from executor.
+		// Default svcAddrRetryCount is 5.
+		svcAddrRetryCount int
+	}
 
-// A layer on top of http.DefaultTransport, with retries.
-type RetryingRoundTripper struct {
-	funcHandler *functionHandler
-}
+	// A layer on top of http.DefaultTransport, with retries.
+	RetryingRoundTripper struct {
+		funcHandler *functionHandler
+	}
+
+	// To keep the request body open during retries, we create an interface with Close operation being a no-op.
+	// Details : https://github.com/flynn/flynn/pull/875
+	fakeCloseReadCloser struct {
+		io.ReadCloser
+	}
+
+	svcEntryRecord struct {
+		svcUrl    *url.URL
+		fromCache bool
+	}
+)
 
 func init() {
 	// just seeding the random number for getting the canary function
 	rand.Seed(time.Now().UnixNano())
 }
 
-// To keep the request body open during retries, we create an interface with Close operation being a no-op.
-// Details : https://github.com/flynn/flynn/pull/875
-type fakeCloseReadCloser struct {
-	io.ReadCloser
-}
-
 func (w *fakeCloseReadCloser) Close() error {
 	return nil
 }
@@ -228,7 +236,8 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt
 			return nil, fission.MakeError(http.StatusInternalServerError, err.Error())
 		}
 
-		// retry to get service url again
+		// service url maybe nil if router cannot find one in cache,
+		// so here we retry to get service url again
 		if serviceUrl == nil {
 			time.Sleep(executingTimeout)
 			continue
@@ -479,27 +488,34 @@ func (fh *functionHandler) getServiceEntry() (serviceUrl *url.URL, serviceUrlFro
 
 	// cache miss or nil entry in cache
 
-	// To prevent multiple update requests will be sent to executor and make executor overloaded,
-	// the first goroutine is responsible to update the service url entry, all other goroutines
-	// for the same function will wait until first goroutine finished.
-	serviceUrl, serviceUrlFromCache, err = fh.svcAddrUpdateLocks.RunOnce(
-		fh.function,
-		func(firstToTheLock bool) (u *url.URL, fromCache bool, err error) {
-
+	// Use throttle to limit the total amount of requests sent
+	// to the executor to prevent it from overloaded.
+	recordObj, err := fh.svcAddrUpdateThrottler.RunOnce(
+		crd.CacheKey(fh.function),
+		func(firstToTheLock bool) (interface{}, error) {
+			var u *url.URL
 			// Get service entry from executor and update cache if its the first goroutine
 			if firstToTheLock { // first to the service url
 				log.Printf("Calling getServiceForFunction for function: %s", fh.function.Name)
 				u, err = fh.getServiceEntryFromExecutor()
-				if err == nil && u != nil {
-					// add the address in router's cache
-					log.Printf("Assigning service url: %s for function: %s", u, fh.function.Name)
-					fh.fmap.assign(fh.function, u)
+				if err != nil {
+					log.Printf("Error getting service url from executor: %v", err)
+					return nil, err
 				}
+				// add the address in router's cache
+				log.Printf("Assigning service url: %s for function: %s", u, fh.function.Name)
+				fh.fmap.assign(fh.function, u)
 			} else {
 				u, err = fh.getServiceEntryFromCache()
+				if err != nil {
+					return nil, err
+				}
 			}
 
-			return u, firstToTheLock, err
+			return svcEntryRecord{
+				svcUrl:    u,
+				fromCache: firstToTheLock,
+			}, err
 		},
 	)
 	if err != nil {
@@ -508,7 +524,12 @@ func (fh *functionHandler) getServiceEntry() (serviceUrl *url.URL, serviceUrlFro
 		return nil, false, err
 	}
 
-	return serviceUrl, serviceUrlFromCache, err
+	record, ok := recordObj.(svcEntryRecord)
+	if !ok {
+		return nil, false, errors.Errorf("Received unknown service record type")
+	}
+
+	return record.svcUrl, record.fromCache, nil
 }
 
 // getServiceEntryFromCache returns service url entry returns from cache

router/httpTriggers.go

@@ -32,6 +32,7 @@ import (
 	"github.com/fission/fission"
 	"github.com/fission/fission/crd"
 	executorClient "github.com/fission/fission/executor/client"
+	"github.com/fission/fission/throttler"
 )
 
 type HTTPTriggerSet struct {
@@ -53,11 +54,11 @@ type HTTPTriggerSet struct {
 	updateRouterRequestChannel chan struct{}
 	tsRoundTripperParams       *tsRoundTripperParams
 	isDebugEnv                 bool
-	svcAddrUpdateLocks         *svcAddrUpdateLocks
+	svcAddrUpdateThrottler     *throttler.Throttler
 }
 
 func makeHTTPTriggerSet(fmap *functionServiceMap, frmap *functionRecorderMap, trmap *triggerRecorderMap, fissionClient *crd.FissionClient,
-	kubeClient *kubernetes.Clientset, executor *executorClient.Client, crdClient *rest.RESTClient, params *tsRoundTripperParams, isDebugEnv bool, locks *svcAddrUpdateLocks) (*HTTPTriggerSet, k8sCache.Store, k8sCache.Store) {
+	kubeClient *kubernetes.Clientset, executor *executorClient.Client, crdClient *rest.RESTClient, params *tsRoundTripperParams, isDebugEnv bool, actionThrottler *throttler.Throttler) (*HTTPTriggerSet, k8sCache.Store, k8sCache.Store) {
 
 	httpTriggerSet := &HTTPTriggerSet{
 		functionServiceMap:         fmap,
@@ -69,7 +70,7 @@ func makeHTTPTriggerSet(fmap *functionServiceMap, frmap *functionRecorderMap, tr
 		updateRouterRequestChannel: make(chan struct{}),
 		tsRoundTripperParams:       params,
 		isDebugEnv:                 isDebugEnv,
-		svcAddrUpdateLocks:         locks,
+		svcAddrUpdateThrottler:     actionThrottler,
 	}
 	var tStore, fnStore, rStore k8sCache.Store
 	var tController, fnController k8sCache.Controller
@@ -156,7 +157,7 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router {
 			tsRoundTripperParams:     ts.tsRoundTripperParams,
 			recorderName:             recorderName,
 			isDebugEnv:               ts.isDebugEnv,
-			svcAddrUpdateLocks:       ts.svcAddrUpdateLocks,
+			svcAddrUpdateThrottler:   ts.svcAddrUpdateThrottler,
 		}
 
 		// The functionHandler for HTTP trigger with fn reference type "FunctionReferenceTypeFunctionName",
@@ -204,15 +205,15 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router {
 		}
 
 		fh := &functionHandler{
-			fmap:                 ts.functionServiceMap,
-			frmap:                ts.recorderSet.functionRecorderMap,
-			trmap:                ts.recorderSet.triggerRecorderMap,
-			function:             &m,
-			executor:             ts.executor,
-			tsRoundTripperParams: ts.tsRoundTripperParams,
-			recorderName:         recorderName,
-			isDebugEnv:           ts.isDebugEnv,
-			svcAddrUpdateLocks:   ts.svcAddrUpdateLocks,
+			fmap:                   ts.functionServiceMap,
+			frmap:                  ts.recorderSet.functionRecorderMap,
+			trmap:                  ts.recorderSet.triggerRecorderMap,
+			function:               &m,
+			executor:               ts.executor,
+			tsRoundTripperParams:   ts.tsRoundTripperParams,
+			recorderName:           recorderName,
+			isDebugEnv:             ts.isDebugEnv,
+			svcAddrUpdateThrottler: ts.svcAddrUpdateThrottler,
 		}
 		muxRouter.HandleFunc(fission.UrlForFunction(function.Metadata.Name, function.Metadata.Namespace), fh.handler)
 	}

router/router.go

@@ -54,6 +54,7 @@ import (
 	"github.com/fission/fission"
 	"github.com/fission/fission/crd"
 	executorClient "github.com/fission/fission/executor/client"
+	"github.com/fission/fission/throttler"
 )
 
 // request url ---[mux]---> Function(name,uid) ----[fmap]----> k8s service url
@@ -150,7 +151,7 @@ func Start(port int, executorUrl string) {
 		keepAlive:         keepAlive,
 		maxRetries:        maxRetries,
 		svcAddrRetryCount: svcAddrRetryCount,
-	}, isDebugEnv, MakeUpdateLocks(svcAddrUpdateTimeout))
+	}, isDebugEnv, throttler.MakeThrottler(svcAddrUpdateTimeout))
 
 	resolver := makeFunctionReferenceResolver(fnStore)
 

router/router_test.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/fission/fission"
 	"github.com/fission/fission/crd"
+	"github.com/fission/fission/throttler"
 )
 
 func TestRouter(t *testing.T) {
@@ -57,7 +58,7 @@ func TestRouter(t *testing.T) {
 			timeoutExponent: 2,
 			keepAlive:       30 * time.Second,
 			maxRetries:      10,
-		}, false, MakeUpdateLocks(30*time.Second))
+		}, false, throttler.MakeThrottler(30*time.Second))
 	triggerUrl := "/foo"
 	triggers.triggers = append(triggers.triggers,
 		crd.HTTPTrigger{

router/svcAddrUpdateLock.go

@@ -1,192 +0,0 @@
-/*
-Copyright 2018 The Fission 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.
-*/
-
-package router
-
-import (
-	"net/url"
-	"sync"
-	"time"
-
-	"github.com/pkg/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-
-	"github.com/fission/fission/crd"
-)
-
-type svcAddrUpdateOperation int
-
-const (
-	GET svcAddrUpdateOperation = iota
-	DELETE
-	EXPIRE
-)
-
-type (
-	// svcAddrUpdateLock is the lock that will be used when
-	// gorountine tries to update service address entry.
-	svcAddrUpdateLock struct {
-		wg         *sync.WaitGroup
-		ctimestamp time.Time // creation time of lock
-		timeExpiry time.Duration
-	}
-
-	svcAddrUpdateLocks struct {
-		requestChan    chan *svcAddrUpdateRequest
-		locks          map[string]*svcAddrUpdateLock
-		lockTimeExpiry time.Duration
-	}
-
-	svcAddrUpdateRequest struct {
-		requestType  svcAddrUpdateOperation
-		responseChan chan *svcAddrUpdateResponse
-		fnMeta       *metav1.ObjectMeta
-	}
-
-	svcAddrUpdateResponse struct {
-		lock           *svcAddrUpdateLock
-		firstGoroutine bool // denote this goroutine is the first goroutine
-	}
-)
-
-func (l *svcAddrUpdateLock) isOld() bool {
-	return time.Since(l.ctimestamp) > l.timeExpiry
-}
-
-func (l *svcAddrUpdateLock) Wait() error {
-	ch := make(chan struct{})
-
-	go func(wg *sync.WaitGroup, ch chan struct{}) {
-		wg.Wait()
-		close(ch)
-	}(l.wg, ch)
-
-	select {
-	case <-ch:
-		return nil
-	case <-time.After(l.timeExpiry):
-		return errors.New("Error waiting for svcAddrUpdateLock to be released: Exceeded timeout")
-	}
-}
-
-func MakeUpdateLocks(timeExpiry time.Duration) *svcAddrUpdateLocks {
-	locks := &svcAddrUpdateLocks{
-		requestChan:    make(chan *svcAddrUpdateRequest),
-		locks:          make(map[string]*svcAddrUpdateLock),
-		lockTimeExpiry: timeExpiry,
-	}
-	go locks.service()
-	go locks.expiryService()
-	return locks
-}
-
-func (ul *svcAddrUpdateLocks) service() {
-
-	for {
-		req := <-ul.requestChan
-
-		switch req.requestType {
-		case GET:
-			key := crd.CacheKey(req.fnMeta)
-			lock, ok := ul.locks[key]
-			if ok && !lock.isOld() {
-				req.responseChan <- &svcAddrUpdateResponse{
-					lock: lock, firstGoroutine: false,
-				}
-				continue
-			} else if ok && lock.isOld() {
-				// in case that one goroutine occupy the update lock for long time
-				lock.wg.Done()
-			}
-
-			lock = &svcAddrUpdateLock{
-				wg:         &sync.WaitGroup{},
-				ctimestamp: time.Now(),
-				timeExpiry: ul.lockTimeExpiry,
-			}
-
-			lock.wg.Add(1)
-
-			ul.locks[key] = lock
-
-			req.responseChan <- &svcAddrUpdateResponse{
-				lock: lock, firstGoroutine: true,
-			}
-
-		case DELETE:
-			key := crd.CacheKey(req.fnMeta)
-			lock, ok := ul.locks[key]
-			if ok {
-				lock.wg.Done()
-				delete(ul.locks, key)
-			}
-
-		case EXPIRE:
-			for k, v := range ul.locks {
-				if v.isOld() {
-					delete(ul.locks, k)
-					v.wg.Done()
-				}
-			}
-		}
-	}
-}
-
-func (locks *svcAddrUpdateLocks) RunOnce(fnMeta *metav1.ObjectMeta,
-	callbackFunc func(bool) (*url.URL, bool, error)) (*url.URL, bool, error) {
-
-	ch := make(chan *svcAddrUpdateResponse)
-	locks.requestChan <- &svcAddrUpdateRequest{
-		requestType:  GET,
-		responseChan: ch,
-		fnMeta:       fnMeta,
-	}
-	resp := <-ch
-
-	if resp.firstGoroutine {
-		// release update lock so that other goroutines can take over the responsibility
-		// of updating the service map if failed.
-		defer func() {
-			go locks.Done(fnMeta)
-		}()
-	} else {
-		// wait for the first goroutine to update the service entry
-		err := resp.lock.Wait()
-		if err != nil {
-			return nil, false, err
-		}
-	}
-
-	return callbackFunc(resp.firstGoroutine)
-}
-
-func (locks *svcAddrUpdateLocks) Done(fnMeta *metav1.ObjectMeta) {
-	locks.requestChan <- &svcAddrUpdateRequest{
-		requestType: DELETE,
-		fnMeta:      fnMeta,
-	}
-}
-
-// expiryService periodically expires time-out locks.
-// Normally, we don't need to do this just in case any of goroutine didn't release lock.
-func (locks *svcAddrUpdateLocks) expiryService() {
-	for {
-		time.Sleep(time.Minute)
-		locks.requestChan <- &svcAddrUpdateRequest{
-			requestType: EXPIRE,
-		}
-	}
-}

throttler/throttler.go

@@ -0,0 +1,227 @@
+/*
+Copyright 2018 The Fission 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.
+*/
+
+package throttler
+
+import (
+	"sync"
+	"time"
+
+	"github.com/pkg/errors"
+)
+
+type throttlerOperationType int
+
+const (
+	GET throttlerOperationType = iota
+	DELETE
+	EXPIRE
+)
+
+type (
+	// actionLock is a lock that indicates whether a resource with
+	// certain key is being updated or not.
+	actionLock struct {
+		wg         *sync.WaitGroup
+		ctimestamp time.Time // creation time of lock
+		timeExpiry time.Duration
+	}
+
+	// Throttler is a simple throttling mechanism that provides the abil-
+	// ity to limit the total amount of requests to do the same thing at
+	// the same time.
+	//
+	// In router, for example, multiple goroutines may try to get the la-
+	// test service URL from executor when there is no service URL entry
+	// in the cache and caused executor overloaded because of receiving
+	// massive requests. With throttler, we can easily limit there is at
+	// most one requests being sent to executor.
+	Throttler struct {
+		requestChan    chan *request
+		locks          map[string]*actionLock
+		lockTimeExpiry time.Duration
+	}
+
+	request struct {
+		requestType  throttlerOperationType
+		responseChan chan *response
+		resourceKey  string
+	}
+
+	response struct {
+		lock           *actionLock
+		firstGoroutine bool // denote this goroutine is the first goroutine
+	}
+)
+
+func (l *actionLock) isOld() bool {
+	return time.Since(l.ctimestamp) > l.timeExpiry
+}
+
+func (l *actionLock) wait() error {
+	ch := make(chan struct{})
+
+	go func(wg *sync.WaitGroup, ch chan struct{}) {
+		wg.Wait()
+		close(ch)
+	}(l.wg, ch)
+
+	select {
+	case <-ch:
+		return nil
+	case <-time.After(l.timeExpiry):
+		return errors.New("Error waiting for actionLock to be released: Exceeded timeout")
+	}
+}
+
+// MakeThrottler returns a throttler that able to limit total amounts of goroutines from
+// doing the same thing at the same time.
+func MakeThrottler(timeExpiry time.Duration) *Throttler {
+	tr := &Throttler{
+		requestChan:    make(chan *request),
+		locks:          make(map[string]*actionLock),
+		lockTimeExpiry: timeExpiry,
+	}
+	go tr.service()
+	go tr.expiryService()
+	return tr
+}
+
+func (tr *Throttler) service() {
+	for {
+		req := <-tr.requestChan
+
+		switch req.requestType {
+		case GET:
+			lock, ok := tr.locks[req.resourceKey]
+			if ok && !lock.isOld() {
+				req.responseChan <- &response{
+					lock: lock, firstGoroutine: false,
+				}
+				continue
+			} else if ok && lock.isOld() {
+				// in case that one goroutine occupy the update lock for long time
+				lock.wg.Done()
+			}
+
+			lock = &actionLock{
+				wg:         &sync.WaitGroup{},
+				ctimestamp: time.Now(),
+				timeExpiry: tr.lockTimeExpiry,
+			}
+
+			lock.wg.Add(1)
+
+			tr.locks[req.resourceKey] = lock
+
+			req.responseChan <- &response{
+				lock: lock, firstGoroutine: true,
+			}
+
+		case DELETE:
+			lock, ok := tr.locks[req.resourceKey]
+			if ok {
+				lock.wg.Done()
+				delete(tr.locks, req.resourceKey)
+			}
+
+		case EXPIRE:
+			for k, v := range tr.locks {
+				if v.isOld() {
+					delete(tr.locks, k)
+					v.wg.Done()
+				}
+			}
+		}
+	}
+}
+
+// RunOnce accepts two arguments:
+// 1. function metadata:
+//   It's used to check whether the actionLock of a function exists.
+//
+//   If not exists, an actionLock will be inserted into the map with
+//   passing key. Then, throttler pass true to callbackFunc to indi-
+//   cate this goroutine is the first goroutine and is responsible for
+//   calling backend service, like updating service URL entry in router.
+//
+//   If exists, the goroutines have to wait until the first goroutine
+//   finishes the update process.
+//
+// 2. callback function:
+//   The callback function is a function accepts one bool argument which
+//   indicates whether a goroutine is responsible for getting/updating a
+//   resource from backend service or waiting for the first goroutine to
+//   be finished.
+//
+//   Example callback function:
+//
+//   func(firstToTheLock bool) (interface{}, error) {
+// 	   var u *url.URL
+//
+//	   if firstToTheLock { // first to the service url
+//		   // Call to backend services then do something else.
+//		   // For example, get service url from executor then update router cache.
+//	   } else {
+//		   // Do something here.
+//		   // For example, get service url from cache
+//	   }
+//
+//	   return AnythingYouWant{}, error
+//   }
+
+func (tr *Throttler) RunOnce(resourceKey string,
+	callbackFunc func(bool) (interface{}, error)) (interface{}, error) {
+
+	ch := make(chan *response)
+	tr.requestChan <- &request{
+		requestType:  GET,
+		responseChan: ch,
+		resourceKey:  resourceKey,
+	}
+	resp := <-ch
+
+	// if we are not the first one, wait for the first goroutine to finish its task
+	if !resp.firstGoroutine {
+		err := resp.lock.wait()
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	// release actionLock so that other goroutines can take over the responsibility if failed.
+	defer func() {
+		go func() {
+			tr.requestChan <- &request{
+				requestType: DELETE,
+				resourceKey: resourceKey,
+			}
+		}()
+	}()
+
+	return callbackFunc(resp.firstGoroutine)
+}
+
+// expiryService periodically expires time-out locks.
+// Normally, we don't need to do this just in case any of goroutine didn't release lock.
+func (tr *Throttler) expiryService() {
+	for {
+		time.Sleep(time.Minute)
+		tr.requestChan <- &request{
+			requestType: EXPIRE,
+		}
+	}
+}