Canary deployments for fission functions. (#892)

router/functionHandler.go

@@ -19,8 +19,10 @@ package router
 import (
 	"bytes"
 	"fmt"
+	"github.com/gorilla/mux"
+	"github.com/satori/go.uuid"
 	"io/ioutil"
-	"log"
+	"math/rand"
 	"net"
 	"net/http"
 	"net/http/httputil"
@@ -28,9 +30,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/gorilla/mux"
-	"github.com/satori/go.uuid"
-	"github.com/sirupsen/logrus"
+	log "github.com/sirupsen/logrus"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/fission/fission"
@@ -47,14 +47,16 @@ type tsRoundTripperParams struct {
 }
 
 type functionHandler struct {
-	fmap                 *functionServiceMap
-	frmap                *functionRecorderMap
-	trmap                *triggerRecorderMap
-	executor             *executorClient.Client
-	function             *metav1.ObjectMeta
-	httpTrigger          *crd.HTTPTrigger
-	tsRoundTripperParams *tsRoundTripperParams
-	recorderName         string
+	fmap                     *functionServiceMap
+	frmap                    *functionRecorderMap
+	trmap                    *triggerRecorderMap
+	executor                 *executorClient.Client
+	function                 *metav1.ObjectMeta
+	httpTrigger              *crd.HTTPTrigger
+	functionMetadataMap      map[string]*metav1.ObjectMeta
+	fnWeightDistributionList []FunctionWeightDistribution
+	tsRoundTripperParams     *tsRoundTripperParams
+	recorderName             string
 }
 
 // A layer on top of http.DefaultTransport, with retries.
@@ -62,6 +64,11 @@ type RetryingRoundTripper struct {
 	funcHandler *functionHandler
 }
 
+func init() {
+	// just seeding the random number for getting the canary function
+	rand.Seed(time.Now().UnixNano())
+}
+
 // RoundTrip is a custom transport with retries for http requests that forwards the request to the right serviceUrl, obtained
 // from router's cache or from executor if router entry is stale.
 //
@@ -108,7 +115,7 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt
 
 			rdr1.Read(p)
 			postedBody = string(p)
-			logrus.Info(fmt.Sprintf("%v", postedBody))
+			log.Info(fmt.Sprintf("%v", postedBody))
 			req.Body = rdr2
 		}
 	}
@@ -136,6 +143,7 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt
 	serviceUrl, err = roundTripper.funcHandler.fmap.lookup(roundTripper.funcHandler.function)
 	if err != nil || serviceUrl == nil {
 		// cache miss or nil entry in cache
+		log.Printf("Setting needExecutor to true for function : %s", roundTripper.funcHandler.function.Name)
 		needExecutor = true
 	}
 
@@ -149,6 +157,7 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt
 			service, err := roundTripper.funcHandler.executor.GetServiceForFunction(
 				roundTripper.funcHandler.function)
 			if err != nil {
+				log.Printf("Err from GetServiceForFunction : %v", err)
 				// We might want a specific error code or header for fission failures as opposed to
 				// user function bugs.
 				return nil, err
@@ -161,6 +170,7 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt
 			}
 
 			// add the address in router's cache
+			log.Printf("assigning serviceUrl : %s for function : %s", service, roundTripper.funcHandler.function.Name)
 			roundTripper.funcHandler.fmap.assign(roundTripper.funcHandler.function, serviceUrl)
 
 			// flag denotes that service was not obtained from cache, instead, created just now by executor
@@ -271,7 +281,7 @@ func (fh *functionHandler) tapService(serviceUrl *url.URL) {
 	fh.executor.TapService(serviceUrl)
 }
 
