From 8b8e8c65390b066095bdab4f2c357f38f5f70e9c Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Tue, 1 Nov 2016 18:11:21 -0700 Subject: [PATCH 1/5] Trim server url in client init --- controller/client/client.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/controller/client/client.go b/controller/client/client.go index 61a84d39..e14c6886 100644 --- a/controller/client/client.go +++ b/controller/client/client.go @@ -24,6 +24,7 @@ import ( "fmt" "io/ioutil" "net/http" + "strings" log "github.com/Sirupsen/logrus" @@ -37,7 +38,7 @@ type ( ) func MakeClient(serverUrl string) *Client { - return &Client{Url: serverUrl} + return &Client{Url: strings.TrimSuffix(serverUrl, "/")} } func (c *Client) delete(relativeUrl string) error { From 1cf531038aea193f34880b40a43f254885002d16 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Tue, 1 Nov 2016 18:11:46 -0700 Subject: [PATCH 2/5] fission.Error constructor from http response --- error.go | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/error.go b/error.go index 564439b1..7f8f04e6 100644 --- a/error.go +++ b/error.go @@ -18,6 +18,7 @@ package fission import ( "fmt" + "net/http" ) func (e Error) Error() string { @@ -28,6 +29,25 @@ func MakeError(code int, msg string) Error { return Error{Code: errorCode(code), Message: msg} } +func MakeErrorFromHTTP(resp *http.Response) error { + if resp.StatusCode == 200 { + return nil + } + + var errCode int + switch resp.StatusCode { + case 403: + errCode = ErrorNotAuthorized + case 404: + errCode = ErrorNotFound + case 400: + errCode = ErrorInvalidArgument + default: + errCode = ErrorInternal + } + return MakeError(errCode, resp.Status) +} + func (err Error) HTTPStatus() int { var code int switch err.Code { From 6e86182e19b168fe7375ab0f07db09c6a780d36a Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Tue, 1 Nov 2016 18:14:23 -0700 Subject: [PATCH 3/5] Poolmgr API tweaks + convenient Start function --- poolmgr/api.go | 11 ++++---- poolmgr/poolmgr.go | 65 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 70 insertions(+), 6 deletions(-) create mode 100644 poolmgr/poolmgr.go diff --git a/poolmgr/api.go b/poolmgr/api.go index d457650d..7928168f 100644 --- a/poolmgr/api.go +++ b/poolmgr/api.go @@ -58,7 +58,7 @@ func MakeAPI(gpm *GenericPoolManager, controller *controllerclient.Client) *API } } -func (api *API) lookupApi(w http.ResponseWriter, r *http.Request) { +func (api *API) getServiceForFunctionApi(w http.ResponseWriter, r *http.Request) { body, err := ioutil.ReadAll(r.Body) if err != nil { http.Error(w, "Failed to read request", 500) @@ -73,15 +73,14 @@ func (api *API) lookupApi(w http.ResponseWriter, r *http.Request) { return } - serviceUrl, err := api.lookup(&m) + serviceName, err := api.getServiceForFunction(&m) if err != nil { code, msg := fission.GetHTTPError(err) log.Printf("Error: %v: %v", code, msg) http.Error(w, msg, code) } - // return serviceUrl - w.Write([]byte(serviceUrl)) + w.Write([]byte(serviceName)) } func (api *API) getFunctionEnv(m *fission.Metadata) (*fission.Environment, error) { @@ -112,7 +111,7 @@ func (api *API) getFunctionEnv(m *fission.Metadata) (*fission.Environment, error return env, nil } -func (api *API) lookup(m *fission.Metadata) (string, error) { +func (api *API) getServiceForFunction(m *fission.Metadata) (string, error) { // Check function -> svc map result, err := api.functionService.Get(m) if err == nil { @@ -153,7 +152,7 @@ func (api *API) lookup(m *fission.Metadata) (string, error) { func (api *API) Serve(port int) { r := mux.NewRouter() - r.HandleFunc("/v1/lookup", api.lookupApi).Methods("GET") + r.HandleFunc("/v1/getServiceForFunction", api.getServiceForFunctionApi).Methods("POST") address := fmt.Sprintf(":%v", port) log.Printf("starting poolmgr at port %v", port) diff --git a/poolmgr/poolmgr.go b/poolmgr/poolmgr.go new file mode 100644 index 00000000..6ac3ba4b --- /dev/null +++ b/poolmgr/poolmgr.go @@ -0,0 +1,65 @@ +/* +Copyright 2016 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package poolmgr + +import ( + "log" + "strings" + + "k8s.io/client-go/1.4/kubernetes" + "k8s.io/client-go/1.4/rest" + + controllerclient "github.com/platform9/fission/controller/client" +) + +// Get a kubernetes client using the pod's service account. This only +// works when we're running inside a kubernetes cluster. +func getKubernetesClient() (*kubernetes.Clientset, error) { + // creates the in-cluster config + config, err := rest.InClusterConfig() + if err != nil { + log.Printf("Error getting kubernetes client config: %v", err) + return nil, err + } + + // creates the clientset + clientset, err := kubernetes.NewForConfig(config) + if err != nil { + log.Printf("Error getting kubernetes client: %v", err) + return nil, err + } + + return clientset, nil +} + +func StartPoolmgr(controllerUrl string, namespace string, port int) error { + controllerUrl = strings.TrimSuffix(controllerUrl, "/") + controllerClient := controllerclient.MakeClient(controllerUrl) + + kubernetesClient, err := getKubernetesClient() + if err != nil { + log.Printf("Failed to get kubernetes client: %v", err) + return err + } + + gpm := MakeGenericPoolManager(controllerUrl, kubernetesClient, namespace) + + api := MakeAPI(gpm, controllerClient) + go api.Serve(port) + + return nil +} From 2b15b726d4ff07d21360cd9cf92426198ed0205e Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Tue, 1 Nov 2016 18:14:38 -0700 Subject: [PATCH 4/5] Poolmgr client --- poolmgr/client/client.go | 60 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 60 insertions(+) create mode 100644 poolmgr/client/client.go diff --git a/poolmgr/client/client.go b/poolmgr/client/client.go new file mode 100644 index 00000000..3556f059 --- /dev/null +++ b/poolmgr/client/client.go @@ -0,0 +1,60 @@ +/* +Copyright 2016 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package client + +import ( + "net/http" + "strings" + + "bytes" + "encoding/json" + "github.com/platform9/fission" + "io/ioutil" +) + +type Client struct { + poolmgrUrl string +} + +func MakeClient(poolmgrUrl string) *Client { + return &Client{poolmgrUrl: strings.TrimSuffix(poolmgrUrl, "/")} +} + +func (c *Client) GetServiceForFunction(metadata *fission.Metadata) (string, error) { + url := c.poolmgrUrl + "/v1/getServiceForFunction" + body, err := json.Marshal(metadata) + if err != nil { + return "", err + } + + resp, err := http.Post(url, "application/json", bytes.NewReader(body)) + if err != nil { + return "", err + } + defer resp.Body.Close() + + if resp.StatusCode != 200 { + return "", fission.MakeErrorFromHTTP(resp) + } + + svcName, err := ioutil.ReadAll(resp.Body) + if err != nil { + return "", err + } + + return string(svcName), nil +} From 3d5e52dc3d05a23303412b2ecbdfecfdf429aac3 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Tue, 1 Nov 2016 18:22:04 -0700 Subject: [PATCH 5/5] Router integration with poolmgr and controller Router uses poolmgr to specialize pods when necessary. Router uses controller to get the list of triggers to listen for. For now this integration is pretty crappy -- we just poll the controller every few seconds and cache the result. The right way would be to have some sort of watch API on the controller and use that. Or maybe share access to etcd directly. --- router/functionHandler.go | 27 ++++++++++++----- router/httpTriggers.go | 63 ++++++++++++++++++++++++++++----------- router/mutablemux.go | 2 +- router/router.go | 13 ++++++-- router/router_test.go | 2 +- 5 files changed, 79 insertions(+), 28 deletions(-) diff --git a/router/functionHandler.go b/router/functionHandler.go index 37f2cb6c..dd85a393 100644 --- a/router/functionHandler.go +++ b/router/functionHandler.go @@ -17,29 +17,40 @@ limitations under the License. package router import ( - "errors" + "fmt" "log" "net/http" "net/http/httputil" "net/url" "github.com/platform9/fission" + poolmgrClient "github.com/platform9/fission/poolmgr/client" ) type functionHandler struct { - fmap *functionServiceMap - poolManagerUrl string - Function fission.Metadata + fmap *functionServiceMap + poolmgr *poolmgrClient.Client + Function fission.Metadata } -func (*functionHandler) getServiceForFunction() (*url.URL, error) { - return nil, errors.New("not implemented") +func (fh *functionHandler) getServiceForFunction() (*url.URL, error) { + // call poolmgr, get a url for a function + svcName, err := fh.poolmgr.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 } func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request *http.Request) { 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) serviceUrl, poolErr := fh.getServiceForFunction() if poolErr != nil { log.Printf("Failed to get service for function (%v,%v): %v", @@ -55,9 +66,11 @@ func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request * } // Proxy off our request to the serviceUrl, and send the response back. - // TODO: As an optimization we may want to cache proxies too -- this would get us + // 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", req.URL) + // send this request to serviceurl req.URL.Scheme = serviceUrl.Scheme req.URL.Host = serviceUrl.Host diff --git a/router/httpTriggers.go b/router/httpTriggers.go index 5731ab5f..2ee95c9e 100644 --- a/router/httpTriggers.go +++ b/router/httpTriggers.go @@ -17,47 +17,76 @@ limitations under the License. package router import ( + "log" + "time" + "github.com/gorilla/mux" + "github.com/platform9/fission" + controllerClient "github.com/platform9/fission/controller/client" + poolmgrClient "github.com/platform9/fission/poolmgr/client" ) type HTTPTriggerSet struct { *functionServiceMap *mutableRouter - controllerUrl string - poolManagerUrl string - triggers []fission.HTTPTrigger + controller *controllerClient.Client + poolmgr *poolmgrClient.Client + triggers []fission.HTTPTrigger } -func makeHTTPTriggerSet(fmap *functionServiceMap, controllerUrl string, poolManagerUrl string) *HTTPTriggerSet { +func makeHTTPTriggerSet(fmap *functionServiceMap, controller *controllerClient.Client, poolmgr *poolmgrClient.Client) *HTTPTriggerSet { triggers := make([]fission.HTTPTrigger, 1) return &HTTPTriggerSet{ functionServiceMap: fmap, triggers: triggers, - controllerUrl: controllerUrl, - poolManagerUrl: poolManagerUrl, + controller: controller, + poolmgr: poolmgr, } } -func (triggers *HTTPTriggerSet) subscribeRouter(mr *mutableRouter) { - triggers.mutableRouter = mr - mr.updateRouter(triggers.getRouterFromTriggers()) - go triggers.watchTriggers() +func (ts *HTTPTriggerSet) subscribeRouter(mr *mutableRouter) { + ts.mutableRouter = mr + mr.updateRouter(ts.getRouterFromTriggers()) + go ts.watchTriggers() } -func (triggers *HTTPTriggerSet) getRouterFromTriggers() *mux.Router { +func (ts *HTTPTriggerSet) getRouterFromTriggers() *mux.Router { muxRouter := mux.NewRouter() - for _, trigger := range triggers.triggers { + for _, trigger := range ts.triggers { fh := &functionHandler{ - fmap: triggers.functionServiceMap, - Function: trigger.Function, - poolManagerUrl: triggers.poolManagerUrl, + fmap: ts.functionServiceMap, + Function: trigger.Function, + poolmgr: ts.poolmgr, } muxRouter.HandleFunc(trigger.UrlPattern, fh.handler) } return muxRouter } -func (triggers *HTTPTriggerSet) watchTriggers() { - // watch controller for updates to triggers and update the router accordingly +func (ts *HTTPTriggerSet) watchTriggers() { + if ts.controller == nil { + return + } + + failureCount := 0 + maxFailures := 5 + + // Watch controller for updates to triggers and update the router accordingly. + // TODO change this to use a watch API; or maybe even watch etcd directly. + for { + triggers, err := ts.controller.HTTPTriggerList() + if err != nil { + failureCount += 1 + if failureCount >= maxFailures { + log.Fatalf("Failed to connect to controller after %v retries: %v", failureCount, err) + } + } + log.Printf("Updating router, %v triggers", len(triggers)) + + ts.triggers = triggers + ts.mutableRouter.updateRouter(ts.getRouterFromTriggers()) + + time.Sleep(3 * time.Second) + } } diff --git a/router/mutablemux.go b/router/mutablemux.go index 0f1a4b5c..6e0354c2 100644 --- a/router/mutablemux.go +++ b/router/mutablemux.go @@ -49,6 +49,6 @@ func (mr *mutableRouter) ServeHTTP(responseWriter http.ResponseWriter, request * } func (mr *mutableRouter) updateRouter(newHandler *mux.Router) { - log.Print("Updating router") + log.Println("Updating router") mr.router.Store(newHandler) } diff --git a/router/router.go b/router/router.go index 41037ef0..90e8dd05 100644 --- a/router/router.go +++ b/router/router.go @@ -41,8 +41,13 @@ package router import ( "fmt" - "github.com/gorilla/mux" "net/http" + + "github.com/gorilla/mux" + + controllerClient "github.com/platform9/fission/controller/client" + poolmgrClient "github.com/platform9/fission/poolmgr/client" + "log" ) // request url ---[mux]---> Function(name,uid) ----[fmap]----> k8s service url @@ -64,6 +69,10 @@ func serve(port int, httpTriggerSet *HTTPTriggerSet) { func Start(port int, controllerUrl string, poolmgrUrl string) { fmap := makeFunctionServiceMap() - triggers := makeHTTPTriggerSet(fmap, controllerUrl, poolmgrUrl) + controller := controllerClient.MakeClient(controllerUrl) + poolmgr := poolmgrClient.MakeClient(poolmgrUrl) + + triggers := makeHTTPTriggerSet(fmap, controller, poolmgr) + log.Printf("Starting router at port %v\n", port) serve(port, triggers) } diff --git a/router/router_test.go b/router/router_test.go index db6c8c59..521c6cd4 100644 --- a/router/router_test.go +++ b/router/router_test.go @@ -33,7 +33,7 @@ func TestRouter(t *testing.T) { fmap.assign(fn, testServiceUrl) - triggers := makeHTTPTriggerSet(fmap, "", "") + triggers := makeHTTPTriggerSet(fmap, nil, nil) triggerUrl := "/foo" triggers.triggers = append(triggers.triggers, fission.HTTPTrigger{UrlPattern: triggerUrl, Function: *fn})