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
This commit is contained in:
Ta-Ching Chen
committed by
Soam Vasani
Soam Vasani
parent
1665235c14
commit
a13015e75a
@@ -111,3 +111,21 @@ 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.
|
||||
|
||||
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:
|
||||
|
||||
|
||||
|
||||
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.
|
||||
+44
@@ -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 <pod name>
|
||||
```
|
||||
|
||||
@@ -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
|
||||
=================
|
||||
|
||||
|
||||
@@ -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=<port> [--controllerUrl=<url> --poolmgrUrl=<url>]
|
||||
fission-bundle --poolmgrPort=<port> [--controllerUrl=<url> --namespace=<namespace>]
|
||||
fission-bundle --kubewatcher [--controllerUrl=<url> --routerUrl=<url>]
|
||||
fission-bundle --logger
|
||||
Options:
|
||||
--controllerPort=<port> Port that the controller should listen on.
|
||||
--routerPort=<port> Port that the router should listen on.
|
||||
@@ -91,6 +98,7 @@ Options:
|
||||
--filepath=<filepath> Directory to store functions in.
|
||||
--namespace=<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 {}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
|
||||
Generated
+12
-13
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
Executable
+2
@@ -0,0 +1,2 @@
|
||||
#!/bin/sh
|
||||
docker build -t fission-daemonset-fluentd:latest .
|
||||
@@ -0,0 +1,48 @@
|
||||
<match fluent.**>
|
||||
type null
|
||||
</match>
|
||||
|
||||
<source>
|
||||
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
|
||||
</source>
|
||||
|
||||
<match fission.**>
|
||||
type record_reformer
|
||||
enable_ruby false
|
||||
tag log
|
||||
<record>
|
||||
namespace ${tag_parts[4]}
|
||||
pod ${tag_parts[5]}
|
||||
container ${tag_parts[6]}
|
||||
funcname ${tag_parts[7]}
|
||||
funcuid ${tag_parts[8]}
|
||||
</record>
|
||||
</match>
|
||||
|
||||
<match **>
|
||||
@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
|
||||
</match>
|
||||
Executable
+27
@@ -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 $@
|
||||
@@ -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)))
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
)
|
||||
@@ -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()
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user