diff --git a/charts/fission-all/templates/deployment.yaml b/charts/fission-all/templates/deployment.yaml index 2902b028..36438d51 100644 --- a/charts/fission-all/templates/deployment.yaml +++ b/charts/fission-all/templates/deployment.yaml @@ -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 diff --git a/charts/fission-all/values.yaml b/charts/fission-all/values.yaml index d3a384fe..86ee877f 100644 --- a/charts/fission-all/values.yaml +++ b/charts/fission-all/values.yaml @@ -10,7 +10,7 @@ serviceType: LoadBalancer image: fission/fission-bundle ## Image pull policy -pullPolicy: IfNotPresent +pullPolicy: IfNotPresent ## Fission image version imageTag: 0.4.0rc diff --git a/fission-bundle/main.go b/fission-bundle/main.go index c652819b..b2890bb4 100644 --- a/fission-bundle/main.go +++ b/fission-bundle/main.go @@ -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=] fission-bundle --storageServicePort= --filePath= fission-bundle --builderMgrPort= [--storageSvcUrl=] [--envbuilder-namespace=] - fission-bundle --logger fission-bundle --timer [--routerUrl=] fission-bundle --mqt [--routerUrl=] Options: @@ -138,7 +131,6 @@ 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. --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) } diff --git a/fission/logdb/influxdb.go b/fission/logdb/influxdb.go index fded6b29..63fd4b35 100644 --- a/fission/logdb/influxdb.go +++ b/fission/logdb/influxdb.go @@ -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 }) } } diff --git a/logger/fluentd/Dockerfile b/logger/fluentd/Dockerfile index 600fb492..9d154135 100644 --- a/logger/fluentd/Dockerfile +++ b/logger/fluentd/Dockerfile @@ -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 && \ diff --git a/logger/fluentd/build.sh b/logger/fluentd/build.sh index 098a765f..06aa0b98 100755 --- a/logger/fluentd/build.sh +++ b/logger/fluentd/build.sh @@ -1,2 +1,2 @@ #!/bin/sh -docker build -t fission-daemonset-fluentd:latest . +docker build -t fission-daemonset-fluentd:latest . diff --git a/logger/fluentd/fluent.conf b/logger/fluentd/fluent.conf index e2299b44..a10f99ae 100644 --- a/logger/fluentd/fluent.conf +++ b/logger/fluentd/fluent.conf @@ -4,26 +4,32 @@ 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 + + type kubernetes_metadata + + + + + type flatten_hash + separator _ + + - 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 record_reformer + enable_ruby false + tag log + + funcuid ${kubernetes_labels_functionUid} + @@ -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 - \ No newline at end of file + diff --git a/logger/logger.go b/logger/logger.go deleted file mode 100644 index 21047786..00000000 --- a/logger/logger.go +++ /dev/null @@ -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))) -} diff --git a/poolmgr/gp.go b/poolmgr/gp.go index 47ec8e8b..613f9e1f 100644 --- a/poolmgr/gp.go +++ b/poolmgr/gp.go @@ -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() - } - }() -} diff --git a/test/build_and_test.sh b/test/build_and_test.sh index e0306f5d..bef72533 100755 --- a/test/build_and_test.sh +++ b/test/build_and_test.sh @@ -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 diff --git a/test/test_utils.sh b/test/test_utils.sh index b538346e..e28cc60c 100755 --- a/test/test_utils.sh +++ b/test/test_utils.sh @@ -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 diff --git a/test/tests/test_logging/log.js b/test/tests/test_logging/log.js new file mode 100644 index 00000000..6546bdfc --- /dev/null +++ b/test/tests/test_logging/log.js @@ -0,0 +1,8 @@ + +module.exports = async function(context) { + console.log("log test log test log test") + return { + status: 200, + body: "Log, test!\n" + }; +} diff --git a/test/tests/test_logging/test_function_logs.sh b/test/tests/test_logging/test_function_logs.sh new file mode 100755 index 00000000..b605d4ee --- /dev/null +++ b/test/tests/test_logging/test_function_logs.sh @@ -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."