Merge pull request #3 from platform9/router

Router, initial merge.

This doesn't yet talk to the Controller or PoolManager APIs.

src/router/functionHandler.go

@@ -0,0 +1,73 @@
+/*
+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 router
+
+import (
+	"errors"
+	"log"
+	"net/http"
+	"net/http/httputil"
+	"net/url"
+)
+
+type functionHandler struct {
+	fmap           *functionServiceMap
+	poolManagerUrl string
+	function
+}
+
+func (*functionHandler) getServiceForFunction() (*url.URL, error) {
+	return nil, errors.New("not implemented")
+}
+
+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.
+		serviceUrl, poolErr := fh.getServiceForFunction()
+		if poolErr != nil {
+			// now we're really screwed
+			log.Printf("Failed to get service for function (%v,%v): %v",
+				fh.function.name, fh.function.uid, poolErr)
+			responseWriter.WriteHeader(500) // TODO: make this smarter based on the actual error
+			return
+		}
+
+		// add it to the map
+		fh.fmap.assign(&fh.function, serviceUrl)
+	}
+
+	// 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
+	// connection reuse and possibly better performance
+	director := func(req *http.Request) {
+		// send this request to serviceurl
+		req.URL.Scheme = serviceUrl.Scheme
+		req.URL.Host = serviceUrl.Host
+		req.URL.Path = serviceUrl.Path
+		// leave the query string intact (req.URL.RawQuery)
+
+		if _, ok := req.Header["User-Agent"]; !ok {
+			// explicitly disable User-Agent so it's not set to default value
+			req.Header.Set("User-Agent", "")
+		}
+	}
+	proxy := &httputil.ReverseProxy{Director: director}
+	proxy.ServeHTTP(responseWriter, request)
+
+	// TODO: handle failures and possibly retry here.
+}

src/router/functionHandler_test.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 router
+
+import (
+	"log"
+	"net/http"
+	"testing"
+	//	"net/http/httputil"
+	"net/http/httptest"
+	"net/url"
+)
+
+func createBackendService(testResponseString string) *url.URL {
+	backendServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Write([]byte(testResponseString))
+	}))
+
+	backendURL, err := url.Parse(backendServer.URL)
+	if err != nil {
+		panic("error parsing url")
+	}
+	return backendURL
+}
+
+/*
+   1. Create a service at some URL
+   2. Add it to the function service map
+   3. Create a http server with some trigger url pointed at function handler
+   4. Send a request to that server, ensure it reaches the first service.
+*/
+func TestFunctionProxying(t *testing.T) {
+	testResponseString := "hi"
+	backendURL := createBackendService(testResponseString)
+	log.Printf("Created backend svc at %v", backendURL)
+
+	fn := &function{name: "foo", uid: "xxx"}
+	fmap := makeFunctionServiceMap()
+	fmap.assign(fn, backendURL)
+
+	fh := &functionHandler{fmap: fmap, function: *fn}
+	functionHandlerServer := httptest.NewServer(http.HandlerFunc(fh.handler))
+	fhURL := functionHandlerServer.URL
+
+	testRequest(fhURL, testResponseString)
+}

src/router/functionServiceMap.go

