Use a cache key of IP/port

Gopkg.toml

@@ -4,7 +4,7 @@
 
 [[constraint]]
   name = "github.com/openfaas/faas"
-  version = "0.18.0"
+  version = "0.18.2"
 
 [[constraint]]
   name = "github.com/openfaas/faas-provider"

handlers/delete.go

@@ -10,8 +10,8 @@ import (
 	"net/http"
 
 	"github.com/gorilla/mux"
-
 	"github.com/openfaas/faas/gateway/requests"
+
 	log "github.com/sirupsen/logrus"
 )
 

handlers/info.go

@@ -9,9 +9,9 @@ import (
 )
 
 const (
-	//OrchestrationIdentifier identifier string for provider orchestration
+	// OrchestrationIdentifier identifier string for provider orchestration
 	OrchestrationIdentifier = "federation"
-	//ProviderName name of the provider
+	// ProviderName name of the provider
 	ProviderName = "faas-federation"
 )
 

handlers/logs.go

@@ -0,0 +1,52 @@
+// Copyright (c) OpenFaaS Author(s) 2019. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+package handlers
+
+import (
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net/http"
+
+	"github.com/openfaas-incubator/faas-federation/routing"
+	log "github.com/sirupsen/logrus"
+)
+
+// MakeLogHandler to read logs from an endpoint
+func MakeLogHandler(proxy http.HandlerFunc, providerLookup routing.ProviderLookup) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		log.Info("log handler")
+
+		query := r.URL.Query()
+		name := query.Get("name")
+		uri, err := providerLookup.Resolve(name)
+
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+
+		uriPath := uri.String() + "/system/logs?name=" + name
+
+		log.Infof("URI forwarding logs to: %s", uriPath)
+
+		req, _ := http.NewRequest(http.MethodGet, uriPath, nil)
+		res, resErr := http.DefaultClient.Do(req)
+
+		if resErr != nil {
+			http.Error(w, resErr.Error(), http.StatusInternalServerError)
+			return
+		}
+
+		if res.StatusCode != http.StatusOK && res.StatusCode != http.StatusContinue {
+			if res.Body != nil {
+				defer res.Body.Close()
+			}
+			http.Error(w, fmt.Sprintf("Incorrect HTTP status code: %d", res.StatusCode), http.StatusInternalServerError)
+			return
+		}
+
+		io.Copy(w, ioutil.NopCloser(res.Body))
+	}
+}

handlers/proxy.go

@@ -33,9 +33,8 @@ func MakeProxyHandler(proxy http.HandlerFunc) http.HandlerFunc {
 		functionName := strings.Split(r.URL.Path, "/")[2]
 		pathVars["name"] = functionName
 		pathVars["params"] = r.URL.Path
+		log.Infof("proxy request to: %s %s", functionName, r.URL.String())
 		proxy.ServeHTTP(w, r)
-
-		log.Infof("proxy request for function %s path %s", functionName, r.URL.String())
 	}
 }
 

handlers/reader.go

@@ -18,6 +18,7 @@ func MakeFunctionReader(providers []string) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 
 		log.Info("read request")
