From 3707b95edb8e9da141d40db5f0c376a35b20f55a 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 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{