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 { 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 { 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/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 +} 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 +} 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})