diff --git a/pkg/controller/client/function.go b/pkg/controller/client/function.go index 98fe083f..7ca7c6c6 100644 --- a/pkg/controller/client/function.go +++ b/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 +} diff --git a/pkg/controller/functionApi.go b/pkg/controller/functionApi.go index 6d3b467e..e96395be 100644 --- a/pkg/controller/functionApi.go +++ b/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 - podList, err := a.kubernetesClient.CoreV1().Pods(ns).List(metav1.ListOptions{LabelSelector: selector}) + selector := map[string]string{ + 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 - jtime := pods[j].ObjectMeta.CreationTimestamp.Time - return itime.After(jtime) + rv1, _ := strconv.ParseInt(pods[i].ObjectMeta.ResourceVersion, 10, 32) + rv2, _ := strconv.ParseInt(pods[j].ObjectMeta.ResourceVersion, 10, 32) + return rv1 > rv2 }) - podLogOpts := apiv1.PodLogOptions{Container: envName} // Only the env container, not fetcher - var podLogsReq *restclient.Request - if len(pods) > 0 { - podLogsReq = a.kubernetesClient.CoreV1().Pods(ns).GetLogs(pods[0].ObjectMeta.Name, &podLogOpts) - } else { + if len(pods) <= 0 { 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) - return - } - defer podLogs.Close() - - _, err = io.Copy(w, podLogs) - if err != nil { - a.respondWithError(w, err) + a.respondWithError(w, errors.Wrapf(err, "error getting container logs")) 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 +} diff --git a/pkg/fission-cli/cmd/function/test.go b/pkg/fission-cli/cmd/function/test.go index 401ab129..b97d1eef 100644 --- a/pkg/fission-cli/cmd/function/test.go +++ b/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() - if deadline := input.Duration(flagkey.FnTestTimeout); deadline > 0 { - var closeCtx func() - ctx, closeCtx = context.WithTimeout(ctx, deadline) - defer closeCtx() - } + ctx, closeCtx := context.WithTimeout(context.Background(), input.Duration(flagkey.FnTestTimeout)) + defer closeCtx() - headers := input.StringSlice(flagkey.FnTestHeader) - - resp, err := doHTTPRequest(ctx, input.String(flagkey.HtMethod), functionUrl.String(), input.String(flagkey.FnTestBody), headers) + resp, err := doHTTPRequest(ctx, functionUrl.String(), + input.StringSlice(flagkey.FnTestHeader), + 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)) - err = printPodLogs(input) + console.Errorf("Error calling function %s: %d; Please try again or fix the error: %s\n", m.Name, resp.StatusCode, string(body)) + 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) - return Log(input) + console.Errorf("Error getting function logs from controller: %v. Try to get logs from log database.", err) + err = Log(input) + if err != nil { + return errors.Wrapf(err, "error retrieving function log from log database") + } + } else { + console.Info(log) } - - return nil + return errors.New("error getting function response") } -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 { - fnName := input.String(flagkey.FnName) - - u, err := util.GetApplicationUrl("application=fission-api") +func printPodLogs(client *client.Client, fnMeta *metav1.ObjectMeta) (string, error) { + reader, statusCode, err := client.FunctionPodLogs(fnMeta) 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") - } - 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") + return "", errors.Wrap(err, "error reading the response body") } - httpClient := http.Client{} - resp, err := httpClient.Do(req) - if err != nil { - return errors.Wrap(err, "execute get logs request") + if statusCode != http.StatusOK { + return string(body), errors.Errorf("error getting logs from controller, status code: '%v'", statusCode) } - defer resp.Body.Close() - 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 + return string(body), nil } diff --git a/pkg/fission-cli/console/log.go b/pkg/fission-cli/console/log.go index 1fdd2132..e4373113 100644 --- a/pkg/fission-cli/console/log.go +++ b/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))) }