diff --git a/common.go b/common.go index f8229fba..b7f72435 100644 --- a/common.go +++ b/common.go @@ -87,3 +87,35 @@ func MergeContainerSpecs(specs ...*apiv1.Container) apiv1.Container { } return *result } + +// IsNetworkDialError returns true if its a network dial error +func IsNetworkDialError(err error) bool { + netErr, ok := err.(net.Error) + if !ok { + return false + } + netOpErr, ok := netErr.(*net.OpError) + if !ok { + return false + } + if netOpErr.Op == "dial" { + return true + } + return false +} + +// IsReadyPod checks that all containers in a pod are ready and returns true if so +func IsReadyPod(pod *apiv1.Pod) bool { + // since its a utility function, just ensuring there is no nil pointer exception + if pod == nil { + return false + } + + for _, cStatus := range pod.Status.ContainerStatuses { + if !cStatus.Ready { + return false + } + } + + return true +} diff --git a/executor/api.go b/executor/api.go index 7b3d05df..23631e10 100644 --- a/executor/api.go +++ b/executor/api.go @@ -57,13 +57,27 @@ func (executor *Executor) getServiceForFunctionApi(w http.ResponseWriter, r *htt w.Write([]byte(serviceName)) } +// getServiceForFunction first checks if this function's service is cached, if yes, it validates the address. +// if it's a valid address, just returns it. +// else, invalidates its cache entry and makes a new request to create a service for this function and finally responds +// with new address or error. +// +// checking for the validity of the address causes a little more over-head than desired. but, it ensures that +// stale addresses are not returned to the router. +// 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(m *metav1.ObjectMeta) (string, error) { // Check function -> svc cache log.Printf("[%v] Checking for cached function service", m.Name) fsvc, err := executor.fsCache.GetByFunction(m) if err == nil { - // Cached, return svc address - return fsvc.Address, nil + if executor.isValidAddress(fsvc) { + // Cached, return svc address + return fsvc.Address, nil + } else { + log.Printf("[%v] Deleting cache entry for invalid address : %s", m.Name, fsvc.Address) + executor.fsCache.DeleteEntry(fsvc) + } } respChan := make(chan *createFuncServiceResponse) diff --git a/executor/executor.go b/executor/executor.go index fced4e8d..a476f45d 100644 --- a/executor/executor.go +++ b/executor/executor.go @@ -119,6 +119,15 @@ func (executor *Executor) serveCreateFuncServices() { } } +func (executor *Executor) getFunctionExecutorType(meta *metav1.ObjectMeta) (fission.ExecutorType, error) { + fn, err := executor.fissionClient.Functions(meta.Namespace).Get(meta.Name) + if err != nil { + return "", err + } + return fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType, nil + +} + func (executor *Executor) createServiceForFunction(meta *metav1.ObjectMeta) (*fscache.FuncSvc, error) { log.Printf("[%v] No cached function service found, creating one", meta.Name) @@ -129,14 +138,12 @@ func (executor *Executor) createServiceForFunction(meta *metav1.ObjectMeta) (*fs return nil, err } - fn, err := executor.fissionClient. - Functions(meta.Namespace). - Get(meta.Name) + executorType, err := executor.getFunctionExecutorType(meta) if err != nil { return nil, err } - switch fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType { + switch executorType { case fission.ExecutorTypeNewdeploy: fs, err := executor.ndm.GetFuncSvc(meta) return fs, err @@ -182,6 +189,15 @@ func (executor *Executor) getFunctionEnv(m *metav1.ObjectMeta) (*crd.Environment return env, nil } +// isValidAddress invokes isValidService or isValidPod depending on the type of executor +func (executor *Executor) isValidAddress(fsvc *fscache.FuncSvc) bool { + if fsvc.Executor == fscache.NEWDEPLOY { + return executor.ndm.IsValidService(fsvc.Address) + } else { + return executor.gpm.IsValidPod(fsvc.KubernetesObjects, fsvc.Address) + } +} + func dumpStackTrace() { debug.PrintStack() } diff --git a/executor/fscache/functionServiceCache.go b/executor/fscache/functionServiceCache.go index 8bab0ff2..91183943 100644 --- a/executor/fscache/functionServiceCache.go +++ b/executor/fscache/functionServiceCache.go @@ -247,14 +247,18 @@ func (fsc *FunctionServiceCache) _touchByAddress(address string) error { return nil } +func (fsc *FunctionServiceCache) DeleteEntry(fsvc *FuncSvc) { + fsc.byFunction.Delete(crd.CacheKey(fsvc.Function)) + fsc.byAddress.Delete(fsvc.Address) + fsc.byFunctionUID.Delete(fsvc.Function.UID) +} + func (fsc *FunctionServiceCache) DeleteOld(fsvc *FuncSvc, minAge time.Duration) (bool, error) { if time.Since(fsvc.Atime) < minAge { return false, nil } - fsc.byFunction.Delete(crd.CacheKey(fsvc.Function)) - fsc.byAddress.Delete(fsvc.Address) - fsc.byFunctionUID.Delete(fsvc.Function.UID) + fsc.DeleteEntry(fsvc) return true, nil } diff --git a/executor/newdeploy/newdeploymgr.go b/executor/newdeploy/newdeploymgr.go index 541877ee..92d9bb08 100644 --- a/executor/newdeploy/newdeploymgr.go +++ b/executor/newdeploy/newdeploymgr.go @@ -22,6 +22,7 @@ import ( "log" "os" "strconv" + "strings" "time" "github.com/pkg/errors" @@ -553,3 +554,18 @@ func (deploy *NewDeploy) updateKubeObjRefRV(fsvc *fscache.FuncSvc, objKind strin func updateStatus(fn *crd.Function, err error, message string) { log.Printf(message, err) } + +// IsValidService does a get on the service address to ensure it's a valid service. returns true if it is, else false. +func (deploy *NewDeploy) IsValidService(svc string) bool { + service := strings.Split(svc, ".") + if len(service) == 0 { + return false + } + svcObj, err := deploy.kubernetesClient.CoreV1().Services(service[1]).Get(service[0], metav1.GetOptions{}) + if err == nil { + log.Printf("Valid service address : %s", svcObj.Spec.ClusterIP) + return true + } + + return false +} diff --git a/executor/poolmgr/gpm.go b/executor/poolmgr/gpm.go index fdd95e9c..8f14ef20 100644 --- a/executor/poolmgr/gpm.go +++ b/executor/poolmgr/gpm.go @@ -21,10 +21,12 @@ import ( "log" "os" "strconv" + "strings" "time" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes" + "k8s.io/client-go/pkg/api" k8sCache "k8s.io/client-go/tools/cache" "github.com/fission/fission" @@ -217,3 +219,18 @@ func (gpm *GenericPoolManager) getEnvPoolsize(env *crd.Environment) int32 { } return poolsize } + +// IsValidPod checks if pod is not deleted and that it has the address passed as the argument. Also checks that all the +// containers in it are reporting a ready status for the healthCheck. +func (gpm *GenericPoolManager) IsValidPod(kubeObjects []api.ObjectReference, podAddress string) bool { + for _, obj := range kubeObjects { + if obj.Kind == "pod" { + pod, err := gpm.kubernetesClient.CoreV1().Pods(obj.Namespace).Get(obj.Name, metav1.GetOptions{}) + if err == nil && strings.Contains(podAddress, pod.Status.PodIP) && fission.IsReadyPod(pod) { + log.Printf("Valid pod address : %s", podAddress) + return true + } + } + } + return false +} diff --git a/router/functionHandler.go b/router/functionHandler.go index 2e26148d..271f49fb 100644 --- a/router/functionHandler.go +++ b/router/functionHandler.go @@ -28,6 +28,7 @@ import ( "github.com/gorilla/mux" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "github.com/fission/fission" executorClient "github.com/fission/fission/executor/client" ) @@ -37,45 +38,139 @@ type functionHandler struct { function *metav1.ObjectMeta } -func (fh *functionHandler) getServiceForFunction() (*url.URL, error) { - // call executor, get a url for a function - svcName, err := fh.executor.GetServiceForFunction(fh.function) - if err != nil { - return nil, err - } - svcUrl, err := url.Parse(fmt.Sprintf("http://%v", svcName)) - if err != nil { - return nil, err - } - return svcUrl, nil -} - // A layer on top of http.DefaultTransport, with retries. type RetryingRoundTripper struct { - maxRetries int - initalTimeout time.Duration + maxRetries int + initialTimeout time.Duration + funcHandler *functionHandler } -func (rrt RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { - timeout := rrt.initalTimeout - transport := http.DefaultTransport.(*http.Transport) +// RoundTrip is a custom transport with retries for http requests that forwards the request to the right serviceUrl, obtained +// from router's cache or from executor if router entry is stale. +// +// It first checks if the service address for this function came from router's cache. +// If it didn't, it makes a request to executor to get a new service for function. If that succeeds, it adds the address +// to it's cache and makes a request to that address with transport.RoundTrip call. +// Initial requests to new k8s services sometimes seem to fail, but retries work. So, it retries with an exponential +// back-off for maxRetries times. +// +// Else if it came from the cache, it makes a transport.RoundTrip with that cached address. If the response received is +// a network dial error (which means that the pod doesn't exist anymore), it removes the cache entry and makes a request +// to executor to get a new service for function. It then retries transport.RoundTrip with the new address. +// +// At any point in time, if the response received from transport.RoundTrip is other than dial network error, it is +// relayed as-is to the user, without any retries. +// +// While this RoundTripper handles the case where a previously cached address of the function pod isn't valid anymore +// (probably because the pod got deleted somehow), by making a request to executor to get a new service for this function, +// it doesn't handle a case where a newly specialized pod gets deleted just after the GetServiceForFunction succeeds. +// In such a case, the RoundTripper will retry requests against the new address and give up after maxRetries. +// However, the subsequent http call for this function will ensure the cache is invalidated. +// +// If GetServiceForFunction returns an error or if RoundTripper exits with an error, it get's translated into 502 +// inside ServeHttp function of the reverseProxy. +// 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 serviceUrl *url.URL - // Do max-1 retries; the last one uses default transport timeouts - for i := rrt.maxRetries - 1; i > 0; i-- { - // update timeout in transport + // set the timeout for transport context + timeout := roundTripper.initialTimeout + transport := http.DefaultTransport.(*http.Transport) + defer transport.CloseIdleConnections() + + // 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 + needExecutor = true + } + + for i := 0; i < roundTripper.maxRetries-1; i++ { + if needExecutor { + log.Printf("Calling getServiceForFunction for function: %s", roundTripper.funcHandler.function.Name) + + // send a request to executor to specialize a new pod + service, err := roundTripper.funcHandler.executor.GetServiceForFunction( + roundTripper.funcHandler.function) + if err != nil { + // We might want a specific error code or header for fission failures as opposed to + // user function bugs. + return nil, err + } + + // parse the address into url + serviceUrl, err = url.Parse(fmt.Sprintf("http://%v", service)) + if err != nil { + return nil, err + } + + // add the address in router's cache + 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 + } + + // 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: timeout, KeepAlive: 30 * time.Second, }).DialContext - resp, err := transport.RoundTrip(req) + // forward the request to the function service + resp, err = transport.RoundTrip(req) if err == nil { + // if transport.RoundTrip succeeds and it was a cached entry, then tapService + if !serviceUrlFromExecutor { + go roundTripper.funcHandler.tapService(serviceUrl) + } + // return response back to user return resp, nil } - timeout *= time.Duration(2) - log.Printf("Retrying request to %v in %v", req.URL.Host, timeout) - time.Sleep(timeout) + // 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, timeout) + timeout *= time.Duration(2) + time.Sleep(timeout) + 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 + } } // finally, one more retry with the default timeout @@ -90,8 +185,6 @@ func (fh *functionHandler) tapService(serviceUrl *url.URL) { } func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request *http.Request) { - reqStartTime := time.Now() - // retrieve url params and add them to request header vars := mux.Vars(request) for k, v := range vars { @@ -101,71 +194,23 @@ func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request * // System Params MetadataToHeaders(HEADERS_FISSION_FUNCTION_PREFIX, fh.function, request) - // cache lookup - serviceUrl, err := fh.fmap.lookup(fh.function) - if err != nil { - // Cache miss: request the Pool Manager to make a new service. - log.Printf("Not cached, getting new service for %v", fh.function) - - var poolErr error - serviceUrl, poolErr = fh.getServiceForFunction() - if poolErr != nil { - log.Printf("Failed to get service for function %v: %v", fh.function.Name, poolErr) - // We might want a specific error code or header for fission - // failures as opposed to user function bugs. - http.Error(responseWriter, "Internal server error (fission)", 500) - return - } - - // add it to the map - fh.fmap.assign(fh.function, serviceUrl) - } else { - // if we're using our cache, asynchronously tell - // executor we're using this service - go fh.tapService(serviceUrl) - } - - // Proxy off our request to the serviceUrl, and send the response back. // TODO: As an optimization we may want to cache proxies too -- this might get us // connection reuse and possibly better performance director := func(req *http.Request) { - log.Printf("Proxying request for %v to %v", req.URL, serviceUrl.Host) - - // send this request to serviceurl - 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. - 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 - - // leave the query string intact (req.URL.RawQuery) - if _, ok := req.Header["User-Agent"]; !ok { // explicitly disable User-Agent so it's not set to default value req.Header.Set("User-Agent", "") } } - // Initial requests to new k8s services sometimes seem to - // fail, but retries work. So use a transport that does retries. proxy := &httputil.ReverseProxy{ Director: director, - Transport: RetryingRoundTripper{ - maxRetries: 10, - initalTimeout: 50 * time.Millisecond, + Transport: &RetryingRoundTripper{ + initialTimeout: 50 * time.Millisecond, + maxRetries: 10, + funcHandler: fh, }, } - delay := time.Since(reqStartTime) - if delay > 100*time.Millisecond { - log.Printf("Request delay for %v: %v", serviceUrl, delay) - } + proxy.ServeHTTP(responseWriter, request) } diff --git a/router/functionHandler_test.go b/router/functionHandler_test.go index 18a4d95b..5b1962ce 100644 --- a/router/functionHandler_test.go +++ b/router/functionHandler_test.go @@ -54,7 +54,9 @@ func TestFunctionProxying(t *testing.T) { fmap := makeFunctionServiceMap(0) fmap.assign(fn, backendURL) - fh := &functionHandler{fmap: fmap, function: fn} + fh := &functionHandler{fmap: fmap, + function: fn, + } functionHandlerServer := httptest.NewServer(http.HandlerFunc(fh.handler)) fhURL := functionHandlerServer.URL diff --git a/router/functionServiceMap.go b/router/functionServiceMap.go index 9871d2a7..74d47b14 100644 --- a/router/functionServiceMap.go +++ b/router/functionServiceMap.go @@ -75,3 +75,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) +} diff --git a/test/tests/test_router_cache_invalidation.sh b/test/tests/test_router_cache_invalidation.sh new file mode 100755 index 00000000..080c4927 --- /dev/null +++ b/test/tests/test_router_cache_invalidation.sh @@ -0,0 +1,61 @@ +#!/bin/bash + +set -euo pipefail + +# 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. + +PYTHON_RUNTIME_IMAGE=gcr.io/fission-ci/python-env:test +fn=python-func-$(date +%s) + +log "Pre-test cleanup" +fission env delete --name python || true + +log "Creating python env" +fission env create --name python --image $PYTHON_RUNTIME_IMAGE +trap "fission env delete --name python" EXIT + +log "Creating hello.py" +mkdir testDir-$fn +printf 'def main():\n return "Hello, world!"' > testDir-$fn/hello.py +trap "rm -rf testDir-$fn" EXIT + +log "Creating function " $fn +fission fn create --name $fn --env python --code testDir-$fn/hello.py +trap "fission fn delete --name $fn" EXIT + +log "rm testDir-$fn" +rm -rf testDir-$fn + +log "Waiting for router to update cache" +sleep 3 + +log "Creating route" +fission route create --function $fn --url /$fn --method GET + +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 \ No newline at end of file