-func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request *http.Request) {
+func (fh functionHandler) handler(responseWriter http.ResponseWriter, request *http.Request) {
 	// retrieve url params and add them to request header
 	vars := mux.Vars(request)
 	for k, v := range vars {
@@ -286,6 +296,18 @@ func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request *
 		log.Print("Record request with ReqUID: ", reqUID)
 	}
 
+	if fh.httpTrigger != nil && fh.httpTrigger.Spec.FunctionReference.Type == fission.FunctionReferenceTypeFunctionWeights {
+		// canary deployment. need to determine the function to send request to now
+		fnMetadata := getCanaryBackend(fh.functionMetadataMap, fh.fnWeightDistributionList)
+		if fnMetadata == nil {
+			log.Printf("Error getting canary backend ")
+			// TODO : write error to responseWrite and return response
+			return
+		}
+		fh.function = fnMetadata
+		log.Debugf("chosen fnBackend's metadata : %+v", fh.function)
+	}
+
 	// system params
 	MetadataToHeaders(HEADERS_FISSION_FUNCTION_PREFIX, fh.function, request)
 
@@ -299,9 +321,44 @@ func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request *
 	proxy := &httputil.ReverseProxy{
 		Director: director,
 		Transport: &RetryingRoundTripper{
-			funcHandler: fh,
+			funcHandler: &fh,
 		},
 	}
 
 	proxy.ServeHTTP(responseWriter, request)
 }
+
+// findCeil picks a function from the functionWeightDistribution list based on the
+// random number generated. It uses the prefix calculated for the function weights.
+func findCeil(randomNumber int, wtDistrList []FunctionWeightDistribution) string {
+	low := 0
+	high := len(wtDistrList) - 1
+
+	for {
+		if low >= high {
+			break
+		}
+
+		mid := low + high/2
+		if randomNumber >= wtDistrList[mid].sumPrefix {
+			low = mid + 1
+		} else {
+			high = mid
+		}
+	}
+
+	if wtDistrList[low].sumPrefix >= randomNumber {
+		return wtDistrList[low].name
+	} else {
+		return ""
+	}
+}
+
+// picks a function to route to based on a random number generated
+func getCanaryBackend(fnMetadatamap map[string]*metav1.ObjectMeta, fnWtDistributionList []FunctionWeightDistribution) *metav1.ObjectMeta {
+	randomNumber := rand.Intn(fnWtDistributionList[len(fnWtDistributionList)-1].sumPrefix + 1)
+
+	fnName := findCeil(randomNumber, fnWtDistributionList)
+
+	return fnMetadatamap[fnName]
+}

router/functionHandler_test.go

@@ -24,6 +24,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/fission/fission"
+	"github.com/fission/fission/crd"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -55,6 +57,19 @@ func TestFunctionProxying(t *testing.T) {
 	fmap := makeFunctionServiceMap(0)
 	fmap.assign(fn, backendURL)
 
+	httpTrigger := &crd.HTTPTrigger{
+		Metadata: metav1.ObjectMeta{
+			Name:            "xxx",
+			Namespace:       metav1.NamespaceDefault,
+			ResourceVersion: "1234",
+		},
+		Spec: fission.HTTPTriggerSpec{
+			FunctionReference: fission.FunctionReference{
+				Type: fission.FunctionReferenceTypeFunctionName,
+			},
+		},
+	}
+
 	fh := &functionHandler{fmap: fmap,
 		function: fn,
 		tsRoundTripperParams: &tsRoundTripperParams{
@@ -63,6 +78,7 @@ func TestFunctionProxying(t *testing.T) {
 			keepAlive:       30 * time.Second,
 			maxRetries:      10,
 		},
+		httpTrigger: httpTrigger,
 	}
 	functionHandlerServer := httptest.NewServer(http.HandlerFunc(fh.handler))
 	fhURL := functionHandlerServer.URL

router/functionReferenceResolver.go

@@ -45,25 +45,33 @@ type (
 
 	resolveResultType int
 
-	// resolveResult is the result of resolving a function reference; for now
-	// it's just the metadata of one function, but in the future could support
+	FunctionWeightDistribution struct {
+		name      string
+		weight    int
+		sumPrefix int
+	}
+
+	// resolveResult is the result of resolving a function reference;
+	// it could be the metadata of one function or
 	// a distribution of requests across two functions.
 	resolveResult struct {
 		resolveResultType
-		functionMetadata *metav1.ObjectMeta
+		functionMetadataMap        map[string]*metav1.ObjectMeta
+		functionWtDistributionList []FunctionWeightDistribution
 	}
 
-	// namespacedFunctionReference is just a function reference plus a
-	// namespace. Since a function reference works on names, it's only
-	// meaningful within a namespace.
-	namespacedFunctionReference struct {
-		namespace         string
-		functionReference fission.FunctionReference
+	// namespacedTriggerReference is just a trigger reference plus a
+	// namespace.
+	namespacedTriggerReference struct {
+		namespace              string
+		triggerName            string
+		triggerResourceVersion string
 	}
 )
 
 const (
 	resolveResultSingleFunction = iota
+	resolveResultMultipleFunctions
 )
 
 func makeFunctionReferenceResolver(store k8sCache.Store) *functionReferenceResolver {
@@ -89,15 +97,12 @@ func makeK8SCache(crdClient *rest.RESTClient) (k8sCache.Store, k8sCache.Controll
 		k8sCache.ResourceEventHandlerFuncs{})
 }
 
-// resolve translates a namespace and a function reference to resolveResult.
-// The resolveResult for now is just a function's metadata. In the future, some
-// function ref types may resolve to two functions rather than just one
-// (e.g. for incremental deployment), which will make the resolveResult a bit
-// more complex.
-func (frr *functionReferenceResolver) resolve(namespace string, fr *fission.FunctionReference) (*resolveResult, error) {
-	nfr := namespacedFunctionReference{
-		namespace:         namespace,
-		functionReference: *fr,
+// resolve translates a trigger's function reference to a resolveResult.
+func (frr *functionReferenceResolver) resolve(trigger crd.HTTPTrigger) (*resolveResult, error) {
+	nfr := namespacedTriggerReference{
+		namespace:              trigger.Metadata.Namespace,
+		triggerName:            trigger.Metadata.Name,
+		triggerResourceVersion: trigger.Metadata.ResourceVersion,
 	}
 
 	// check cache
@@ -110,14 +115,21 @@ func (frr *functionReferenceResolver) resolve(namespace string, fr *fission.Func
 	// resolve on cache miss
 	var rr *resolveResult
 
-	switch fr.Type {
+	switch trigger.Spec.FunctionReference.Type {
 	case fission.FunctionReferenceTypeFunctionName:
-		rr, err = frr.resolveByName(namespace, fr.Name)
+		rr, err = frr.resolveByName(nfr.namespace, trigger.Spec.FunctionReference.Name)
 		if err != nil {
 			return nil, err
 		}
+
+	case fission.FunctionReferenceTypeFunctionWeights:
+		rr, err = frr.resolveByFunctionWeights(nfr.namespace, &trigger.Spec.FunctionReference)
+		if err != nil {
+			return nil, err
+		}
+
 	default:
-		return nil, fmt.Errorf("Unrecognized function reference type %v", fr.Type)
+		return nil, fmt.Errorf("Unrecognized function reference type %v", trigger.Spec.FunctionReference.Type)
 	}
 
 	// cache resolve result
@@ -143,25 +155,71 @@ func (frr *functionReferenceResolver) resolveByName(namespace, name string) (*re
 	}
 
 	f := obj.(*crd.Function)
+	functionMetadataMap := make(map[string]*metav1.ObjectMeta, 1)
+	functionMetadataMap[f.Metadata.Name] = &f.Metadata
+
 	rr := resolveResult{
-		resolveResultType: resolveResultSingleFunction,
-		functionMetadata:  &f.Metadata,
+		resolveResultType:   resolveResultSingleFunction,
+		functionMetadataMap: functionMetadataMap,
 	}
+
 	return &rr, nil
 }
 
-func (frr *functionReferenceResolver) delete(namespace string, fr *fission.FunctionReference) error {
-	nfr := namespacedFunctionReference{
-		namespace:         namespace,
-		functionReference: *fr,
+func (frr *functionReferenceResolver) resolveByFunctionWeights(namespace string, fr *fission.FunctionReference) (*resolveResult, error) {
+
+	functionMetadataMap := make(map[string]*metav1.ObjectMeta, 0)
+	fnWtDistrList := make([]FunctionWeightDistribution, 0)
+	sumPrefix := 0
+
+	for functionName, functionWeight := range fr.FunctionWeights {
+		// get function from cache
+		obj, isExist, err := frr.store.Get(&crd.Function{
+			Metadata: metav1.ObjectMeta{
+				Namespace: namespace,
+				Name:      functionName,
+			},
+		})
+		if err != nil {
+			return nil, err
+		}
+		if !isExist {
+			return nil, fmt.Errorf("function %v does not exist", functionName)
+		}
+
+		f := obj.(*crd.Function)
+		functionMetadataMap[f.Metadata.Name] = &f.Metadata
+		sumPrefix = sumPrefix + functionWeight
+		fnWtDistrList = append(fnWtDistrList, FunctionWeightDistribution{
+			name:      functionName,
+			weight:    functionWeight,
+			sumPrefix: sumPrefix,
+		})
+
+	}
+
+	rr := resolveResult{
+		resolveResultType:          resolveResultMultipleFunctions,
+		functionMetadataMap:        functionMetadataMap,
+		functionWtDistributionList: fnWtDistrList,
+	}
+
+	return &rr, nil
+}
+
+func (frr *functionReferenceResolver) delete(namespace string, triggerName, triggerRV string) error {
+	nfr := namespacedTriggerReference{
+		namespace:              namespace,
+		triggerName:            triggerName,
+		triggerResourceVersion: triggerRV,
 	}
 	return frr.refCache.Delete(nfr)
 }
 
-func (frr *functionReferenceResolver) copy() map[namespacedFunctionReference]resolveResult {
-	cache := make(map[namespacedFunctionReference]resolveResult)
+func (frr *functionReferenceResolver) copy() map[namespacedTriggerReference]resolveResult {
+	cache := make(map[namespacedTriggerReference]resolveResult)
 	for k, v := range frr.refCache.Copy() {
-		key := k.(namespacedFunctionReference)
+		key := k.(namespacedTriggerReference)
 		val := v.(resolveResult)
 		cache[key] = val
 	}

router/httpTriggers.go

@@ -119,7 +119,7 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router {
 		trigger := ts.triggers[i]
 
 		// resolve function reference
-		rr, err := ts.resolver.resolve(trigger.Metadata.Namespace, &trigger.Spec.FunctionReference)
+		rr, err := ts.resolver.resolve(trigger)
 		if err != nil {
 			// Unresolvable function reference. Report the error via
 			// the trigger's status.
@@ -135,22 +135,27 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router {
 			recorderName = recorder.Spec.Name
 		}
 
-		//log.Printf("The trigger %v should be recorded: %v", trigger.Metadata.Name, doRecord)
-
-		if rr.resolveResultType != resolveResultSingleFunction {
+		if rr.resolveResultType != resolveResultSingleFunction && rr.resolveResultType != resolveResultMultipleFunctions {
 			// not implemented yet
 			log.Panicf("resolve result type not implemented (%v)", rr.resolveResultType)
 		}
 
 		fh := &functionHandler{
-			fmap:                 ts.functionServiceMap,
-			frmap:                ts.recorderSet.functionRecorderMap,
-			trmap:                ts.recorderSet.triggerRecorderMap,
-			function:             rr.functionMetadata,
-			executor:             ts.executor,
-			httpTrigger:          &trigger,
-			tsRoundTripperParams: ts.tsRoundTripperParams,
-			recorderName:         recorderName,
+			fmap:                     ts.functionServiceMap,
+			frmap:                    ts.recorderSet.functionRecorderMap,
+			trmap:                    ts.recorderSet.triggerRecorderMap,
+			executor:                 ts.executor,
+			httpTrigger:              &trigger,
+			functionMetadataMap:      rr.functionMetadataMap,
+			fnWeightDistributionList: rr.functionWtDistributionList,
+			tsRoundTripperParams:     ts.tsRoundTripperParams,
+			recorderName:             recorderName,
+		}
+
+		if rr.resolveResultType == resolveResultSingleFunction {
+			for _, metadata := range fh.functionMetadataMap {
+				fh.function = metadata
+			}
 		}
 
 		ht := muxRouter.HandleFunc(trigger.Spec.RelativeURL, fh.handler)
@@ -270,12 +275,16 @@ func (ts *HTTPTriggerSet) initFunctionController() (k8sCache.Store, k8sCache.Con
 
 				// update resolver function reference cache
 				for key, rr := range ts.resolver.copy() {
-					if key.functionReference.Name == fn.Metadata.Name &&
-						rr.functionMetadata.ResourceVersion != fn.Metadata.ResourceVersion {
-						err := ts.resolver.delete(key.namespace, &key.functionReference)
+					if key.namespace == fn.Metadata.Namespace &&
+						rr.functionMetadataMap[fn.Metadata.Name] != nil &&
+						rr.functionMetadataMap[fn.Metadata.Name].ResourceVersion != fn.Metadata.ResourceVersion {
+						// invalidate resolver cache
+						log.Printf("Invalidating resolver cache")
+						err := ts.resolver.delete(key.namespace, key.triggerName, key.triggerResourceVersion)
 						if err != nil {
 							log.Printf("Error deleting functionReferenceResolver cache: %v", err)
 						}
+
 						break
 					}
 				}

router/router.go

@@ -124,13 +124,12 @@ func Start(port int, executorUrl string) {
 		log.Fatalf("Failed to parse max retry times: %v", err)
 	}
 
-	triggers, _, fnStore := makeHTTPTriggerSet(fmap, frmap, trmap, fissionClient, kubeClient, executor, restClient,
-		&tsRoundTripperParams{
-			timeout:         timeout,
-			timeoutExponent: timeoutExponent,
-			keepAlive:       keepAlive,
-			maxRetries:      maxRetries,
-		})
+	triggers, _, fnStore := makeHTTPTriggerSet(fmap, frmap, trmap, fissionClient, kubeClient, executor, restClient, &tsRoundTripperParams{
+		timeout:         timeout,
+		timeoutExponent: timeoutExponent,
+		keepAlive:       keepAlive,
+		maxRetries:      maxRetries,
+	})
 
 	resolver := makeFunctionReferenceResolver(fnStore)
 

router/router_test.go

@@ -46,18 +46,6 @@ func TestRouter(t *testing.T) {
 	fmap := makeFunctionServiceMap(0)
 	fmap.assign(fn, testServiceUrl)
 
-	// set up the resolver's cache for this function
-	frr := makeFunctionReferenceResolver(nil)
-	nfr := namespacedFunctionReference{
-		namespace:         metav1.NamespaceDefault,
-		functionReference: fr,
-	}
-	rr := resolveResult{
-		resolveResultType: resolveResultSingleFunction,
-		functionMetadata:  fn,
-	}
-	frr.refCache.Set(nfr, rr)
-
 	frmap := makeFunctionRecorderMap(time.Minute)
 
 	trmap := makeTriggerRecorderMap(time.Minute)
@@ -74,8 +62,9 @@ func TestRouter(t *testing.T) {
 	triggers.triggers = append(triggers.triggers,
 		crd.HTTPTrigger{
 			Metadata: metav1.ObjectMeta{
-				Name:      "xxx",
-				Namespace: metav1.NamespaceDefault,
+				Name:            "xxx",
+				Namespace:       metav1.NamespaceDefault,
+				ResourceVersion: "1234",
 			},
 			Spec: fission.HTTPTriggerSpec{
 				RelativeURL:       triggerUrl,
@@ -84,6 +73,23 @@ func TestRouter(t *testing.T) {
 			},
 		})
 
+	// set up the resolver's cache for this function
+	frr := makeFunctionReferenceResolver(nil)
+	nfr := namespacedTriggerReference{
+		namespace:              metav1.NamespaceDefault,
+		triggerName:            "xxx",
+		triggerResourceVersion: "1234",
+	}
+
+	fnMetaMap := make(map[string]*metav1.ObjectMeta, 1)
+	fnMetaMap[fn.Name] = fn
+
+	rr := resolveResult{
+		resolveResultType:   resolveResultSingleFunction,
+		functionMetadataMap: fnMetaMap,
+	}
+	frr.refCache.Set(nfr, rr)
+
 	// run the router
 	port := 4242
 	ctx, cancel := context.WithCancel(context.Background())