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() + } + }() +}