Fix CLI unable to get pod logs from controller (#1451)

pkg/controller/client/function.go

@@ -19,10 +19,15 @@ package client
 import (
 	"encoding/json"
 	"fmt"
+	"io"
+	"net/http"
+	"net/url"
 
+	"github.com/pkg/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
+	"github.com/fission/fission/pkg/fission-cli/console"
 )
 
 func (c *Client) FunctionCreate(f *fv1.Function) (*metav1.ObjectMeta, error) {
@@ -152,3 +157,29 @@ func (c *Client) FunctionList(functionNamespace string) ([]fv1.Function, error)
 
 	return funcs, nil
 }
+
+func (c *Client) FunctionPodLogs(m *metav1.ObjectMeta) (io.ReadCloser, int, error) {
+	relativeUrl := fmt.Sprintf("functions/%v", m.Name)
+	relativeUrl += fmt.Sprintf("?namespace=%v", m.Namespace)
+
+	queryURL, err := url.Parse(c.Url)
+	if err != nil {
+		return nil, 0, errors.Wrapf(err, "error parsing the base URL '%v'", c.Url)
+	}
+	queryURL.Path = fmt.Sprintf("/proxy/logs/%s", m.Name)
+
+	console.Verbose(2, fmt.Sprintf("Try to get pod logs from controller '%v'", queryURL.String()))
+
+	req, err := http.NewRequest(http.MethodPost, queryURL.String(), nil)
+	if err != nil {
+		return nil, 0, errors.Wrap(err, "error creating logs request")
+	}
+
+	httpClient := http.Client{}
+	resp, err := httpClient.Do(req)
+	if err != nil {
+		return nil, 0, errors.Wrap(err, "error executing get logs request")
+	}
+
+	return resp.Body, resp.StatusCode, nil
+}

pkg/controller/functionApi.go

@@ -18,12 +18,15 @@ package controller
 
 import (
 	"encoding/json"
+	"fmt"
 	"io"
 	"io/ioutil"
 	"net/http"
 	"net/http/httputil"
 	"net/url"
 	"sort"
+	"strconv"
+	"strings"
 
 	"github.com/emicklei/go-restful"
 	restfulspec "github.com/emicklei/go-restful-openapi"
@@ -33,10 +36,12 @@ import (
 	"go.uber.org/zap"
 	apiv1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	restclient "k8s.io/client-go/rest"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/kubernetes"
 
 	fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
 	ferror "github.com/fission/fission/pkg/error"
+	"github.com/fission/fission/pkg/types"
 )
 
 func RegisterFunctionRoute(ws *restful.WebService) {
@@ -280,27 +285,35 @@ func (a *API) FunctionLogsApiPost(w http.ResponseWriter, r *http.Request) {
 func (a *API) FunctionPodLogs(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
 	fnName := vars["function"]
-	ns := vars["namespace"]
+
+	ns := a.extractQueryParamFromRequest(r, "namespace")
+	podNs := "fission-function"
 
 	if len(ns) == 0 {
-		ns = "fission-function"
+		ns = metav1.NamespaceDefault
+	} else if ns != metav1.NamespaceDefault {
+		// If the function namespace is "default", executor
+		// will create function pods under "fission-function".
+		// Otherwise, the function pod will be created under
+		// the same namespace of function.
+		podNs = ns
 	}
 
 	f, err := a.fissionClient.Functions(ns).Get(fnName)
-	if err != nil {
-		a.respondWithError(w, err)
-		return
-	}
-	envName := f.Spec.Environment.Name
-
 	if err != nil {
 		a.respondWithError(w, err)
 		return
 	}
 
 	// Get function Pods first
-	selector := "functionName=" + fnName
+	selector := map[string]string{
-	podList, err := a.kubernetesClient.CoreV1().Pods(ns).List(metav1.ListOptions{LabelSelector: selector})
+		types.FUNCTION_UID:          string(f.Metadata.UID),
+		types.ENVIRONMENT_NAME:      f.Spec.Environment.Name,
+		types.ENVIRONMENT_NAMESPACE: f.Spec.Environment.Namespace,
+	}
+	podList, err := a.kubernetesClient.CoreV1().Pods(podNs).List(metav1.ListOptions{
+		LabelSelector: labels.Set(selector).AsSelector().String(),
+	})
 	if err != nil {
 		a.respondWithError(w, err)
 		return
@@ -309,30 +322,47 @@ func (a *API) FunctionPodLogs(w http.ResponseWriter, r *http.Request) {
 	// Get the logs for last Pod executed
 	pods := podList.Items
 	sort.Slice(pods, func(i, j int) bool {
-		itime := pods[i].ObjectMeta.CreationTimestamp.Time
+		rv1, _ := strconv.ParseInt(pods[i].ObjectMeta.ResourceVersion, 10, 32)
-		jtime := pods[j].ObjectMeta.CreationTimestamp.Time
+		rv2, _ := strconv.ParseInt(pods[j].ObjectMeta.ResourceVersion, 10, 32)
-		return itime.After(jtime)
+		return rv1 > rv2
 	})
 
-	podLogOpts := apiv1.PodLogOptions{Container: envName} // Only the env container, not fetcher
+	if len(pods) <= 0 {
-	var podLogsReq *restclient.Request
-	if len(pods) > 0 {
-		podLogsReq = a.kubernetesClient.CoreV1().Pods(ns).GetLogs(pods[0].ObjectMeta.Name, &podLogOpts)
-	} else {
 		a.respondWithError(w, errors.New("no active pods found"))
 		return
 	}
 
-	podLogs, err := podLogsReq.Stream()
+	// get the pod with highest resource version
+	err = getContainerLog(a.kubernetesClient, w, f, &pods[0])
 	if err != nil {
-		a.respondWithError(w, err)
+		a.respondWithError(w, errors.Wrapf(err, "error getting container logs"))
-		return
-	}
-	defer podLogs.Close()
-
-	_, err = io.Copy(w, podLogs)
-	if err != nil {
-		a.respondWithError(w, err)
 		return
 	}
 }
+
+func getContainerLog(kubernetesClient *kubernetes.Clientset, w http.ResponseWriter, fn *fv1.Function, pod *apiv1.Pod) error {
+	seq := strings.Repeat("=", 35)
+
+	for _, container := range pod.Spec.Containers {
+		podLogOpts := apiv1.PodLogOptions{Container: container.Name} // Only the env container, not fetcher
+		podLogsReq := kubernetesClient.CoreV1().Pods(pod.Namespace).GetLogs(pod.ObjectMeta.Name, &podLogOpts)
+
+		podLogs, err := podLogsReq.Stream()
+		if err != nil {
+			return errors.Wrapf(err, "error streaming pod log")
+		}
+
+		msg := fmt.Sprintf("\n%v\nFunction: %v\nEnvironment: %v\nNamespace: %v\nPod: %v\nContainer: %v\nNode: %v\n%v\n", seq,
+			fn.Metadata.Name, fn.Spec.Environment.Name, pod.Namespace, pod.Name, container.Name, pod.Spec.NodeName, seq)
+		w.Write([]byte(msg))
+
+		_, err = io.Copy(w, podLogs)
+		if err != nil {
+			return errors.Wrapf(err, "error copying pod log")
+		}
+
+		podLogs.Close()
+	}
+
+	return nil
+}

pkg/fission-cli/cmd/function/test.go

@@ -102,16 +102,13 @@ func (opts *TestSubCommand) do(input cli.Input) error {
 		functionUrl.RawQuery = query.Encode()
 	}
 
-	ctx := context.Background()
+	ctx, closeCtx := context.WithTimeout(context.Background(), input.Duration(flagkey.FnTestTimeout))
-	if deadline := input.Duration(flagkey.FnTestTimeout); deadline > 0 {
+	defer closeCtx()
-		var closeCtx func()
-		ctx, closeCtx = context.WithTimeout(ctx, deadline)
-		defer closeCtx()
-	}
 
-	headers := input.StringSlice(flagkey.FnTestHeader)
+	resp, err := doHTTPRequest(ctx, functionUrl.String(),
-
+		input.StringSlice(flagkey.FnTestHeader),
-	resp, err := doHTTPRequest(ctx, input.String(flagkey.HtMethod), functionUrl.String(), input.String(flagkey.FnTestBody), headers)
+		input.String(flagkey.HtMethod),
+		input.String(flagkey.FnTestBody))
 	if err != nil {
 		return err
 	}
@@ -123,21 +120,25 @@ func (opts *TestSubCommand) do(input cli.Input) error {
 	}
 
 	if resp.StatusCode < 400 {
-		fmt.Print(string(body))
+		os.Stdout.Write(body)
 		return nil
 	}
 
-	fmt.Printf("Error calling function %s: %d; Please try again or fix the error: %s", m.Name, resp.StatusCode, string(body))
+	console.Errorf("Error calling function %s: %d; Please try again or fix the error: %s\n", m.Name, resp.StatusCode, string(body))
-	err = printPodLogs(input)
+	log, err := printPodLogs(opts.client, m)
 	if err != nil {
-		fmt.Printf("Error getting function logs from pod: %v. Try to get logs from log database", err)
+		console.Errorf("Error getting function logs from controller: %v. Try to get logs from log database.", err)
-		return Log(input)
+		err = Log(input)
+		if err != nil {
+			return errors.Wrapf(err, "error retrieving function log from log database")
+		}
+	} else {
+		console.Info(log)
 	}
-
+	return errors.New("error getting function response")
-	return nil
 }
 
-func doHTTPRequest(ctx context.Context, method, url, body string, headers []string) (*http.Response, error) {
+func doHTTPRequest(ctx context.Context, url string, headers []string, method, body string) (*http.Response, error) {
 	method, err := httptrigger.GetMethod(method)
 	if err != nil {
 		return nil, err
@@ -163,41 +164,21 @@ func doHTTPRequest(ctx context.Context, method, url, body string, headers []stri
 	return resp, nil
 }
 
-func printPodLogs(input cli.Input) error {
+func printPodLogs(client *client.Client, fnMeta *metav1.ObjectMeta) (string, error) {
-	fnName := input.String(flagkey.FnName)
+	reader, statusCode, err := client.FunctionPodLogs(fnMeta)
-
-	u, err := util.GetApplicationUrl("application=fission-api")
 	if err != nil {
-		return err
+		return "", errors.Wrap(err, "error executing get logs request")
 	}
+	defer reader.Close()
 
-	queryURL, err := url.Parse(u)
+	body, err := ioutil.ReadAll(reader)
 	if err != nil {
-		return errors.Wrap(err, "error parsing the base URL")
+		return "", errors.Wrap(err, "error reading the response body")
-	}
-	queryURL.Path = fmt.Sprintf("/proxy/logs/%s", fnName)
-
-	req, err := http.NewRequest(http.MethodPost, queryURL.String(), nil)
-	if err != nil {
-		return errors.Wrap(err, "error creating logs request")
 	}
 
-	httpClient := http.Client{}
+	if statusCode != http.StatusOK {
-	resp, err := httpClient.Do(req)
+		return string(body), errors.Errorf("error getting logs from controller, status code: '%v'", statusCode)
-	if err != nil {
-		return errors.Wrap(err, "execute get logs request")
 	}
 
-	defer resp.Body.Close()
+	return string(body), nil
-	if resp.StatusCode != http.StatusOK {
-		return errors.New("get logs from pod directly")
-	}
-
-	body, err := ioutil.ReadAll(resp.Body)
-	if err != nil {
-		return errors.Wrap(err, "read the response body")
-	}
-
-	fmt.Println(string(body))
-	return nil
 }

pkg/fission-cli/console/log.go

@@ -33,6 +33,11 @@ func Error(msg interface{}) {
 	os.Stderr.WriteString(fmt.Sprintf("%v: %v\n", color.RedString("Error"), trimNewline(msg)))
 }
 
+func Errorf(format string, args ...interface{}) {
+	msg := fmt.Sprintf(format, args...)
+	os.Stderr.WriteString(fmt.Sprintf("%v: %v\n", color.RedString("Error"), trimNewline(msg)))
+}
+
 func Warn(msg interface{}) {
 	os.Stdout.WriteString(fmt.Sprintf("%v: %v\n", color.YellowString("Warning"), trimNewline(msg)))
 }