diff --git a/charts/fission-all/templates/deployment.yaml b/charts/fission-all/templates/deployment.yaml index 4e40d958..b890ae18 100644 --- a/charts/fission-all/templates/deployment.yaml +++ b/charts/fission-all/templates/deployment.yaml @@ -200,6 +200,10 @@ spec: value: {{ .Values.routerRoundTripKeepAliveTime | default "30s" | quote }} - name: ROUTER_ROUND_TRIP_MAX_RETRIES value: {{ .Values.routerRoundTripMaxRetries | default 10 | quote }} + - name: ROUTER_ROUND_TRIP_SVC_ADDRESS_MAX_RETRIES + value: {{ .Values.routerRoundTripSvcAddressMaxRetries | default 5 | quote }} + - name: ROUTER_ROUND_TRIP_SVC_ADDRESS_UPDATE_TIMEOUT + value: {{ .Values.routerRoundTripSvcAddressUpdateTimeout | default 30 | quote }} - name: DEBUG_ENV value: {{ .Values.debugEnv | quote }} readinessProbe: diff --git a/charts/fission-core/templates/deployment.yaml b/charts/fission-core/templates/deployment.yaml index c562dee8..2f90c74c 100644 --- a/charts/fission-core/templates/deployment.yaml +++ b/charts/fission-core/templates/deployment.yaml @@ -197,6 +197,10 @@ spec: value: {{ .Values.routerRoundTripKeepAliveTime | default "30s" | quote }} - name: ROUTER_ROUND_TRIP_MAX_RETRIES value: {{ .Values.routerRoundTripMaxRetries | default 10 | quote }} + - name: ROUTER_ROUND_TRIP_SVC_ADDRESS_MAX_RETRIES + value: {{ .Values.routerRoundTripSvcAddressMaxRetries | default 5 | quote }} + - name: ROUTER_ROUND_TRIP_SVC_ADDRESS_UPDATE_TIMEOUT + value: {{ .Values.routerRoundTripSvcAddressUpdateTimeout | default 30 | quote }} - name: DEBUG_ENV value: {{ .Values.debugEnv | quote }} readinessProbe: diff --git a/executor/executor.go b/executor/executor.go index 60b2fb82..d4345705 100644 --- a/executor/executor.go +++ b/executor/executor.go @@ -142,13 +142,6 @@ func (executor *Executor) getFunctionExecutorType(meta *metav1.ObjectMeta) (fiss func (executor *Executor) createServiceForFunction(meta *metav1.ObjectMeta) (*fscache.FuncSvc, error) { log.Printf("[%v] No cached function service found, creating one", meta.Name) - // from Func -> get Env - log.Printf("[%v] getting environment for function", meta.Name) - env, err := executor.getFunctionEnv(meta) - if err != nil { - return nil, err - } - executorType, err := executor.getFunctionExecutorType(meta) if err != nil { return nil, err @@ -161,6 +154,13 @@ func (executor *Executor) createServiceForFunction(meta *metav1.ObjectMeta) (*fs case fission.ExecutorTypeNewdeploy: fsvc, fsvcErr = executor.ndm.GetFuncSvc(meta) default: + // from Func -> get Env + log.Printf("[%v] getting environment for function", meta.Name) + env, err := executor.getFunctionEnv(meta) + if err != nil { + return nil, err + } + pool, err := executor.gpm.GetPool(env) if err != nil { return nil, err diff --git a/router/functionHandler.go b/router/functionHandler.go index 71e5670a..d456f9f3 100644 --- a/router/functionHandler.go +++ b/router/functionHandler.go @@ -29,8 +29,8 @@ import ( "time" "github.com/gorilla/mux" + "github.com/pkg/errors" "github.com/satori/go.uuid" - log "github.com/sirupsen/logrus" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -49,7 +49,19 @@ type tsRoundTripperParams struct { timeout time.Duration timeoutExponent int keepAlive time.Duration - maxRetries 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 + + // 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 } type functionHandler struct { @@ -64,6 +76,7 @@ type functionHandler struct { tsRoundTripperParams *tsRoundTripperParams recorderName string isDebugEnv bool + svcAddrUpdateLocks *svcAddrUpdateLocks } // A layer on top of http.DefaultTransport, with retries. @@ -103,8 +116,9 @@ func init() { // Earlier, GetServiceForFunction was called inside handler function and fission explicitly set http status code to 500 // if it returned an error. func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *http.Response, err error) { - var needExecutor, serviceUrlFromExecutor bool + var serviceUrlFromCache bool var serviceUrl *url.URL + var retryCounter int // Set forwarded host header if not exists addForwardedHostHeader(req) @@ -130,11 +144,13 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt } } + fnMeta := roundTripper.funcHandler.function + // Metrics stuff startTime := time.Now() funcMetricLabels := &functionLabels{ - namespace: roundTripper.funcHandler.function.Namespace, - name: roundTripper.funcHandler.function.Name, + namespace: fnMeta.Namespace, + name: fnMeta.Name, } httpMetricLabels := &httpLabels{ method: req.Method, @@ -146,22 +162,150 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt // set the timeout for transport context transport := http.DefaultTransport.(*http.Transport) + // Disables caching, Please refer to issue and specifically comment: https://github.com/fission/fission/issues/723#issuecomment-398781995 transport.DisableKeepAlives = true - // cache lookup to get serviceUrl - serviceUrl, err = roundTripper.funcHandler.fmap.lookup(roundTripper.funcHandler.function) - if err != nil || serviceUrl == nil { - // cache miss or nil entry in cache - log.Printf("Setting needExecutor to true for function : %s", roundTripper.funcHandler.function.Name) - needExecutor = true - } - executingTimeout := roundTripper.funcHandler.tsRoundTripperParams.timeout for i := 0; i < roundTripper.funcHandler.tsRoundTripperParams.maxRetries-1; i++ { - if needExecutor { - log.Printf("Calling getServiceForFunction for function: %s", roundTripper.funcHandler.function.Name) + + // cache lookup to get serviceUrl + serviceUrl, err = roundTripper.funcHandler.fmap.lookup(fnMeta) + if err != nil { + e, ok := err.(fission.Error) + if (ok && e.Code != fission.ErrorNotFound) || !ok { + if ok { + err = errors.Wrap(err, fmt.Sprintf("Error getting function %v;s service entry from cache", fnMeta.Name)) + } else { + err = errors.Wrap(err, "Unknown error when looking up service entry") + } + return nil, err + } + } else { + serviceUrlFromCache = true + } + + if serviceUrl != nil { + // modify the request to reflect the service url + // this service url may have come from the cache lookup or from executor response + req.URL.Scheme = serviceUrl.Scheme + req.URL.Host = serviceUrl.Host + + // To keep the function run container simple, it + // doesn't do any routing. In the future if we have + // multiple functions per container, we could use the + // function metadata here. + // leave the query string intact (req.URL.RawQuery) + req.URL.Path = "/" + + // Overwrite request host with internal host, + // or request will be blocked in some situations + // (e.g. istio-proxy) + req.Host = serviceUrl.Host + + // over-riding default settings. + transport.DialContext = (&net.Dialer{ + Timeout: executingTimeout, + KeepAlive: roundTripper.funcHandler.tsRoundTripperParams.keepAlive, + }).DialContext + + overhead := time.Since(startTime) + + // tapService before invoking roundTrip for the serviceUrl + if serviceUrlFromCache { + go roundTripper.funcHandler.tapService(serviceUrl) + } + + // forward the request to the function service + resp, err = transport.RoundTrip(req) + if err == nil { + // Track metrics + httpMetricLabels.code = resp.StatusCode + funcMetricLabels.cached = serviceUrlFromCache + + functionCallCompleted(funcMetricLabels, httpMetricLabels, + overhead, time.Since(startTime), resp.ContentLength) + + if len(roundTripper.funcHandler.recorderName) > 0 { + if roundTripper.funcHandler.httpTrigger != nil { + trigger := roundTripper.funcHandler.httpTrigger.Metadata.Name + redis.Record( + trigger, + roundTripper.funcHandler.recorderName, + req.Header.Get("X-Fission-ReqUID"), req, originalUrl, postedBody, resp, fnMeta.Namespace, + time.Now().UnixNano(), + ) + } else { + log.Println("No http trigger attached for recorder: %v", roundTripper.funcHandler.recorderName) + } + } + + // return response back to user + return resp, nil + } + + // if transport.RoundTrip returns a non-network dial error, then relay it back to user + if !fission.IsNetworkDialError(err) { + err = errors.Wrapf(err, "Error sending request to function %v", fnMeta.Name) + return resp, err + } + + // dial timeout or dial network errors goes here + + // The reason for request failure may vary from case to case. + // After some investigation, found most of the failure are due to + // network timeout or target function is under heavy workload. In + // such cases, if router keeps trying to get new function service + // will increase executor burden and cause 502 error. + // + // The "retryCounter" was introduced to solve this problem by retrying + // requests for "limited threshold". Once a request's retryCounter higher + // than the predefined threshold, reset retryCounter and remove service + // cache, then retry to get new svc record from executor again. + if retryCounter < roundTripper.funcHandler.tsRoundTripperParams.svcAddrRetryCount { + retryCounter++ + + executingTimeout = executingTimeout * time.Duration(roundTripper.funcHandler.tsRoundTripperParams.timeoutExponent) + + log.Printf("request to %s errored out. backing off for %v before retrying", + req.URL.Host, executingTimeout) + + time.Sleep(executingTimeout) + + if serviceUrlFromCache { + continue + } + } else { + // if transport.RoundTrip returns a network dial error and serviceUrl was from cache, + // it means, the entry in router cache is stale, so invalidate it. + log.Printf("request to %s errored out. removing function : %s from router's cache "+ + "and requesting a new service for function", + req.URL.Host, fnMeta.Name) + roundTripper.funcHandler.fmap.remove(fnMeta) + retryCounter = 0 + } + } + + // break directly if we still fail at the last round + if i >= roundTripper.funcHandler.tsRoundTripperParams.maxRetries-1 { + break + } + + // cache miss or nil entry in cache + lock, ableToUpdateCache := roundTripper.funcHandler.grabUpdateEntryLock(fnMeta) + + if !ableToUpdateCache { + // This goroutine wait for update of service map to finish. + err = lock.Wait() + if err != nil { + log.Println(errors.Wrap(err, + fmt.Sprintf("Error updating service address entry for function %v_%v", fnMeta.Name, fnMeta.Namespace))) + } + } else { + // This goroutine is the first one to grab update lock + + log.Printf("Calling getServiceForFunction for function: %s", fnMeta.Name) // send a request to executor to specialize a new pod service, err := roundTripper.funcHandler.executor.GetServiceForFunction( @@ -169,7 +313,7 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt if err != nil { statusCode, errMsg := fission.GetHTTPError(err) - log.Printf("Err from GetServiceForFunction : %v : %v", statusCode, errMsg) + log.Printf("Err from GetServiceForFunction for function (%v): %v : %v", roundTripper.funcHandler.function, statusCode, errMsg) // We might want a specific error code or header for fission failures as opposed to // user function bugs. @@ -186,118 +330,37 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt }, nil } + roundTripper.funcHandler.releaseUpdateEntryLock(fnMeta) return nil, err } // parse the address into url serviceUrl, err = url.Parse(fmt.Sprintf("http://%v", service)) if err != nil { + log.Printf("Error parsing service url (%v): %v", serviceUrl, err) + roundTripper.funcHandler.releaseUpdateEntryLock(fnMeta) return nil, err } // add the address in router's cache - log.Printf("assigning serviceUrl : %s for function : %s", serviceUrl, roundTripper.funcHandler.function.Name) + log.Printf("Assigning serviceUrl : %s for function : %s", serviceUrl, roundTripper.funcHandler.function.Name) roundTripper.funcHandler.fmap.assign(roundTripper.funcHandler.function, serviceUrl) // flag denotes that service was not obtained from cache, instead, created just now by executor - serviceUrlFromExecutor = true - } + serviceUrlFromCache = false - // modify the request to reflect the service url - // this service url may have come from the cache lookup or from executor response - req.URL.Scheme = serviceUrl.Scheme - req.URL.Host = serviceUrl.Host - - // To keep the function run container simple, it - // doesn't do any routing. In the future if we have - // multiple functions per container, we could use the - // function metadata here. - // leave the query string intact (req.URL.RawQuery) - req.URL.Path = "/" - - // Overwrite request host with internal host, - // or request will be blocked in some situations - // (e.g. istio-proxy) - req.Host = serviceUrl.Host - - // over-riding default settings. - transport.DialContext = (&net.Dialer{ - Timeout: executingTimeout, - KeepAlive: roundTripper.funcHandler.tsRoundTripperParams.keepAlive, - }).DialContext - - overhead := time.Since(startTime) - - // tapService before invoking roundTrip for the serviceUrl - if !serviceUrlFromExecutor { - go roundTripper.funcHandler.tapService(serviceUrl) - } - - // forward the request to the function service - resp, err = transport.RoundTrip(req) - if err == nil { - // Track metrics - httpMetricLabels.code = resp.StatusCode - funcMetricLabels.cached = !serviceUrlFromExecutor - - functionCallCompleted(funcMetricLabels, httpMetricLabels, - overhead, time.Since(startTime), resp.ContentLength) - - // if transport.RoundTrip succeeds and it was a cached entry, then tapService - if !serviceUrlFromExecutor { - go roundTripper.funcHandler.tapService(serviceUrl) - } - - trigger := "" - if roundTripper.funcHandler.httpTrigger != nil { - trigger = roundTripper.funcHandler.httpTrigger.Metadata.Name - } else { - log.Println("No trigger attached.") // Wording? - } - - if len(roundTripper.funcHandler.recorderName) > 0 { - redis.Record( - trigger, - roundTripper.funcHandler.recorderName, - req.Header.Get("X-Fission-ReqUID"), req, originalUrl, postedBody, resp, roundTripper.funcHandler.function.Namespace, - time.Now().UnixNano(), - ) - } - - // return response back to user - return resp, nil - } - - // if transport.RoundTrip returns a non-network dial error, then relay it back to user - if !fission.IsNetworkDialError(err) { - return resp, err - } - - // means its a newly created service and it returned a network dial error. - // just retry after backing off for timeout period. - if serviceUrlFromExecutor { - log.Printf("request to %s errored out. backing off for %v before retrying", - req.URL.Host, executingTimeout) - - executingTimeout = executingTimeout * time.Duration(roundTripper.funcHandler.tsRoundTripperParams.timeoutExponent) - time.Sleep(executingTimeout) - - needExecutor = false - continue - } else { - // if transport.RoundTrip returns a network dial error and serviceUrl was from cache, - // it means, the entry in router cache is stale, so invalidate it. - // also set needExecutor to true so a new service can be requested for function. - log.Printf("request to %s errored out. removing function : %s from router's cache "+ - "and requesting a new service for function", - req.URL.Host, roundTripper.funcHandler.function.Name) - roundTripper.funcHandler.fmap.remove(roundTripper.funcHandler.function) - needExecutor = true + roundTripper.funcHandler.releaseUpdateEntryLock(fnMeta) } } // finally, one more retry with the default timeout - return http.DefaultTransport.RoundTrip(req) + resp, err = http.DefaultTransport.RoundTrip(req) + if err != nil { + log.Printf("Error getting response from function %v: %v", + fnMeta.Name, err) + } + + return resp, err } func (fh *functionHandler) tapService(serviceUrl *url.URL) { @@ -429,3 +492,15 @@ func addForwardedHostHeader(req *http.Request) { req.Header.Set(FORWARDED, host) req.Header.Set(X_FORWARDED_HOST, req.Host) } + +// grabUpdateEntryLock helps goroutine to grab update lock for updating function service cache. +// If the update lock exists, return old svcAddrUpdateLock. +func (fh *functionHandler) grabUpdateEntryLock(fnMeta *metav1.ObjectMeta) (lock *svcAddrUpdateLock, ableToUpdateCache bool) { + return fh.svcAddrUpdateLocks.Get(fnMeta) +} + +// releaseUpdateEntryLock release update lock so that other goroutines can take over the responsibility +// of updating the service map. +func (fh *functionHandler) releaseUpdateEntryLock(fnMeta *metav1.ObjectMeta) { + fh.svcAddrUpdateLocks.Delete(fnMeta) +} diff --git a/router/httpTriggers.go b/router/httpTriggers.go index 8ce2e7b2..60001b2e 100644 --- a/router/httpTriggers.go +++ b/router/httpTriggers.go @@ -53,10 +53,12 @@ type HTTPTriggerSet struct { updateRouterRequestChannel chan struct{} tsRoundTripperParams *tsRoundTripperParams isDebugEnv bool + svcAddrUpdateLocks *svcAddrUpdateLocks } func makeHTTPTriggerSet(fmap *functionServiceMap, frmap *functionRecorderMap, trmap *triggerRecorderMap, fissionClient *crd.FissionClient, - kubeClient *kubernetes.Clientset, executor *executorClient.Client, crdClient *rest.RESTClient, params *tsRoundTripperParams, isDebugEnv bool) (*HTTPTriggerSet, k8sCache.Store, k8sCache.Store) { + kubeClient *kubernetes.Clientset, executor *executorClient.Client, crdClient *rest.RESTClient, params *tsRoundTripperParams, isDebugEnv bool, locks *svcAddrUpdateLocks) (*HTTPTriggerSet, k8sCache.Store, k8sCache.Store) { + httpTriggerSet := &HTTPTriggerSet{ functionServiceMap: fmap, triggers: []crd.HTTPTrigger{}, @@ -67,6 +69,7 @@ func makeHTTPTriggerSet(fmap *functionServiceMap, frmap *functionRecorderMap, tr updateRouterRequestChannel: make(chan struct{}), tsRoundTripperParams: params, isDebugEnv: isDebugEnv, + svcAddrUpdateLocks: locks, } var tStore, fnStore, rStore k8sCache.Store var tController, fnController k8sCache.Controller @@ -153,8 +156,16 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router { tsRoundTripperParams: ts.tsRoundTripperParams, recorderName: recorderName, isDebugEnv: ts.isDebugEnv, + svcAddrUpdateLocks: ts.svcAddrUpdateLocks, } + // The functionHandler for HTTP trigger with fn reference type "FunctionReferenceTypeFunctionName", + // it's function metadata is set here. + + // The functionHandler For HTTP trigger with fn reference type "FunctionReferenceTypeFunctionWeights", + // it's function metadata is decided dynamically before proxying the request in order to support canary + // deployment. For more details, please check "handler" function of functionHandler. + if rr.resolveResultType == resolveResultSingleFunction { for _, metadata := range fh.functionMetadataMap { fh.function = metadata @@ -201,6 +212,7 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router { tsRoundTripperParams: ts.tsRoundTripperParams, recorderName: recorderName, isDebugEnv: ts.isDebugEnv, + svcAddrUpdateLocks: ts.svcAddrUpdateLocks, } muxRouter.HandleFunc(fission.UrlForFunction(function.Metadata.Name, function.Metadata.Namespace), fh.handler) } diff --git a/router/router.go b/router/router.go index 7e7cf670..4876ac32 100644 --- a/router/router.go +++ b/router/router.go @@ -129,12 +129,28 @@ func Start(port int, executorUrl string) { log.Fatalf("Failed to parse DEBUG_ENV: %v", err) } + // svcAddrRetryCount is the max times for RetryingRoundTripper to retry with a specific service address + svcAddrRetryCount, err := strconv.Atoi(os.Getenv("ROUTER_ROUND_TRIP_SVC_ADDRESS_MAX_RETRIES")) + if err != nil { + svcAddrRetryCount = 5 + log.Printf("Failed to parse svc address retry conunt, set it to default value(5): %v", err) + } + + // svcAddrUpdateTimeout is the timeout setting for a goroutine to wait for the update of a service entry. + // If the update process cannot be done within the timeout window, consider it failed. + svcAddrUpdateTimeout, err := time.ParseDuration(os.Getenv("ROUTER_ROUND_TRIP_SVC_ADDRESS_UPDATE_TIMEOUT")) + if err != nil { + svcAddrUpdateTimeout = 30 * time.Second + log.Printf("Failed to parse svc address update timeout, set it to default value(30): %v", err) + } + triggers, _, fnStore := makeHTTPTriggerSet(fmap, frmap, trmap, fissionClient, kubeClient, executor, restClient, &tsRoundTripperParams{ - timeout: timeout, - timeoutExponent: timeoutExponent, - keepAlive: keepAlive, - maxRetries: maxRetries, - }, isDebugEnv) + timeout: timeout, + timeoutExponent: timeoutExponent, + keepAlive: keepAlive, + maxRetries: maxRetries, + svcAddrRetryCount: svcAddrRetryCount, + }, isDebugEnv, MakeUpdateLocks(svcAddrUpdateTimeout)) resolver := makeFunctionReferenceResolver(fnStore) diff --git a/router/router_test.go b/router/router_test.go index d8f3f202..6b179e5e 100644 --- a/router/router_test.go +++ b/router/router_test.go @@ -57,7 +57,7 @@ func TestRouter(t *testing.T) { timeoutExponent: 2, keepAlive: 30 * time.Second, maxRetries: 10, - }, false) + }, false, MakeUpdateLocks(30*time.Second)) triggerUrl := "/foo" triggers.triggers = append(triggers.triggers, crd.HTTPTrigger{ diff --git a/router/svcAddrUpdateLock.go b/router/svcAddrUpdateLock.go new file mode 100644 index 00000000..af592202 --- /dev/null +++ b/router/svcAddrUpdateLock.go @@ -0,0 +1,174 @@ +/* +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 ( + "errors" + "sync" + "time" + + 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 + loaded bool // denote the lock for same function already exists + } +) + +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, loaded: 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, loaded: 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) Get(fnMeta *metav1.ObjectMeta) (lock *svcAddrUpdateLock, ableToUpdate bool) { + ch := make(chan *svcAddrUpdateResponse) + locks.requestChan <- &svcAddrUpdateRequest{ + requestType: GET, + responseChan: ch, + fnMeta: fnMeta, + } + resp := <-ch + return resp.lock, resp.loaded +} + +func (locks *svcAddrUpdateLocks) Delete(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, + } + } +}