@@ -0,0 +1,111 @@
+/*
+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 router
+
+import (
+	"errors"
+	"log"
+	"net/url"
+)
+
+type requestType int
+
+const (
+	LOOKUP   requestType = iota // lookup the map
+	ASSIGN                      // assign function
+	NEXT_GEN                    // increment current generation
+	SWEEP                       // delete all but the current generation
+)
+
+type functionServiceMapResponse struct {
+	serviceUrl url.URL
+	error
+}
+type functionServiceMapRequest struct {
+	function
+	serviceUrl url.URL
+	requestType
+	responseChannel chan<- functionServiceMapResponse
+}
+type functionServiceMapEntry struct {
+	serviceUrl url.URL
+	generation uint64
+}
+
+type functionServiceMap struct {
+	// map (funcname, uid) -> url
+	svc               map[function]functionServiceMapEntry
+	currentGeneration uint64
+	requestChannel    chan *functionServiceMapRequest
+}
+
+func makeFunctionServiceMap() *functionServiceMap {
+	fmap := &functionServiceMap{}
+	fmap.requestChannel = make(chan *functionServiceMapRequest)
+	fmap.svc = make(map[function]functionServiceMapEntry)
+	go fmap.functionServiceMapWork()
+	return fmap
+}
+
+func (fmap *functionServiceMap) functionServiceMapWork() {
+	for {
+		req := <-fmap.requestChannel
+		switch req.requestType {
+		case LOOKUP:
+			e, present := fmap.svc[req.function]
+			if present {
+				req.responseChannel <- functionServiceMapResponse{serviceUrl: e.serviceUrl}
+			} else {
+				req.responseChannel <- functionServiceMapResponse{error: errors.New("not found")}
+			}
+		case ASSIGN:
+			fmap.svc[req.function] =
+				functionServiceMapEntry{serviceUrl: req.serviceUrl, generation: fmap.currentGeneration}
+			// no response
+		case NEXT_GEN:
+			fmap.currentGeneration++
+			// no response
+		case SWEEP:
+			log.Panic("not implemented")
+		default:
+			log.Panic("bad request")
+		}
+	}
+}
+
+func (fmap *functionServiceMap) lookup(f *function) (*url.URL, error) {
+	respChannel := make(chan functionServiceMapResponse)
+	fmap.requestChannel <- &functionServiceMapRequest{function: *f, requestType: LOOKUP, responseChannel: respChannel}
+	resp := <-respChannel
+	if resp.error != nil {
+		return nil, resp.error
+	} else {
+		return &resp.serviceUrl, nil
+	}
+}
+
+func (fmap *functionServiceMap) assign(f *function, serviceUrl *url.URL) {
+	fmap.requestChannel <- &functionServiceMapRequest{function: *f, serviceUrl: *serviceUrl, requestType: ASSIGN}
+}
+
+func (fmap *functionServiceMap) nextGen() {
+	fmap.requestChannel <- &functionServiceMapRequest{requestType: NEXT_GEN}
+}
+
+func (fmap *functionServiceMap) sweep() {
+	fmap.requestChannel <- &functionServiceMapRequest{requestType: SWEEP}
+}

src/router/functionServiceMap_test.go

@@ -0,0 +1,47 @@
+/*
+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 router
+
+import (
+	"net/url"
+	"testing"
+)
+
+func TestFunctionServiceMap(t *testing.T) {
+	m := makeFunctionServiceMap()
+	fn := &function{name: "foo", uid: "012"}
+	u, err := url.Parse("/foo012")
+	if err != nil {
+		t.Errorf("can't parse url")
+	}
+
+	m.assign(fn, u)
+
+	v, err := m.lookup(fn)
+	if err != nil {
+		t.Errorf("Lookup error: %v", err)
+	}
+	if *v != *u {
+		t.Errorf("Expected %#v, got %#v", u, v)
+	}
+
+	fn.name = "bar"
+	_, err2 := m.lookup(fn)
+	if err2 == nil {
+		t.Errorf("No error on missing entry")
+	}
+}

src/router/httpTriggers.go

@@ -0,0 +1,62 @@
+/*
+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 router
+
+import (
+	"github.com/gorilla/mux"
+)
+
+type HTTPTriggerSet struct {
+	*functionServiceMap
+	*mutableRouter
+	controllerUrl  string
+	poolManagerUrl string
+	triggers       []httptrigger
+}
+
+func makeHTTPTriggerSet(fmap *functionServiceMap, controllerUrl string, poolManagerUrl string) *HTTPTriggerSet {
+	triggers := make([]httptrigger, 1)
+	return &HTTPTriggerSet{
+		functionServiceMap: fmap,
+		triggers:           triggers,
+		controllerUrl:      controllerUrl,
+		poolManagerUrl:     poolManagerUrl,
+	}
+}
+
+func (triggers *HTTPTriggerSet) subscribeRouter(mr *mutableRouter) {
+	triggers.mutableRouter = mr
+	mr.updateRouter(triggers.getRouterFromTriggers())
+	go triggers.watchTriggers()
+}
+
+func (triggers *HTTPTriggerSet) getRouterFromTriggers() *mux.Router {
+	muxRouter := mux.NewRouter()
+	for _, trigger := range triggers.triggers {
+		fh := &functionHandler{
+			fmap:           triggers.functionServiceMap,
+			function:       trigger.function,
+			poolManagerUrl: triggers.poolManagerUrl,
+		}
+		muxRouter.HandleFunc(trigger.urlPattern, fh.handler)
+	}
+	return muxRouter
+}
+
+func (triggers *HTTPTriggerSet) watchTriggers() {
+	// watch controller for updates to triggers and update the router accordingly
+}

src/router/mutablemux.go

@@ -0,0 +1,54 @@
+/*
+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 router
+
+import (
+	"github.com/gorilla/mux"
+	"log"
+	"net/http"
+	"sync/atomic"
+)
+
+//
+// mutableRouter wraps the mux router, and allows the router to be
+// atomically changed.
+//
+
+type mutableRouter struct {
+	router atomic.Value // mux.Router
+}
+
+func NewMutableRouter(handler *mux.Router) *mutableRouter {
+	mr := mutableRouter{}
+	mr.router.Store(handler)
+	return &mr
+}
+
+func (mr *mutableRouter) ServeHTTP(responseWriter http.ResponseWriter, request *http.Request) {
+	// Atomically grab the underlying mux router and call it.
+	routerValue := mr.router.Load()
+	router, ok := routerValue.(*mux.Router)
+	if !ok {
+		log.Panic("Invalid router type")
+	}
+	router.ServeHTTP(responseWriter, request)
+}
+
+func (mr *mutableRouter) updateRouter(newHandler *mux.Router) {
+	log.Print("Updating router")
+	mr.router.Store(newHandler)
+}

src/router/mutablemux_test.go

@@ -0,0 +1,94 @@
+/*
+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 router
+
+import (
+	"github.com/gorilla/mux"
+	"log"
+	"net/http"
+	"testing"
+	"time"
+)
+
+func OldHandler(responseWriter http.ResponseWriter, request *http.Request) {
+	responseWriter.Write([]byte("old handler"))
+}
+func NewHandler(responseWriter http.ResponseWriter, request *http.Request) {
+	responseWriter.Write([]byte("new handler"))
+}
+
+func verifyRequest(expectedResponse string) {
+	targetUrl := "http://localhost:3333"
+	testRequest(targetUrl, expectedResponse)
+}
+
+func startServer(mr *mutableRouter) {
+	http.ListenAndServe(":3333", mr)
+}
+
+func spamServer(quit chan bool) {
+	i := 0
+	for {
+		select {
+		case <-quit:
+			break
+		default:
+			i = i + 1
+			resp, err := http.Get("http://localhost:3333")
+			if err != nil {
+				log.Panicf("failed to make get request %v: %v", i, err)
+			}
+			resp.Body.Close()
+		}
+	}
+}
+
+func TestMutableMux(t *testing.T) {
+	// make a simple mutable router
+	log.Print("Create mutable router")
+	muxRouter := mux.NewRouter()
+	muxRouter.HandleFunc("/", OldHandler)
+	mr := NewMutableRouter(muxRouter)
+
+	// start http server
+	log.Print("Start http server")
+	go startServer(mr)
+
+	// continuously make requests, panic if any fails
+	time.Sleep(100 * time.Millisecond)
+	q := make(chan bool)
+	go spamServer(q)
+
+	time.Sleep(5 * time.Millisecond)
+
+	// connect and verify old handler
+	log.Print("Verify old handler")
+	verifyRequest("old handler")
+
+	// change the muxer
+	log.Print("Change mux router")
+	newMuxRouter := mux.NewRouter()
+	newMuxRouter.HandleFunc("/", NewHandler)
+	mr.updateRouter(newMuxRouter)
+
+	// connect and verify the new handler
+	log.Print("Verify new handler")
+	verifyRequest("new handler")
+
+	q <- true
+	time.Sleep(100 * time.Millisecond)
+}

src/router/router.go

@@ -0,0 +1,102 @@
+/*
+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.
+*/
+
+/*
+
+This is the Fission Router package.
+
+Its job is to:
+
+  1. Keep track of HTTP triggers and their mappings to functions
+
+     Use the controller API to get and watch this state.
+
+  2. Given a function, get a reference to a routable function run service
+
+     Use the ContainerPoolManager API to get a service backed by one
+     or more function run containers.  The container(s) backing the
+     service may be newly created, or they might be reused.  The only
+     requirement is that one or more containers backs the service.
+
+  3. Forward the request to the service, and send the response back.
+
+     Plain ol HTTP.
+
+*/
+
+package router
+
+import (
+	"fmt"
+	"github.com/gorilla/mux"
+	flag "github.com/ogier/pflag"
+	"net/http"
+)
+
+type (
+	function struct {
+		name string
+		uid  string
+	}
+
+	httptrigger struct {
+		urlPattern string
+		function
+	}
+
+	options struct {
+		port           int
+		poolManagerUrl string
+		controllerUrl  string
+		//...
+	}
+)
+
+// request url ---[mux]---> function(name,uid) ----[fmap]----> k8s service url
+
+// request url ---[trigger]---> function(name, deployment) ----[deployment]----> function(name, uid) ----[pool mgr]---> k8s service url
+
+func router(httpTriggerSet *HTTPTriggerSet) *mutableRouter {
+	muxRouter := mux.NewRouter()
+	mr := NewMutableRouter(muxRouter)
+	httpTriggerSet.subscribeRouter(mr)
+	return mr
+}
+
+func server(port int, httpTriggerSet *HTTPTriggerSet) {
+	mr := router(httpTriggerSet)
+	url := fmt.Sprintf(":%v", port)
+	http.ListenAndServe(url, mr)
+}
+
+func getOptions() *options {
+	options := &options{}
+
+	flag.IntVar(&options.port, "port", 80, "Port to listen on")
+
+	// default to using dns service discovery
+	flag.StringVar(&options.poolManagerUrl, "poolmanager_url", "http://poolmanager/", "URL for the PoolManager service")
+	flag.StringVar(&options.controllerUrl, "controller_url", "http://controller/", "URL for the controller service")
+
+	return options
+}
+
+func main() {
+	options := getOptions()
+	fmap := makeFunctionServiceMap()
+	triggers := makeHTTPTriggerSet(fmap, options.controllerUrl, options.poolManagerUrl)
+	server(options.port, triggers)
+}

