Refactor logging to remove logger, use fluentd with kubernetes filter (#380)

Remove the `logger` container from the logging daemonset. 

Remove the outgoing call from the poolmgr to the logger.  Use Fluentd's Kubernetes filter to add function name and UID to influx metadata.  

This means fluentd now figures out when to start collecting function logs on its own, without being informed by poolmgr.  This is great for other execution strategies, and for autoscaling, where fission isn't in direct control of function pod creation.

Also adds an integration test to make sure logging keeps working.
This commit is contained in:
prithviramesh
2017-11-14 18:08:57 -08:00
committed by Soam Vasani
parent 74e05ba3a1
commit 8b40ad0b33
13 changed files with 126 additions and 310 deletions
+5 -22
View File
@@ -348,7 +348,7 @@ metadata:
svc: influxdb
chart: "{{ .Chart.Name }}-{{ .Chart.Version }}"
spec:
type: ClusterIP
type: ClusterIP
ports:
- port: 8086
targetPort: 8086
@@ -400,25 +400,6 @@ spec:
svc: logger
spec:
containers:
- name: logger
image: "{{ .Values.image }}:{{ .Values.imageTag }}"
imagePullPolicy: {{ .Values.pullPolicy }}
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: {{ .Values.logger.fluentdImage }}
imagePullPolicy: {{ .Values.pullPolicy }}
@@ -439,9 +420,11 @@ spec:
secretKeyRef:
name: influxdb
key: password
- name: FLUENTD_PATH
value: /var/log/containers/*{{.Values.functionNamespace}}*.log
volumeMounts:
- name: container-log
mountPath: /var/log/containers
mountPath: /var/log/
readOnly: true
- name: docker-log
mountPath: /var/lib/docker/containers
@@ -453,7 +436,7 @@ spec:
volumes:
- name: container-log
hostPath:
path: /var/log/containers
path: /var/log/
- name: docker-log
hostPath:
path: /var/lib/docker/containers
+1 -1
View File
@@ -10,7 +10,7 @@ serviceType: LoadBalancer
image: fission/fission-bundle
## Image pull policy
pullPolicy: IfNotPresent
pullPolicy: IfNotPresent
## Fission image version
imageTag: 0.4.0rc
-12
View File
@@ -9,7 +9,6 @@ import (
"github.com/fission/fission/buildermgr"
"github.com/fission/fission/controller"
"github.com/fission/fission/kubewatcher"
"github.com/fission/fission/logger"
"github.com/fission/fission/mqtrigger"
"github.com/fission/fission/poolmgr"
"github.com/fission/fission/router"
@@ -41,11 +40,6 @@ func runKubeWatcher(routerUrl string) {
}
}
func runLogger() {
logger.Start()
log.Fatalf("Error: Logger exited.")
}
func runTimer(routerUrl string) {
err := timer.Start(routerUrl)
if err != nil {
@@ -122,7 +116,6 @@ Usage:
fission-bundle --kubewatcher [--routerUrl=<url>]
fission-bundle --storageServicePort=<port> --filePath=<filePath>
fission-bundle --builderMgrPort=<port> [--storageSvcUrl=<url>] [--envbuilder-namespace=<namespace>]
fission-bundle --logger
fission-bundle --timer [--routerUrl=<url>]
fission-bundle --mqt [--routerUrl=<url>]
Options:
@@ -138,7 +131,6 @@ 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.
--timer Start Timer.
--mqt Start message queue trigger.
`
@@ -174,10 +166,6 @@ Options:
runKubeWatcher(routerUrl)
}
if arguments["--logger"] == true {
runLogger()
}
if arguments["--timer"] == true {
runTimer(routerUrl)
}
+8 -7
View File
@@ -96,14 +96,15 @@ func (influx InfluxDB) GetLogs(filter LogFilter) ([]LogEntry, error) {
log.Fatal(err)
}
logEntries = append(logEntries, LogEntry{
//The attributes of the LogEntry are selected as relative to their position in InfluxDB's line protocol response
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),
Container: row[2].(string), //docker_container_id
FuncName: row[8].(string), //kubernetes_labels_functionName
FuncUid: row[3].(string), //funcuid
Message: strings.TrimSuffix(row[17].(string), "\n"), //log field
Namespace: row[14].(string), //kubernetes_namespace_name
Pod: row[15].(string), //kubernetes_pod_name
Stream: row[18].(string), //stream
})
}
}
+3 -1
View File
@@ -35,7 +35,7 @@ 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 curl ca-certificates gcc g++ 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 && \
@@ -45,6 +45,8 @@ RUN apt-get -qq update && \
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 && \
td-agent-gem install --no-document fluent-plugin-kubernetes_metadata_filter && \
td-agent-gem install --no-document fluent-plugin-flatten-hash && \
# Remove build tools
apt-get remove -y -qq gcc make && \
apt-get autoremove -y -qq && \
+1 -1
View File
@@ -1,2 +1,2 @@
#!/bin/sh
docker build -t fission-daemonset-fluentd:latest .
docker build -t fission-daemonset-fluentd:latest .
+20 -14
View File
@@ -4,26 +4,32 @@
<source>
type tail
format json
format json
time_key time
path /var/log/fission/*.log
path "#{ENV['FLUENTD_PATH']}"
time_format %Y-%m-%dT%H:%M:%S.%NZ
tag fission.*
read_from_head true
refresh_interval 5
</source>
<filter fission.**>
type kubernetes_metadata
</filter>
<filter fission.**>
type flatten_hash
separator _
</filter>
<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>
type record_reformer
enable_ruby false
tag log
<record>
funcuid ${kubernetes_labels_functionUid}
</record>
</match>
<match **>
@@ -35,7 +41,7 @@
password "#{ENV['INFLUXDB_PASSWD']}"
use_ssl false
time_precision s
tag_keys ["funcuid", "pod"]
tag_keys ["funcuid"]
sequence_tag _seq
buffer_type file
buffer_path /var/log/fission/fluentd.buffer
@@ -45,4 +51,4 @@
retry_limit 10
retry_wait 1.0
num_threads 2
</match>
</match>
-216
View File
@@ -1,216 +0,0 @@
/*
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"
"github.com/fission/fission"
"github.com/gorilla/handlers"
"github.com/gorilla/mux"
log "github.com/sirupsen/logrus"
"k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/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.CoreV1().Pods(namespace).Get(pod, v1.GetOptions{})
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 logger at port %s", address)
log.Fatal(http.ListenAndServe(address, handlers.LoggingHandler(os.Stdout, r)))
}
-33
View File
@@ -43,7 +43,6 @@ import (
"github.com/fission/fission/crd"
"github.com/fission/fission/environments/fetcher"
fetcherClient "github.com/fission/fission/environments/fetcher/client"
"github.com/fission/fission/logger"
)
const POOLMGR_INSTANCEID_LABEL string = "poolmgrInstanceId"
@@ -348,9 +347,6 @@ func (gp *GenericPool) specializePod(pod *apiv1.Pod, metadata *metav1.ObjectMeta
return err
}
// Tell logging helper about this function invocation
gp.setupLogging(pod, metadata)
// get function run container to specialize
log.Printf("[%v] specializing pod", metadata.Name)
@@ -682,32 +678,3 @@ 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 *apiv1.Pod, metadata *metav1.ObjectMeta) {
logReq := logger.LogRequest{
Namespace: pod.Namespace,
Pod: pod.Name,
Container: gp.env.Metadata.Name,
FuncName: metadata.Name,
FuncUid: string(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()
}
}()
}
+4 -1
View File
@@ -13,6 +13,7 @@ source $(dirname $0)/test_utils.sh
REPO=gcr.io/fission-ci
IMAGE=$REPO/fission-bundle
FETCHER_IMAGE=$REPO/fetcher
FLUENTD_IMAGE=gcr.io/fission-ci/fluentd
TAG=test
build_and_push_fission_bundle $IMAGE:$TAG
@@ -27,6 +28,8 @@ build_and_push_env_runtime $ENV $REPO/$ENV-env:$TAG
build_and_push_env_builder $ENV $REPO/$ENV-env-builder:$TAG
build_and_push_fluentd $FLUENTD_IMAGE:$TAG
build_fission_cli
install_and_test $IMAGE $TAG $FETCHER_IMAGE $TAG
install_and_test $IMAGE $TAG $FETCHER_IMAGE $TAG $FLUENTD_IMAGE $TAG
+18 -2
View File
@@ -56,6 +56,19 @@ build_builder() {
popd
}
build_and_push_fluentd(){
image_tag=$1
pushd $ROOT/logger/fluentd
docker build -t $image_tag .
gcloud_login
gcloud docker -- push $image_tag
popd
}
build_and_push_env_runtime() {
env=$1
image_tag=$2
@@ -110,11 +123,12 @@ helm_install_fission() {
fetcherImageTag=$5
controllerNodeport=$6
routerNodeport=$7
fluentdImage=$8
ns=f-$id
fns=f-func-$id
helmVars=image=$image,imageTag=$imageTag,fetcherImage=$fetcherImage,fetcherImageTag=$fetcherImageTag,functionNamespace=$fns,controllerPort=$controllerNodeport,routerPort=$routerNodeport,pullPolicy=Always,analytics=false
helmVars=image=$image,imageTag=$imageTag,fetcherImage=$fetcherImage,fetcherImageTag=$fetcherImageTag,functionNamespace=$fns,controllerPort=$controllerNodeport,routerPort=$routerNodeport,pullPolicy=Always,analytics=false,logger.fluentdImage=$fluentdImage
helm_setup
@@ -280,6 +294,8 @@ install_and_test() {
imageTag=$2
fetcherImage=$3
fetcherImageTag=$4
fluentdImage=$5
fluentdImageTag=$6
controllerPort=31234
routerPort=31235
@@ -288,7 +304,7 @@ install_and_test() {
id=$(generate_test_id)
trap "helm_uninstall_fission $id" EXIT
if ! helm_install_fission $id $image $imageTag $fetcherImage $fetcherImageTag $controllerPort $routerPort
if ! helm_install_fission $id $image $imageTag $fetcherImage $fetcherImageTag $controllerPort $routerPort $fluentdImage:$fluentdImageTag
then
dump_logs $id
exit 1
+8
View File
@@ -0,0 +1,8 @@
module.exports = async function(context) {
console.log("log test log test log test")
return {
status: 200,
body: "Log, test!\n"
};
}
+58
View File
@@ -0,0 +1,58 @@
#!/bin/bash
set -euo pipefail
ROOT=$(dirname $0)/../..
fn=nodejs-logtest
function cleanup {
echo "Cleanup route"
var=$(fission route list | grep $fn | awk '{print $1;}')
fission route delete --name $var
fission function delete --name $fn
}
# Create a hello world function in nodejs, test it with an http trigger
echo "Pre-test cleanup"
fission env delete --name nodejs || true
echo "Creating nodejs env"
fission env create --name nodejs --image fission/node-env
trap "fission env delete --name nodejs" EXIT
echo "Creating function"
fission fn create --name $fn --env nodejs --code log.js
trap "fission fn delete --name $fn" EXIT
echo "Creating route"
fission route create --function $fn --url /logtest --method GET
echo "Waiting for router to catch up"
sleep 3
echo "Doing 4 HTTP GET on the function's route"
curl http://$FISSION_ROUTER/logtest
curl http://$FISSION_ROUTER/logtest
curl http://$FISSION_ROUTER/logtest
curl http://$FISSION_ROUTER/logtest
echo "Grabbing logs, should have 4 calls in logs"
sleep 15
echo "woke up"
logs=$(fission function logs --name $fn)
num=$(echo "$logs" | grep 'log test' | wc -l)
echo $num
if [ $num -ne 4 ]
then
echo "Test Failed"
trap cleanup EXIT
fi
cleanup
echo "All done."