Scale functions with namespace option

gateway/build.sh

@@ -42,4 +42,4 @@ docker build --build-arg https_proxy=$https_proxy --build-arg http_proxy=$http_p
   --build-arg VERSION="${VERSION:-dev}" \
   --build-arg GOARM="${GOARM}" \
   --build-arg ARCH="${arch}" \
-  -t $NS/gateway:$eTAG . -f $dockerfile --no-cache
+  -t $NS/gateway:$eTAG . -f $dockerfile

gateway/handlers/alerthandler.go

@@ -16,7 +16,7 @@ import (
 )
 
 // MakeAlertHandler handles alerts from Prometheus Alertmanager
-func MakeAlertHandler(service scaling.ServiceQuery) http.HandlerFunc {
+func MakeAlertHandler(service scaling.ServiceQuery, defaultNamespace string) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 
 		log.Println("Alert received.")
@@ -42,7 +42,7 @@ func MakeAlertHandler(service scaling.ServiceQuery) http.HandlerFunc {
 			return
 		}
 
-		errors := handleAlerts(&req, service)
+		errors := handleAlerts(&req, service, defaultNamespace)
 		if len(errors) > 0 {
 			log.Println(errors)
 			var errorOutput string
@@ -58,10 +58,10 @@ func MakeAlertHandler(service scaling.ServiceQuery) http.HandlerFunc {
 	}
 }
 
-func handleAlerts(req *requests.PrometheusAlert, service scaling.ServiceQuery) []error {
+func handleAlerts(req *requests.PrometheusAlert, service scaling.ServiceQuery, defaultNamespace string) []error {
 	var errors []error
 	for _, alert := range req.Alerts {
-		if err := scaleService(alert, service); err != nil {
+		if err := scaleService(alert, service, defaultNamespace); err != nil {
 			log.Println(err)
 			errors = append(errors, err)
 		}
@@ -70,12 +70,13 @@ func handleAlerts(req *requests.PrometheusAlert, service scaling.ServiceQuery) [
 	return errors
 }
 
-func scaleService(alert requests.PrometheusInnerAlert, service scaling.ServiceQuery) error {
+func scaleService(alert requests.PrometheusInnerAlert, service scaling.ServiceQuery, defaultNamespace string) error {
 	var err error
-	serviceName := alert.Labels.FunctionName
+
+	serviceName, namespace := getNamespace(defaultNamespace, alert.Labels.FunctionName)
 
 	if len(serviceName) > 0 {
-		queryResponse, getErr := service.GetReplicas(serviceName)
+		queryResponse, getErr := service.GetReplicas(serviceName, namespace)
 		if getErr == nil {
 			status := alert.Status
 
@@ -86,7 +87,7 @@ func scaleService(alert requests.PrometheusInnerAlert, service scaling.ServiceQu
 				return nil
 			}
 
-			updateErr := service.SetReplicas(serviceName, newReplicas)
+			updateErr := service.SetReplicas(serviceName, namespace, newReplicas)
 			if updateErr != nil {
 				err = updateErr
 			}

gateway/handlers/baseurlresolver_test.go

@@ -5,8 +5,10 @@ package handlers
 
 import (
 	"fmt"
+	"log"
 	"net/http"
 	"net/url"
+	"strings"
 	"testing"
 )
 
@@ -27,6 +29,28 @@ func TestSingleHostBaseURLResolver(t *testing.T) {
 
 const watchdogPort = 8080
 
+func TestFunctionAsHostBaseURLResolver_WithNamespaceOverride(t *testing.T) {
+
+	suffix := "openfaas-fn.local.cluster.svc."
+	namespace := "openfaas-fn"
+	newNS := "production-fn"
+
+	r := FunctionAsHostBaseURLResolver{FunctionSuffix: suffix, FunctionNamespace: namespace}
+
+	req, _ := http.NewRequest(http.MethodGet, "http://localhost/function/hello."+newNS, nil)
+
+	resolved := r.Resolve(req)
+
+	newSuffix := strings.Replace(suffix, namespace, newNS, -1)
+
+	want := fmt.Sprintf("http://hello.%s:%d", newSuffix, watchdogPort)
+	log.Println(want)
+	if resolved != want {
+		t.Logf("r.Resolve failed, want: %s got: %s", want, resolved)
+		t.Fail()
+	}
+}
+
 func TestFunctionAsHostBaseURLResolver_WithSuffix(t *testing.T) {
 	suffix := "openfaas-fn.local.cluster.svc."
 	r := FunctionAsHostBaseURLResolver{FunctionSuffix: suffix}
@@ -35,7 +59,7 @@ func TestFunctionAsHostBaseURLResolver_WithSuffix(t *testing.T) {
 
 	resolved := r.Resolve(req)
 	want := fmt.Sprintf("http://hello.%s:%d", suffix, watchdogPort)
-
+	log.Println(want)
 	if resolved != want {
 		t.Logf("r.Resolve failed, want: %s got: %s", want, resolved)
 		t.Fail()

gateway/handlers/forwarding_proxy.go

@@ -179,7 +179,8 @@ func (s SingleHostBaseURLResolver) Resolve(r *http.Request) string {
 
 // FunctionAsHostBaseURLResolver resolves URLs using a function from the URL as a host
 type FunctionAsHostBaseURLResolver struct {
-	FunctionSuffix string
+	FunctionSuffix    string
+	FunctionNamespace string
 }
 
 // Resolve the base URL for a request
@@ -188,8 +189,13 @@ func (f FunctionAsHostBaseURLResolver) Resolve(r *http.Request) string {
 
 	const watchdogPort = 8080
 	var suffix string
+
 	if len(f.FunctionSuffix) > 0 {
-		suffix = "." + f.FunctionSuffix
+		if index := strings.LastIndex(svcName, "."); index > -1 && len(svcName) > index+1 {
+			suffix = strings.Replace(f.FunctionSuffix, f.FunctionNamespace, "", -1)
+		} else {
+			suffix = "." + f.FunctionSuffix
+		}
 	}
 
 	return fmt.Sprintf("http://%s%s:%d", svcName, suffix, watchdogPort)

gateway/handlers/forwarding_proxy_test.go

@@ -135,6 +135,11 @@ func Test_getServiceName(t *testing.T) {
 			url:         "/function/testFunc",
 			serviceName: "testFunc",
 		},
+		{
+			name:        "includes namespace",
+			url:         "/function/test1.fn",
+			serviceName: "test1.fn",
+		},
 		{
 			name:        "can handle request with trailing slash",
 			url:         "/function/testFunc/",

gateway/handlers/function_prefix_trimming_url_path_transformer_test.go

@@ -32,6 +32,30 @@ func Test_Transform_RemovesFunctionPrefixWithSingleParam(t *testing.T) {
 	}
 }
 
+func Test_Transform_RemovesFunctionPrefixWithDotInName(t *testing.T) {
+
+	req, _ := http.NewRequest(http.MethodGet, "/function/figlet.fn", nil)
+	transformer := FunctionPrefixTrimmingURLPathTransformer{}
+	want := ""
+	got := transformer.Transform(req)
+
+	if want != got {
+		t.Errorf("want: %s, got: %s", want, got)
+	}
+}
+
+func Test_Transform_RemovesFunctionPrefixWithDotInNameAndPath(t *testing.T) {
+
+	req, _ := http.NewRequest(http.MethodGet, "/function/figlet.fn/employees", nil)
+	transformer := FunctionPrefixTrimmingURLPathTransformer{}
+	want := "/employees"
+	got := transformer.Transform(req)
+
+	if want != got {
+		t.Errorf("want: %s, got: %s", want, got)
+	}
+}
+
 func Test_Transform_RemovesFunctionPrefixWithParams(t *testing.T) {
 
 	req, _ := http.NewRequest(http.MethodGet, "/function/figlet/employees/100", nil)

gateway/handlers/namespaces_test.go

@@ -0,0 +1,45 @@
+// Copyright (c) Alex Ellis 2017. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+package handlers
+
+import "testing"
+
+func Test_getNamespace_Default(t *testing.T) {
+	root, ns := getNamespace("openfaas-fn", "figlet.openfaas-fn")
+	wantRoot := "figlet"
+	wantNs := "openfaas-fn"
+
+	if root != wantRoot {
+		t.Errorf("function root: want %s, got %s", wantRoot, root)
+	}
+	if ns != wantNs {
+		t.Errorf("function ns: want %s, got %s", wantNs, ns)
+	}
+}
+
+func Test_getNamespace_Override(t *testing.T) {
+	root, ns := getNamespace("fn", "figlet.fn")
+	wantRoot := "figlet"
+	wantNs := "fn"
+
+	if root != wantRoot {
+		t.Errorf("function root: want %s, got %s", wantRoot, root)
+	}
+	if ns != wantNs {
+		t.Errorf("function ns: want %s, got %s", wantNs, ns)
+	}
+}
+
+func Test_getNamespace_Empty(t *testing.T) {
+	root, ns := getNamespace("", "figlet")
+	wantRoot := "figlet"
+	wantNs := ""
+
+	if root != wantRoot {
+		t.Errorf("function root: want %s, got %s", wantRoot, root)
+	}
+	if ns != wantNs {
+		t.Errorf("function ns: want %s, got %s", wantNs, ns)
+	}
+}

gateway/handlers/notifiers.go

@@ -66,7 +66,7 @@ func getServiceName(urlValue string) string {
 	forward := "/function/"
 	if strings.HasPrefix(urlValue, forward) {
 		// With a path like `/function/xyz/rest/of/path?q=a`, the service
-		// name we wish to locate is just the `xyz` portion.  With a postive
+		// name we wish to locate is just the `xyz` portion.  With a positive
 		// match on the regex below, it will return a three-element slice.
 		// The item at index `0` is the same as `urlValue`, at `1`
 		// will be the service name we need, and at `2` the rest of the path.

gateway/handlers/scaling.go

@@ -7,36 +7,45 @@ import (
 	"fmt"
 	"log"
 	"net/http"
+	"strings"
 
 	"github.com/openfaas/faas/gateway/scaling"
 )
 
+func getNamespace(defaultNamespace, fullName string) (string, string) {
+	if index := strings.LastIndex(fullName, "."); index > -1 {
+		return fullName[:index], fullName[index+1:]
+	}
+	return fullName, defaultNamespace
+}
+
 // MakeScalingHandler creates handler which can scale a function from
 // zero to N replica(s). After scaling the next http.HandlerFunc will
 // be called. If the function is not ready after the configured
 // amount of attempts / queries then next will not be invoked and a status
 // will be returned to the client.
-func MakeScalingHandler(next http.HandlerFunc, config scaling.ScalingConfig) http.HandlerFunc {
+func MakeScalingHandler(next http.HandlerFunc, config scaling.ScalingConfig, defaultNamespace string) http.HandlerFunc {
 
 	scaler := scaling.NewFunctionScaler(config)
 
 	return func(w http.ResponseWriter, r *http.Request) {
 
-		functionName := getServiceName(r.URL.String())
-		res := scaler.Scale(functionName)
+		functionName, namespace := getNamespace(defaultNamespace, getServiceName(r.URL.String()))
+
+		res := scaler.Scale(functionName, namespace)
 
 		if !res.Found {
-			errStr := fmt.Sprintf("error finding function %s: %s", functionName, res.Error.Error())
-			log.Printf("Scaling: %s", errStr)
+			errStr := fmt.Sprintf("error finding function %s.%s: %s", functionName, namespace, res.Error.Error())
+			log.Printf("Scaling: %s\n", errStr)
 
 			w.WriteHeader(http.StatusNotFound)
 			w.Write([]byte(errStr))
 			return
 		}
 
 		if res.Error != nil {
-			errStr := fmt.Sprintf("error finding function %s: %s", functionName, res.Error.Error())
-			log.Printf("Scaling: %s", errStr)
+			errStr := fmt.Sprintf("error finding function %s.%s: %s", functionName, namespace, res.Error.Error())
+			log.Printf("Scaling: %s\n", errStr)
 
 			w.WriteHeader(http.StatusInternalServerError)
 			w.Write([]byte(errStr))
@@ -48,6 +57,6 @@ func MakeScalingHandler(next http.HandlerFunc, config scaling.ScalingConfig) htt
 			return
 		}
 
-		log.Printf("[Scale] function=%s 0=>N timed-out after %f seconds", functionName, res.Duration.Seconds())
+		log.Printf("[Scale] function=%s.%s 0=>N timed-out after %f seconds\n", functionName, namespace, res.Duration.Seconds())
 	}
 }

gateway/plugin/external.go

@@ -54,20 +54,21 @@ type ExternalServiceQuery struct {
 
 // ScaleServiceRequest request scaling of replica
 type ScaleServiceRequest struct {
-	ServiceName string `json:"serviceName"`
-	Replicas    uint64 `json:"replicas"`
+	ServiceName      string `json:"serviceName"`
+	ServiceNamespace string `json:"serviceNamespace"`
+	Replicas         uint64 `json:"replicas"`
 }
 
 // GetReplicas replica count for function
-func (s ExternalServiceQuery) GetReplicas(serviceName string) (scaling.ServiceQueryResponse, error) {
+func (s ExternalServiceQuery) GetReplicas(serviceName, serviceNamespace string) (scaling.ServiceQueryResponse, error) {
 	start := time.Now()
 
 	var err error
 	var emptyServiceQueryResponse scaling.ServiceQueryResponse
 
 	function := types.FunctionStatus{}
 
-	urlPath := fmt.Sprintf("%ssystem/function/%s", s.URL.String(), serviceName)
+	urlPath := fmt.Sprintf("%ssystem/function/%s?namespace=%s", s.URL.String(), serviceName, serviceNamespace)
 
 	req, _ := http.NewRequest(http.MethodGet, urlPath, nil)
 
@@ -91,8 +92,10 @@ func (s ExternalServiceQuery) GetReplicas(serviceName string) (scaling.ServiceQu
 			if err != nil {
 				log.Println(urlPath, err)
 			}
+			log.Printf("GetReplicas [%s.%s] took: %fs", serviceName, serviceNamespace, time.Since(start).Seconds())
+
 		} else {
-			log.Printf("GetReplicas took: %fs", time.Since(start).Seconds())
+			log.Printf("GetReplicas [%s.%s] took: %fs, code: %d\n", serviceName, serviceNamespace, time.Since(start).Seconds(), res.StatusCode)
 			return emptyServiceQueryResponse, fmt.Errorf("server returned non-200 status code (%d) for function, %s", res.StatusCode, serviceName)
 		}
 	}
@@ -115,8 +118,7 @@ func (s ExternalServiceQuery) GetReplicas(serviceName string) (scaling.ServiceQu
 			log.Printf("Bad Scaling Factor: %d, is not in range of [0 - 100]. Will fallback to %d", extractedScalingFactor, scalingFactor)
 		}
 	}
-
-	log.Printf("GetReplicas took: %fs", time.Since(start).Seconds())
+	log.Printf("GetReplicas [%s.%s] took: %fs", serviceName, serviceNamespace, time.Since(start).Seconds())
 
 	return scaling.ServiceQueryResponse{
 		Replicas:          function.Replicas,
@@ -128,20 +130,22 @@ func (s ExternalServiceQuery) GetReplicas(serviceName string) (scaling.ServiceQu
 }
 
 // SetReplicas update the replica count
-func (s ExternalServiceQuery) SetReplicas(serviceName string, count uint64) error {
+func (s ExternalServiceQuery) SetReplicas(serviceName, serviceNamespace string, count uint64) error {
 	var err error
 
 	scaleReq := ScaleServiceRequest{
-		ServiceName: serviceName,
-		Replicas:    count,
+		ServiceName:      serviceName,
+		Replicas:         count,
+		ServiceNamespace: serviceNamespace,
 	}
 
 	requestBody, err := json.Marshal(scaleReq)
 	if err != nil {
 		return err
 	}
 
-	urlPath := fmt.Sprintf("%ssystem/scale-function/%s", s.URL.String(), serviceName)
+	start := time.Now()
+	urlPath := fmt.Sprintf("%ssystem/scale-function/%s?namespace=%s", s.URL.String(), serviceName, serviceNamespace)
 	req, _ := http.NewRequest(http.MethodPost, urlPath, bytes.NewReader(requestBody))
 
 	if s.AuthInjector != nil {
@@ -163,6 +167,8 @@ func (s ExternalServiceQuery) SetReplicas(serviceName string, count uint64) erro
 		err = fmt.Errorf("error scaling HTTP code %d, %s", res.StatusCode, urlPath)
 	}
 
+	log.Printf("SetReplicas [%s.%s] took: %fs", serviceName, serviceNamespace, time.Since(start).Seconds())
+
 	return err
 }