+
 		functions, err := routing.ReadServices(providers)
 		if err != nil {
 			log.Printf("Error getting service list: %s\n", err.Error())
@@ -34,6 +35,7 @@ func MakeFunctionReader(providers []string) http.HandlerFunc {
 
 		functionBytes, _ := json.Marshal(result)
 		w.Header().Set("Content-Type", "application/json")
+
 		w.WriteHeader(http.StatusOK)
 		w.Write(functionBytes)
 	}

handlers/replicas.go

@@ -8,6 +8,7 @@ import (
 	"net/http"
 
 	"github.com/gorilla/mux"
+	"github.com/openfaas-incubator/faas-federation/routing"
 	types "github.com/openfaas/faas-provider/types"
 	log "github.com/sirupsen/logrus"
 )
@@ -21,16 +22,25 @@ func MakeReplicaUpdater() http.HandlerFunc {
 }
 
 // MakeReplicaReader reads the amount of replicas for a deployment
-func MakeReplicaReader() http.HandlerFunc {
+func MakeReplicaReader(provider routing.ProviderLookup) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 		log.Info("read replicas")
 
 		vars := mux.Vars(r)
 		functionName := vars["name"]
 
+		res, ok := provider.GetFunction(functionName)
+
+		if !ok {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+
 		found := &types.FunctionStatus{}
 		found.Name = functionName
 		found.AvailableReplicas = 1
+		found.Annotations = res.Annotations
+		found.Labels = res.Labels
 
 		functionBytes, _ := json.Marshal(found)
 		w.Header().Set("Content-Type", "application/json")

main.go

@@ -64,11 +64,12 @@ func main() {
 		DeleteHandler:  handlers.MakeDeleteHandler(proxyFunc),
 		DeployHandler:  handlers.MakeDeployHandler(proxyFunc, providerLookup),
 		FunctionReader: handlers.MakeFunctionReader(cfg.Providers),
-		ReplicaReader:  handlers.MakeReplicaReader(),
+		ReplicaReader:  handlers.MakeReplicaReader(providerLookup),
 		ReplicaUpdater: handlers.MakeReplicaUpdater(),
 		UpdateHandler:  handlers.MakeUpdateHandler(proxyFunc, providerLookup),
 		HealthHandler:  handlers.MakeHealthHandler(),
 		InfoHandler:    handlers.MakeInfoHandler(version.BuildVersion(), version.GitCommitSHA),
+		LogHandler:     handlers.MakeLogHandler(proxyFunc, providerLookup),
 	}
 
 	bootstrapConfig := bootTypes.FaaSConfig{

routing/provider.go

@@ -55,7 +55,8 @@ func NewDefaultProviderRouting(providers []string, defaultProvider string) (Prov
 }
 
 func (d *defaultProviderRouting) ReloadCache() error {
-	log.Info("reloading cache starting...")
+	log.Info("reloading cache started...")
+
 	var urls []string
 	for _, v := range d.providers {
 		urls = append(urls, v.String())
@@ -83,6 +84,7 @@ func (d *defaultProviderRouting) ReloadCache() error {
 
 func (d *defaultProviderRouting) Resolve(functionName string) (providerURI *url.URL, err error) {
 	f, ok := d.GetFunction(functionName)
+
 	if !ok {
 		log.Warnf("can not find function %s in cache map, will attempt cache reload", functionName)
 		if err := d.ReloadCache(); err != nil {
@@ -95,6 +97,8 @@ func (d *defaultProviderRouting) Resolve(functionName string) (providerURI *url.
 		}
 	}
 
+	log.Infof("Fn: %s, annotations: %v", functionName, f.Annotations)
+
 	c, ok := (*f.Annotations)[federationProviderNameConstraint]
 	if !ok {
 		log.Infof("%s constraint not found using default provider %s", federationProviderNameConstraint, d.defaultProvider.String())
@@ -126,6 +130,7 @@ func ensureAnnotation(f *types.FunctionDeployment, defaultValue string) {
 
 func (d *defaultProviderRouting) matchBasedOnName(v string) *url.URL {
 	for _, u := range d.providers {
+
 		if strings.EqualFold(getHostNameWithoutPorts(u), v) {
 			return u
 		}
@@ -135,7 +140,8 @@ func (d *defaultProviderRouting) matchBasedOnName(v string) *url.URL {
 }
 
 func getHostNameWithoutPorts(v *url.URL) string {
-	return strings.Split(v.Host, ":")[0]
+	// return strings.Split(v.Host, ":")[0]
+	return v.String()
 }
 
 func (d *defaultProviderRouting) AddFunction(f *types.FunctionDeployment) {

routing/provider_test.go

@@ -28,8 +28,8 @@ func Test_defaultProviderRouting_Resolve(t *testing.T) {
 			name: "provider a is resolved",
 			fields: fields{
 				cache: map[string]*types.FunctionDeployment{
-					"echo": {Service: "echo", Annotations: &map[string]string{federationProviderNameConstraint: "faas-provider-a"}},
-					"cat":  {Service: "cat", Annotations: &map[string]string{federationProviderNameConstraint: "faas-provider-b"}},
+					"echo": {Service: "echo", Annotations: &map[string]string{federationProviderNameConstraint: "faas-provider-a:8080"}},
+					"cat":  {Service: "cat", Annotations: &map[string]string{federationProviderNameConstraint: "faas-provider-b:8080"}},
 				},
 				providers: map[string]*url.URL{
 					"faas-provider-a": parseURL("http://faas-provider-a:8080"),
@@ -42,12 +42,12 @@ func Test_defaultProviderRouting_Resolve(t *testing.T) {
 			name: "provider b is resolved",
 			fields: fields{
 				cache: map[string]*types.FunctionDeployment{
-					"echo": {Service: "echo", Annotations: &map[string]string{federationProviderNameConstraint: "faas-provider-a"}},
-					"cat":  {Service: "cat", Annotations: &map[string]string{federationProviderNameConstraint: "faas-provider-b"}},
+					"echo": {Service: "echo", Annotations: &map[string]string{federationProviderNameConstraint: "http://faas-provider-a:8080"}},
+					"cat":  {Service: "cat", Annotations: &map[string]string{federationProviderNameConstraint: "http://faas-provider-b:8080"}},
 				},
 				providers: map[string]*url.URL{
-					"faas-provider-a": parseURL("http://faas-provider-a:8080"),
-					"faas-provider-b": parseURL("http://faas-provider-b:8080"),
+					"http://faas-provider-a:8080": parseURL("http://faas-provider-a:8080"),
+					"http://faas-provider-b:8080": parseURL("http://faas-provider-b:8080"),
 				},
 				defaultProvider: "http://faas-provider-a:8080",
 			}, args: args{functionName: "cat"}, wantProviderHostName: "faas-provider-b:8080", wantErr: false,
@@ -56,7 +56,7 @@ func Test_defaultProviderRouting_Resolve(t *testing.T) {
 			name: "default provider is resolved, when constraint is missing",
 			fields: fields{
 				cache: map[string]*types.FunctionDeployment{
-					"echo": {Service: "echo", Annotations: &map[string]string{federationProviderNameConstraint: "faas-provider-a"}},
+					"echo": {Service: "echo", Annotations: &map[string]string{federationProviderNameConstraint: "faas-provider-a:8080"}},
 					"cat":  {Service: "cat", Annotations: &map[string]string{}},
 				},
 				providers: map[string]*url.URL{

routing/services.go

@@ -26,15 +26,20 @@ func ReadServices(providers []string) (*ReadServicesResult, error) {
 	}
 
 	results := Get(urls, len(providers))
-	serviceResult := &ReadServicesResult{Providers: map[string][]*types.FunctionStatus{}}
+
+	serviceResult := &ReadServicesResult{
+		Providers: map[string][]*types.FunctionStatus{},
+	}
+
 	for _, v := range results {
 		if v.Err != nil {
 			log.Errorf("error fetching function list for %s. %v", providers[v.Index], v.Err)
 			break
 		}
 
 		if v.Response.StatusCode > 399 {
-			log.Errorf("unexpected error code %d while fetching function list for %s. %v", v.Response.StatusCode, providers[v.Index], v.Err)
+			log.Errorf("unexpected error code %d while fetching function list for %s. %v",
+				v.Response.StatusCode, providers[v.Index], v.Err)
 			break
 		}
 
@@ -44,7 +49,8 @@ func ReadServices(providers []string) (*ReadServicesResult, error) {
 			return nil, fmt.Errorf("error reading response for %s. %v", providers[v.Index], err)
 		}
 
-		_ = v.Response.Body.Close()
+		defer v.Response.Body.Close()
+
 		err = json.Unmarshal(functionBytes, &function)
 		if err != nil {
 			return nil, fmt.Errorf("error unmarshalling response for %s. %v", providers[v.Index], err)
@@ -56,24 +62,24 @@ func ReadServices(providers []string) (*ReadServicesResult, error) {
 	return serviceResult, nil
 }
 
-func createToRequest(request *types.FunctionDeployment) *types.FunctionStatus {
-	return &types.FunctionStatus{
-		Name:              request.Service,
-		Annotations:       request.Annotations,
-		EnvProcess:        request.EnvProcess,
-		Image:             request.Image,
-		Labels:            request.Labels,
-		AvailableReplicas: 1,
-		Replicas:          1,
-	}
-}
+// func createToRequest(request *types.FunctionDeployment) *types.FunctionStatus {
+// 	return &types.FunctionStatus{
+// 		Name:              request.Service,
+// 		Annotations:       request.Annotations,
+// 		EnvProcess:        request.EnvProcess,
+// 		Image:             request.Image,
+// 		Labels:            request.Labels,
+// 		AvailableReplicas: 1,
+// 		Replicas:          1,
+// 	}
+// }
 
 func requestToCreate(f *types.FunctionStatus) *types.FunctionDeployment {
 	return &types.FunctionDeployment{
 		Service:     f.Name,
-		Annotations: f.Annotations,
-		Labels:      f.Labels,
 		Image:       f.Image,
 		EnvProcess:  f.EnvProcess,
+		Annotations: f.Annotations,
+		Labels:      f.Labels,
 	}
 }