diff --git a/src/router/functionHandler.go b/src/router/functionHandler.go new file mode 100644 index 00000000..8b93d89d --- /dev/null +++ b/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. +} diff --git a/src/router/functionHandler_test.go b/src/router/functionHandler_test.go new file mode 100644 index 00000000..72bfa1e7 --- /dev/null +++ b/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) +} diff --git a/src/router/functionServiceMap.go b/src/router/functionServiceMap.go new file mode 100644 index 00000000..d667de67 --- /dev/null +++ b/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} +} diff --git a/src/router/functionServiceMap_test.go b/src/router/functionServiceMap_test.go new file mode 100644 index 00000000..48b12c52 --- /dev/null +++ b/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") + } +} diff --git a/src/router/httpTriggers.go b/src/router/httpTriggers.go new file mode 100644 index 00000000..7cdcb38b --- /dev/null +++ b/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 +} diff --git a/src/router/mutablemux.go b/src/router/mutablemux.go new file mode 100644 index 00000000..0f1a4b5c --- /dev/null +++ b/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) +} diff --git a/src/router/mutablemux_test.go b/src/router/mutablemux_test.go new file mode 100644 index 00000000..a8e7df31 --- /dev/null +++ b/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) +} diff --git a/src/router/router.go b/src/router/router.go new file mode 100644 index 00000000..c1e77e35 --- /dev/null +++ b/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) +} diff --git a/src/router/router_test.go b/src/router/router_test.go new file mode 100644 index 00000000..556a2849 --- /dev/null +++ b/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) +} diff --git a/src/router/util_test.go b/src/router/util_test.go new file mode 100644 index 00000000..86322b09 --- /dev/null +++ b/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") + } +}