diff --git a/pkg/router/functionHandler.go b/pkg/router/functionHandler.go index 0c26f71d..67b50a1a 100644 --- a/pkg/router/functionHandler.go +++ b/pkg/router/functionHandler.go @@ -36,6 +36,7 @@ import ( k8stypes "k8s.io/apimachinery/pkg/types" fv1 "github.com/fission/fission/pkg/apis/core/v1" + "github.com/fission/fission/pkg/crd" ferror "github.com/fission/fission/pkg/error" "github.com/fission/fission/pkg/error/network" executorClient "github.com/fission/fission/pkg/executor/client" @@ -103,6 +104,11 @@ type ( fakeCloseReadCloser struct { io.ReadCloser } + + svcEntryRecord struct { + svcURL *url.URL + cacheHit bool + } ) func init() { @@ -188,7 +194,7 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re // trying to get new service url from cache/executor. if retryCounter == 0 { // get function service url from cache or executor - roundTripper.serviceURL, err = roundTripper.funcHandler.getServiceEntryFromExecutor() + roundTripper.serviceURL, roundTripper.urlFromCache, err = roundTripper.funcHandler.getServiceEntry() if err != nil { // We might want a specific error code or header for fission failures as opposed to // user function bugs. @@ -209,7 +215,12 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re }, nil } return nil, ferror.MakeError(http.StatusInternalServerError, err.Error()) - + } + if roundTripper.serviceURL == nil { + roundTripper.logger.Warn("serviceURL is empty for function, retrying", zap.Duration("executingTimeout", executingTimeout)) + time.Sleep(executingTimeout) + executingTimeout = executingTimeout * time.Duration(roundTripper.funcHandler.tsRoundTripperParams.timeoutExponent) + continue } if roundTripper.funcHandler.function.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypePoolmgr { defer func(fn *fv1.Function, serviceURL *url.URL) { @@ -312,6 +323,9 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re roundTripper.logger.Debug(fmt.Sprintf( "retry counter exceeded pre-defined threshold of %v", roundTripper.funcHandler.tsRoundTripperParams.svcAddrRetryCount)) + if roundTripper.urlFromCache { + roundTripper.funcHandler.removeServiceEntryFromCache() + } retryCounter = 0 } @@ -533,15 +547,46 @@ func (fh functionHandler) unTapService(fn *fv1.Function, serviceUrl *url.URL) er return nil } -// getServiceEntryFromExecutor returns service url entry returns from executor -func (fh functionHandler) getServiceEntryFromExecutor() (*url.URL, error) { +// getServiceEntryFromCache returns service url entry returns from cache +func (fh functionHandler) getServiceEntryFromCache() (serviceUrl *url.URL, err error) { + // cache lookup to get serviceUrl + serviceUrl, err = fh.fmap.lookup(&fh.function.ObjectMeta) + if err != nil { + var errMsg string + + e, ok := err.(ferror.Error) + if !ok { + errMsg = fmt.Sprintf("Unknown error when looking up service entry: %v", err) + } else { + // Ignore ErrorNotFound error here, it's an expected error, + // roundTripper will try to get service url later. + if e.Code == ferror.ErrorNotFound { + return nil, nil + } + errMsg = fmt.Sprintf("Error getting function %v;s service entry from cache: %v", fh.function.ObjectMeta.Name, err) + } + return nil, ferror.MakeError(http.StatusInternalServerError, errMsg) + } + return serviceUrl, nil +} + +// addServiceEntryToCache add service url entry to cache +func (fn functionHandler) addServiceEntryToCache(serviceUrl *url.URL) { + fn.fmap.assign(&fn.function.ObjectMeta, serviceUrl) +} + +// removeServiceEntryFromCache removes service url entry from cache +func (fn functionHandler) removeServiceEntryFromCache() { + fn.fmap.remove(&fn.function.ObjectMeta) +} + +func (fh functionHandler) getServiceEntryFromExecutor() (serviceUrl *url.URL, err error) { // send a request to executor to specialize a new pod fh.logger.Debug("function timeout specified", zap.Int("timeout", fh.function.Spec.FunctionTimeout)) timeout := 30 * time.Second if fh.function.Spec.FunctionTimeout > 0 { timeout = time.Second * time.Duration(fh.function.Spec.FunctionTimeout) } - ctx, cancel := context.WithTimeout(context.Background(), timeout) defer cancel() service, err := fh.executor.GetServiceForFunction(ctx, fh.function) @@ -554,17 +599,59 @@ func (fh functionHandler) getServiceEntryFromExecutor() (*url.URL, error) { zap.Int("status_code", statusCode)) return nil, err } - // parse the address into url - serviceURL, err := url.Parse(fmt.Sprintf("http://%v", service)) + svcURL, err := url.Parse(fmt.Sprintf("http://%v", service)) if err != nil { fh.logger.Error("error parsing service url", zap.Error(err), - zap.String("service_url", serviceURL.String())) + zap.String("service_url", svcURL.String())) return nil, err } + return svcURL, err +} - return serviceURL, nil +// getServiceEntryFromExecutor returns service url entry returns from executor +func (fh functionHandler) getServiceEntry() (svcURL *url.URL, cacheHit bool, err error) { + if fh.function.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypePoolmgr { + svcURL, err = fh.getServiceEntryFromExecutor() + return svcURL, false, err + } + // Check if service URL present in cache + svcURL, err = fh.getServiceEntryFromCache() + if err == nil && svcURL != nil { + return svcURL, true, nil + } else if err != nil { + return nil, false, err + } + + fnMeta := &fh.function.ObjectMeta + recordObj, err := fh.svcAddrUpdateThrottler.RunOnce( + crd.CacheKey(fnMeta), + func(firstToTheLock bool) (interface{}, error) { + if !firstToTheLock { + svcURL, err := fh.getServiceEntryFromCache() + if err != nil { + return nil, err + } + return svcEntryRecord{svcURL: svcURL, cacheHit: true}, err + } + svcURL, err = fh.getServiceEntryFromExecutor() + if err != nil { + return nil, err + } + fh.addServiceEntryToCache(svcURL) + return svcEntryRecord{ + svcURL: svcURL, + cacheHit: false, + }, nil + }, + ) + + record, ok := recordObj.(svcEntryRecord) + if !ok { + return nil, false, fmt.Errorf("received unknown service record type") + } + return record.svcURL, record.cacheHit, err } // getProxyErrorHandler returns a reverse proxy error handler diff --git a/pkg/router/functionServiceMap.go b/pkg/router/functionServiceMap.go index 0fed6a70..d73ca621 100644 --- a/pkg/router/functionServiceMap.go +++ b/pkg/router/functionServiceMap.go @@ -77,3 +77,8 @@ func (fmap *functionServiceMap) assign(f *metav1.ObjectMeta, serviceURL *url.URL // ignore error } } + +func (fmap *functionServiceMap) remove(f *metav1.ObjectMeta) error { + mk := keyFromMetadata(f) + return fmap.cache.Delete(*mk) +}