From a13015e75a9611f1fb33fe84a1dd0ab2c603aea2 Mon Sep 17 00:00:00 2001 From: Ta-Ching Chen Date: Thu, 23 Mar 2017 06:59:42 +0800 Subject: [PATCH] Add function logs support (#53) (#131) Add function log aggregation and persistence using Fluentd and InfluxDB. Fluentd and a helper sidecar run as a daemonset. The poolmgr sets up logging for each function pod, using the helper sidecar. Fluentd forwards logs to InfluxDB, which is run as a deployment and service. The client CLI directly queries InfluxDB for logs. Fluentd supports many outputs besides InfluxDB, so we aren't very tied to InfluxDB. The setup is somewhat manual, which we should be able to improve by integrating this into the helm chart. Diagram of component interactions: https://cloud.githubusercontent.com/assets/202578/23100399/b0e3ea00-f6ba-11e6-8f2f-6588cfef2e84.png --- Documentation/Architecture.md | 20 +++- INSTALL.md | 44 +++++++ README.md | 3 + fission-bundle/main.go | 12 ++ fission-cloud.yaml | 19 +++ fission-logger.yaml | 122 +++++++++++++++++++ fission-nodeport.yaml | 19 +++ fission/function.go | 134 +++++++++++++++++++++ fission/logdb/influxdb.go | 118 +++++++++++++++++++ fission/logdb/logdb.go | 68 +++++++++++ fission/main.go | 9 ++ glide.lock | 25 ++-- glide.yaml | 4 + logger/fluentd/Dockerfile | 68 +++++++++++ logger/fluentd/build.sh | 2 + logger/fluentd/fluent.conf | 48 ++++++++ logger/fluentd/run.sh | 27 +++++ logger/logger.go | 215 ++++++++++++++++++++++++++++++++++ logger/types.go | 35 ++++++ poolmgr/gp.go | 51 ++++++++ 20 files changed, 1029 insertions(+), 14 deletions(-) create mode 100644 fission-logger.yaml create mode 100644 fission/logdb/influxdb.go create mode 100644 fission/logdb/logdb.go create mode 100644 logger/fluentd/Dockerfile create mode 100755 logger/fluentd/build.sh create mode 100644 logger/fluentd/fluent.conf create mode 100755 logger/fluentd/run.sh create mode 100644 logger/logger.go create mode 100644 logger/types.go diff --git a/Documentation/Architecture.md b/Documentation/Architecture.md index 0be7879e..8b65469b 100644 --- a/Documentation/Architecture.md +++ b/Documentation/Architecture.md @@ -110,4 +110,22 @@ be loaded with any function. When poolmgr needs to create a service for a function, it calls fetcher to fetch the function. Fetcher downloads the function into a volume shared between fetcher and this environment container. Poolmgr -then requests the container to load the function. \ No newline at end of file +then requests the container to load the function. + +Logger +----------- + +Logger helps to forward function logs to centralized db service for log +persistence. Currently only influxdb is supported to store logs. +Following is a diagram describe how log service works: + +![Logger Diagram](https://cloud.githubusercontent.com/assets/202578/23100399/b0e3ea00-f6ba-11e6-8f2f-6588cfef2e84.png) + +1. Pool manager choose a pod from pool to execute user function +2. Pool manager makes a HTTP POST to logger helper, once helper receives +the request it creates a symlink to container log for fluentd. +3. Fluentd reads log from symlink and pipes to influxdb +4. `fission function logs ...` retrieve event logs from influxdb with +optional log filter +5. Pool manager removes function pod from pool +6. Pool manager asks logger helper to stop piping logs, logger removes symlink. \ No newline at end of file diff --git a/INSTALL.md b/INSTALL.md index 6309b187..a0b52b75 100644 --- a/INSTALL.md +++ b/INSTALL.md @@ -7,6 +7,7 @@ * [Get and Run Fission: GKE or other Cloud](#get-and-run-fission-gke-or-other-cloud) * [Install the client CLI](#install-the-client-cli) * [Run an example](#run-an-example) + * [Enable Persistent Function Logs (Optional)](#enable-persistent-function-logs) ## Running Fission on your Cluster @@ -110,3 +111,46 @@ Finally, you're ready to use Fission! $ curl http://$FISSION_ROUTER/hello Hello, world! ``` + + +### 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: + +``` + $ 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: + +``` + $ fission function logs --name hello +``` + +You can also list the all the pods that have hosted the function +(including ones that aren't alive any more) and view logs for a +particular pod: + +``` + $ fission function pods --name hello + + $ fission function logs --name hello --pod +``` diff --git a/README.md b/README.md index 8f39732a..7b8775a0 100644 --- a/README.md +++ b/README.md @@ -182,6 +182,9 @@ Finally, you're ready to use Fission! Hello, world! ``` +You can also set up persistence for logs: [instructions here](INSTALL.md). + + Compiling Fission ================= diff --git a/fission-bundle/main.go b/fission-bundle/main.go index cec0759e..010615b1 100644 --- a/fission-bundle/main.go +++ b/fission-bundle/main.go @@ -7,6 +7,7 @@ import ( "github.com/docopt/docopt-go" "github.com/fission/fission/controller" "github.com/fission/fission/kubewatcher" + "github.com/fission/fission/logger" "github.com/fission/fission/poolmgr" "github.com/fission/fission/router" ) @@ -47,6 +48,11 @@ func runKubeWatcher(controllerUrl, routerUrl string) { } } +func runLogger() { + logger.Start() + log.Fatalf("Error: Logger exited.") +} + func getPort(portArg interface{}) int { portArgStr := portArg.(string) port, err := strconv.Atoi(portArgStr) @@ -80,6 +86,7 @@ Usage: fission-bundle --routerPort= [--controllerUrl= --poolmgrUrl=] fission-bundle --poolmgrPort= [--controllerUrl= --namespace=] fission-bundle --kubewatcher [--controllerUrl= --routerUrl=] + fission-bundle --logger Options: --controllerPort= Port that the controller should listen on. --routerPort= Port that the router should listen on. @@ -91,6 +98,7 @@ Options: --filepath= Directory to store functions in. --namespace= Kubernetes namespace in which to run function containers. Defaults to 'fission-function'. --kubewatcher Start Kubernetes events watcher. + --logger Start logger. ` arguments, err := docopt.Parse(usage, nil, true, "fission-bundle", false) if err != nil { @@ -123,5 +131,9 @@ Options: runKubeWatcher(controllerUrl, routerUrl) } + if arguments["--logger"] == true { + runLogger() + } + select {} } diff --git a/fission-cloud.yaml b/fission-cloud.yaml index ca83a089..0b58433c 100644 --- a/fission-cloud.yaml +++ b/fission-cloud.yaml @@ -30,3 +30,22 @@ spec: nodePort: 31313 selector: svc: controller + +--- +# If you want to switch to external db service, please checkout +# sample Kubernetes Service config in fission-logger.yaml. +apiVersion: v1 +kind: Service +metadata: + name: influxdb + namespace: fission + labels: + svc: influxdb +spec: + type: LoadBalancer + ports: + - port: 8086 + targetPort: 8086 + nodePort: 31315 + selector: + svc: influxdb \ No newline at end of file diff --git a/fission-logger.yaml b/fission-logger.yaml new file mode 100644 index 00000000..3b4bcc5a --- /dev/null +++ b/fission-logger.yaml @@ -0,0 +1,122 @@ +# Sample Kubernetes Service config for external db service. +# +# apiVersion: v1 +# kind: Service +# metadata: +# name: influxdb +# namespace: fission +# spec: +# ports: +# - name: "8086" +# port: 8086 +# targetPort: 8086 +# protocol: TCP + +# --- +# apiVersion: v1 +# kind: Endpoints +# metadata: +# name: influxdb +# namespace: fission +# subsets: +# - addresses: +# - ip: replace.influxdb.host.here +# ports: +# - name: "8086" +# port: 8086 +# protocol: TCP + +apiVersion: extensions/v1beta1 +kind: Deployment +metadata: + name: influxdb + namespace: fission +spec: + replicas: 1 + template: + metadata: + labels: + svc: influxdb + spec: + containers: + - name: influxdb + image: tutum/influxdb + env: + - name: PRE_CREATE_DB + value: fissionFunctionLog + # Create a random username/password for InfluxDB here, and + # repeat it in the fluentd spec below. + - name: ADMIN_USER + value: "" + - name: INFLUXDB_INIT_PWD + value: "" + +--- +apiVersion: extensions/v1beta1 +kind: DaemonSet +metadata: + name: logger + namespace: fission +spec: + template: + metadata: + labels: + svc: logger + spec: + containers: + - name: logger + image: fission/fission-bundle + imagePullPolicy: IfNotPresent + command: ["/fission-bundle"] + args: ["--logger"] + volumeMounts: + - name: container-log + mountPath: /var/log/containers + readOnly: true + - name: docker-log + mountPath: /var/lib/docker/containers + readOnly: true + - name: fission-log + mountPath: /var/log/fission + readOnly: false + ports: + - containerPort: 1234 + hostPort: 1234 + protocol: TCP + - name: fluentd + image: fission/fluentd + imagePullPolicy: IfNotPresent + env: + - name: INFLUXDB_ADDRESS + value: influxdb + - name: INFLUXDB_PORT + value: "8086" + - name: INFLUXDB_DBNAME + value: "fissionFunctionLog" + # Username/password for fluentd to push logs to + # influxdb. This should match the influxdb + # username/passowrd defined above. + - name: INFLUXDB_USERNAME + value: "" + - name: INFLUXDB_PASSWD + value: "" + volumeMounts: + - name: container-log + mountPath: /var/log/containers + readOnly: true + - name: docker-log + mountPath: /var/lib/docker/containers + readOnly: true + - name: fission-log + mountPath: /var/log/fission + readOnly: false + volumes: + - name: container-log + hostPath: + path: /var/log/containers + - name: docker-log + hostPath: + path: /var/lib/docker/containers + - name: fission-log + hostPath: + path: /var/log/fission diff --git a/fission-nodeport.yaml b/fission-nodeport.yaml index ecad9fc8..ea9fcbc9 100644 --- a/fission-nodeport.yaml +++ b/fission-nodeport.yaml @@ -30,3 +30,22 @@ spec: nodePort: 31313 selector: svc: controller + +--- +# If you want to switch to external db service, please checkout +# sample Kubernetes Service config in fission-logger.yaml. +apiVersion: v1 +kind: Service +metadata: + name: influxdb + namespace: fission + labels: + svc: influxdb +spec: + type: NodePort + ports: + - port: 8086 + targetPort: 8086 + nodePort: 31315 + selector: + svc: influxdb \ No newline at end of file diff --git a/fission/function.go b/fission/function.go index 24233400..49fd0871 100644 --- a/fission/function.go +++ b/fission/function.go @@ -17,6 +17,7 @@ limitations under the License. package main import ( + "context" "fmt" "io/ioutil" "net/http" @@ -24,11 +25,13 @@ import ( "os/exec" "strings" "text/tabwriter" + "time" "github.com/satori/go.uuid" "github.com/urfave/cli" "github.com/fission/fission" + "github.com/fission/fission/fission/logdb" ) func fnFetchCode(filePath string) []byte { @@ -276,3 +279,134 @@ func fnEdit(c *cli.Context) error { fmt.Printf("function %v updated, new uuid: %v\n", newfn.Name, newfn.Uid) return nil } + +func fnLogs(c *cli.Context) error { + client := getClient(c.GlobalString("server")) + + fnName := c.String("name") + if len(fnName) == 0 { + 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 + } + + fnPod := c.String("pod") + m := &fission.Metadata{Name: fnName} + + 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) + if err != nil { + fatal("failed to connect log database") + } + + requestChan := make(chan struct{}) + responseChan := make(chan struct{}) + ctx := context.Background() + + go func(ctx context.Context, requestChan, responseChan chan struct{}) { + t := time.Unix(0, 0*int64(time.Millisecond)) + for { + select { + case <-requestChan: + logFilter := logdb.LogFilter{ + Pod: fnPod, + Function: f.Metadata.Name, + FuncUid: f.Metadata.Uid, + Since: t, + } + logEntries, err := logDB.GetLogs(logFilter) + if err != nil { + fatal("failed to query logs") + } + for _, logEntry := range logEntries { + if c.Bool("d") { + fmt.Printf("Timestamp: %s\nNamespace: %s\nFunction Name: %s\nFunction ID: %s\nPod: %s\nContainer: %s\nStream: %s\nLog: %s\n---\n", + logEntry.Timestamp, logEntry.Namespace, logEntry.FuncName, logEntry.FuncUid, logEntry.Pod, logEntry.Container, logEntry.Stream, logEntry.Message) + } else { + fmt.Printf("[%s] %s\n", logEntry.Timestamp, logEntry.Message) + } + t = logEntry.Timestamp + } + responseChan <- struct{}{} + case <-ctx.Done(): + return + } + } + }(ctx, requestChan, responseChan) + + for { + requestChan <- struct{}{} + <-responseChan + if !c.Bool("f") { + ctx.Done() + return nil + } + time.Sleep(1 * time.Second) + } +} + +func fnPods(c *cli.Context) error { + client := getClient(c.GlobalString("server")) + + fnName := c.String("name") + if len(fnName) == 0 { + 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 + } + + m := &fission.Metadata{Name: fnName} + + 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) + if err != nil { + fatal("failed to connect log database") + } + + logFilter := logdb.LogFilter{ + Function: f.Metadata.Name, + FuncUid: f.Metadata.Uid, + } + pods, err := logDB.GetPods(logFilter) + if err != nil { + fatal("failed to get pods of function") + return err + } + for _, pod := range pods { + fmt.Println(pod) + } + + return err +} diff --git a/fission/logdb/influxdb.go b/fission/logdb/influxdb.go new file mode 100644 index 00000000..421bb0b8 --- /dev/null +++ b/fission/logdb/influxdb.go @@ -0,0 +1,118 @@ +/* +Copyright 2016 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package logdb + +import ( + "log" + "time" + + "fmt" + + "strings" + + influxdbClient "github.com/influxdata/influxdb/client/v2" +) + +const ( + INFLUXDB_DATABASE = "fissionFunctionLog" +) + +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 +} + +type InfluxDB struct { + dbClient influxdbClient.Client +} + +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) + if err != nil /*|| response.Err != ""*/ { + return []string{}, err + } + pods := []string{} + for _, r := range response.Results { + for _, series := range r.Series { + for _, pod := range series.Tags { + pods = append(pods, pod) + } + } + } + return pods, nil +} + +func (influx InfluxDB) GetLogs(filter LogFilter) ([]LogEntry, error) { + timestamp := filter.Since.UnixNano() + var query string + 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) + } else { + query = fmt.Sprintf("select * from \"log\" where \"funcuid\" = '%s' AND \"time\" > %d ORDER BY time ASC", filter.FuncUid, timestamp) + } + + logEntries := []LogEntry{} + response, err := influx.query(query) + if err != nil { + return logEntries, nil + } + for _, r := range response.Results { + for _, series := range r.Series { + for _, row := range series.Values { + t, err := time.Parse(time.RFC3339, row[0].(string)) + if err != nil { + log.Fatal(err) + } + logEntries = append(logEntries, LogEntry{ + Timestamp: t, + Container: row[2].(string), + FuncName: row[3].(string), + FuncUid: row[4].(string), + Message: strings.TrimSuffix(row[5].(string), "\n"), + Namespace: row[6].(string), + Pod: row[7].(string), + Stream: row[8].(string), + }) + } + } + } + return logEntries, nil +} + +func (influx InfluxDB) query(queryCmd string) (*influxdbClient.Response, error) { + q := influxdbClient.Query{ + Command: queryCmd, + Database: INFLUXDB_DATABASE, + } + return influx.dbClient.Query(q) +} diff --git a/fission/logdb/logdb.go b/fission/logdb/logdb.go new file mode 100644 index 00000000..0caefd59 --- /dev/null +++ b/fission/logdb/logdb.go @@ -0,0 +1,68 @@ +/* +Copyright 2016 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package logdb + +import ( + "time" + + log "github.com/Sirupsen/logrus" +) + +const ( + INFLUXDB = "influxdb" +) + +type LogDatabase interface { + GetPods(LogFilter) ([]string, error) + GetLogs(LogFilter) ([]LogEntry, error) +} + +type LogFilter struct { + Pod string + Function string + FuncUid string + Since time.Time +} + +type LogEntry struct { + Timestamp time.Time + Message string + Stream string + Container string + Namespace string + FuncName string + FuncUid string + Pod string +} + +type DBConfig struct { + DBType string + Endpoint string + Username string + Password string +} + +func GetLogDB(cnf DBConfig) (LogDatabase, error) { + switch cnf.DBType { + case INFLUXDB: + return NewInfluxDB(cnf) + } + log.WithFields(log.Fields{ + "FISSION_LOGDB_URL": cnf.Endpoint, + }).Fatalf("FISSION_LOGDB_URL is incorrect, now only support %s", INFLUXDB) + return nil, nil +} diff --git a/fission/main.go b/fission/main.go index fb0c8265..f74efa45 100644 --- a/fission/main.go +++ b/fission/main.go @@ -41,6 +41,13 @@ func main() { fnCodeFlag := cli.StringFlag{Name: "code", Usage: "local path or URL for source code"} fnPackageFlag := cli.StringFlag{Name: "package", Usage: "local path or URL for binary package"} fnUidFlag := cli.StringFlag{Name: "uid", Usage: "function uid, optional (use latest if unspecified)"} + 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}, @@ -49,6 +56,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}, } // httptriggers diff --git a/glide.lock b/glide.lock index 48787d97..b88c5d10 100644 --- a/glide.lock +++ b/glide.lock @@ -1,5 +1,5 @@ -hash: fb2f38693c5f2fade3db9434c266743d0f9ea9b5f878b44e79763e248136dcf2 -updated: 2016-11-14T01:06:12.764919869-08:00 +hash: 6b4c050c1cd5a7cbb99141348020c882d088b30a9fe9cf92b78a7f706e59da4c +updated: 2017-02-22T23:15:19.862789385+08:00 imports: - name: github.com/blang/semver version: 60ec3488bfea7cca02b021d106d9911120d25fe9 @@ -69,9 +69,15 @@ imports: - name: github.com/gorilla/context version: 08b5f424b9271eedf6f9f0ce86cb9396ed337a42 - name: github.com/gorilla/handlers - version: ee54c7b44cab12289237fb8631314790076e728b + version: 3a5767ca75ece5f7f1440b1d16975247f8d8b221 - name: github.com/gorilla/mux - version: 0eeaf8392f5b04950925b8a69fe70f110fa7cbfc + version: 392c28fe23e1c45ddba891b0320b3b5df220beea +- name: github.com/influxdata/influxdb + version: b7bb7e8359642b6e071735b50ae41f5eb343fd42 + subpackages: + - client/v2 + - models + - pkg/escape - name: github.com/jonboulle/clockwork version: 2eee05ed794112d45db504eb05aa693efd2b8b09 - name: github.com/juju/ratelimit @@ -91,7 +97,7 @@ imports: - name: github.com/satori/go.uuid version: 879c5887cd475cd7864858769793b2ceb0d44feb - name: github.com/Sirupsen/logrus - version: d26492970760ca5d33129d2d799e34be5c4782eb + version: c078b1e43f58d563c74cebe63c85789e76ddb627 - name: github.com/spf13/pflag version: 08b1a584251b5b62f458943640fc8ebd4d50aaa5 - name: github.com/ugorji/go @@ -99,7 +105,7 @@ imports: subpackages: - codec - name: github.com/urfave/cli - version: a14d7d367bc02b1f57d88de97926727f2d936387 + version: 0bdeddeeb0f650497d603c4ad7b20cfe685682f6 - name: golang.org/x/net version: 6acef71eb69611914f7a30939ea9f6e194c78172 subpackages: @@ -158,13 +164,6 @@ imports: - name: k8s.io/client-go version: 843f7c4f28b1f647f664f883697107d5c02c5acc subpackages: - - 1.4/kubernetes - - 1.4/pkg/api - - 1.4/pkg/api/v1 - - 1.4/pkg/apis/extensions/v1beta1 - - 1.4/pkg/labels - - 1.4/pkg/util/intstr - - 1.4/rest - 1.5/discovery - 1.5/kubernetes - 1.5/kubernetes/typed/apps/v1alpha1 diff --git a/glide.yaml b/glide.yaml index b587349c..80463c84 100644 --- a/glide.yaml +++ b/glide.yaml @@ -29,3 +29,7 @@ import: - 1.5/pkg/labels - 1.5/pkg/util/intstr - 1.5/rest +- package: github.com/influxdata/influxdb + version: v1.2.0 + subpackages: + - client/v2 diff --git a/logger/fluentd/Dockerfile b/logger/fluentd/Dockerfile new file mode 100644 index 00000000..600fb492 --- /dev/null +++ b/logger/fluentd/Dockerfile @@ -0,0 +1,68 @@ +# This file originally came from official Kubernetes GitHub repository. +# You can reach original file with the following link: +# https://github.com/kubernetes/kubernetes/tree/42fbf93fb0bb48d0592e2aa08c5ce6d28ab6d4b0/cluster/addons/fluentd-gcp/fluentd-gcp-image + +# Modification: +# 1. add plugin "fluent-plugin-influxdb" for influxdb support + +# Copyright 2016 The Kubernetes Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# This Dockerfile will build an image that is configured +# to use Fluentd to collect all Docker container log files +# and then cause them to be ingested using the Google Cloud +# Logging API. This configuration assumes that the host performning +# the collection is a VM that has been created with a logging.write +# scope and that the Logging API has been enabled for the project +# in the Google Developer Console. + +FROM gcr.io/google_containers/ubuntu-slim:0.6 + + +# Disable prompts from apt +ENV DEBIAN_FRONTEND noninteractive + +# Install build tools +RUN apt-get -qq update && \ + apt-get install -y -qq curl ca-certificates gcc make bash sudo && \ + apt-get install -y -qq --reinstall lsb-base lsb-release && \ + # Install logging agent and required gems + /usr/bin/curl -sSL https://toolbelt.treasuredata.com/sh/install-ubuntu-xenial-td-agent2.sh | sh && \ + sed -i -e "s/USER=td-agent/USER=root/" -e "s/GROUP=td-agent/GROUP=root/" /etc/init.d/td-agent && \ + td-agent-gem install --no-document fluent-plugin-record-reformer -v 0.8.2 && \ + td-agent-gem install --no-document fluent-plugin-systemd -v 0.0.5 && \ + td-agent-gem install --no-document fluent-plugin-google-cloud -v 0.5.2 && \ + td-agent-gem install --no-document fluent-plugin-detect-exceptions -v 0.0.4 && \ + td-agent-gem install --no-document fluent-plugin-influxdb && \ + # Remove build tools + apt-get remove -y -qq gcc make && \ + apt-get autoremove -y -qq && \ + apt-get clean -qq && \ + # Remove unnecessary files + rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/* \ + /opt/td-agent/embedded/share/doc \ + /opt/td-agent/embedded/share/gtk-doc \ + /opt/td-agent/embedded/lib/postgresql \ + /opt/td-agent/embedded/bin/postgres \ + /opt/td-agent/embedded/share/postgresql \ + /etc/td-agent/td-agent.conf + +# Copy the Fluentd configuration file for logging Docker container logs. +COPY fluent.conf /etc/td-agent/td-agent.conf + +# Copy the entrypoint for the container +COPY run.sh /run.sh + +# Start Fluentd to pick up our config that watches Docker container logs. +CMD /run.sh $FLUENTD_ARGS diff --git a/logger/fluentd/build.sh b/logger/fluentd/build.sh new file mode 100755 index 00000000..098a765f --- /dev/null +++ b/logger/fluentd/build.sh @@ -0,0 +1,2 @@ +#!/bin/sh +docker build -t fission-daemonset-fluentd:latest . diff --git a/logger/fluentd/fluent.conf b/logger/fluentd/fluent.conf new file mode 100644 index 00000000..e2299b44 --- /dev/null +++ b/logger/fluentd/fluent.conf @@ -0,0 +1,48 @@ + + type null + + + + type tail + format json + time_key time + path /var/log/fission/*.log + time_format %Y-%m-%dT%H:%M:%S.%NZ + tag fission.* + read_from_head true + refresh_interval 5 + + + + type record_reformer + enable_ruby false + tag log + + namespace ${tag_parts[4]} + pod ${tag_parts[5]} + container ${tag_parts[6]} + funcname ${tag_parts[7]} + funcuid ${tag_parts[8]} + + + + + @type influxdb + host "#{ENV['INFLUXDB_ADDRESS']}" + port "#{ENV['INFLUXDB_PORT']}" + dbname "#{ENV['INFLUXDB_DBNAME']}" + user "#{ENV['INFLUXDB_USERNAME']}" + password "#{ENV['INFLUXDB_PASSWD']}" + use_ssl false + time_precision s + tag_keys ["funcuid", "pod"] + sequence_tag _seq + buffer_type file + buffer_path /var/log/fission/fluentd.buffer + buffer_chunk_limit 128m + buffer_queue_limit 256 + flush_interval 5 + retry_limit 10 + retry_wait 1.0 + num_threads 2 + \ No newline at end of file diff --git a/logger/fluentd/run.sh b/logger/fluentd/run.sh new file mode 100755 index 00000000..d5e3036b --- /dev/null +++ b/logger/fluentd/run.sh @@ -0,0 +1,27 @@ +# This file originally came from official Kubernetes GitHub repository. +# You can reach original file with the following link: +# https://github.com/kubernetes/kubernetes/tree/42fbf93fb0bb48d0592e2aa08c5ce6d28ab6d4b0/cluster/addons/fluentd-gcp/fluentd-gcp-image + +#!/bin/sh + +# Copyright 2016 The Kubernetes Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# For systems without journald +mkdir -p /var/log/journal + +LD_PRELOAD=/opt/td-agent/embedded/lib/libjemalloc.so +RUBY_GC_HEAP_OLDOBJECT_LIMIT_FACTOR=0.9 + +/usr/sbin/td-agent $@ diff --git a/logger/logger.go b/logger/logger.go new file mode 100644 index 00000000..e55ef759 --- /dev/null +++ b/logger/logger.go @@ -0,0 +1,215 @@ +/* +Copyright 2016 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package logger + +import ( + "encoding/json" + "fmt" + "io/ioutil" + "net/http" + "os" + "path/filepath" + "strings" + + log "github.com/Sirupsen/logrus" + "github.com/fission/fission" + "github.com/gorilla/handlers" + "github.com/gorilla/mux" + "k8s.io/client-go/1.5/kubernetes" + "k8s.io/client-go/1.5/rest" +) + +func makelogRequestTracker() logRequestTracker { + return logRequestTracker{ + logMap: make(map[string]LogRequest), + } +} + +func (l logRequestTracker) Add(logReq LogRequest) { + l.Lock() + l.logMap[logReq.Pod] = logReq + l.Unlock() +} + +func (l logRequestTracker) Get(pod string) LogRequest { + l.RLock() + logReq, ok := l.logMap[pod] + l.RUnlock() + if ok { + return logReq + } + return LogRequest{} +} + +func (l logRequestTracker) Remove(logReq LogRequest) { + l.Lock() + delete(l.logMap, logReq.Pod) + l.Unlock() +} + +// Get a kubernetes client using the pod's service account. This only +// works when we're running inside a kubernetes cluster. +func getKubernetesClient() (*kubernetes.Clientset, error) { + // creates the in-cluster config + config, err := rest.InClusterConfig() + if err != nil { + log.Printf("Error getting kubernetes client config: %v", err) + return nil, err + } + + // creates the clientset + clientset, err := kubernetes.NewForConfig(config) + if err != nil { + log.Printf("Error getting kubernetes client: %v", err) + return nil, err + } + + return clientset, nil +} + +// make sure that the targetPath is a legitimate path for security purpose +func validateFilePath(targetPath string, expectedPathPrefix string) bool { + targetPath = filepath.Clean(targetPath) + return strings.HasPrefix(targetPath, expectedPathPrefix) +} + +// The ContainerID is consist of container engine type (docker://) and uuid of container. +// (e.g., docker://f4ca66baaa715030e20273aaf5232635a144165f1cd8e34ca5175064c245b679) +// This function tries to extract container uuid from ContainerID. +func parseContainerString(containerID string) (string, error) { + // Trim the quotes and split the type and ID. + parts := strings.Split(strings.Trim(containerID, "\""), "://") + if len(parts) != 2 { + return "", fmt.Errorf("invalid container ID: %q", containerID) + } + _, ID := parts[0], parts[1] + return ID, nil +} + +func getcontainerID(kubeClient *kubernetes.Clientset, namespace, pod, container string) (string, error) { + podInfo, err := kubeClient.Core().Pods(namespace).Get(pod) + if err != nil { + log.Printf("Failed to get pod info: %v", err) + return "", err + } + var containerID string + for _, c := range podInfo.Status.ContainerStatuses { + if c.Name == container { + containerID, err = parseContainerString(c.ContainerID) + if err != nil { + log.Printf("Failed to get container id: %v", err) + return "", err + } + return containerID, nil + } + } + return "", fission.MakeError(404, "no matching container is found") +} + +func getContainerLogPath(logReq LogRequest) (string, bool) { + logPath := fmt.Sprintf("/var/lib/docker/containers/%s/%s-json.log", logReq.ContainerID, logReq.ContainerID) + if !validateFilePath(logPath, "/var/lib/docker/containers") { + return "", false + } + return logPath, true +} + +func getFissionLogSymlinkPath(logReq LogRequest) (string, bool) { + // pass function related information through a symlink name + logSymLink := fmt.Sprintf("/var/log/fission/%s.%s.%s.%s.%s.log", logReq.Namespace, logReq.Pod, logReq.ContainerID, logReq.FuncName, logReq.FuncUid) + if !validateFilePath(logSymLink, "/var/log/fission") { + return "", false + } + return logSymLink, true +} + +func createLogSymlink(w http.ResponseWriter, r *http.Request) { + body, err := ioutil.ReadAll(r.Body) + if err != nil { + http.Error(w, "Failed to read request", 500) + return + } + logReq := LogRequest{} + if err = json.Unmarshal(body, &logReq); err != nil { + w.Write([]byte(fmt.Sprintf("%v", err))) + return + } + + kubernetesClient, err := getKubernetesClient() + if err != nil { + log.Warningf("Failed to get kubernetes client: %v", err) + w.WriteHeader(http.StatusInternalServerError) + return + } + + containerID, err := getcontainerID(kubernetesClient, logReq.Namespace, logReq.Pod, logReq.Container) + if err != nil || containerID == "" { + log.Warningf("Failed to get container id: %v", err) + w.WriteHeader(http.StatusInternalServerError) + return + } + + logReq.ContainerID = containerID + containerLogFilePath, isValidLogPath := getContainerLogPath(logReq) + fissionLogSymlinkPath, isValidSymlinkPath := getFissionLogSymlinkPath(logReq) + if !isValidLogPath || !isValidSymlinkPath { + w.WriteHeader(http.StatusBadRequest) + return + } + + err = os.Symlink(containerLogFilePath, fissionLogSymlinkPath) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + return + } + logInfo.Add(logReq) + w.WriteHeader(http.StatusOK) +} + +func removeLogSymlink(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + pod := vars["pod"] + logReq := logInfo.Get(pod) + if logReq.Pod == "" { + w.WriteHeader(http.StatusInternalServerError) + return + } + + fissionLogSymlinkPath, isValidSymlinkPath := getFissionLogSymlinkPath(logReq) + if !isValidSymlinkPath { + w.WriteHeader(http.StatusBadRequest) + return + } + err := os.Remove(fissionLogSymlinkPath) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + return + } + w.WriteHeader(http.StatusOK) +} + +var logInfo logRequestTracker + +func Start() { + logInfo = makelogRequestTracker() + r := mux.NewRouter() + r.HandleFunc("/v1/log", createLogSymlink).Methods("POST") + r.HandleFunc("/v1/log/{pod}", removeLogSymlink).Methods("DELETE") + address := fmt.Sprintf(":%v", 1234) + log.Printf("starting poolmgr at port %s", address) + log.Fatal(http.ListenAndServe(address, handlers.LoggingHandler(os.Stdout, r))) +} diff --git a/logger/types.go b/logger/types.go new file mode 100644 index 00000000..51bbd070 --- /dev/null +++ b/logger/types.go @@ -0,0 +1,35 @@ +/* +Copyright 2016 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package logger + +import "sync" + +type ( + LogRequest struct { + Namespace string `json:"namespace"` + Pod string `json:"pod"` + Container string `json:"container"` + FuncName string `json:"funcname"` + FuncUid string `json:"funcuid"` + ContainerID string `json:"-"` + } + + logRequestTracker struct { + sync.RWMutex + logMap map[string]LogRequest + } +) diff --git a/poolmgr/gp.go b/poolmgr/gp.go index 49678a23..7938af77 100644 --- a/poolmgr/gp.go +++ b/poolmgr/gp.go @@ -18,6 +18,7 @@ package poolmgr import ( "bytes" + "encoding/json" "errors" "fmt" "log" @@ -37,6 +38,7 @@ import ( "k8s.io/client-go/1.5/pkg/util/intstr" "github.com/fission/fission" + "github.com/fission/fission/logger" ) const POOLMGR_INSTANCEID_LABEL string = "poolmgrInstanceId" @@ -265,6 +267,9 @@ func (gp *GenericPool) specializePod(pod *v1.Pod, metadata *fission.Metadata) er return errors.New(fmt.Sprintf("Error from fetcher: %v", resp.Status)) } + // Tell logging helper about this function invocation + gp.setupLogging(pod, metadata) + // get function run container to specialize log.Printf("[%v] specializing pod", metadata) specializeUrl := fmt.Sprintf("http://%v:8888/specialize", podIP) @@ -490,6 +495,23 @@ func (gp *GenericPool) CleanupFunctionService(podName string) error { return nil } + pod, err := gp.kubernetesClient.Core().Pods(gp.namespace).Get(podName) + if err != nil { + return err + } + + loggerUrl := fmt.Sprintf("http://%s:1234/v1/log/%s", pod.Spec.NodeName, pod.Name) + req, err := http.NewRequest("DELETE", loggerUrl, nil) + resp, err := http.DefaultClient.Do(req) + if err != nil { + log.Printf("Error from %s daemonset logger: %v", pod.Spec.NodeName, err) + } else { + if resp.StatusCode != 200 { + log.Printf("Received not http 200(OK) status from %s daemonset logger: %s", pod.Spec.NodeName, resp.Status) + } + resp.Body.Close() + } + // delete pod err = gp.kubernetesClient.Core().Pods(gp.namespace).Delete(podName, nil) if err != nil { @@ -556,3 +578,32 @@ func (gp *GenericPool) destroy() error { return nil } + +// Calls the logging daemonset pod on the node where the given pod is +// running. +func (gp *GenericPool) setupLogging(pod *v1.Pod, metadata *fission.Metadata) { + logReq := logger.LogRequest{ + Namespace: pod.Namespace, + Pod: pod.Name, + Container: gp.env.Metadata.Name, + FuncName: metadata.Name, + FuncUid: metadata.Uid, + } + reqbody, err := json.Marshal(logReq) + if err != nil { + log.Printf("Error creating log request") + return + } + go func() { + loggerUrl := fmt.Sprintf("http://%s:1234/v1/log", pod.Status.HostIP) + resp, err := http.Post(loggerUrl, "application/json", bytes.NewReader(reqbody)) + if err != nil { + log.Printf("Error connecting to %s log daemonset pod: %v", pod.Spec.NodeName, err) + } else { + if resp.StatusCode != 200 { + log.Printf("Error from %s log daemonset pod: %s", pod.Spec.NodeName, resp.Status) + } + resp.Body.Close() + } + }() +}