From 990d943f21eda078f4fb7899f7cc47facd819edc Mon Sep 17 00:00:00 2001 From: Ta-Ching Chen Date: Wed, 31 May 2017 05:19:22 +0800 Subject: [PATCH] Retrieve function logs from controller (#207) Establish a proxy server from the controller to the log database. Redirect query commands send from client to database then proxy back the db response. Use parameter binding instead of fmt.Sprintf to prevent SQL injection. --- INSTALL.md | 21 +------- controller/api.go | 43 +++++++++++++--- controller/functionApi.go | 36 ++++++++++++- fission/function.go | 31 +++-------- fission/logdb/influxdb.go | 106 ++++++++++++++++++++++++++------------ fission/logdb/logdb.go | 17 ++---- fission/main.go | 7 +-- 7 files changed, 158 insertions(+), 103 deletions(-) diff --git a/INSTALL.md b/INSTALL.md index b0b54760..7acdbd15 100644 --- a/INSTALL.md +++ b/INSTALL.md @@ -129,30 +129,13 @@ Finally, you're ready to use Fission! ### Enable Persistent Function Logs (Optional) Fission uses InfluxDB to store logs and fluentd to forward them from -function pods into InfluxDB. To setup both InfluxDB and fluentd: - -Edit `fission-logger.yaml` to add a username and password for the -Influxdb deployment. Then: +function pods into InfluxDB. ``` $ kubectl create -f fission-logger.yaml ``` -On the client side, - -If you're using minikube or a local cluster: - -``` -$ export FISSION_LOGDB=http://$(minikube ip):31315 -``` - -If you're using GKE or other cloud: - -``` -$ export FISSION_LOGDB=http://$(kubectl --namespace fission get svc influxdb -o=jsonpath='{..ip}'):8086 -``` - -That's it for setup. You can now use this to view function logs: +That's it for the basic setup. You can now use following command to view function logs: ``` $ fission function logs --name hello diff --git a/controller/api.go b/controller/api.go index c6342277..53545df3 100644 --- a/controller/api.go +++ b/controller/api.go @@ -21,21 +21,31 @@ import ( "net/http" "os" "runtime/debug" + "strings" log "github.com/Sirupsen/logrus" "github.com/gorilla/handlers" "github.com/gorilla/mux" "github.com/fission/fission" + "github.com/fission/fission/fission/logdb" ) -type API struct { - FunctionStore - HTTPTriggerStore - TimeTriggerStore - EnvironmentStore - WatchStore -} +type ( + API struct { + FunctionStore + HTTPTriggerStore + TimeTriggerStore + EnvironmentStore + WatchStore + } + + logDBConfig struct { + httpURL string + username string + password string + } +) func MakeAPI(rs *ResourceStore) *API { api := &API{ @@ -64,6 +74,23 @@ func (api *API) respondWithError(w http.ResponseWriter, err error) { http.Error(w, msg, code) } +func (api *API) getLogDBConfig(dbType string) logDBConfig { + dbType = strings.ToUpper(dbType) + // retrieve db auth config from the env + url := os.Getenv(fmt.Sprintf("%s_URL", dbType)) + if url == "" { + // set up default database url + url = logdb.INFLUXDB_URL + } + username := os.Getenv(fmt.Sprintf("%s_USERNAME", dbType)) + password := os.Getenv(fmt.Sprintf("%s_PASSWORD", dbType)) + return logDBConfig{ + httpURL: url, + username: username, + password: password, + } +} + func (api *API) HomeHandler(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json; charset=utf-8") fmt.Fprintf(w, "{\"message\": \"Fission API\", \"version\": \"0.1.0\"}\n") @@ -103,6 +130,8 @@ func (api *API) Serve(port int) { r.HandleFunc("/v1/triggers/time/{timeTrigger}", api.TimeTriggerApiUpdate).Methods("PUT") r.HandleFunc("/v1/triggers/time/{timeTrigger}", api.TimeTriggerApiDelete).Methods("DELETE") + r.HandleFunc("/proxy/{dbType}", api.FunctionLogsApiPost).Methods("POST") + address := fmt.Sprintf(":%v", port) log.WithFields(log.Fields{"port": port}).Info("Server started") diff --git a/controller/functionApi.go b/controller/functionApi.go index 6cbf7df8..5a49428a 100644 --- a/controller/functionApi.go +++ b/controller/functionApi.go @@ -17,14 +17,16 @@ limitations under the License. package controller import ( + "encoding/base64" + "encoding/json" "io/ioutil" "net/http" + "net/http/httputil" + "net/url" - "encoding/json" log "github.com/Sirupsen/logrus" "github.com/gorilla/mux" - "encoding/base64" "github.com/fission/fission" ) @@ -171,3 +173,33 @@ func (api *API) FunctionApiDelete(w http.ResponseWriter, r *http.Request) { api.respondWithSuccess(w, []byte("")) } + +// FunctionLogsApiPost establishes a proxy server to log database, and redirect +// query command send from client to database then proxy back the db response. +func (api *API) FunctionLogsApiPost(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + // get dbType from url + dbType := vars["dbType"] + + // find correspond db http url + dbCnf := api.getLogDBConfig(dbType) + + svcUrl, err := url.Parse(dbCnf.httpURL) + if err != nil { + log.Printf("Failed to establish proxy server for function logs: %v", err) + } + // set up proxy server director + director := func(req *http.Request) { + // only replace url Scheme and Host to remote influxDB + // and leave query string intact + req.URL.Scheme = svcUrl.Scheme + req.URL.Host = svcUrl.Host + req.URL.Path = svcUrl.Path + // set up http basic auth for database authentication + req.SetBasicAuth(dbCnf.username, dbCnf.password) + } + proxy := &httputil.ReverseProxy{ + Director: director, + } + proxy.ServeHTTP(w, r) +} diff --git a/fission/function.go b/fission/function.go index 49fd0871..f8d10595 100644 --- a/fission/function.go +++ b/fission/function.go @@ -288,11 +288,6 @@ func fnLogs(c *cli.Context) error { fatal("Need name of function, use --name") } - dbHost := c.String("dbhost") - if len(dbHost) == 0 { - fatal("Need host address of log database, use --dbhost") - } - dbType := c.String("dbtype") if len(dbType) == 0 { dbType = logdb.INFLUXDB @@ -304,13 +299,9 @@ func fnLogs(c *cli.Context) error { f, err := client.FunctionGet(m) checkErr(err, "get function") - auth := logdb.DBConfig{ - DBType: dbType, - Endpoint: dbHost, - Username: c.String("username"), - Password: c.String("password"), - } - logDB, err := logdb.GetLogDB(auth) + // client first send db query to controller, then controller will + // establishe a proxy server that bridges the client and the database. + logDB, err := logdb.GetLogDB(dbType, c.GlobalString("server")) if err != nil { fatal("failed to connect log database") } @@ -369,11 +360,6 @@ func fnPods(c *cli.Context) error { fatal("Need name of function, use --name") } - dbHost := c.String("dbhost") - if len(dbHost) == 0 { - fatal("Need host address of log database, use --dbhost") - } - dbType := c.String("dbtype") if len(dbType) == 0 { dbType = logdb.INFLUXDB @@ -384,13 +370,9 @@ func fnPods(c *cli.Context) error { f, err := client.FunctionGet(m) checkErr(err, "get function") - auth := logdb.DBConfig{ - DBType: dbType, - Endpoint: dbHost, - Username: c.String("username"), - Password: c.String("password"), - } - logDB, err := logdb.GetLogDB(auth) + // client first sends db query to the controller, then the controller + // will establish a proxy server that bridges the client and the database. + logDB, err := logdb.GetLogDB(dbType, c.GlobalString("server")) if err != nil { fatal("failed to connect log database") } @@ -404,6 +386,7 @@ func fnPods(c *cli.Context) error { fatal("failed to get pods of function") return err } + fmt.Printf("NAME\t\n") for _, pod := range pods { fmt.Println(pod) } diff --git a/fission/logdb/influxdb.go b/fission/logdb/influxdb.go index 421bb0b8..fded6b29 100644 --- a/fission/logdb/influxdb.go +++ b/fission/logdb/influxdb.go @@ -17,47 +17,40 @@ limitations under the License. package logdb import ( + "encoding/json" + "fmt" "log" + "net/http" + "net/url" + "strings" "time" - "fmt" - - "strings" - influxdbClient "github.com/influxdata/influxdb/client/v2" + + "github.com/fission/fission" ) const ( INFLUXDB_DATABASE = "fissionFunctionLog" + INFLUXDB_URL = "http://influxdb:8086/query" ) -func NewInfluxDB(cnf DBConfig) (InfluxDB, error) { - dbClient, err := influxdbClient.NewHTTPClient(influxdbClient.HTTPConfig{ - Addr: cnf.Endpoint, - Username: cnf.Username, - Password: cnf.Password, - }) - if err != nil { - log.Fatal(err) - return InfluxDB{}, err - } - - return InfluxDB{ - dbClient: dbClient, - }, nil +func NewInfluxDB(serverURL string) (InfluxDB, error) { + return InfluxDB{endpoint: serverURL}, nil } type InfluxDB struct { - dbClient influxdbClient.Client + endpoint string } func (influx InfluxDB) GetPods(filter LogFilter) ([]string, error) { - query := fmt.Sprintf("select * from \"log\" where \"funcuid\" = '%s' group by \"pod\"", filter.FuncUid) - q := influxdbClient.Query{ - Command: query, - Database: INFLUXDB_DATABASE, - } - response, err := influx.dbClient.Query(q) + parameters := make(map[string]interface{}) + parameters["funcuid"] = filter.FuncUid + + queryCmd := "select * from \"log\" where \"funcuid\" = $funcuid group by \"pod\"" + query := influxdbClient.NewQueryWithParameters(queryCmd, INFLUXDB_DATABASE, "", parameters) + + response, err := influx.query(query) if err != nil /*|| response.Err != ""*/ { return []string{}, err } @@ -74,13 +67,22 @@ func (influx InfluxDB) GetPods(filter LogFilter) ([]string, error) { func (influx InfluxDB) GetLogs(filter LogFilter) ([]LogEntry, error) { timestamp := filter.Since.UnixNano() - var query string + var queryCmd string + + // please check "Example 4: Bind a parameter in the WHERE clause to specific tag value" + // at https://docs.influxdata.com/influxdb/v1.2/tools/api/ + parameters := make(map[string]interface{}) + parameters["funcuid"] = filter.FuncUid + parameters["time"] = timestamp + if filter.Pod != "" { - query = fmt.Sprintf("select * from \"log\" where \"funcuid\" = '%s' AND \"pod\" = '%s' AND \"time\" > %d ORDER BY time ASC", filter.FuncUid, filter.Pod, timestamp) + queryCmd = "select * from \"log\" where \"funcuid\" = $funcuid AND \"pod\" = $pod AND \"time\" > $time ORDER BY time ASC" + parameters["pod"] = filter.Pod } else { - query = fmt.Sprintf("select * from \"log\" where \"funcuid\" = '%s' AND \"time\" > %d ORDER BY time ASC", filter.FuncUid, timestamp) + queryCmd = "select * from \"log\" where \"funcuid\" = $funcuid AND \"time\" > $time ORDER BY time ASC" } + query := influxdbClient.NewQueryWithParameters(queryCmd, INFLUXDB_DATABASE, "", parameters) logEntries := []LogEntry{} response, err := influx.query(query) if err != nil { @@ -109,10 +111,48 @@ func (influx InfluxDB) GetLogs(filter LogFilter) ([]LogEntry, error) { return logEntries, nil } -func (influx InfluxDB) query(queryCmd string) (*influxdbClient.Response, error) { - q := influxdbClient.Query{ - Command: queryCmd, - Database: INFLUXDB_DATABASE, +func (influx InfluxDB) query(query influxdbClient.Query) (*influxdbClient.Response, error) { + queryURL, err := url.Parse(influx.endpoint) + if err != nil { + return nil, err } - return influx.dbClient.Query(q) + // connect to controller first, then controller will redirect our query command + // to influxdb and proxy back the db response. + queryURL.Path = fmt.Sprintf("/proxy/%s", INFLUXDB) + req, err := http.NewRequest("POST", queryURL.String(), nil) + if err != nil { + return nil, err + } + + parametersBytes, err := json.Marshal(query.Parameters) + if err != nil { + return nil, err + } + + // set up http URL query string + params := req.URL.Query() + params.Set("q", query.Command) + params.Set("db", query.Database) + params.Set("params", string(parametersBytes)) + req.URL.RawQuery = params.Encode() + + httpClient := http.Client{Timeout: 5 * time.Second} + resp, err := httpClient.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fission.MakeErrorFromHTTP(resp) + } + + // decode influxdb response + response := influxdbClient.Response{} + decoder := json.NewDecoder(resp.Body) + decoder.UseNumber() + if decoder.Decode(&response) != nil { + return nil, fmt.Errorf("Failed to decode influxdb response: %v", err) + } + return &response, nil } diff --git a/fission/logdb/logdb.go b/fission/logdb/logdb.go index 0caefd59..8bb4eaf4 100644 --- a/fission/logdb/logdb.go +++ b/fission/logdb/logdb.go @@ -49,20 +49,11 @@ type LogEntry struct { Pod string } -type DBConfig struct { - DBType string - Endpoint string - Username string - Password string -} - -func GetLogDB(cnf DBConfig) (LogDatabase, error) { - switch cnf.DBType { +func GetLogDB(dbType string, serverURL string) (LogDatabase, error) { + switch dbType { case INFLUXDB: - return NewInfluxDB(cnf) + return NewInfluxDB(serverURL) } - log.WithFields(log.Fields{ - "FISSION_LOGDB_URL": cnf.Endpoint, - }).Fatalf("FISSION_LOGDB_URL is incorrect, now only support %s", INFLUXDB) + log.Fatalf("Log database type is incorrect, now only support %s", INFLUXDB) return nil, nil } diff --git a/fission/main.go b/fission/main.go index 9fb0fa19..301b827d 100644 --- a/fission/main.go +++ b/fission/main.go @@ -44,10 +44,7 @@ func main() { fnPodFlag := cli.StringFlag{Name: "pod", Usage: "function pod name, optional (use latest if unspecified)"} fnFollowFlag := cli.BoolFlag{Name: "follow, f", Usage: "specify if the logs should be streamed"} fnDetailFlag := cli.BoolFlag{Name: "detail, d", Usage: "display detailed information"} - fnLogDBHostFlag := cli.StringFlag{Name: "dbhost", Usage: "log database host to connect to", EnvVar: "FISSION_LOGDB"} fnLogDBTypeFlag := cli.StringFlag{Name: "dbtype", Usage: "log database type, e.g. influxdb (currently only influxdb is supported)"} - fnUserNameFlag := cli.StringFlag{Name: "username, u", Usage: "username for connecting log database"} - fnPasswordFlag := cli.StringFlag{Name: "password, p", Usage: "password for connecting log database"} fnSubcommands := []cli.Command{ {Name: "create", Usage: "Create new function (and optionally, an HTTP route to it)", Flags: []cli.Flag{fnNameFlag, fnEnvNameFlag, fnCodeFlag, fnPackageFlag, htUrlFlag, htMethodFlag}, Action: fnCreate}, {Name: "get", Usage: "Get function source code", Flags: []cli.Flag{fnNameFlag, fnUidFlag}, Action: fnGet}, @@ -56,8 +53,8 @@ func main() { {Name: "update", Usage: "Update function source code", Flags: []cli.Flag{fnNameFlag, fnEnvNameFlag, fnCodeFlag, fnPackageFlag}, Action: fnUpdate}, {Name: "delete", Usage: "Delete function", Flags: []cli.Flag{fnNameFlag, fnUidFlag}, Action: fnDelete}, {Name: "list", Usage: "List all functions", Flags: []cli.Flag{}, Action: fnList}, - {Name: "logs", Usage: "Display funtion logs", Flags: []cli.Flag{fnNameFlag, fnPodFlag, fnFollowFlag, fnDetailFlag, fnLogDBHostFlag, fnLogDBTypeFlag, fnUserNameFlag, fnPasswordFlag}, Action: fnLogs}, - {Name: "pods", Usage: "Display funtion pods", Flags: []cli.Flag{fnNameFlag, fnLogDBHostFlag, fnLogDBTypeFlag, fnUserNameFlag, fnPasswordFlag}, Action: fnPods}, + {Name: "logs", Usage: "Display funtion logs", Flags: []cli.Flag{fnNameFlag, fnPodFlag, fnFollowFlag, fnDetailFlag, fnLogDBTypeFlag}, Action: fnLogs}, + {Name: "pods", Usage: "Display funtion pods", Flags: []cli.Flag{fnNameFlag, fnLogDBTypeFlag}, Action: fnPods}, } // httptriggers