diff --git a/router/functionHandler.go b/router/functionHandler.go index 65f7897c..c53f4759 100644 --- a/router/functionHandler.go +++ b/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 diff --git a/router/httpTriggers.go b/router/httpTriggers.go index 60001b2e..12be67ea 100644 --- a/router/httpTriggers.go +++ b/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) } diff --git a/router/router.go b/router/router.go index 4876ac32..7b00af13 100644 --- a/router/router.go +++ b/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) diff --git a/router/router_test.go b/router/router_test.go index 6b179e5e..8e34839d 100644 --- a/router/router_test.go +++ b/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{ diff --git a/router/svcAddrUpdateLock.go b/router/svcAddrUpdateLock.go deleted file mode 100644 index 3227ded5..00000000 --- a/router/svcAddrUpdateLock.go +++ /dev/null @@ -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, - } - } -} diff --git a/throttler/throttler.go b/throttler/throttler.go new file mode 100644 index 00000000..bf0a579a --- /dev/null +++ b/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, + } + } +}