Archives bigger than 256K size need env variable for uploading (#697)

For archives bigger than 256K, the Storage service was called from the client side, this needed few environment variables to be set. This change uses port forwarding to achieve the same and does not need environment variables to be set.

charts/fission-all/templates/deployment.yaml

@@ -126,6 +126,10 @@ spec:
         env:
           - name: FISSION_FUNCTION_NAMESPACE
             value: "{{ .Values.functionNamespace }}"
+          - name: POD_NAMESPACE
+            valueFrom:
+              fieldRef:
+                fieldPath: metadata.namespace
         readinessProbe:
           httpGet:
             path: "/healthz"
@@ -595,6 +599,7 @@ spec:
     metadata:
       labels:
         svc: storagesvc
+        application: fission-storage
     spec:
       containers:
       - name: storagesvc

charts/fission-all/templates/svc.yaml

@@ -65,6 +65,7 @@ metadata:
   name: storagesvc
   labels:
     svc: storagesvc
+    application: fission-storage
     chart: "{{ .Chart.Name }}-{{ .Chart.Version }}"
 spec:
   type: ClusterIP

charts/fission-core/templates/deployment.yaml

@@ -127,6 +127,10 @@ spec:
         env:
           - name: FISSION_FUNCTION_NAMESPACE
             value: "{{ .Values.functionNamespace }}"
+          - name: POD_NAMESPACE
+            valueFrom:
+              fieldRef:
+                fieldPath: metadata.namespace
         readinessProbe:
           httpGet:
             path: "/healthz"
@@ -335,6 +339,7 @@ spec:
     metadata:
       labels:
         svc: storagesvc
+        application: fission-storage
     spec:
       containers:
       - name: storagesvc

charts/fission-core/templates/svc.yaml

@@ -44,6 +44,7 @@ metadata:
   name: storagesvc
   labels:
     svc: storagesvc
+    application: fission-storage
     chart: "{{ .Chart.Name }}-{{ .Chart.Version }}"
 spec:
   type: ClusterIP

controller/api.go

@@ -35,6 +35,15 @@ import (
 	"github.com/fission/fission/fission/logdb"
 )
 
+var podNamespace string
+
+func init() {
+	podNamespace = os.Getenv("POD_NAMESPACE")
+	if podNamespace == "" {
+		podNamespace = "fission"
+	}
+}
+
 type (
 	API struct {
 		fissionClient     *crd.FissionClient
@@ -167,6 +176,18 @@ func (api *API) HealthHandler(w http.ResponseWriter, r *http.Request) {
 	w.WriteHeader(http.StatusOK)
 }
 
+func (api *API) GetSvcName(w http.ResponseWriter, r *http.Request) {
+	appLabelSelector := "application=" + r.URL.Query().Get("application")
+	services, err := api.kubernetesClient.CoreV1().Services(podNamespace).List(metav1.ListOptions{
+		LabelSelector: appLabelSelector,
+	})
+	if err != nil || len(services.Items) > 1 || len(services.Items) == 0 {
+		api.respondWithError(w, err)
+	}
+	service := services.Items[0]
+	fmt.Fprintf(w, service.Name+"."+podNamespace)
+}
+
 func (api *API) Serve(port int) {
 	r := mux.NewRouter()
 	r.HandleFunc("/healthz", api.HealthHandler).Methods("GET")
@@ -223,6 +244,7 @@ func (api *API) Serve(port int) {
 	r.HandleFunc("/proxy/storage/v1/archive", api.StorageServiceProxy)
 	r.HandleFunc("/proxy/logs/{function}", api.FunctionPodLogs).Methods("POST")
 	r.HandleFunc("/proxy/workflows-apiserver/{path:.*}", api.WorkflowApiserverProxy)
+	r.HandleFunc("/proxy/svcname", api.GetSvcName).Queries("application", "").Methods("GET")
 
 	address := fmt.Sprintf(":%v", port)
 

controller/client/core.go

@@ -19,6 +19,7 @@ package client
 import (
 	"encoding/json"
 	"fmt"
+	"io/ioutil"
 	"net/http"
 
 	apiv1 "k8s.io/api/core/v1"
@@ -72,3 +73,27 @@ func (c *Client) ConfigMapGet(m *metav1.ObjectMeta) (*apiv1.ConfigMap, error) {
 
 	return &configMap, nil
 }
+
+func (c *Client) GetSvcURL(label string) (string, error) {
+	url := fmt.Sprintf("%s/proxy/svcname?"+label, c.Url)
+
+	resp, err := http.Get(url)
+
+	if err != nil {
+		return "", err
+	}
+
+	if resp == nil {
+		return "", fmt.Errorf("Failed to find service for given label: %v", label)
+	}
+
+	defer resp.Body.Close()
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return "", err
+	}
+
+	storageSvc := string(body)
+
+	return storageSvc, err
+}

fission/common.go

@@ -149,20 +149,6 @@ func createArchive(client *client.Client, fileName string, specFile string) *fis
 		archive.Type = fission.ArchiveTypeLiteral
 		archive.Literal = contents
 	} else {
-		// make a kubernetes client
-		_, kubeClient, _, err := crd.GetKubernetesClient()
-		if err != nil {
-			log.Fatal(err.Error())
-		}
-
-		fissionNamespace := os.Getenv("FISSION_NAMESPACE")
-
-		// get svc end point for storagesvc
-		service, err := kubeClient.CoreV1().Services(fissionNamespace).Get("storagesvc", metav1.GetOptions{})
-		if err != nil {
-			log.Fatal(fmt.Sprintf("Error getting storage service object from kubernetes :%v", err.Error()))
-		}
-
 		u := strings.TrimSuffix(client.Url, "/") + "/proxy/storage"
 		ssClient := storageSvcClient.MakeClient(u)
 
@@ -170,13 +156,17 @@ func createArchive(client *client.Client, fileName string, specFile string) *fis
 		id, err := ssClient.Upload(fileName, nil)
 		checkErr(err, fmt.Sprintf("upload file %v", fileName))
 
-		// this needs to be storagesvc.fission
-		storageSvcEndpoint := fmt.Sprintf("http://%s.%s/", service.Name, service.Namespace)
-		storageServiceClient := storageSvcClient.MakeClient(storageSvcEndpoint)
-		archiveUrl := storageServiceClient.GetUrl(id)
+		storageSvc, err := client.GetSvcURL("application=fission-storage")
+		storageSvcURL := "http://" + storageSvc
+		checkErr(err, "get fission storage service name")
+
+		// We make a new client with actual URL of Storage service so that the URL is not
+		// pointing to 127.0.0.1 i.e. proxy. DON'T reuse previous ssClient
+		pkgClient := storageSvcClient.MakeClient(storageSvcURL)
+		archiveURL := pkgClient.GetUrl(id)
 
 		archive.Type = fission.ArchiveTypeUrl
-		archive.URL = archiveUrl
+		archive.URL = archiveURL
 
 		csum, err := fileChecksum(fileName)
 		checkErr(err, fmt.Sprintf("calculate checksum for file %v", fileName))

fission/main.go

@@ -48,6 +48,10 @@ func getKubeConfigPath() string {
 }
 
 func getServerUrl() string {
+	return getApplicationUrl("application=fission-api")
+}
+
+func getApplicationUrl(selector string) string {
 	var serverUrl string
 	// Use FISSION_URL env variable if set; otherwise, port-forward to controller.
 	fissionUrl := os.Getenv("FISSION_URL")