Merge pull request #16 from platform9/router-poolmgr

Router integration with poolmgr and controller

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 {

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 {

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(serviceName))
-	w.Write([]byte(serviceUrl))
 }
 
 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)

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
+}

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
+}

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
+	fmap     *functionServiceMap
-	poolManagerUrl string
+	poolmgr  *poolmgrClient.Client
-	Function       fission.Metadata
+	Function fission.Metadata
 }
 
-func (*functionHandler) getServiceForFunction() (*url.URL, error) {
+func (fh *functionHandler) getServiceForFunction() (*url.URL, error) {
-	return nil, errors.New("not implemented")
+	// 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

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
+	controller *controllerClient.Client
-	poolManagerUrl string
+	poolmgr    *poolmgrClient.Client
-	triggers       []fission.HTTPTrigger
+	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,
+		controller:         controller,
-		poolManagerUrl:     poolManagerUrl,
+		poolmgr:            poolmgr,
 	}
 }
 
-func (triggers *HTTPTriggerSet) subscribeRouter(mr *mutableRouter) {
+func (ts *HTTPTriggerSet) subscribeRouter(mr *mutableRouter) {
-	triggers.mutableRouter = mr
+	ts.mutableRouter = mr
-	mr.updateRouter(triggers.getRouterFromTriggers())
+	mr.updateRouter(ts.getRouterFromTriggers())
-	go triggers.watchTriggers()
+	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,
+			fmap:     ts.functionServiceMap,
-			Function:       trigger.Function,
+			Function: trigger.Function,
-			poolManagerUrl: triggers.poolManagerUrl,
+			poolmgr:  ts.poolmgr,
 		}
 		muxRouter.HandleFunc(trigger.UrlPattern, fh.handler)
 	}
 	return muxRouter
 }
 
-func (triggers *HTTPTriggerSet) watchTriggers() {
+func (ts *HTTPTriggerSet) watchTriggers() {
-	// watch controller for updates to triggers and update the router accordingly
+	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)
+	}
 }

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)
 }

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)
 }

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})