diff --git a/charts/fission-all/templates/deployment.yaml b/charts/fission-all/templates/deployment.yaml index a9c7260a..94e2de98 100644 --- a/charts/fission-all/templates/deployment.yaml +++ b/charts/fission-all/templates/deployment.yaml @@ -175,6 +175,14 @@ spec: valueFrom: fieldRef: fieldPath: metadata.namespace + - name: ROUTER_ROUND_TRIP_TIMEOUT + value: {{ .Values.routerRoundTripTimeout | default "50ms" | quote }} + - name: ROUTER_ROUNDTRIP_TIMEOUT_EXPONENT + value: {{ .Values.routerRoundTripTimeoutExponent | default 2 | quote }} + - name: ROUTER_ROUND_TRIP_KEEP_ALIVE_TIME + value: {{ .Values.routerRoundTripKeepAliveTime | default "30s" | quote }} + - name: ROUTER_ROUND_TRIP_MAX_RETRIES + value: {{ .Values.routerRoundTripMaxRetries | default 10 | quote }} readinessProbe: httpGet: path: "/router-healthz" diff --git a/charts/fission-core/templates/deployment.yaml b/charts/fission-core/templates/deployment.yaml index f9a2c60c..d351271e 100644 --- a/charts/fission-core/templates/deployment.yaml +++ b/charts/fission-core/templates/deployment.yaml @@ -172,6 +172,14 @@ spec: valueFrom: fieldRef: fieldPath: metadata.namespace + - name: ROUTER_ROUND_TRIP_TIMEOUT + value: {{ .Values.routerRoundTripTimeout | default "50ms" | quote }} + - name: ROUTER_ROUNDTRIP_TIMEOUT_EXPONENT + value: {{ .Values.routerRoundTripTimeoutExponent | default 2 | quote }} + - name: ROUTER_ROUND_TRIP_KEEP_ALIVE_TIME + value: {{ .Values.routerRoundTripKeepAliveTime | default "30s" | quote }} + - name: ROUTER_ROUND_TRIP_MAX_RETRIES + value: {{ .Values.routerRoundTripMaxRetries | default 10 | quote }} readinessProbe: httpGet: path: "/router-healthz" diff --git a/router/functionHandler.go b/router/functionHandler.go index 68763c69..b08b0426 100644 --- a/router/functionHandler.go +++ b/router/functionHandler.go @@ -33,18 +33,24 @@ import ( executorClient "github.com/fission/fission/executor/client" ) +type tsRoundTripperParams struct { + timeout time.Duration + timeoutExponent int + keepAlive time.Duration + maxRetries int +} + type functionHandler struct { - fmap *functionServiceMap - executor *executorClient.Client - function *metav1.ObjectMeta - httpTrigger *crd.HTTPTrigger + fmap *functionServiceMap + executor *executorClient.Client + function *metav1.ObjectMeta + httpTrigger *crd.HTTPTrigger + tsRoundTripperParams *tsRoundTripperParams } // A layer on top of http.DefaultTransport, with retries. type RetryingRoundTripper struct { - maxRetries int - initialTimeout time.Duration - funcHandler *functionHandler + funcHandler *functionHandler } // RoundTrip is a custom transport with retries for http requests that forwards the request to the right serviceUrl, obtained @@ -92,7 +98,6 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt } // set the timeout for transport context - timeout := roundTripper.initialTimeout transport := http.DefaultTransport.(*http.Transport) // Disables caching, Please refer to issue and specifically comment: https://github.com/fission/fission/issues/723#issuecomment-398781995 transport.DisableKeepAlives = true @@ -104,7 +109,9 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt needExecutor = true } - for i := 0; i < roundTripper.maxRetries-1; i++ { + executingTimeout := roundTripper.funcHandler.tsRoundTripperParams.timeout + + for i := 0; i < roundTripper.funcHandler.tsRoundTripperParams.maxRetries-1; i++ { if needExecutor { log.Printf("Calling getServiceForFunction for function: %s", roundTripper.funcHandler.function.Name) @@ -149,8 +156,8 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt // over-riding default settings. transport.DialContext = (&net.Dialer{ - Timeout: timeout, - KeepAlive: 30 * time.Second, + Timeout: executingTimeout, + KeepAlive: roundTripper.funcHandler.tsRoundTripperParams.keepAlive, }).DialContext overhead := time.Since(startTime) @@ -183,9 +190,11 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt // 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) + req.URL.Host, executingTimeout) + + executingTimeout = executingTimeout * time.Duration(roundTripper.funcHandler.tsRoundTripperParams.timeoutExponent) + time.Sleep(executingTimeout) + needExecutor = false continue } else { @@ -231,9 +240,7 @@ func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request * proxy := &httputil.ReverseProxy{ Director: director, Transport: &RetryingRoundTripper{ - initialTimeout: 50 * time.Millisecond, - maxRetries: 10, - funcHandler: fh, + funcHandler: fh, }, } diff --git a/router/functionHandler_test.go b/router/functionHandler_test.go index 5b1962ce..5f4f6a14 100644 --- a/router/functionHandler_test.go +++ b/router/functionHandler_test.go @@ -22,6 +22,7 @@ import ( "net/http/httptest" "net/url" "testing" + "time" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) @@ -56,6 +57,12 @@ func TestFunctionProxying(t *testing.T) { fh := &functionHandler{fmap: fmap, function: fn, + tsRoundTripperParams: &tsRoundTripperParams{ + timeout: 50 * time.Millisecond, + timeoutExponent: 2, + keepAlive: 30 * time.Second, + maxRetries: 10, + }, } functionHandlerServer := httptest.NewServer(http.HandlerFunc(fh.handler)) fhURL := functionHandlerServer.URL diff --git a/router/httpTriggers.go b/router/httpTriggers.go index 5a80fa0c..51bba411 100644 --- a/router/httpTriggers.go +++ b/router/httpTriggers.go @@ -49,17 +49,20 @@ type HTTPTriggerSet struct { functions []crd.Function funcStore k8sCache.Store funcController k8sCache.Controller + + tsRoundTripperParams *tsRoundTripperParams } func makeHTTPTriggerSet(fmap *functionServiceMap, fissionClient *crd.FissionClient, - kubeClient *kubernetes.Clientset, executor *executorClient.Client, crdClient *rest.RESTClient) (*HTTPTriggerSet, k8sCache.Store, k8sCache.Store) { + kubeClient *kubernetes.Clientset, executor *executorClient.Client, crdClient *rest.RESTClient, params *tsRoundTripperParams) (*HTTPTriggerSet, k8sCache.Store, k8sCache.Store) { httpTriggerSet := &HTTPTriggerSet{ - functionServiceMap: fmap, - triggers: []crd.HTTPTrigger{}, - fissionClient: fissionClient, - kubeClient: kubeClient, - executor: executor, - crdClient: crdClient, + functionServiceMap: fmap, + triggers: []crd.HTTPTrigger{}, + fissionClient: fissionClient, + kubeClient: kubeClient, + executor: executor, + crdClient: crdClient, + tsRoundTripperParams: params, } var tStore, fnStore k8sCache.Store var tController, fnController k8sCache.Controller @@ -121,10 +124,11 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router { } fh := &functionHandler{ - fmap: ts.functionServiceMap, - function: rr.functionMetadata, - executor: ts.executor, - httpTrigger: &trigger, + fmap: ts.functionServiceMap, + function: rr.functionMetadata, + executor: ts.executor, + httpTrigger: &trigger, + tsRoundTripperParams: ts.tsRoundTripperParams, } ht := muxRouter.HandleFunc(trigger.Spec.RelativeURL, fh.handler) @@ -152,9 +156,10 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router { for _, function := range ts.functions { m := function.Metadata fh := &functionHandler{ - fmap: ts.functionServiceMap, - function: &m, - executor: ts.executor, + fmap: ts.functionServiceMap, + function: &m, + executor: ts.executor, + tsRoundTripperParams: ts.tsRoundTripperParams, } muxRouter.HandleFunc(fission.UrlForFunction(function.Metadata.Name, function.Metadata.Namespace), fh.handler) } diff --git a/router/router.go b/router/router.go index 9ea146b7..19ade766 100644 --- a/router/router.go +++ b/router/router.go @@ -44,6 +44,8 @@ import ( "fmt" "log" "net/http" + "os" + "strconv" "time" "github.com/gorilla/mux" @@ -97,7 +99,34 @@ func Start(port int, executorUrl string) { restClient := fissionClient.GetCrdClient() executor := executorClient.MakeClient(executorUrl) - triggers, _, fnStore := makeHTTPTriggerSet(fmap, fissionClient, kubeClient, executor, restClient) + + timeout, err := time.ParseDuration(os.Getenv("ROUTER_ROUND_TRIP_TIMEOUT")) + if err != nil { + log.Fatalf("Failed to parse timeout: %v", err) + } + + timeoutExponent, err := strconv.Atoi(os.Getenv("ROUTER_ROUNDTRIP_TIMEOUT_EXPONENT")) + if err != nil { + log.Fatalf("Failed to parse timeout exponent: %v", err) + } + + keepAlive, err := time.ParseDuration(os.Getenv("ROUTER_ROUND_TRIP_KEEP_ALIVE_TIME")) + if err != nil { + log.Fatalf("Failed to parse keep alive time: %v", err) + } + + maxRetries, err := strconv.Atoi(os.Getenv("ROUTER_ROUND_TRIP_MAX_RETRIES")) + if err != nil { + log.Fatalf("Failed to parse max retry times: %v", err) + } + + triggers, _, fnStore := makeHTTPTriggerSet(fmap, fissionClient, kubeClient, executor, restClient, + &tsRoundTripperParams{ + timeout: timeout, + timeoutExponent: timeoutExponent, + keepAlive: keepAlive, + maxRetries: maxRetries, + }) resolver := makeFunctionReferenceResolver(fnStore) go serveMetric() diff --git a/router/router_test.go b/router/router_test.go index 9a6c61d9..23068903 100644 --- a/router/router_test.go +++ b/router/router_test.go @@ -59,7 +59,13 @@ func TestRouter(t *testing.T) { frr.refCache.Set(nfr, rr) // HTTP trigger set with a trigger for this function - triggers, _, _ := makeHTTPTriggerSet(fmap, nil, nil, nil, nil) + triggers, _, _ := makeHTTPTriggerSet(fmap, nil, nil, nil, nil, + &tsRoundTripperParams{ + timeout: 50 * time.Millisecond, + timeoutExponent: 2, + keepAlive: 30 * time.Second, + maxRetries: 10, + }) triggerUrl := "/foo" triggers.triggers = append(triggers.triggers, crd.HTTPTrigger{