Update router cache on new function version

Addresses issue #51. Makes versioning in poolmgr explicit, so its
cache needs no invalidation on version update.  Functions in poolmgr
are always cached by name and UID.

Router now updates implictly versioned routes with the latest version
of a function.

This means that users will see requests to implicitly-versioned routes
go to the latest version of a function within 3 seconds.  (Those 3 sec
will go away when we use a real watch instead of polling the
controller.)

There's no change in behaviour for routes that explictly specify a
function UID.

poolmgr/api.go

@@ -119,11 +119,19 @@ func (api *API) getFunctionEnv(m *fission.Metadata) (*fission.Environment, error
 }
 
 func (api *API) getServiceForFunction(m *fission.Metadata) (string, error) {
-	// Check function -> svc map
+	// Make sure we have the full metadata.  This ensures that
+	// poolmgr does not implicitly interpret empty-UID as latest
+	// version.
+	if len(m.Uid) == 0 {
+		return "", fission.MakeError(fission.ErrorInvalidArgument,
+			fmt.Sprintf("invalid metadata for function %v", m.Name))
+	}
+
+	// Check function -> svc cache
 	log.Printf("[%v] Checking for cached function service", m.Name)
 	fsvc, err := api.fsCache.GetByFunction(m)
 	if err == nil {
-		// Cached, return svc name
+		// Cached, return svc address
 		return fsvc.address, nil
 	}
 

router/httpTriggers.go

@@ -18,8 +18,6 @@ package router
 
 import (
 	"log"
-	"os"
-	"strconv"
 	"time"
 
 	"github.com/gorilla/mux"
@@ -57,11 +55,22 @@ func (ts *HTTPTriggerSet) subscribeRouter(mr *mutableRouter) {
 func (ts *HTTPTriggerSet) getRouter() *mux.Router {
 	muxRouter := mux.NewRouter()
 
+	// make a name -> latest version map
+	latestVersions := make(map[string]string)
+	for _, f := range ts.functions {
+		latestVersions[f.Metadata.Name] = f.Metadata.Uid
+	}
+
 	// HTTP triggers setup by the user
 	for _, trigger := range ts.triggers {
+		m := trigger.Function
+		if len(m.Uid) == 0 {
+			// explicitly use the latest function version
+			m.Uid = latestVersions[m.Name]
+		}
 		fh := &functionHandler{
 			fmap:     ts.functionServiceMap,
-			Function: trigger.Function,
+			Function: m,
 			poolmgr:  ts.poolmgr,
 		}
 		muxRouter.HandleFunc(trigger.UrlPattern, fh.handler)
@@ -87,25 +96,10 @@ func (ts *HTTPTriggerSet) watchTriggers() {
 	}
 
 	// the number of connection failures we'll accept before quitting
-	var err error
 	maxFailures := 5
-	maxFailuresEnv := os.Getenv("FISSION_ROUTER_MAX_FAILURES")
-	if len(maxFailuresEnv) != 0 {
-		maxFailures, err = strconv.Atoi(maxFailuresEnv)
-		if err != nil {
-			log.Fatalf("FISSION_ROUTER_MAX_FAILURES must be an integer, found %v", maxFailuresEnv)
-		}
-	}
 
 	// amount of time to sleep between polling calls
-	pollSleepSec := 3
-	pollSleepEnv := os.Getenv("FISSION_ROUTER_POLL_SLEEP_SECONDS")
-	if len(pollSleepEnv) != 0 {
-		pollSleepSec, err = strconv.Atoi(pollSleepEnv)
-		if err != nil {
-			log.Fatalf("FISSION_ROUTER_POLL_SLEEP_SECONDS must be an integer, found %v", pollSleepEnv)
-		}
-	}
+	pollSleepDuration := 3 * time.Second
 
 	// 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.
@@ -117,7 +111,7 @@ func (ts *HTTPTriggerSet) watchTriggers() {
 			if failureCount >= maxFailures {
 				log.Fatalf("Failed to connect to controller after %v retries: %v", failureCount, err)
 			}
-			time.Sleep(time.Duration(pollSleepSec) * time.Second)
+			time.Sleep(pollSleepDuration)
 			continue
 		}
 		ts.triggers = triggers
@@ -129,6 +123,6 @@ func (ts *HTTPTriggerSet) watchTriggers() {
 		ts.functions = functions
 
 		ts.mutableRouter.updateRouter(ts.getRouter())
-		time.Sleep(time.Duration(pollSleepSec) * time.Second)
+		time.Sleep(pollSleepDuration)
 	}
 }