From 6d2fe08973390031e2a16e895b51af2a1915d15f Mon Sep 17 00:00:00 2001 From: Ta-Ching Chen Date: Mon, 25 Nov 2019 18:44:48 +0800 Subject: [PATCH] Allow to tap multiple function services at one time (#1434) The router taps function service one by one which is inefficient and increases the burden of executor. This PR aggregates all requests into one to solve the problem mentioned above. --- pkg/executor/api.go | 46 ++++++++++++- pkg/executor/client/client.go | 91 +++++++++++++++++-------- pkg/router/functionHandler.go | 56 +++++++-------- pkg/router/functionHandler_test.go | 12 ++-- pkg/router/functionReferenceResolver.go | 18 ++--- pkg/router/httpTriggers.go | 21 +++--- pkg/router/router_test.go | 16 +++-- 7 files changed, 169 insertions(+), 91 deletions(-) diff --git a/pkg/executor/api.go b/pkg/executor/api.go index ff486644..2cb57934 100644 --- a/pkg/executor/api.go +++ b/pkg/executor/api.go @@ -20,6 +20,8 @@ import ( "context" "encoding/json" "fmt" + "github.com/hashicorp/go-multierror" + "github.com/pkg/errors" "io/ioutil" "net/http" "strings" @@ -32,6 +34,7 @@ import ( fv1 "github.com/fission/fission/pkg/apis/fission.io/v1" ferror "github.com/fission/fission/pkg/error" + "github.com/fission/fission/pkg/executor/client" ) func (executor *Executor) getServiceForFunctionApi(w http.ResponseWriter, r *http.Request) { @@ -115,6 +118,7 @@ func (executor *Executor) getServiceForFunction(fn *fv1.Function) (string, error } // find funcSvc and update its atime +// TODO: Deprecated tapService func (executor *Executor) tapService(w http.ResponseWriter, r *http.Request) { body, err := ioutil.ReadAll(r.Body) if err != nil { @@ -137,6 +141,45 @@ func (executor *Executor) tapService(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) } +// find funcSvc and update its atime +func (executor *Executor) tapServices(w http.ResponseWriter, r *http.Request) { + body, err := ioutil.ReadAll(r.Body) + if err != nil { + executor.logger.Error("failed to read tap service request", zap.Error(err)) + http.Error(w, "Failed to read request", http.StatusInternalServerError) + return + } + + tapSvcReqs := []client.TapServiceRequest{} + err = json.Unmarshal(body, &tapSvcReqs) + if err != nil { + executor.logger.Error("failed to decode tap service request", + zap.Error(err), + zap.String("request-payload", string(body))) + http.Error(w, "Failed to decode tap service request", http.StatusBadRequest) + return + } + + errs := &multierror.Error{} + for _, req := range tapSvcReqs { + svcHost := strings.TrimPrefix(req.ServiceUrl, "http://") + err = executor.fsCache.TouchByAddress(svcHost) + if err != nil { + errs = multierror.Append(errs, + errors.Wrapf(err, "'%v' failed to tap function '%v/%v' with service url '%v'", + req.FnMetadata.Namespace, req.FnMetadata.Name, req.ServiceUrl, req.FnExecutorType)) + } + } + + if errs.ErrorOrNil() != nil { + executor.logger.Error("error tapping function service", zap.Error(errs)) + http.Error(w, "Not found", http.StatusNotFound) + return + } + + w.WriteHeader(http.StatusOK) +} + func (executor *Executor) healthHandler(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) } @@ -144,7 +187,8 @@ func (executor *Executor) healthHandler(w http.ResponseWriter, r *http.Request) 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") + 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") return r } diff --git a/pkg/executor/client/client.go b/pkg/executor/client/client.go index 45e70df7..06e9c1f3 100644 --- a/pkg/executor/client/client.go +++ b/pkg/executor/client/client.go @@ -26,29 +26,36 @@ import ( "strings" "time" + fv1 "github.com/fission/fission/pkg/apis/fission.io/v1" + ferror "github.com/fission/fission/pkg/error" "github.com/pkg/errors" "go.opencensus.io/plugin/ochttp" "go.uber.org/zap" "golang.org/x/net/context/ctxhttp" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - - ferror "github.com/fission/fission/pkg/error" ) -type Client struct { - logger *zap.Logger - executorUrl string - tappedByUrl map[string]bool - requestChan chan string - httpClient *http.Client -} +type ( + Client struct { + logger *zap.Logger + executorUrl string + tappedByUrl map[string]TapServiceRequest + requestChan chan TapServiceRequest + httpClient *http.Client + } + TapServiceRequest struct { + FnMetadata metav1.ObjectMeta + FnExecutorType fv1.ExecutorType + ServiceUrl string + } +) func MakeClient(logger *zap.Logger, executorUrl string) *Client { c := &Client{ logger: logger.Named("executor_client"), executorUrl: strings.TrimSuffix(executorUrl, "/"), - tappedByUrl: make(map[string]bool), - requestChan: make(chan string), + tappedByUrl: make(map[string]TapServiceRequest), + requestChan: make(chan TapServiceRequest), httpClient: &http.Client{ Transport: &ochttp.Transport{}, }, @@ -87,40 +94,64 @@ func (c *Client) service() { ticker := time.NewTicker(time.Second * 5) for { select { - case serviceUrl := <-c.requestChan: - c.tappedByUrl[serviceUrl] = true + case svcReq := <-c.requestChan: + c.tappedByUrl[svcReq.ServiceUrl] = svcReq case <-ticker.C: - urls := c.tappedByUrl - c.tappedByUrl = make(map[string]bool) - if len(urls) > 0 { - go func() { - for u := range urls { - err := c._tapService(u) - if err != nil { - c.logger.Error("error tapping function service address", zap.Error(err), zap.String("address", u)) - } - } - c.logger.Debug("tapped services in batch", zap.Int("service_count", len(urls))) - }() + if len(c.tappedByUrl) == 0 { + continue } + + urls := c.tappedByUrl + c.tappedByUrl = make(map[string]TapServiceRequest) + + go func() { + svcReqs := []TapServiceRequest{} + for _, req := range urls { + svcReqs = append(svcReqs, req) + } + c.logger.Debug("tapped services in batch", zap.Int("service_count", len(urls))) + + err := c._tapService(svcReqs) + if err != nil { + c.logger.Error("error tapping function service address", zap.Error(err)) + } + }() } } } -func (c *Client) TapService(serviceUrl *url.URL) { - c.requestChan <- serviceUrl.String() +func (c *Client) TapService(fnMeta metav1.ObjectMeta, executorType fv1.ExecutorType, serviceUrl *url.URL) { + c.requestChan <- TapServiceRequest{ + FnMetadata: metav1.ObjectMeta{ + Name: fnMeta.Name, + Namespace: fnMeta.Namespace, + ResourceVersion: fnMeta.ResourceVersion, + UID: fnMeta.UID, + }, + FnExecutorType: executorType, + // service url is for executor to know which + // pod/service is currently used to serve user function. + ServiceUrl: serviceUrl.String(), + } } -func (c *Client) _tapService(serviceUrlStr string) error { - executorUrl := c.executorUrl + "/v2/tapService" +func (c *Client) _tapService(tapSvcReqs []TapServiceRequest) error { + executorUrl := c.executorUrl + "/v2/tapServices" - resp, err := http.Post(executorUrl, "application/octet-stream", bytes.NewReader([]byte(serviceUrlStr))) + body, err := json.Marshal(tapSvcReqs) + if err != nil { + return err + } + + resp, err := http.Post(executorUrl, "application/json", bytes.NewReader(body)) if err != nil { return err } defer resp.Body.Close() + if resp.StatusCode != 200 { return ferror.MakeErrorFromHTTP(resp) } + return nil } diff --git a/pkg/router/functionHandler.go b/pkg/router/functionHandler.go index 6666f8ca..851482f1 100644 --- a/pkg/router/functionHandler.go +++ b/pkg/router/functionHandler.go @@ -54,9 +54,9 @@ type ( logger *zap.Logger fmap *functionServiceMap executor *executorClient.Client - function *metav1.ObjectMeta + function *fv1.Function httpTrigger *fv1.HTTPTrigger - functionMetadataMap map[string]*metav1.ObjectMeta + functionMap map[string]*fv1.Function fnWeightDistributionList []FunctionWeightDistribution tsRoundTripperParams *tsRoundTripperParams isDebugEnv bool @@ -150,7 +150,7 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re // Set forwarded host header if not exists roundTripper.addForwardedHostHeader(req) - fnMeta := roundTripper.funcHandler.function + fnMeta := &roundTripper.funcHandler.function.Metadata // Metrics stuff startTime := time.Now() @@ -239,7 +239,7 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re // tapService before invoking roundTrip for the serviceUrl if serviceUrlFromCache { - go roundTripper.funcHandler.tapService(serviceUrl) + go roundTripper.funcHandler.tapService(roundTripper.funcHandler.function, serviceUrl) } // modify the request to reflect the service url @@ -388,25 +388,25 @@ func (roundTripper *RetryingRoundTripper) closeContext() { } } -func (fh *functionHandler) tapService(serviceUrl *url.URL) { +func (fh *functionHandler) tapService(fn *fv1.Function, serviceUrl *url.URL) { if fh.executor == nil { return } - fh.executor.TapService(serviceUrl) + fh.executor.TapService(fn.Metadata, fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType, serviceUrl) } func (fh functionHandler) handler(responseWriter http.ResponseWriter, request *http.Request) { if fh.httpTrigger != nil && fh.httpTrigger.Spec.FunctionReference.Type == types.FunctionReferenceTypeFunctionWeights { // canary deployment. need to determine the function to send request to now - fnMetadata := getCanaryBackend(fh.functionMetadataMap, fh.fnWeightDistributionList) - if fnMetadata == nil { + fn := getCanaryBackend(fh.functionMap, fh.fnWeightDistributionList) + if fn == nil { fh.logger.Error("could not get canary backend", - zap.Any("metadataMap", fh.functionMetadataMap), + zap.Any("fnMap", fh.functionMap), zap.Any("distributionList", fh.fnWeightDistributionList)) // TODO : write error to responseWrite and return response return } - fh.function = fnMetadata + fh.function = fn fh.logger.Debug("chosen function backend's metadata", zap.Any("metadata", fh.function)) } @@ -414,7 +414,7 @@ func (fh functionHandler) handler(responseWriter http.ResponseWriter, request *h setPathInfoToHeader(request) // system params - setFunctionMetadataToHeader(fh.function, request) + setFunctionMetadataToHeader(&fh.function.Metadata, request) director := func(req *http.Request) { if _, ok := req.Header["User-Agent"]; !ok { @@ -423,7 +423,7 @@ func (fh functionHandler) handler(responseWriter http.ResponseWriter, request *h } } - fnTimeout := fh.functionTimeoutMap[fh.function.GetUID()] + fnTimeout := fh.functionTimeoutMap[fh.function.Metadata.GetUID()] if fnTimeout == 0 { fnTimeout = fv1.DEFAULT_FUNCTION_TIMEOUT } @@ -437,7 +437,7 @@ func (fh functionHandler) handler(responseWriter http.ResponseWriter, request *h proxy := &httputil.ReverseProxy{ Director: director, Transport: rrt, - ErrorHandler: getProxyErrorHandler(fh.logger, fh.function), + ErrorHandler: getProxyErrorHandler(fh.logger, &fh.function.Metadata), } defer func() { @@ -483,12 +483,10 @@ func findCeil(randomNumber int, wtDistrList []FunctionWeightDistribution) string } // picks a function to route to based on a random number generated -func getCanaryBackend(fnMetadatamap map[string]*metav1.ObjectMeta, fnWtDistributionList []FunctionWeightDistribution) *metav1.ObjectMeta { +func getCanaryBackend(fnMap map[string]*fv1.Function, fnWtDistributionList []FunctionWeightDistribution) *fv1.Function { randomNumber := rand.Intn(fnWtDistributionList[len(fnWtDistributionList)-1].sumPrefix + 1) - fnName := findCeil(randomNumber, fnWtDistributionList) - - return fnMetadatamap[fnName] + return fnMap[fnName] } // getProxyErrorHandler returns a reverse proxy error handler @@ -574,28 +572,30 @@ func (fh *functionHandler) getServiceEntry() (serviceUrl *url.URL, serviceUrlFro ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() + fnMeta := &fh.function.Metadata + // 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), + 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", fh.function.Name)) + 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", fh.function.Name)) + 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", fh.function.Name)) - fh.fmap.assign(fh.function, u) + zap.String("function_name", fnMeta.Name)) + fh.fmap.assign(fnMeta, u) } else { u, err = fh.getServiceEntryFromCache() if err != nil { @@ -613,9 +613,9 @@ func (fh *functionHandler) getServiceEntry() (serviceUrl *url.URL, serviceUrlFro e := "error updating service address entry for function" fh.logger.Error(e, zap.Error(err), - zap.String("function_name", fh.function.Name), - zap.String("function_namespace", fh.function.Namespace)) - return nil, false, errors.Wrapf(err, "%s %s_%s", e, fh.function.Name, fh.function.Namespace) + 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) } record, ok := recordObj.(svcEntryRecord) @@ -629,7 +629,7 @@ func (fh *functionHandler) getServiceEntry() (serviceUrl *url.URL, serviceUrlFro // 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) + serviceUrl, err = fh.fmap.lookup(&fh.function.Metadata) if err != nil { var errMsg string @@ -642,7 +642,7 @@ func (fh *functionHandler) getServiceEntryFromCache() (serviceUrl *url.URL, err if e.Code == ferror.ErrorNotFound { return nil, nil } - errMsg = fmt.Sprintf("Error getting function %v;s service entry from cache: %v", fh.function.Name, err) + errMsg = fmt.Sprintf("Error getting function %v;s service entry from cache: %v", fh.function.Metadata.Name, err) } return nil, ferror.MakeError(http.StatusInternalServerError, errMsg) } @@ -652,7 +652,7 @@ func (fh *functionHandler) getServiceEntryFromCache() (serviceUrl *url.URL, err // getServiceEntryFromExecutor returns service url entry returns from executor func (fh *functionHandler) getServiceEntryFromExecutor(ctx context.Context) (*url.URL, error) { // send a request to executor to specialize a new pod - service, err := fh.executor.GetServiceForFunction(ctx, fh.function) + service, err := fh.executor.GetServiceForFunction(ctx, &fh.function.Metadata) if err != nil { statusCode, errMsg := ferror.GetHTTPError(err) fh.logger.Error("error from GetServiceForFunction", diff --git a/pkg/router/functionHandler_test.go b/pkg/router/functionHandler_test.go index 9826e634..a6ce0697 100644 --- a/pkg/router/functionHandler_test.go +++ b/pkg/router/functionHandler_test.go @@ -57,12 +57,12 @@ func TestFunctionProxying(t *testing.T) { backendURL := createBackendService(testResponseString) log.Printf("Created backend svc at %v", backendURL) - fn := &metav1.ObjectMeta{Name: "foo", Namespace: metav1.NamespaceDefault} + fnMeta := metav1.ObjectMeta{Name: "foo", Namespace: metav1.NamespaceDefault} logger, err := zap.NewDevelopment() panicIf(err) fmap := makeFunctionServiceMap(logger, 0) - fmap.assign(fn, backendURL) + fmap.assign(&fnMeta, backendURL) httpTrigger := &fv1.HTTPTrigger{ Metadata: metav1.ObjectMeta{ @@ -78,9 +78,11 @@ func TestFunctionProxying(t *testing.T) { } fh := &functionHandler{ - logger: logger, - fmap: fmap, - function: fn, + logger: logger, + fmap: fmap, + function: &fv1.Function{ + Metadata: metav1.ObjectMeta{Name: "foo", Namespace: metav1.NamespaceDefault}, + }, tsRoundTripperParams: &tsRoundTripperParams{ timeout: 50 * time.Millisecond, timeoutExponent: 2, diff --git a/pkg/router/functionReferenceResolver.go b/pkg/router/functionReferenceResolver.go index 7a760828..eb41913d 100644 --- a/pkg/router/functionReferenceResolver.go +++ b/pkg/router/functionReferenceResolver.go @@ -50,7 +50,7 @@ type ( // a distribution of requests across two functions. resolveResult struct { resolveResultType - functionMetadataMap map[string]*metav1.ObjectMeta + functionMap map[string]*fv1.Function functionWtDistributionList []FunctionWeightDistribution } @@ -134,12 +134,13 @@ func (frr *functionReferenceResolver) resolveByName(namespace, name string) (*re } f := obj.(*fv1.Function) - functionMetadataMap := make(map[string]*metav1.ObjectMeta, 1) - functionMetadataMap[f.Metadata.Name] = &f.Metadata + functionMap := map[string]*fv1.Function{ + f.Metadata.Name: f, + } rr := resolveResult{ - resolveResultType: resolveResultSingleFunction, - functionMetadataMap: functionMetadataMap, + resolveResultType: resolveResultSingleFunction, + functionMap: functionMap, } return &rr, nil @@ -147,7 +148,7 @@ func (frr *functionReferenceResolver) resolveByName(namespace, name string) (*re func (frr *functionReferenceResolver) resolveByFunctionWeights(namespace string, fr *fv1.FunctionReference) (*resolveResult, error) { - functionMetadataMap := make(map[string]*metav1.ObjectMeta) + functionMap := make(map[string]*fv1.Function) fnWtDistrList := make([]FunctionWeightDistribution, 0) sumPrefix := 0 @@ -167,19 +168,18 @@ func (frr *functionReferenceResolver) resolveByFunctionWeights(namespace string, } f := obj.(*fv1.Function) - functionMetadataMap[f.Metadata.Name] = &f.Metadata + functionMap[f.Metadata.Name] = f sumPrefix = sumPrefix + functionWeight fnWtDistrList = append(fnWtDistrList, FunctionWeightDistribution{ name: functionName, weight: functionWeight, sumPrefix: sumPrefix, }) - } rr := resolveResult{ resolveResultType: resolveResultMultipleFunctions, - functionMetadataMap: functionMetadataMap, + functionMap: functionMap, functionWtDistributionList: fnWtDistrList, } diff --git a/pkg/router/httpTriggers.go b/pkg/router/httpTriggers.go index 1bcdb6f5..1d5a7569 100644 --- a/pkg/router/httpTriggers.go +++ b/pkg/router/httpTriggers.go @@ -142,7 +142,7 @@ func (ts *HTTPTriggerSet) getRouter(fnTimeoutMap map[types.UID]int) *mux.Router fmap: ts.functionServiceMap, executor: ts.executor, httpTrigger: &trigger, - functionMetadataMap: rr.functionMetadataMap, + functionMap: rr.functionMap, fnWeightDistributionList: rr.functionWtDistributionList, tsRoundTripperParams: ts.tsRoundTripperParams, isDebugEnv: ts.isDebugEnv, @@ -158,8 +158,8 @@ func (ts *HTTPTriggerSet) getRouter(fnTimeoutMap map[types.UID]int) *mux.Router // deployment. For more details, please check "handler" function of functionHandler. if rr.resolveResultType == resolveResultSingleFunction { - for _, metadata := range fh.functionMetadataMap { - fh.function = metadata + for _, fn := range fh.functionMap { + fh.function = fn } } @@ -185,20 +185,19 @@ func (ts *HTTPTriggerSet) getRouter(fnTimeoutMap map[types.UID]int) *mux.Router // Internal triggers for each function by name. Non-http // triggers route into these. - for _, function := range ts.functions { - m := function.Metadata - + for i := range ts.functions { + fn := ts.functions[i] fh := &functionHandler{ - logger: ts.logger.Named(m.Name), + logger: ts.logger.Named(fn.Metadata.Name), fmap: ts.functionServiceMap, - function: &m, + function: &fn, executor: ts.executor, tsRoundTripperParams: ts.tsRoundTripperParams, isDebugEnv: ts.isDebugEnv, svcAddrUpdateThrottler: ts.svcAddrUpdateThrottler, functionTimeoutMap: fnTimeoutMap, } - muxRouter.HandleFunc(utils.UrlForFunction(function.Metadata.Name, function.Metadata.Namespace), fh.handler) + muxRouter.HandleFunc(utils.UrlForFunction(fn.Metadata.Name, fn.Metadata.Namespace), fh.handler) } // Healthz endpoint for the router. @@ -263,8 +262,8 @@ func (ts *HTTPTriggerSet) initFunctionController() (k8sCache.Store, k8sCache.Con // update resolver function reference cache for key, rr := range ts.resolver.copy() { if key.namespace == fn.Metadata.Namespace && - rr.functionMetadataMap[fn.Metadata.Name] != nil && - rr.functionMetadataMap[fn.Metadata.Name].ResourceVersion != fn.Metadata.ResourceVersion { + rr.functionMap[fn.Metadata.Name] != nil && + rr.functionMap[fn.Metadata.Name].Metadata.ResourceVersion != fn.Metadata.ResourceVersion { // invalidate resolver cache ts.logger.Debug("invalidating resolver cache") err := ts.resolver.delete(key.namespace, key.triggerName, key.triggerResourceVersion) diff --git a/pkg/router/router_test.go b/pkg/router/router_test.go index 00f8d4f6..87889e17 100644 --- a/pkg/router/router_test.go +++ b/pkg/router/router_test.go @@ -32,12 +32,12 @@ import ( func TestRouter(t *testing.T) { // metadata for a fake function - fn := &metav1.ObjectMeta{Name: "foo", Namespace: metav1.NamespaceDefault} + fnMeta := metav1.ObjectMeta{Name: "foo", Namespace: metav1.NamespaceDefault} // and a reference to it fr := fv1.FunctionReference{ Type: types.FunctionReferenceTypeFunctionName, - Name: fn.Name, + Name: fnMeta.Name, } // start a fake service @@ -49,7 +49,7 @@ func TestRouter(t *testing.T) { // set up the cache with this fake service fmap := makeFunctionServiceMap(logger, 0) - fmap.assign(fn, testServiceUrl) + fmap.assign(&fnMeta, testServiceUrl) // HTTP trigger set with a trigger for this function triggers, _, _ := makeHTTPTriggerSet(logger, fmap, nil, nil, nil, nil, @@ -81,12 +81,14 @@ func TestRouter(t *testing.T) { triggerResourceVersion: "1234", } - fnMetaMap := make(map[string]*metav1.ObjectMeta, 1) - fnMetaMap[fn.Name] = fn + fnMetaMap := make(map[string]*fv1.Function, 1) + fnMetaMap[fnMeta.Name] = &fv1.Function{ + Metadata: fnMeta, + } rr := resolveResult{ - resolveResultType: resolveResultSingleFunction, - functionMetadataMap: fnMetaMap, + resolveResultType: resolveResultSingleFunction, + functionMap: fnMetaMap, } frr.refCache.Set(nfr, rr)