src/router/router_test.go

@@ -0,0 +1,44 @@
+/*
+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 router
+
+import (
+	"fmt"
+	"testing"
+	"time"
+)
+
+func TestRouter(t *testing.T) {
+	fmap := makeFunctionServiceMap()
+	fn := &function{name: "foo", uid: "xxx"}
+
+	testResponseString := "hi"
+	testServiceUrl := createBackendService(testResponseString)
+
+	fmap.assign(fn, testServiceUrl)
+
+	triggers := makeHTTPTriggerSet(fmap, "", "")
+	triggerUrl := "/foo"
+	triggers.triggers = append(triggers.triggers, httptrigger{triggerUrl, *fn})
+
+	port := 4242
+	go server(port, triggers)
+	time.Sleep(100 * time.Millisecond)
+
+	testUrl := fmt.Sprintf("http://localhost:%v%v", port, triggerUrl)
+	testRequest(testUrl, testResponseString)
+}

src/router/util_test.go

@@ -0,0 +1,30 @@
+package router
+
+import (
+	"io/ioutil"
+	"log"
+	"net/http"
+)
+
+func testRequest(targetUrl string, expectedResponse string) {
+	resp, err := http.Get(targetUrl)
+	if err != nil {
+		log.Panicf("failed to make get request: %v", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != 200 {
+		log.Panicf("response status: %v", resp.StatusCode)
+	}
+
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		log.Panic("failed to read response")
+	}
+
+	bodyStr := string(body)
+	log.Printf("Server responded with %v", bodyStr)
+	if bodyStr != expectedResponse {
+		log.Panic("Unexpected response")
+	}
+}