From 404634eac68c6214f64b83ea761096d23fa3bffd Mon Sep 17 00:00:00 2001 From: xiekeyang Date: Fri, 13 Jul 2018 03:08:52 +0800 Subject: [PATCH] Round Tripper of Fission Router: parameters be configurable (#713) The Round Tripper parameters of timeout, keep alive time and Max retry times is configured in router system environment. And they are set to Round Tripper when router service initializing. This setup new nested structure `tsRoundTripperParams` to transfer them. --- charts/fission-all/templates/deployment.yaml | 8 ++++ charts/fission-core/templates/deployment.yaml | 8 ++++ router/functionHandler.go | 41 +++++++++++-------- router/functionHandler_test.go | 7 ++++ router/httpTriggers.go | 33 ++++++++------- router/router.go | 31 +++++++++++++- router/router_test.go | 8 +++- 7 files changed, 103 insertions(+), 33 deletions(-) diff --git a/charts/fission-all/templates/deployment.yaml b/charts/fission-all/templates/deployment.yaml index a9c7260a0..94e2de984 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 f9a2c60cc..d351271e0 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 68763c69a..b08b04263 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 5b1962ced..5f4f6a145 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 5a80fa0c7..51bba411a 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 9ea146b79..19ade766f 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 9a6c61d99..23068903c 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{