From 746c51901da590cff09317dbe59aa19241211812 Mon Sep 17 00:00:00 2001 From: Vishal Date: Tue, 31 Oct 2017 11:02:15 +0530 Subject: [PATCH] Test functions 236 (#355) A single CLI command to invoke a function, print results, and get logs if the function fails. Meant to be an easy way to do testing from the CLI. --- controller/api.go | 3 ++ controller/functionApi.go | 68 +++++++++++++++++++++++++++++++++++++++ controller/tpr.go | 4 +-- fission/common.go | 36 +++++++++++++++++++++ fission/function.go | 62 +++++++++++++++++++++++++++++++++++ fission/main.go | 4 +++ 6 files changed, 175 insertions(+), 2 deletions(-) diff --git a/controller/api.go b/controller/api.go index cf2e05c7..2a263bee 100644 --- a/controller/api.go +++ b/controller/api.go @@ -27,6 +27,7 @@ import ( "github.com/gorilla/mux" log "github.com/sirupsen/logrus" kerrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/client-go/kubernetes" "github.com/fission/fission" "github.com/fission/fission/fission/logdb" @@ -36,6 +37,7 @@ import ( type ( API struct { fissionClient *tpr.FissionClient + kubernetesClient *kubernetes.Clientset storageServiceUrl string builderManagerUrl string workflowApiUrl string @@ -178,6 +180,7 @@ func (api *API) Serve(port int) { r.HandleFunc("/proxy/storage/v1/archive", api.StorageServiceProxy) r.HandleFunc("/proxy/buildermgr/v1/build", api.BuilderManagerBuildProxy) r.HandleFunc("/proxy/buildermgr/v1/builder", api.BuilderManagerEnvBuilderProxy) + r.HandleFunc("/proxy/logs/{function}", api.FunctionPodLogs).Methods("POST") r.HandleFunc("/proxy/workflows-apiserver/{path:.*}", api.WorkflowApiserverProxy) address := fmt.Sprintf(":%v", port) diff --git a/controller/functionApi.go b/controller/functionApi.go index b75c959a..e6283626 100644 --- a/controller/functionApi.go +++ b/controller/functionApi.go @@ -18,14 +18,20 @@ package controller import ( "encoding/json" + "errors" + "io" "io/ioutil" "net/http" "net/http/httputil" "net/url" + "sort" "github.com/gorilla/mux" log "github.com/sirupsen/logrus" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/pkg/api" + "k8s.io/client-go/pkg/api/v1" + restclient "k8s.io/client-go/rest" "github.com/fission/fission" "github.com/fission/fission/tpr" @@ -188,3 +194,65 @@ func (a *API) FunctionLogsApiPost(w http.ResponseWriter, r *http.Request) { } proxy.ServeHTTP(w, r) } + +// FunctionPodLogs : Get logs for a function directly from pod +func (a *API) FunctionPodLogs(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + fnName := vars["function"] + ns := vars["namespace"] + + if len(ns) == 0 { + ns = "fission-function" + } + + f, err := a.fissionClient.Functions(api.NamespaceDefault).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.Core().Pods(ns).List(metav1.ListOptions{LabelSelector: selector}) + if err != nil { + a.respondWithError(w, err) + return + } + + // 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) + }) + + podLogOpts := v1.PodLogOptions{Container: envName} // Only the env container, not fetcher + var podLogsReq *restclient.Request + if len(pods) > 0 { + podLogsReq = a.kubernetesClient.Core().Pods(ns).GetLogs(pods[0].ObjectMeta.Name, &podLogOpts) + } else { + a.respondWithError(w, errors.New("No active pods found")) + return + } + + podLogs, err := podLogsReq.Stream() + if err != nil { + a.respondWithError(w, err) + return + } + defer podLogs.Close() + + _, err = io.Copy(w, podLogs) + if err != nil { + a.respondWithError(w, err) + return + } + return +} diff --git a/controller/tpr.go b/controller/tpr.go index f1533849..82a6f6c8 100644 --- a/controller/tpr.go +++ b/controller/tpr.go @@ -24,11 +24,11 @@ import ( ) func makeTPRBackedAPI() (*API, error) { - fissionClient, _, err := tpr.MakeFissionClient() + fissionClient, kubernetesClient, err := tpr.MakeFissionClient() if err != nil { return nil, err } - return &API{fissionClient: fissionClient}, nil + return &API{fissionClient: fissionClient, kubernetesClient: kubernetesClient}, nil } func validateResourceName(name string) error { diff --git a/fission/common.go b/fission/common.go index 1e56677d..ab705777 100644 --- a/fission/common.go +++ b/fission/common.go @@ -17,7 +17,9 @@ limitations under the License. package main import ( + "errors" "fmt" + "net/http" "os" "strings" @@ -29,6 +31,10 @@ func fatal(msg string) { os.Exit(1) } +func warn(msg string) { + os.Stderr.WriteString(msg + "\n") +} + func getClient(serverUrl string) *client.Client { if len(serverUrl) == 0 { @@ -50,3 +56,33 @@ func checkErr(err error, msg string) { fatal(fmt.Sprintf("Failed to %v: %v", msg, err)) } } + +func httpRequest(method, url, body string, headers []string) *http.Response { + if method == "" { + method = "GET" + } + + if method != http.MethodGet && + method != http.MethodDelete && + method != http.MethodPost && + method != http.MethodPut { + fatal(fmt.Sprintf("Invalid HTTP method '%s'.", method)) + } + + req, err := http.NewRequest(method, url, strings.NewReader(body)) + checkErr(err, "create HTTP request") + + for _, header := range headers { + headerKeyValue := strings.SplitN(header, ":", 2) + if len(headerKeyValue) != 2 { + checkErr(errors.New(""), "create request without appropriate headers") + } + req.Header.Set(headerKeyValue[0], headerKeyValue[1]) + } + + client := &http.Client{} + resp, err := client.Do(req) + checkErr(err, "execute HTTP request") + + return resp +} diff --git a/fission/function.go b/fission/function.go index 7fff60b5..9e28decf 100644 --- a/fission/function.go +++ b/fission/function.go @@ -20,9 +20,12 @@ import ( "context" "crypto/sha256" "encoding/hex" + "errors" "fmt" "io" "io/ioutil" + "net/http" + "net/url" "os" "strings" "text/tabwriter" @@ -134,6 +137,34 @@ func getContents(filePath string) []byte { return code } +func printPodLogs(c *cli.Context) error { + fnName := c.String("name") + if len(fnName) == 0 { + fatal("Need --name argument.") + } + + queryURL, err := url.Parse(c.GlobalString("server")) + checkErr(err, "parse the base URL") + queryURL.Path = fmt.Sprintf("/proxy/logs/%s", fnName) + + req, err := http.NewRequest("POST", queryURL.String(), nil) + checkErr(err, "create logs request") + + httpClient := http.Client{} + resp, err := httpClient.Do(req) + checkErr(err, "execute get logs request") + + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return errors.New("get logs from pod directly") + } + + body, err := ioutil.ReadAll(resp.Body) + checkErr(err, "read the response body") + fmt.Println(string(body)) + return nil +} + func fnCreate(c *cli.Context) error { client := getClient(c.GlobalString("server")) @@ -383,6 +414,7 @@ func fnList(c *cli.Context) error { } func fnLogs(c *cli.Context) error { + client := getClient(c.GlobalString("server")) fnName := c.String("name") @@ -497,3 +529,33 @@ func fnPods(c *cli.Context) error { return err } + +func fnTest(c *cli.Context) error { + fnName := c.String("name") + routerURL := os.Getenv("FISSION_ROUTER") + if len(routerURL) == 0 { + fatal("Need FISSION_ROUTER set to your fission router.") + } + + url := fmt.Sprintf("http://%s/fission-function/%s", routerURL, fnName) + + resp := httpRequest(c.String("method"), url, c.String("body"), c.StringSlice("header")) + if resp.StatusCode < 400 { + body, err := ioutil.ReadAll(resp.Body) + checkErr(err, "Function test") + fmt.Print(string(body)) + defer resp.Body.Close() + return nil + } + + body, err := ioutil.ReadAll(resp.Body) + checkErr(err, "read log response from pod") + fmt.Printf("Error calling function %v: %v %v", fnName, resp.StatusCode, string(body)) + defer resp.Body.Close() + err = printPodLogs(c) + if err != nil { + fnLogs(c) + } + + return nil +} diff --git a/fission/main.go b/fission/main.go index 1c0b1188..ecb011c9 100644 --- a/fission/main.go +++ b/fission/main.go @@ -47,8 +47,11 @@ func main() { fnFollowFlag := cli.BoolFlag{Name: "follow, f", Usage: "specify if the logs should be streamed"} fnDetailFlag := cli.BoolFlag{Name: "detail, d", Usage: "display detailed information"} fnLogDBTypeFlag := cli.StringFlag{Name: "dbtype", Usage: "log database type, e.g. influxdb (currently only influxdb is supported)"} + fnBodyFlag := cli.StringFlag{Name: "body, b", Usage: "request body"} + fnHeaderFlag := cli.StringSliceFlag{Name: "header, H", Usage: "request headers"} fnEntryPointFlag := cli.StringFlag{Name: "entrypoint", Usage: "entry point for environment v2 to load with"} fnBuildCmdFlag := cli.StringFlag{Name: "buildcmd", Usage: "build command for builder to run with"} + fnSubcommands := []cli.Command{ {Name: "create", Usage: "Create new function (and optionally, an HTTP route to it)", Flags: []cli.Flag{fnNameFlag, fnEnvNameFlag, fnCodeFlag, fnPackageFlag, fnSrcArchiveFlag, fnDeployArchiveFlag, fnEntryPointFlag, fnBuildCmdFlag, htUrlFlag, htMethodFlag}, Action: fnCreate}, {Name: "get", Usage: "Get function source code", Flags: []cli.Flag{fnNameFlag}, Action: fnGet}, @@ -58,6 +61,7 @@ func main() { {Name: "list", Usage: "List all functions", Flags: []cli.Flag{}, Action: fnList}, {Name: "logs", Usage: "Display function logs", Flags: []cli.Flag{fnNameFlag, fnPodFlag, fnFollowFlag, fnDetailFlag, fnLogDBTypeFlag}, Action: fnLogs}, {Name: "pods", Usage: "Display function pods", Flags: []cli.Flag{fnNameFlag, fnLogDBTypeFlag}, Action: fnPods}, + {Name: "test", Usage: "Test a function", Flags: []cli.Flag{fnNameFlag, fnEnvNameFlag, fnCodeFlag, fnPackageFlag, fnSrcArchiveFlag, htMethodFlag, fnBodyFlag, fnHeaderFlag}, Action: fnTest}, } // httptriggers