apply gofmt

This commit is contained in:
Soam Vasani
2016-09-01 16:32:32 -07:00
parent 2c872642f3
commit 8c56818e0a
8 changed files with 69 additions and 77 deletions
+6 -6
View File
@@ -17,15 +17,15 @@ limitations under the License.
package router package router
import ( import (
"errors"
"log" "log"
"net/http" "net/http"
"net/http/httputil" "net/http/httputil"
"errors"
"net/url" "net/url"
) )
type functionHandler struct { type functionHandler struct {
fmap *functionServiceMap fmap *functionServiceMap
poolManagerUrl string poolManagerUrl string
function function
} }
@@ -36,13 +36,13 @@ func (*functionHandler) getServiceForFunction() (*url.URL, error) {
func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request *http.Request) { func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request *http.Request) {
serviceUrl, err := fh.fmap.lookup(&fh.function) serviceUrl, err := fh.fmap.lookup(&fh.function)
if (err != nil) { if err != nil {
// Cache miss: request the Pool Manager to make a new service. // Cache miss: request the Pool Manager to make a new service.
serviceUrl, poolErr := fh.getServiceForFunction() serviceUrl, poolErr := fh.getServiceForFunction()
if (poolErr != nil) { if poolErr != nil {
// now we're really screwed // now we're really screwed
log.Printf("Failed to get service for function (%v,%v): %v", log.Printf("Failed to get service for function (%v,%v): %v",
fh.function.name, fh.function.uid, poolErr); fh.function.name, fh.function.uid, poolErr)
responseWriter.WriteHeader(500) // TODO: make this smarter based on the actual error responseWriter.WriteHeader(500) // TODO: make this smarter based on the actual error
return return
} }
@@ -54,7 +54,7 @@ func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request *
// Proxy off our request to the serviceUrl, and send the response back. // 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 would get us
// connection reuse and possibly better performance // connection reuse and possibly better performance
director := func(req *http.Request) { director := func(req *http.Request) {
// send this request to serviceurl // send this request to serviceurl
req.URL.Scheme = serviceUrl.Scheme req.URL.Scheme = serviceUrl.Scheme
req.URL.Host = serviceUrl.Host req.URL.Host = serviceUrl.Host
+9 -10
View File
@@ -18,14 +18,13 @@ package router
import ( import (
"log" "log"
"testing"
"net/http" "net/http"
// "net/http/httputil" "testing"
// "net/http/httputil"
"net/http/httptest" "net/http/httptest"
"net/url" "net/url"
) )
func createBackendService(testResponseString string) *url.URL { func createBackendService(testResponseString string) *url.URL {
backendServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { backendServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Write([]byte(testResponseString)) w.Write([]byte(testResponseString))
@@ -39,23 +38,23 @@ func createBackendService(testResponseString string) *url.URL {
} }
/* /*
1. Create a service at some URL 1. Create a service at some URL
2. Add it to the function service map 2. Add it to the function service map
3. Create a http server with some trigger url pointed at function handler 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. 4. Send a request to that server, ensure it reaches the first service.
*/ */
func TestFunctionProxying(t *testing.T) { func TestFunctionProxying(t *testing.T) {
testResponseString := "hi" testResponseString := "hi"
backendURL := createBackendService(testResponseString) backendURL := createBackendService(testResponseString)
log.Printf("Created backend svc at %v", backendURL) log.Printf("Created backend svc at %v", backendURL)
fn := &function{name: "foo", uid: "xxx"} fn := &function{name: "foo", uid: "xxx"}
fmap := makeFunctionServiceMap() fmap := makeFunctionServiceMap()
fmap.assign(fn, backendURL) fmap.assign(fn, backendURL)
fh := &functionHandler{fmap: fmap, function: *fn} fh := &functionHandler{fmap: fmap, function: *fn}
functionHandlerServer := httptest.NewServer(http.HandlerFunc(fh.handler)) functionHandlerServer := httptest.NewServer(http.HandlerFunc(fh.handler))
fhURL := functionHandlerServer.URL fhURL := functionHandlerServer.URL
testRequest(fhURL, testResponseString) testRequest(fhURL, testResponseString)
} }
+20 -23
View File
@@ -23,11 +23,12 @@ import (
) )
type requestType int type requestType int
const ( const (
LOOKUP requestType = iota // lookup the map LOOKUP requestType = iota // lookup the map
ASSIGN // assign function ASSIGN // assign function
NEXT_GEN // increment current generation NEXT_GEN // increment current generation
SWEEP // delete all but the current generation SWEEP // delete all but the current generation
) )
type functionServiceMapResponse struct { type functionServiceMapResponse struct {
@@ -35,7 +36,7 @@ type functionServiceMapResponse struct {
error error
} }
type functionServiceMapRequest struct { type functionServiceMapRequest struct {
function function
serviceUrl url.URL serviceUrl url.URL
requestType requestType
responseChannel chan<- functionServiceMapResponse responseChannel chan<- functionServiceMapResponse
@@ -47,12 +48,12 @@ type functionServiceMapEntry struct {
type functionServiceMap struct { type functionServiceMap struct {
// map (funcname, uid) -> url // map (funcname, uid) -> url
svc map[function]functionServiceMapEntry svc map[function]functionServiceMapEntry
currentGeneration uint64 currentGeneration uint64
requestChannel chan *functionServiceMapRequest requestChannel chan *functionServiceMapRequest
} }
func makeFunctionServiceMap() (*functionServiceMap) { func makeFunctionServiceMap() *functionServiceMap {
fmap := &functionServiceMap{} fmap := &functionServiceMap{}
fmap.requestChannel = make(chan *functionServiceMapRequest) fmap.requestChannel = make(chan *functionServiceMapRequest)
fmap.svc = make(map[function]functionServiceMapEntry) fmap.svc = make(map[function]functionServiceMapEntry)
@@ -62,25 +63,25 @@ func makeFunctionServiceMap() (*functionServiceMap) {
func (fmap *functionServiceMap) functionServiceMapWork() { func (fmap *functionServiceMap) functionServiceMapWork() {
for { for {
req := <- fmap.requestChannel req := <-fmap.requestChannel
switch req.requestType { switch req.requestType {
case LOOKUP: case LOOKUP:
e, present := fmap.svc[req.function] e, present := fmap.svc[req.function]
if present { if present {
req.responseChannel <- functionServiceMapResponse{ serviceUrl: e.serviceUrl } req.responseChannel <- functionServiceMapResponse{serviceUrl: e.serviceUrl}
} else { } else {
req.responseChannel <- functionServiceMapResponse{ error: errors.New("not found") } req.responseChannel <- functionServiceMapResponse{error: errors.New("not found")}
} }
case ASSIGN: case ASSIGN:
fmap.svc[req.function] = fmap.svc[req.function] =
functionServiceMapEntry{ serviceUrl: req.serviceUrl, generation: fmap.currentGeneration } functionServiceMapEntry{serviceUrl: req.serviceUrl, generation: fmap.currentGeneration}
// no response // no response
case NEXT_GEN: case NEXT_GEN:
fmap.currentGeneration++ fmap.currentGeneration++
// no response // no response
case SWEEP: case SWEEP:
log.Panic("not implemented") log.Panic("not implemented")
default: default:
log.Panic("bad request") log.Panic("bad request")
} }
} }
@@ -88,10 +89,9 @@ func (fmap *functionServiceMap) functionServiceMapWork() {
func (fmap *functionServiceMap) lookup(f *function) (*url.URL, error) { func (fmap *functionServiceMap) lookup(f *function) (*url.URL, error) {
respChannel := make(chan functionServiceMapResponse) respChannel := make(chan functionServiceMapResponse)
fmap.requestChannel <- fmap.requestChannel <- &functionServiceMapRequest{function: *f, requestType: LOOKUP, responseChannel: respChannel}
&functionServiceMapRequest{ function: *f, requestType: LOOKUP, responseChannel: respChannel }
resp := <-respChannel resp := <-respChannel
if (resp.error != nil) { if resp.error != nil {
return nil, resp.error return nil, resp.error
} else { } else {
return &resp.serviceUrl, nil return &resp.serviceUrl, nil
@@ -99,16 +99,13 @@ func (fmap *functionServiceMap) lookup(f *function) (*url.URL, error) {
} }
func (fmap *functionServiceMap) assign(f *function, serviceUrl *url.URL) { func (fmap *functionServiceMap) assign(f *function, serviceUrl *url.URL) {
fmap.requestChannel <- fmap.requestChannel <- &functionServiceMapRequest{function: *f, serviceUrl: *serviceUrl, requestType: ASSIGN}
&functionServiceMapRequest{ function: *f, serviceUrl: *serviceUrl, requestType: ASSIGN }
} }
func (fmap *functionServiceMap) nextGen() { func (fmap *functionServiceMap) nextGen() {
fmap.requestChannel <- fmap.requestChannel <- &functionServiceMapRequest{requestType: NEXT_GEN}
&functionServiceMapRequest{ requestType: NEXT_GEN }
} }
func (fmap *functionServiceMap) sweep() { func (fmap *functionServiceMap) sweep() {
fmap.requestChannel <- fmap.requestChannel <- &functionServiceMapRequest{requestType: SWEEP}
&functionServiceMapRequest{ requestType: SWEEP }
} }
+6 -6
View File
@@ -17,31 +17,31 @@ limitations under the License.
package router package router
import ( import (
"testing"
"net/url" "net/url"
"testing"
) )
func TestFunctionServiceMap(t *testing.T) { func TestFunctionServiceMap(t *testing.T) {
m := makeFunctionServiceMap() m := makeFunctionServiceMap()
fn := &function{ name: "foo", uid: "012" } fn := &function{name: "foo", uid: "012"}
u, err := url.Parse("/foo012") u, err := url.Parse("/foo012")
if (err != nil) { if err != nil {
t.Errorf("can't parse url") t.Errorf("can't parse url")
} }
m.assign(fn, u) m.assign(fn, u)
v, err := m.lookup(fn) v, err := m.lookup(fn)
if (err != nil) { if err != nil {
t.Errorf("Lookup error: %v", err) t.Errorf("Lookup error: %v", err)
} }
if (*v != *u) { if *v != *u {
t.Errorf("Expected %#v, got %#v", u, v) t.Errorf("Expected %#v, got %#v", u, v)
} }
fn.name = "bar" fn.name = "bar"
_, err2 := m.lookup(fn) _, err2 := m.lookup(fn)
if (err2 == nil) { if err2 == nil {
t.Errorf("No error on missing entry") t.Errorf("No error on missing entry")
} }
} }
+6 -6
View File
@@ -17,13 +17,13 @@ limitations under the License.
package router package router
import ( import (
"net/http"
"sync/atomic"
"github.com/gorilla/mux" "github.com/gorilla/mux"
"log" "log"
"net/http"
"sync/atomic"
) )
// //
// mutableRouter wraps the mux router, and allows the router to be // mutableRouter wraps the mux router, and allows the router to be
// atomically changed. // atomically changed.
// //
@@ -32,8 +32,8 @@ type mutableRouter struct {
router atomic.Value // mux.Router router atomic.Value // mux.Router
} }
func NewMutableRouter(handler *mux.Router) (*mutableRouter) { func NewMutableRouter(handler *mux.Router) *mutableRouter {
mr := mutableRouter{}; mr := mutableRouter{}
mr.router.Store(handler) mr.router.Store(handler)
return &mr return &mr
} }
@@ -42,7 +42,7 @@ func (mr *mutableRouter) ServeHTTP(responseWriter http.ResponseWriter, request *
// Atomically grab the underlying mux router and call it. // Atomically grab the underlying mux router and call it.
routerValue := mr.router.Load() routerValue := mr.router.Load()
router, ok := routerValue.(*mux.Router) router, ok := routerValue.(*mux.Router)
if (!ok) { if !ok {
log.Panic("Invalid router type") log.Panic("Invalid router type")
} }
router.ServeHTTP(responseWriter, request) router.ServeHTTP(responseWriter, request)
+6 -7
View File
@@ -17,10 +17,10 @@ limitations under the License.
package router package router
import ( import (
"testing"
"net/http"
"github.com/gorilla/mux" "github.com/gorilla/mux"
"log" "log"
"net/http"
"testing"
"time" "time"
) )
@@ -40,17 +40,16 @@ func startServer(mr *mutableRouter) {
http.ListenAndServe(":3333", mr) http.ListenAndServe(":3333", mr)
} }
func spamServer(quit chan bool) { func spamServer(quit chan bool) {
i := 0 i := 0
for { for {
select { select {
case <- quit: case <-quit:
break break
default: default:
i = i + 1 i = i + 1
resp, err := http.Get("http://localhost:3333") resp, err := http.Get("http://localhost:3333")
if (err != nil) { if err != nil {
log.Panicf("failed to make get request %v: %v", i, err) log.Panicf("failed to make get request %v: %v", i, err)
} }
resp.Body.Close() resp.Body.Close()
@@ -64,11 +63,11 @@ func TestMutableMux(t *testing.T) {
muxRouter := mux.NewRouter() muxRouter := mux.NewRouter()
muxRouter.HandleFunc("/", OldHandler) muxRouter.HandleFunc("/", OldHandler)
mr := NewMutableRouter(muxRouter) mr := NewMutableRouter(muxRouter)
// start http server // start http server
log.Print("Start http server") log.Print("Start http server")
go startServer(mr) go startServer(mr)
// continuously make requests, panic if any fails // continuously make requests, panic if any fails
time.Sleep(100 * time.Millisecond) time.Sleep(100 * time.Millisecond)
q := make(chan bool) q := make(chan bool)
+9 -11
View File
@@ -23,7 +23,7 @@ Its job is to:
1. Keep track of HTTP triggers and their mappings to functions 1. Keep track of HTTP triggers and their mappings to functions
Use the controller API to get and watch this state. Use the controller API to get and watch this state.
2. Given a function, get a reference to a routable function run service 2. Given a function, get a reference to a routable function run service
Use the ContainerPoolManager API to get a service backed by one Use the ContainerPoolManager API to get a service backed by one
@@ -37,41 +37,39 @@ Its job is to:
*/ */
package router package router
import ( import (
"fmt" "fmt"
"net/http"
"github.com/gorilla/mux" "github.com/gorilla/mux"
flag "github.com/ogier/pflag" flag "github.com/ogier/pflag"
"net/http"
) )
type ( type (
function struct { function struct {
name string name string
uid string uid string
} }
httptrigger struct { httptrigger struct {
urlPattern string urlPattern string
function function
} }
options struct { options struct {
port int port int
poolManagerUrl string poolManagerUrl string
controllerUrl string controllerUrl string
//... //...
} }
) )
// request url ---[mux]---> function(name,uid) ----[fmap]----> k8s service url // 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 // request url ---[trigger]---> function(name, deployment) ----[deployment]----> function(name, uid) ----[pool mgr]---> k8s service url
func router(httpTriggerSet *HTTPTriggerSet) (*mutableRouter) { func router(httpTriggerSet *HTTPTriggerSet) *mutableRouter {
muxRouter := mux.NewRouter() muxRouter := mux.NewRouter()
mr := NewMutableRouter(muxRouter) mr := NewMutableRouter(muxRouter)
httpTriggerSet.subscribeRouter(mr) httpTriggerSet.subscribeRouter(mr)
@@ -84,7 +82,7 @@ func server(port int, httpTriggerSet *HTTPTriggerSet) {
http.ListenAndServe(url, mr) http.ListenAndServe(url, mr)
} }
func getOptions() (*options) { func getOptions() *options {
options := &options{} options := &options{}
flag.IntVar(&options.port, "port", 80, "Port to listen on") flag.IntVar(&options.port, "port", 80, "Port to listen on")
@@ -92,7 +90,7 @@ func getOptions() (*options) {
// default to using dns service discovery // default to using dns service discovery
flag.StringVar(&options.poolManagerUrl, "poolmanager_url", "http://poolmanager/", "URL for the PoolManager service") 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") flag.StringVar(&options.controllerUrl, "controller_url", "http://controller/", "URL for the controller service")
return options return options
} }
+7 -8
View File
@@ -1,31 +1,30 @@
package router package router
import ( import (
"net/http"
"log"
"io/ioutil" "io/ioutil"
"log"
"net/http"
) )
func testRequest(targetUrl string, expectedResponse string) { func testRequest(targetUrl string, expectedResponse string) {
resp, err := http.Get(targetUrl) resp, err := http.Get(targetUrl)
if (err != nil) { if err != nil {
log.Panicf("failed to make get request: %v", err) log.Panicf("failed to make get request: %v", err)
} }
defer resp.Body.Close() defer resp.Body.Close()
if (resp.StatusCode != 200) { if resp.StatusCode != 200 {
log.Panicf("response status: %v", resp.StatusCode) log.Panicf("response status: %v", resp.StatusCode)
} }
body, err := ioutil.ReadAll(resp.Body) body, err := ioutil.ReadAll(resp.Body)
if (err != nil) { if err != nil {
log.Panic("failed to read response") log.Panic("failed to read response")
} }
bodyStr := string(body) bodyStr := string(body)
log.Printf("Server responded with %v", bodyStr) log.Printf("Server responded with %v", bodyStr)
if (bodyStr != expectedResponse) { if bodyStr != expectedResponse {
log.Panic("Unexpected response") log.Panic("Unexpected response")
} }
} }