diff --git a/go.sum b/go.sum index 173221f0..f9770bb1 100644 --- a/go.sum +++ b/go.sum @@ -34,7 +34,9 @@ github.com/Shopify/sarama v1.23.1/go.mod h1:XLH1GYJnLVE0XCr6KdJGVJRTwY30moWNJ4sE github.com/Shopify/toxiproxy v2.1.4+incompatible h1:TKdv8HiTLgE5wdJuEML90aBgNWsokNbMijUGhmcoBJc= github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI= github.com/airbrake/gobrake v3.6.1+incompatible/go.mod h1:wM4gu3Cn0W0K7GUuVWnlXZU11AGBXMILnrdOU8Kn00o= +github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc h1:cAKDfWh5VpdgMhJosfJnn5/FoN2SRZ4p7fJNX58YPaU= github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= +github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf h1:qet1QNfXsQxTZqLG4oE62mJzwPIB8+Tee4RNCL9ulrY= github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= github.com/apache/thrift v0.12.0 h1:pODnxUFNcjP9UTLZGTdeh+j16A8lJbRvD3rOtrk/7bs= github.com/apache/thrift v0.12.0/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ= @@ -577,6 +579,7 @@ google.golang.org/grpc v1.17.0/go.mod h1:6QZJwpn2B+Zp71q/5VxRsJ6NXXVCE5NRUHRo+f3 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c= google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiqXj38= gopkg.in/airbrake/gobrake.v2 v2.0.9/go.mod h1:/h5ZAUhDkGaJfjzjKLSjv6zCL6O0LLBxU4K+aSYdM/U= +gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc= gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY= diff --git a/pkg/apis/core/v1/types.go b/pkg/apis/core/v1/types.go index 6a71c400..3f319ad1 100644 --- a/pkg/apis/core/v1/types.go +++ b/pkg/apis/core/v1/types.go @@ -349,6 +349,10 @@ type ( // is detected within the idle timeout, the executor will then recycle the // function pod(s) to release resources. IdleTimeout *int `json:"idletimeout,omitempty"` + + // Maximum number of pods to be specialized which will serve requests + // This is optional. If not specified default value will be taken as 5 + Concurrency int `json:"concurrency,omitempty"` } // InvokeStrategy is a set of controls over how the function executes. diff --git a/pkg/error/httperror.go b/pkg/error/httperror.go index ee0292d2..b0c0fe82 100644 --- a/pkg/error/httperror.go +++ b/pkg/error/httperror.go @@ -58,6 +58,8 @@ func MakeErrorFromHTTP(resp *http.Response) error { errCode = ErrorNameExists case http.StatusRequestTimeout: errCode = ErrorRequestTimeout + case http.StatusTooManyRequests: + errCode = ErrorTooManyRequests default: errCode = ErrorInternal } @@ -83,6 +85,8 @@ func (err Error) HTTPStatus() int { code = http.StatusNotFound case ErrorNameExists: code = http.StatusConflict + case ErrorTooManyRequests: + code = http.StatusTooManyRequests default: code = http.StatusInternalServerError } @@ -131,6 +135,7 @@ const ( ErrorChecksumFail ErrorSizeLimitExceeded ErrorRequestTimeout + ErrorTooManyRequests ) // must match order and len of the above const diff --git a/pkg/executor/api.go b/pkg/executor/api.go index 6f7ecb45..0ba9962a 100644 --- a/pkg/executor/api.go +++ b/pkg/executor/api.go @@ -61,6 +61,28 @@ func (executor *Executor) getServiceForFunctionApi(w http.ResponseWriter, r *htt return } + t := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType + et, exists := executor.executorTypes[t] + if !exists { + http.Error(w, fmt.Sprintf("Unknown executor type '%v'", t), http.StatusNotFound) + return + } + + executor.logger.Debug(fmt.Sprintf("active instances: %v", et.GetTotalAvailable(fn))) + conncurrency := fn.Spec.Concurrency + if conncurrency == 0 { + // set to default conncurrency + conncurrency = 5 + executor.logger.Debug(fmt.Sprintf("concurrency specified in function: %v", fn.Spec.Concurrency)) + executor.logger.Debug("setting concurrency to 5") + } + if t == fv1.ExecutorTypePoolmgr && et.GetTotalAvailable(fn) >= conncurrency { + errMsg := fmt.Sprintf("max concurrency reached for %v. All %v instance are active", fn.ObjectMeta.Name, fn.Spec.Concurrency) + executor.logger.Error("error occured", zap.String("error", errMsg)) + http.Error(w, errMsg, http.StatusTooManyRequests) + return + } + serviceName, err := executor.getServiceForFunction(fn) if err != nil { code, msg := ferror.GetHTTPError(err) @@ -85,17 +107,13 @@ func (executor *Executor) getServiceForFunctionApi(w http.ResponseWriter, r *htt // To make it optimal, plan is to add an eager cache invalidator function that watches for pod deletion events and // invalidates the cache entry if the pod address was cached. func (executor *Executor) getServiceForFunction(fn *fv1.Function) (string, error) { + t := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType + et := executor.executorTypes[t] // Check function -> svc cache executor.logger.Debug("checking for cached function service", zap.String("function_name", fn.ObjectMeta.Name), zap.String("function_namespace", fn.ObjectMeta.Namespace)) - t := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType - et, exists := executor.executorTypes[t] - if !exists { - return "", errors.Errorf("Unknown executor type '%v'", t) - } - fsvc, err := et.GetFuncSvcFromCache(fn) if err == nil { if et.IsValid(fsvc) { @@ -181,12 +199,50 @@ func (executor *Executor) healthHandler(w http.ResponseWriter, r *http.Request) w.WriteHeader(http.StatusOK) } +func (executor *Executor) unTapService(w http.ResponseWriter, r *http.Request) { + body, err := ioutil.ReadAll(r.Body) + if err != nil { + http.Error(w, "Failed to read request", http.StatusInternalServerError) + return + } + tapSvcReq := client.TapServiceRequest{} + err = json.Unmarshal(body, &tapSvcReq) + if err != nil { + http.Error(w, "Failed to parse request", http.StatusBadRequest) + return + } + + fn, err := executor.fissionClient.CoreV1().Functions(tapSvcReq.FnMetadata.Namespace).Get(tapSvcReq.FnMetadata.Name, metav1.GetOptions{}) + if err != nil { + if k8serrors.IsNotFound(err) { + http.Error(w, "Failed to find function", http.StatusNotFound) + } else { + http.Error(w, "Failed to get function", http.StatusInternalServerError) + } + return + } + + t := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType + if t != fv1.ExecutorTypePoolmgr { + msg := fmt.Sprintf("Unknown executor type '%v'", t) + http.Error(w, msg, http.StatusBadRequest) + return + } + + et := executor.executorTypes[t] + + et.UnTapService(fn, tapSvcReq.ServiceUrl) + + w.WriteHeader(http.StatusOK) +} + func (executor *Executor) GetHandler() http.Handler { r := mux.NewRouter() r.HandleFunc("/v2/getServiceForFunction", executor.getServiceForFunctionApi).Methods("POST") r.HandleFunc("/v2/tapService", executor.tapService).Methods("POST") // for backward compatibility r.HandleFunc("/v2/tapServices", executor.tapServices).Methods("POST") r.HandleFunc("/healthz", executor.healthHandler).Methods("GET") + r.HandleFunc("/v2/unTapService", executor.unTapService).Methods("POST") return r } diff --git a/pkg/executor/client/client.go b/pkg/executor/client/client.go index 36a3a837..276efc84 100644 --- a/pkg/executor/client/client.go +++ b/pkg/executor/client/client.go @@ -91,6 +91,32 @@ func (c *Client) GetServiceForFunction(ctx context.Context, metadata *metav1.Obj return string(svcName), nil } +func (c *Client) UnTapService(ctx context.Context, fnMeta metav1.ObjectMeta, executorType fv1.ExecutorType, serviceUrl *url.URL) error { + url := c.executorUrl + "/v2/unTapService" + tapSvc := TapServiceRequest{ + FnMetadata: fnMeta, + FnExecutorType: executorType, + ServiceUrl: strings.TrimPrefix(serviceUrl.String(), "http://"), + } + + body, err := json.Marshal(tapSvc) + if err != nil { + return errors.Wrap(err, "could not marshal request body for getting service for function") + } + + resp, err := ctxhttp.Post(ctx, c.httpClient, url, "application/json", bytes.NewReader(body)) + if err != nil { + return errors.Wrap(err, "error posting to getting service for function") + } + defer resp.Body.Close() + + if resp.StatusCode != 200 { + return ferror.MakeErrorFromHTTP(resp) + } + + return nil +} + func (c *Client) service() { ticker := time.NewTicker(time.Second * 5) for { diff --git a/pkg/executor/executor.go b/pkg/executor/executor.go index f9070980..c2d3f388 100644 --- a/pkg/executor/executor.go +++ b/pkg/executor/executor.go @@ -99,6 +99,31 @@ func (executor *Executor) serveCreateFuncServices() { req := <-executor.requestChan fnMetadata := &req.function.ObjectMeta + if req.function.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypePoolmgr { + go func() { + buffer := 10 // add some buffer time for specialization + specializationTimeout := req.function.Spec.InvokeStrategy.ExecutionStrategy.SpecializationTimeout + + // set minimum specialization timeout to avoid illegal input and + // compatibility problem when applying old spec file that doesn't + // have specialization timeout field. + if specializationTimeout < fv1.DefaultSpecializationTimeOut { + specializationTimeout = fv1.DefaultSpecializationTimeOut + } + + fnSpecializationTimeoutContext, cancel := context.WithTimeout(context.Background(), + time.Duration(specializationTimeout+buffer)*time.Second) + defer cancel() + + fsvc, err := executor.createServiceForFunction(fnSpecializationTimeoutContext, req.function) + req.respChan <- &createFuncServiceResponse{ + funcSvc: fsvc, + err: err, + } + }() + continue + } + // Cache miss -- is this first one to request the func? wg, found := executor.fsCreateWg[crd.CacheKey(fnMetadata)] if !found { diff --git a/pkg/executor/executortype/executortype.go b/pkg/executor/executortype/executortype.go index 20c2921b..7a83fd53 100644 --- a/pkg/executor/executortype/executortype.go +++ b/pkg/executor/executortype/executortype.go @@ -18,6 +18,7 @@ package executortype import ( "context" + "go.uber.org/zap" fv1 "github.com/fission/fission/pkg/apis/core/v1" @@ -44,6 +45,9 @@ type ExecutorType interface { // avoid idle pod reaper recycles pods. TapService(serviceUrl string) error + // UnTapService updates the isActive to false + UnTapService(fn *fv1.Function, svcHost string) + // IsValid returns true if a function service is valid. Different executor types // use distinct ways to examine the function service. IsValid(*fscache.FuncSvc) bool @@ -56,4 +60,7 @@ type ExecutorType interface { // CleanupOldExecutorObjects cleans up resources created by old executor instances CleanupOldExecutorObjects() + + // getTotalAvailable returns total active instances of particular function + GetTotalAvailable(*fv1.Function) int } diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go index 8da83b44..177c08d6 100644 --- a/pkg/executor/executortype/newdeploy/newdeploymgr.go +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -153,6 +153,15 @@ func (deploy *NewDeploy) DeleteFuncSvcFromCache(fsvc *fscache.FuncSvc) { deploy.fsCache.DeleteEntry(fsvc) } +func (deploy *NewDeploy) UnTapService(fn *fv1.Function, svcHost string) { + // Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added. +} + +func (deploy *NewDeploy) GetTotalAvailable(fn *fv1.Function) int { + // Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added. + return 0 +} + func (deploy *NewDeploy) TapService(svcHost string) error { err := deploy.fsCache.TouchByAddress(svcHost) if err != nil { diff --git a/pkg/executor/executortype/poolmgr/gp.go b/pkg/executor/executortype/poolmgr/gp.go index b2583c66..4cddcb6b 100644 --- a/pkg/executor/executortype/poolmgr/gp.go +++ b/pkg/executor/executortype/poolmgr/gp.go @@ -703,10 +703,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac Atime: time.Now(), } - _, err = gp.fsCache.Add(*fsvc) - if err != nil { - return nil, err - } + gp.fsCache.AddFunc(*fsvc) gp.fsCache.IncreaseColdStarts(fn.ObjectMeta.Name, string(fn.ObjectMeta.UID)) diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go index 57cf6e6c..718ff9e4 100644 --- a/pkg/executor/executortype/poolmgr/gpm.go +++ b/pkg/executor/executortype/poolmgr/gpm.go @@ -166,11 +166,19 @@ func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, fn *fv1.Function) } func (gpm *GenericPoolManager) GetFuncSvcFromCache(fn *fv1.Function) (*fscache.FuncSvc, error) { - return gpm.fsCache.GetByFunction(&fn.ObjectMeta) + return gpm.fsCache.GetFuncSvc(&fn.ObjectMeta) } func (gpm *GenericPoolManager) DeleteFuncSvcFromCache(fsvc *fscache.FuncSvc) { - gpm.fsCache.DeleteEntry(fsvc) + gpm.fsCache.DeleteFunctionSvc(fsvc) +} + +func (gpm *GenericPoolManager) UnTapService(fn *fv1.Function, svcHost string) { + gpm.fsCache.MarkAvailable(fn, svcHost) +} + +func (gpm *GenericPoolManager) GetTotalAvailable(fn *fv1.Function) int { + return gpm.fsCache.GetTotalAvailable(&fn.ObjectMeta) } func (gpm *GenericPoolManager) TapService(svcHost string) error { @@ -581,7 +589,7 @@ func (gpm *GenericPoolManager) idleObjectReaper() { fnList[fn.ObjectMeta.UID] = fns.Items[i] } - funcSvcs, err := gpm.fsCache.ListOld(pollSleep) + funcSvcs, err := gpm.fsCache.ListOldForPool(pollSleep) if err != nil { gpm.logger.Error("error reaping idle pods", zap.Error(err)) continue @@ -618,7 +626,7 @@ func (gpm *GenericPoolManager) idleObjectReaper() { } go func() { - deleted, err := gpm.fsCache.DeleteOld(fsvc, idlePodReapTime) + deleted, err := gpm.fsCache.DeleteOldPoolCache(fsvc, idlePodReapTime) if err != nil { gpm.logger.Error("error deleting Kubernetes objects for function service", zap.Error(err), diff --git a/pkg/executor/fscache/functionServiceCache.go b/pkg/executor/fscache/functionServiceCache.go index d80cb805..7a8882ba 100644 --- a/pkg/executor/fscache/functionServiceCache.go +++ b/pkg/executor/fscache/functionServiceCache.go @@ -30,6 +30,7 @@ import ( "github.com/fission/fission/pkg/cache" "github.com/fission/fission/pkg/crd" ferror "github.com/fission/fission/pkg/error" + poolcache "github.com/fission/fission/pkg/newcache" ) type fscRequestType int @@ -40,6 +41,7 @@ const ( TOUCH fscRequestType = iota LISTOLD LOG + LISTOLDPOOL ) type ( @@ -56,10 +58,11 @@ type ( } FunctionServiceCache struct { - logger *zap.Logger - byFunction *cache.Cache // function-key -> funcSvc : map[string]*funcSvc - byAddress *cache.Cache // address -> function : map[string]metav1.ObjectMeta - byFunctionUID *cache.Cache // function uid -> function : map[string]metav1.ObjectMeta + logger *zap.Logger + byFunction *cache.Cache // function-key -> funcSvc : map[string]*funcSvc + byAddress *cache.Cache // address -> function : map[string]metav1.ObjectMeta + byFunctionUID *cache.Cache // function uid -> function : map[string]metav1.ObjectMeta + connFunctionCache *poolcache.Cache // function-key -> funcSvc : map[string]*funcSvc requestChannel chan *fscRequest } @@ -91,11 +94,12 @@ func IsNameExistError(err error) bool { func MakeFunctionServiceCache(logger *zap.Logger) *FunctionServiceCache { fsc := &FunctionServiceCache{ - logger: logger.Named("function_service_cache"), - byFunction: cache.MakeCache(0, 0), - byAddress: cache.MakeCache(0, 0), - byFunctionUID: cache.MakeCache(0, 0), - requestChannel: make(chan *fscRequest), + logger: logger.Named("function_service_cache"), + byFunction: cache.MakeCache(0, 0), + byAddress: cache.MakeCache(0, 0), + byFunctionUID: cache.MakeCache(0, 0), + connFunctionCache: poolcache.NewPoolCache(), + requestChannel: make(chan *fscRequest), } go fsc.service() return fsc @@ -131,6 +135,17 @@ func (fsc *FunctionServiceCache) service() { } } fsc.logger.Info("function service cache", zap.Int("item_count", len(funcCopy)), zap.Strings("cache", info)) + case LISTOLDPOOL: + fscs := fsc.connFunctionCache.ListValue() + funcObjects := make([]*FuncSvc, 0) + for _, funcSvc := range fscs { + fsvc := funcSvc.(*FuncSvc) + if time.Since(fsvc.Atime) > req.age { + funcObjects = append(funcObjects, fsvc) + } + } + resp.objects = funcObjects + } req.responseChannel <- resp } @@ -152,6 +167,23 @@ func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc, return &fsvcCopy, nil } +func (fsc *FunctionServiceCache) GetFuncSvc(m *metav1.ObjectMeta) (*FuncSvc, error) { + key := crd.CacheKey(m) + + fsvcI, err := fsc.connFunctionCache.GetValue(key) + if err != nil { + fsc.logger.Info("Not found in Cache") + return nil, err + } + + // update atime + fsvc := fsvcI.(*FuncSvc) + fsvc.Atime = time.Now() + + fsvcCopy := *fsvc + return &fsvcCopy, nil +} + func (fsc *FunctionServiceCache) GetByFunctionUID(uid types.UID) (*FuncSvc, error) { mI, err := fsc.byFunctionUID.Get(uid) if err != nil { @@ -173,6 +205,23 @@ func (fsc *FunctionServiceCache) GetByFunctionUID(uid types.UID) (*FuncSvc, erro return &fsvcCopy, nil } +func (fsc *FunctionServiceCache) AddFunc(fsvc FuncSvc) { + fsc.connFunctionCache.SetValue(crd.CacheKey(fsvc.Function), fsvc.Address, &fsvc) + now := time.Now() + fsvc.Ctime = now + fsvc.Atime = now + + fsc.setFuncAlive(fsvc.Function.Name, string(fsvc.Function.UID), true) +} + +func (fsc *FunctionServiceCache) GetTotalAvailable(m *metav1.ObjectMeta) int { + return fsc.connFunctionCache.GetTotalAvailable(crd.CacheKey(m)) +} + +func (fsc *FunctionServiceCache) MarkAvailable(fn *fv1.Function, svcHost string) { + fsc.connFunctionCache.MarkAvailable(crd.CacheKey(&fn.ObjectMeta), svcHost) +} + func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (*FuncSvc, error) { existing, err := fsc.byFunction.Set(crd.CacheKey(fsvc.Function), &fsvc) if err != nil { @@ -255,6 +304,10 @@ func (fsc *FunctionServiceCache) DeleteEntry(fsvc *FuncSvc) { fsc.setFuncAlive(fsvc.Function.Name, string(fsvc.Function.UID), false) } +func (fsc *FunctionServiceCache) DeleteFunctionSvc(fsvc *FuncSvc) { + fsc.connFunctionCache.DeleteValue(crd.CacheKey(fsvc.Function), fsvc.Address) +} + func (fsc *FunctionServiceCache) DeleteOld(fsvc *FuncSvc, minAge time.Duration) (bool, error) { if time.Since(fsvc.Atime) < minAge { return false, nil @@ -265,6 +318,16 @@ func (fsc *FunctionServiceCache) DeleteOld(fsvc *FuncSvc, minAge time.Duration) return true, nil } +func (fsc *FunctionServiceCache) DeleteOldPoolCache(fsvc *FuncSvc, minAge time.Duration) (bool, error) { + if time.Since(fsvc.Atime) < minAge { + return false, nil + } + + fsc.DeleteFunctionSvc(fsvc) + + return true, nil +} + func (fsc *FunctionServiceCache) ListOld(age time.Duration) ([]*FuncSvc, error) { responseChannel := make(chan *fscResponse) fsc.requestChannel <- &fscRequest{ @@ -276,6 +339,17 @@ func (fsc *FunctionServiceCache) ListOld(age time.Duration) ([]*FuncSvc, error) return resp.objects, resp.error } +func (fsc *FunctionServiceCache) ListOldForPool(age time.Duration) ([]*FuncSvc, error) { + responseChannel := make(chan *fscResponse) + fsc.requestChannel <- &fscRequest{ + requestType: LISTOLDPOOL, + age: age, + responseChannel: responseChannel, + } + resp := <-responseChannel + return resp.objects, resp.error +} + func (fsc *FunctionServiceCache) Log() { fsc.logger.Info("--- FunctionService Cache Contents") responseChannel := make(chan *fscResponse) diff --git a/pkg/executor/fscache/functionServiceCache_test.go b/pkg/executor/fscache/functionServiceCache_test.go index c8d78bcf..7d0d2600 100644 --- a/pkg/executor/fscache/functionServiceCache_test.go +++ b/pkg/executor/fscache/functionServiceCache_test.go @@ -1,6 +1,7 @@ package fscache import ( + "fmt" "log" "testing" "time" @@ -122,3 +123,89 @@ func TestFunctionServiceCache(t *testing.T) { log.Panicf("found fsvc by function uid while expecting empty cache: %v", err) } } + +func TestFunctionServiceNewCache(t *testing.T) { + logger, err := zap.NewDevelopment() + panicIf(err) + + fsc := MakeFunctionServiceCache(logger) + if fsc == nil { + log.Panicf("error creating cache") + } + + var fsvc *FuncSvc + now := time.Now() + + objects := []apiv1.ObjectReference{ + { + Kind: "pod", + Name: "xxx", + APIVersion: "v1", + Namespace: "fission-function", + }, + { + Kind: "pod", + Name: "xxx2", + APIVersion: "v1", + Namespace: "fission-function", + }, + } + + fsvc = &FuncSvc{ + Function: &metav1.ObjectMeta{ + Name: "foo", + UID: "1212", + }, + Environment: &fv1.Environment{ + ObjectMeta: metav1.ObjectMeta{ + Name: "foo-env", + UID: "2323", + }, + Spec: fv1.EnvironmentSpec{ + Version: 1, + Runtime: fv1.Runtime{ + Image: "fission/foo-env", + }, + Builder: fv1.Builder{}, + }, + }, + Address: "xxx", + KubernetesObjects: objects, + Ctime: now, + Atime: now, + } + + fn := &fv1.Function{ + ObjectMeta: metav1.ObjectMeta{ + Name: "foo", + UID: "1212", + }, + } + + fsc.AddFunc(*fsvc) + + active := fsc.GetTotalAvailable(fsvc.Function) + if active != 1 { + logger.Panic(fmt.Sprintln("active instances not matched expected 1, found ", active)) + } + + fsc.MarkAvailable(fn, fsvc.Address) + + if fsc.GetTotalAvailable(fsvc.Function) != 0 { + log.Panicln("active instances not matched") + } + + _, err = fsc.GetFuncSvc(fsvc.Function) + if err != nil { + logger.Panic("received error while retrieving value from cache") + } + + vals, err := fsc.ListOldForPool(30 * time.Second) + if err != nil { + logger.Panic("received error while get list of old values") + } + if len(vals) != 0 { + logger.Panic(fmt.Sprintln("list of old values didn't matched the expected: 1", "received", len(vals))) + } + fsc.DeleteFunctionSvc(fsvc) +} diff --git a/pkg/fission-cli/cmd/function/command.go b/pkg/fission-cli/cmd/function/command.go index b9695979..4584500a 100644 --- a/pkg/fission-cli/cmd/function/command.go +++ b/pkg/fission-cli/cmd/function/command.go @@ -35,7 +35,7 @@ func Commands() *cobra.Command { flag.FnEnvName, flag.FnEntryPoint, flag.FnPkgName, flag.FnExecutorType, flag.FnCfgMap, flag.FnSecret, flag.FnSpecializationTimeout, flag.FnExecutionTimeout, - flag.FnIdleTimeout, + flag.FnIdleTimeout, flag.FnConcurrency, // TODO retired pkg & trigger related flags from function cmd flag.PkgCode, flag.PkgSrcArchive, flag.PkgDeployArchive, @@ -86,7 +86,7 @@ func Commands() *cobra.Command { flag.FnEnvName, flag.FnEntryPoint, flag.FnPkgName, flag.FnExecutorType, flag.FnSecret, flag.FnCfgMap, flag.FnSpecializationTimeout, flag.FnExecutionTimeout, - flag.FnIdleTimeout, + flag.FnIdleTimeout, flag.FnConcurrency, flag.PkgCode, flag.PkgSrcArchive, flag.PkgDeployArchive, flag.PkgSrcChecksum, flag.PkgDeployChecksum, flag.PkgInsecure, diff --git a/pkg/fission-cli/cmd/function/create.go b/pkg/fission-cli/cmd/function/create.go index ec3da96f..95247328 100644 --- a/pkg/fission-cli/cmd/function/create.go +++ b/pkg/fission-cli/cmd/function/create.go @@ -41,6 +41,7 @@ import ( const ( DEFAULT_MIN_SCALE = 1 DEFAULT_TARGET_CPU_PERCENTAGE = 80 + DEFAULT_CONCURRENCY = 5 ) type CreateSubCommand struct { @@ -96,6 +97,11 @@ func (opts *CreateSubCommand) complete(input cli.Input) error { fnIdleTimeout := input.Int(flagkey.FnIdleTimeout) + fnConcurrency := DEFAULT_CONCURRENCY + if input.IsSet(flagkey.FnConcurrency) { + fnConcurrency = input.Int(flagkey.FnConcurrency) + } + pkgName := input.String(flagkey.FnPackageName) secretNames := input.StringSlice(flagkey.FnSecret) @@ -294,6 +300,7 @@ func (opts *CreateSubCommand) complete(input cli.Input) error { InvokeStrategy: *invokeStrategy, FunctionTimeout: fnTimeout, IdleTimeout: &fnIdleTimeout, + Concurrency: fnConcurrency, }, } diff --git a/pkg/fission-cli/cmd/function/update.go b/pkg/fission-cli/cmd/function/update.go index 44dd5c3c..ca680f0a 100644 --- a/pkg/fission-cli/cmd/function/update.go +++ b/pkg/fission-cli/cmd/function/update.go @@ -155,6 +155,10 @@ func (opts *UpdateSubCommand) complete(input cli.Input) error { function.Spec.IdleTimeout = &fnTimeout } + if input.IsSet(flagkey.FnConcurrency) { + function.Spec.Concurrency = input.Int(flagkey.FnConcurrency) + } + if len(pkgName) == 0 { pkgName = function.Spec.Package.PackageRef.Name } diff --git a/pkg/fission-cli/flag/flag.go b/pkg/fission-cli/flag/flag.go index db8a23cc..264ddfa2 100644 --- a/pkg/fission-cli/flag/flag.go +++ b/pkg/fission-cli/flag/flag.go @@ -110,6 +110,7 @@ var ( FnTestHeader = Flag{Type: StringSlice, Name: flagkey.FnTestHeader, Short: "H", Usage: "Request headers"} FnTestQuery = Flag{Type: StringSlice, Name: flagkey.FnTestQuery, Short: "q", Usage: "Request query parameters: -q key1=value1 -q key2=value2"} FnIdleTimeout = Flag{Type: Int, Name: flagkey.FnIdleTimeout, Usage: "The length of time (in seconds) that a function is idle before pod(s) are eligible for recycling", DefaultValue: 120} + FnConcurrency = Flag{Type: Int, Name: flagkey.FnConcurrency, Aliases: []string{"con"}, Usage: "Maximum number of pods specialized concurrently to serve requests", DefaultValue: 5} HtName = Flag{Type: String, Name: flagkey.HtName, Usage: "HTTP trigger name"} HtMethod = Flag{Type: String, Name: flagkey.HtMethod, Usage: "HTTP Method: GET|POST|PUT|DELETE|HEAD", DefaultValue: http.MethodGet} diff --git a/pkg/fission-cli/flag/key/key.go b/pkg/fission-cli/flag/key/key.go index 0f9d82cc..f9b48d2e 100644 --- a/pkg/fission-cli/flag/key/key.go +++ b/pkg/fission-cli/flag/key/key.go @@ -63,6 +63,7 @@ const ( FnTestHeader = "header" FnTestQuery = "query" FnIdleTimeout = "idletimeout" + FnConcurrency = "concurrency" HtName = resourceName HtMethod = "method" diff --git a/pkg/newcache/poolcache.go b/pkg/newcache/poolcache.go new file mode 100644 index 00000000..709cfef1 --- /dev/null +++ b/pkg/newcache/poolcache.go @@ -0,0 +1,219 @@ +/* +Copyright 2016 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 poolcache implements a simple cache implementation having values mapped by two keys. +// As of now this package is only used by poolmanager executor +package poolcache + +import ( + "fmt" + + ferror "github.com/fission/fission/pkg/error" +) + +type requestType int + +const ( + getValue requestType = iota + listValue + getTotalAvailable + setValue + markAvailable + deleteValue +) + +type ( + // value used as "value" in cache + value struct { + val interface{} + isActive bool + } + // Cache is simple cache having two keys [function][address] mapped to value and requestChannel for operation on it + Cache struct { + cache map[interface{}]map[interface{}]*value + requestChannel chan *request + } + + request struct { + requestType + function interface{} + address interface{} + value interface{} + responseChannel chan *response + } + response struct { + error + allValues []interface{} + value interface{} + totalAvailable int + } +) + +// NewPoolCache create a Cache object +func NewPoolCache() *Cache { + c := &Cache{ + cache: make(map[interface{}]map[interface{}]*value), + requestChannel: make(chan *request), + } + go c.service() + return c +} + +func (c *Cache) service() { + for { + req := <-c.requestChannel + resp := &response{} + switch req.requestType { + case getValue: + values, ok := c.cache[req.function] + found := false + if !ok { + resp.error = ferror.MakeError(ferror.ErrorNotFound, + fmt.Sprintf("function Name '%v' not found", req.function)) + } else { + for addr := range values { + if !values[addr].isActive { + // update atime + // mark active + values[addr].isActive = true + resp.value = values[addr].val + found = true + break + } + } + } + if !found { + resp.error = ferror.MakeError(ferror.ErrorNotFound, fmt.Sprintf("funtion '%v' No inactive function found", req.function)) + } + req.responseChannel <- resp + case listValue: + vals := make([]interface{}, 0) + for _, values := range c.cache { + for _, value := range values { + vals = append(vals, value.val) + } + } + resp.allValues = vals + req.responseChannel <- resp + case getTotalAvailable: + if values, ok := c.cache[req.function]; ok { + for addr := range values { + if values[addr].isActive { + resp.totalAvailable++ + } + } + } + req.responseChannel <- resp + case setValue: + if _, ok := c.cache[req.function]; ok { + c.cache[req.function][req.address] = &value{ + val: req.value, + isActive: true, + } + } else { + c.cache[req.function] = make(map[interface{}]*value) + c.cache[req.function][req.address] = &value{ + val: req.value, + isActive: true, + } + } + case markAvailable: + if _, ok := c.cache[req.function]; ok { + if _, ok = c.cache[req.function][req.address]; ok { + c.cache[req.function][req.address].isActive = false + } + } + case deleteValue: + delete(c.cache[req.function], req.address) + req.responseChannel <- resp + default: + resp.error = ferror.MakeError(ferror.ErrorInvalidArgument, + fmt.Sprintf("invalid request type: %v", req.requestType)) + req.responseChannel <- resp + } + } +} + +// GetValue returns a value interface with status inActive else return error +func (c *Cache) GetValue(function interface{}) (interface{}, error) { + respChannel := make(chan *response) + c.requestChannel <- &request{ + requestType: getValue, + function: function, + responseChannel: respChannel, + } + resp := <-respChannel + return resp.value, resp.error +} + +// ListValue returns a list of the function services stored in the Cache +func (c *Cache) ListValue() []interface{} { + respChannel := make(chan *response) + c.requestChannel <- &request{ + requestType: listValue, + responseChannel: respChannel, + } + resp := <-respChannel + return resp.allValues +} + +// GetTotalAvailable returns a total number active function services +func (c *Cache) GetTotalAvailable(function interface{}) int { + respChannel := make(chan *response) + c.requestChannel <- &request{ + requestType: getTotalAvailable, + function: function, + responseChannel: respChannel, + } + resp := <-respChannel + return resp.totalAvailable +} + +// SetValue marks the value at key [function][address] as active(begin used) +func (c *Cache) SetValue(function, address, value interface{}) { + respChannel := make(chan *response) + c.requestChannel <- &request{ + requestType: setValue, + function: function, + address: address, + value: value, + responseChannel: respChannel, + } +} + +// MarkAvailable marks the value at key [function][address] as available +func (c *Cache) MarkAvailable(function, address interface{}) { + respChannel := make(chan *response) + c.requestChannel <- &request{ + requestType: markAvailable, + function: function, + address: address, + responseChannel: respChannel, + } +} + +// DeleteValue deletes the value at key composed of [function][address] +func (c *Cache) DeleteValue(function, address interface{}) error { + respChannel := make(chan *response) + c.requestChannel <- &request{ + requestType: deleteValue, + function: function, + address: address, + responseChannel: respChannel, + } + resp := <-respChannel + return resp.error +} diff --git a/pkg/newcache/poolcache_test.go b/pkg/newcache/poolcache_test.go new file mode 100644 index 00000000..3bdbb97c --- /dev/null +++ b/pkg/newcache/poolcache_test.go @@ -0,0 +1,54 @@ +package poolcache + +import ( + "log" + "testing" + "time" +) + +func checkErr(err error) { + if err != nil { + log.Panicf("err: %v", err) + } +} + +func TestPoolCache(t *testing.T) { + c := NewPoolCache() + + c.SetValue("func", "ip", "value") + + c.SetValue("func2", "ip2", "value2") + + c.SetValue("func2", "ip22", "value22") + + cc := c.ListValue() + if len(cc) != 3 { + log.Panicf("expected 2 items") + } + active := c.GetTotalAvailable("func2") + if active != 2 { + log.Panicf("expected 2 items") + } + + c.DeleteValue("func2", "ip2") + + c.MarkAvailable("func", "ip") + + _, err := c.GetValue("func") + checkErr(err) + + c.DeleteValue("func", "ip") + + _, err = c.GetValue("func") + if err == nil { + log.Panicf("found deleted element") + } + + c.SetValue("expires", "42", "all answers") + + time.Sleep(150 * time.Millisecond) + _, err = c.GetValue("expires") + if err == nil { + log.Panicf("found expired element") + } +} diff --git a/pkg/router/functionHandler.go b/pkg/router/functionHandler.go index 8996a10e..37612f41 100644 --- a/pkg/router/functionHandler.go +++ b/pkg/router/functionHandler.go @@ -35,7 +35,6 @@ 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" @@ -76,8 +75,8 @@ type ( // 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. + // errors router received is higher than svcAddrRetryCount. + // Try to get a new one from executor. // Default svcAddrRetryCount is 5. svcAddrRetryCount int } @@ -98,11 +97,6 @@ type ( fakeCloseReadCloser struct { io.ReadCloser } - - svcEntryRecord struct { - svcUrl *url.URL - fromCache bool - } ) func init() { @@ -148,10 +142,9 @@ func (w *fakeCloseReadCloser) RealClose() error { // 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) (*http.Response, error) { - // Set forwarded host header if not exists + // set the timeout for transport context roundTripper.addForwardedHostHeader(req) - // set the timeout for transport context transport := roundTripper.getDefaultTransport() ocRoundTripper := &ochttp.Transport{Base: transport} @@ -190,11 +183,16 @@ 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, roundTripper.urlFromCache, err = roundTripper.funcHandler.getServiceEntry() + roundTripper.serviceUrl, err = roundTripper.funcHandler.getServiceEntryFromExecutor() if err != nil { // We might want a specific error code or header for fission failures as opposed to // user function bugs. statusCode, errMsg := ferror.GetHTTPError(err) + // if statusCode == http.StatusTooManyRequests { + // time.Sleep(executingTimeout) + // executingTimeout = executingTimeout * time.Duration(roundTripper.funcHandler.tsRoundTripperParams.timeoutExponent) + // continue + // } else { if roundTripper.funcHandler.isDebugEnv { return &http.Response{ StatusCode: statusCode, @@ -208,18 +206,14 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re }, nil } return nil, ferror.MakeError(http.StatusInternalServerError, err.Error()) + // } } - - // service url maybe nil if router cannot find one in cache, - // so here we retry to get service url again - if roundTripper.serviceUrl == nil { - time.Sleep(executingTimeout) - executingTimeout = executingTimeout * time.Duration(roundTripper.funcHandler.tsRoundTripperParams.timeoutExponent) - continue + if roundTripper.funcHandler.function.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypePoolmgr { + defer roundTripper.funcHandler.unTapService(roundTripper.funcHandler.function, roundTripper.serviceUrl) } // modify the request to reflect the service url - // this service url may have come from the cache lookup or from executor response + // this service url comes from executor response req.URL.Scheme = roundTripper.serviceUrl.Scheme req.URL.Host = roundTripper.serviceUrl.Host @@ -254,7 +248,7 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re return resp, nil } - roundTripper.totalRetry += 1 + roundTripper.totalRetry++ if i >= roundTripper.funcHandler.tsRoundTripperParams.maxRetries-1 { // return here if we are in the last round @@ -276,6 +270,7 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re // if transport.RoundTrip returns a non-network dial error (e.g. "context canceled"), then relay it back to user if !isNetDialErr { + roundTripper.logger.Error("encountered non-network dial error", zap.Error(err)) return resp, err } @@ -285,21 +280,7 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re } // Check whether an error is an timeout error ("dial tcp i/o timeout"). - // If it's not a timeout error or retryCounter exceeded pre-defined threshold, - // we assume the entry in router cache is stale, invalidate it. - if !isNetTimeoutErr || retryCounter >= roundTripper.funcHandler.tsRoundTripperParams.svcAddrRetryCount { - if roundTripper.urlFromCache { - // 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. - roundTripper.logger.Debug("request errored out - removing function from router's cache and requesting a new service for function", - zap.String("url", req.URL.Host), - zap.String("function_name", fnMeta.Name), - zap.Error(err)) - - roundTripper.funcHandler.fmap.remove(fnMeta) - } - retryCounter = 0 - } else { + if isNetTimeoutErr { roundTripper.logger.Debug("request errored out - backing off before retrying", zap.String("url", req.URL.Host), zap.String("function_name", fnMeta.Name), @@ -307,6 +288,14 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re retryCounter++ } + // If it's not a timeout error or retryCounter exceeded pre-defined threshold, + if retryCounter >= roundTripper.funcHandler.tsRoundTripperParams.svcAddrRetryCount { + roundTripper.logger.Debug(fmt.Sprintf( + "retry counter exceeded pre-defined threshold of %v", + roundTripper.funcHandler.tsRoundTripperParams.svcAddrRetryCount)) + retryCounter = 0 + } + roundTripper.logger.Debug("Backing off before retrying", zap.Any("backoff_time", executingTimeout), zap.Error(err)) time.Sleep(executingTimeout) executingTimeout = executingTimeout * time.Duration(roundTripper.funcHandler.tsRoundTripperParams.timeoutExponent) @@ -478,12 +467,12 @@ func (roundTripper RetryingRoundTripper) addForwardedHostHeader(req *http.Reques // Format of req.Host is : // We need to extract hostname from it, than // check whether a host is ipv4 or ipv6 or FQDN - reqUrl := fmt.Sprintf("%s://%s", req.Proto, req.Host) - u, err := url.Parse(reqUrl) + reqURL := fmt.Sprintf("%s://%s", req.Proto, req.Host) + u, err := url.Parse(reqURL) if err != nil { roundTripper.logger.Error("error parsing request url while adding forwarded host headers", zap.Error(err), - zap.String("url", reqUrl)) + zap.String("url", reqURL)) return } @@ -508,100 +497,35 @@ func (roundTripper RetryingRoundTripper) addForwardedHostHeader(req *http.Reques req.Header.Set(X_FORWARDED_HOST, req.Host) } -// getServiceEntry is a short-hand for developers to get service url entry that may returns from executor or cache -func (fh *functionHandler) getServiceEntry() (serviceUrl *url.URL, serviceUrlFromCache bool, err error) { - // try to find service url from cache first - serviceUrl, err = fh.getServiceEntryFromCache() - if err == nil && serviceUrl != nil { - return serviceUrl, true, nil - } else if err != nil { - return nil, false, err - } - - // cache miss or nil entry in cache +// unTapservice marks the serviceURL in executor's cache as inactive, so that it can be reused +func (fh functionHandler) unTapService(fn *fv1.Function, serviceUrl *url.URL) error { + fh.logger.Info("UnTapService Called") ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() - - fnMeta := &fh.function.ObjectMeta - - // 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(fnMeta), - 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 - fh.logger.Debug("calling getServiceForFunction", - zap.String("function_name", fnMeta.Name)) - u, err = fh.getServiceEntryFromExecutor(ctx) - if err != nil { - fh.logger.Error("error getting service url from executor", - zap.Error(err), - zap.String("function_name", fnMeta.Name)) - return nil, err - } - // add the address in router's cache - fh.logger.Info("assigning service url for function", - zap.String("url", u.String()), - zap.String("function_name", fnMeta.Name)) - fh.fmap.assign(fnMeta, u) - } else { - u, err = fh.getServiceEntryFromCache() - if err != nil { - return nil, err - } - } - - return svcEntryRecord{ - svcUrl: u, - fromCache: firstToTheLock, - }, err - }, - ) + err := fh.executor.UnTapService(ctx, fn.ObjectMeta, fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType, serviceUrl) if err != nil { - e := "error updating service address entry for function" - fh.logger.Error(e, + statusCode, errMsg := ferror.GetHTTPError(err) + fh.logger.Error("error from UnTapService", zap.Error(err), - zap.String("function_name", fnMeta.Name), - zap.String("function_namespace", fnMeta.Namespace)) - return nil, false, errors.Wrapf(err, "%s %s_%s", e, fnMeta.Name, fnMeta.Namespace) + zap.String("error_message", errMsg), + zap.Any("function", fh.function), + zap.Int("status_code", statusCode)) + return 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 -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 + return nil } // getServiceEntryFromExecutor returns service url entry returns from executor -func (fh functionHandler) getServiceEntryFromExecutor(ctx context.Context) (*url.URL, error) { +func (fh functionHandler) getServiceEntryFromExecutor() (*url.URL, 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.ObjectMeta) if err != nil { statusCode, errMsg := ferror.GetHTTPError(err) diff --git a/pkg/router/functionHandler_test.go b/pkg/router/functionHandler_test.go index 27223387..74a689ad 100644 --- a/pkg/router/functionHandler_test.go +++ b/pkg/router/functionHandler_test.go @@ -19,10 +19,8 @@ package router import ( "context" "errors" - "log" "net/http" "net/http/httptest" - "net/url" "testing" "time" @@ -34,72 +32,6 @@ import ( fv1 "github.com/fission/fission/pkg/apis/core/v1" ) -func createBackendService(testResponseString string) *url.URL { - backendServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - w.Write([]byte(testResponseString)) - })) - - backendURL, err := url.Parse(backendServer.URL) - if err != nil { - panic("error parsing url") - } - return backendURL -} - -/* - 1. Create a service at some URL - 2. Add it to the function service map - 3. Create a http server with some trigger url pointed at function handler - 4. Send a request to that server, ensure it reaches the first service. -*/ -func TestFunctionProxying(t *testing.T) { - testResponseString := "hi" - backendURL := createBackendService(testResponseString) - log.Printf("Created backend svc at %v", backendURL) - - fnMeta := metav1.ObjectMeta{Name: "foo", Namespace: metav1.NamespaceDefault} - - config := zap.NewDevelopmentConfig() - config.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder - logger, err := config.Build() - - panicIf(err) - - fmap := makeFunctionServiceMap(logger, 0) - fmap.assign(&fnMeta, backendURL) - - httpTrigger := &fv1.HTTPTrigger{ - ObjectMeta: metav1.ObjectMeta{ - Name: "xxx", - Namespace: metav1.NamespaceDefault, - ResourceVersion: "1234", - }, - Spec: fv1.HTTPTriggerSpec{ - FunctionReference: fv1.FunctionReference{ - Type: fv1.FunctionReferenceTypeFunctionName, - }, - }, - } - - fh := &functionHandler{ - logger: logger, - fmap: fmap, - function: &fv1.Function{ - ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: metav1.NamespaceDefault}, - }, - tsRoundTripperParams: &tsRoundTripperParams{ - timeout: 50 * time.Millisecond, - timeoutExponent: 2, - maxRetries: 10, - }, - httpTrigger: httpTrigger, - } - functionHandlerServer := httptest.NewServer(http.HandlerFunc(fh.handler)) - fhURL := functionHandlerServer.URL - - testRequest(fhURL, testResponseString) -} - func TestProxyErrorHandler(t *testing.T) { config := zap.NewDevelopmentConfig() config.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder diff --git a/pkg/router/functionServiceMap.go b/pkg/router/functionServiceMap.go index 2cae359a..7a11578f 100644 --- a/pkg/router/functionServiceMap.go +++ b/pkg/router/functionServiceMap.go @@ -77,8 +77,3 @@ 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) -} diff --git a/pkg/router/router_test.go b/pkg/router/router_test.go deleted file mode 100644 index a6e811e5..00000000 --- a/pkg/router/router_test.go +++ /dev/null @@ -1,108 +0,0 @@ -/* -Copyright 2016 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 ( - "context" - "fmt" - "testing" - "time" - - "go.uber.org/zap" - "go.uber.org/zap/zapcore" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - - fv1 "github.com/fission/fission/pkg/apis/core/v1" - "github.com/fission/fission/pkg/throttler" -) - -func TestRouter(t *testing.T) { - // metadata for a fake function - fnMeta := metav1.ObjectMeta{Name: "foo", Namespace: metav1.NamespaceDefault} - - // and a reference to it - fr := fv1.FunctionReference{ - Type: fv1.FunctionReferenceTypeFunctionName, - Name: fnMeta.Name, - } - - // start a fake service - testResponseString := "hi" - testServiceUrl := createBackendService(testResponseString) - - config := zap.NewDevelopmentConfig() - config.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder - logger, err := config.Build() - panicIf(err) - - // set up the cache with this fake service - fmap := makeFunctionServiceMap(logger, 0) - fmap.assign(&fnMeta, testServiceUrl) - - // HTTP trigger set with a trigger for this function - triggers, _, _ := makeHTTPTriggerSet(logger, fmap, nil, nil, nil, nil, - &tsRoundTripperParams{ - timeout: 50 * time.Millisecond, - timeoutExponent: 2, - maxRetries: 10, - }, false, throttler.MakeThrottler(30*time.Second)) - triggerUrl := "/foo" - triggers.triggers = append(triggers.triggers, - fv1.HTTPTrigger{ - ObjectMeta: metav1.ObjectMeta{ - Name: "xxx", - Namespace: metav1.NamespaceDefault, - ResourceVersion: "1234", - }, - Spec: fv1.HTTPTriggerSpec{ - RelativeURL: triggerUrl, - FunctionReference: fr, - Method: "GET", - }, - }) - - // set up the resolver's cache for this function - frr := makeFunctionReferenceResolver(nil) - nfr := namespacedTriggerReference{ - namespace: metav1.NamespaceDefault, - triggerName: "xxx", - triggerResourceVersion: "1234", - } - - fnMetaMap := make(map[string]*fv1.Function, 1) - fnMetaMap[fnMeta.Name] = &fv1.Function{ - ObjectMeta: fnMeta, - } - - rr := resolveResult{ - resolveResultType: resolveResultSingleFunction, - functionMap: fnMetaMap, - } - frr.refCache.Set(nfr, rr) - - // run the router - port := 4242 - tracingSamplingRate := .5 - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - go serve(ctx, logger, port, tracingSamplingRate, triggers, frr, false) - time.Sleep(100 * time.Millisecond) - - // hit the router - testUrl := fmt.Sprintf("http://localhost:%v%v", port, triggerUrl) - testRequest(testUrl, testResponseString) -} diff --git a/test/test_utils.sh b/test/test_utils.sh index 0521883d..cb1300af 100755 --- a/test/test_utils.sh +++ b/test/test_utils.sh @@ -528,7 +528,6 @@ run_all_tests() { $ROOT/test/tests/test_package_command.sh \ $ROOT/test/tests/test_package_checksum.sh \ $ROOT/test/tests/test_pass.sh \ - $ROOT/test/tests/test_router_cache_invalidation.sh \ $ROOT/test/tests/test_specs/test_spec.sh \ $ROOT/test/tests/test_specs/test_spec_multifile.sh \ $ROOT/test/tests/test_specs/test_spec_merge/test_spec_merge.sh \ diff --git a/test/tests/test_router_cache_invalidation.sh b/test/tests/test_router_cache_invalidation.sh deleted file mode 100755 index e9201253..00000000 --- a/test/tests/test_router_cache_invalidation.sh +++ /dev/null @@ -1,72 +0,0 @@ -#!/bin/bash - -set -euo pipefail -source $(dirname $0)/../utils.sh - -TEST_ID=$(generate_test_id) -echo "TEST_ID = $TEST_ID" - -tmp_dir="/tmp/test-$TEST_ID" -mkdir -p $tmp_dir - -cleanup() { - log "Cleaning up..." - clean_resource_by_id $TEST_ID - rm -rf $tmp_dir -} - -if [ -z "${TEST_NOCLEANUP:-}" ]; then - trap cleanup EXIT -else - log "TEST_NOCLEANUP is set; not cleaning up test artifacts afterwards." -fi - -# 1. This test first creates a python function with a route -# 2. Makes a curl request to the route and verifies http.StatusOK is received. -# This step ensures the pod address is cached in router. -# 3. Then, finds the pod that has the function loaded and deletes the pod with grace period 0s. -# This step results in a stale entry in the router cache. -# 4. Finally, makes a curl request again and waits for response http.StatusOK. -# This ensures that router invalidated its cache, made a request to executor to get service for function and retried -# the request against this new address. - -env=python-$TEST_ID -fn=python-func-$TEST_ID - -log "Creating python env" -fission env create --name $env --image $PYTHON_RUNTIME_IMAGE - -log "Creating hello.py" -printf 'def main():\n return "Hello, world!"' > $tmp_dir/hello.py - -log "Creating function " $fn -fission fn create --name $fn --env $env --code $tmp_dir/hello.py - -log "Creating route" -fission route create --function $fn --url /$fn --method GET - -log "Waiting for router to update cache" -sleep 5 - -http_status=`curl -sw "%{http_code}" http://$FISSION_ROUTER/$fn -o /dev/null` -log "http_status: $http_status" -if [ "$http_status" -ne "200" ]; then - log "Something went wrong, http status even before deleting function pod is $http_status" - exit 1 -fi - -log "getting function pod" -funcPod=`kubectl get pods -n $FUNCTION_NAMESPACE -L functionName | grep $fn| tr -s " "| cut -d" " -f1` -log "funcPod : $funcPod" - -kubectl delete pod $funcPod -n $FUNCTION_NAMESPACE --grace-period=0 -log "deleted function pod $funcPod" - -http_status=`curl -sw "%{http_code}" http://$FISSION_ROUTER/$fn -o /dev/null` -log "http_status: $http_status" -if [ "$http_status" -ne "200" ]; then - log "Something went wrong, http status after deleting function pod is $http_status" - exit 1 -fi - -log "Test PASSED"