Handle logs from all pods in function and error condition in fission fn log command (#2634)

* handle error condition in fission fn log command
* use single stream for log exclude fetcher logs
* add all-pods in fn logs command
* update previous stable version

pkg/fission-cli/cmd/function/command.go

@@ -133,7 +133,7 @@ func Commands() *cobra.Command {
 		Required: []flag.Flag{flag.FnName},
 		Optional: []flag.Flag{
 			flag.FnLogFollow, flag.FnLogReverseQuery, flag.FnLogCount,
-			flag.FnLogDetail, flag.FnLogPod, flag.NamespaceFunction, flag.FnLogDBType, flag.NamespacePod},
+			flag.FnLogDetail, flag.FnLogPod, flag.NamespaceFunction, flag.FnLogDBType, flag.NamespacePod, flag.FnLogAllPods},
 	})
 
 	testCmd := &cobra.Command{

pkg/fission-cli/cmd/function/log.go

@@ -17,8 +17,8 @@ limitations under the License.
 package function
 
 import (
+	"bytes"
 	"context"
-	"fmt"
 	"io"
 	"os"
 	"time"
@@ -28,6 +28,7 @@ import (
 
 	"github.com/fission/fission/pkg/fission-cli/cliwrapper/cli"
 	"github.com/fission/fission/pkg/fission-cli/cmd"
+	"github.com/fission/fission/pkg/fission-cli/console"
 	flagkey "github.com/fission/fission/pkg/fission-cli/flag/key"
 	"github.com/fission/fission/pkg/fission-cli/logdb"
 )
@@ -51,6 +52,7 @@ func (opts *LogSubCommand) do(input cli.Input) error {
 
 	logReverseQuery := !input.Bool(flagkey.FnLogFollow) && input.Bool(flagkey.FnLogReverseQuery)
 
+	allPods := input.Bool(flagkey.FnLogAllPods)
 	recordLimit := input.Int(flagkey.FnLogCount)
 	if recordLimit <= 0 {
 		recordLimit = 1000
@@ -74,6 +76,7 @@ func (opts *LogSubCommand) do(input cli.Input) error {
 	requestChan := make(chan struct{})
 	responseChan := make(chan struct{})
 	ctx := input.Context()
+	warn := true
 
 	go func(ctx context.Context, requestChan, responseChan chan struct{}) {
 		t := time.Unix(0, 0*int64(time.Millisecond))
@@ -91,20 +94,28 @@ func (opts *LogSubCommand) do(input cli.Input) error {
 					RecordLimit:    recordLimit,
 					FunctionObject: f,
 					Details:        detail,
+					WarnUser:       warn,
+					AllPods:        allPods,
 				}
 
-				buf, err := logDB.GetLogs(ctx, logFilter)
+				buf := new(bytes.Buffer)
+				err = logDB.GetLogs(ctx, logFilter, buf)
+				t = time.Now().UTC() // next time fetch values from this time
 				if err != nil {
-					fmt.Printf("Error querying logs: %v", err)
+					console.Verbose(2, "error querying logs: %s", err)
+					if dbType == logdb.KUBERNETES { //in case of Kubernetes log we print pod namespace warning once
+						warn = false
+					}
 					responseChan <- struct{}{}
-					return
+					continue
 				}
 				_, err = io.Copy(os.Stdout, buf)
 				if err != nil {
-					return
+					console.Verbose(2, "eror copying logs: %s", err)
+					responseChan <- struct{}{}
+					continue
 				}
 
-				t = time.Now().UTC()            // next time fetch values from this time
 				if dbType == logdb.KUBERNETES { //in case of Kubernetes log we print pods info only once. And then print new logs
 					detail = false
 				}

pkg/fission-cli/flag/flag.go

@@ -130,6 +130,7 @@ var (
 	FnRequestsPerPod        = Flag{Type: Int, Name: flagkey.FnRequestsPerPod, Aliases: []string{"rpp"}, Usage: "Maximum number of concurrent requests that can be served by a specialized pod", DefaultValue: 1}
 	FnOnceOnly              = Flag{Type: Bool, Name: flagkey.FnOnceOnly, Aliases: []string{"yolo"}, Usage: "Specifies if specialized pod will serve exactly one request in its lifetime"}
 	FnSubPath               = Flag{Type: String, Name: flagkey.FnSubPath, Usage: "Sub Path to check if function internally supports routing"}
+	FnLogAllPods            = Flag{Type: Bool, Name: flagkey.FnLogAllPods, Usage: "Get all pod's logs in the function."}
 	// Termination Grace Period configurable at function creation/update only for container functions
 	FnTerminationGracePeriod = Flag{Type: Int64, Name: flagkey.FnGracePeriod, Usage: "Grace time (in seconds) for pod to perform connection draining before termination (default value will be used if negative value is given)", DefaultValue: 360}
 

pkg/fission-cli/flag/key/key.go

@@ -84,6 +84,7 @@ const (
 	FnOnceOnly              = "onceonly"
 	FnSubPath               = "subpath"
 	FnGracePeriod           = "graceperiod"
+	FnLogAllPods            = "all-pods"
 
 	HtName              = resourceName
 	HtMethod            = "method"

pkg/fission-cli/logdb/influxdb.go

@@ -62,7 +62,7 @@ func makeIndexMap(cols []string) map[string]int {
 	return indexMap
 }
 
-func (influx InfluxDB) GetLogs(ctx context.Context, filter LogFilter) (output *bytes.Buffer, err error) {
+func (influx InfluxDB) GetLogs(ctx context.Context, filter LogFilter, output *bytes.Buffer) (err error) {
 	timestamp := filter.Since.UnixNano()
 	var queryCmd string
 
@@ -91,7 +91,7 @@ func (influx InfluxDB) GetLogs(ctx context.Context, filter LogFilter) (output *b
 	logEntries := []LogEntry{}
 	response, err := influx.query(query)
 	if err != nil {
-		return nil, err
+		return err
 	}
 	for _, r := range response.Results {
 		for _, series := range r.Series {
@@ -115,11 +115,11 @@ func (influx InfluxDB) GetLogs(ctx context.Context, filter LogFilter) (output *b
 			for _, row := range series.Values {
 				t, err := time.Parse(time.RFC3339, row[0].(string))
 				if err != nil {
-					return nil, err
+					return err
 				}
 				seqNum, err := strconv.Atoi(row[seq].(string))
 				if err != nil {
-					return nil, err
+					return err
 				}
 				entry := LogEntry{
 					//The attributes of the LogEntry are selected as relative to their position in InfluxDB's line protocol response
@@ -140,23 +140,22 @@ func (influx InfluxDB) GetLogs(ctx context.Context, filter LogFilter) (output *b
 
 	sort.Sort(ByTimestamp(logEntries, filter.Reverse))
 
-	output = new(bytes.Buffer)
 	for _, logEntry := range logEntries {
 		if filter.Details {
 			msg := fmt.Sprintf("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)
 			if _, err := output.WriteString(msg); err != nil {
-				return output, errors.Wrapf(err, "error copying pod log")
+				return errors.Wrapf(err, "error copying pod log")
 			}
 		} else {
 			msg := fmt.Sprintf("[%s] %s\n", logEntry.Timestamp, logEntry.Message)
 			if _, err := output.WriteString(msg); err != nil {
-				return output, errors.Wrapf(err, "error copying pod log")
+				return errors.Wrapf(err, "error copying pod log")
 			}
 		}
 	}
 
-	return output, nil
+	return nil
 }
 
 func (influx InfluxDB) query(query influxdbClient.Query) (*influxdbClient.Response, error) {

pkg/fission-cli/logdb/kubernetes_log.go

@@ -23,7 +23,6 @@ import (
 	"io"
 	"sort"
 	"strconv"
-	"strings"
 
 	"github.com/pkg/errors"
 	v1 "k8s.io/api/core/v1"
@@ -44,9 +43,9 @@ type kubernetesLogs struct {
 	client cmd.Client
 }
 
-func (k kubernetesLogs) GetLogs(ctx context.Context, logFilter LogFilter) (podLogs *bytes.Buffer, err error) {
+func (k kubernetesLogs) GetLogs(ctx context.Context, logFilter LogFilter, podLogs *bytes.Buffer) (err error) {
-	podLogs, err = GetFunctionPodLogs(ctx, k.client, logFilter)
+	err = GetFunctionPodLogs(ctx, k.client, logFilter, podLogs)
-	return podLogs, err
+	return err
 }
 
 func NewKubernetesEndpoint(logDBOptions LogDBOptions) (kubernetesLogs, error) {
@@ -55,7 +54,7 @@ func NewKubernetesEndpoint(logDBOptions LogDBOptions) (kubernetesLogs, error) {
 }
 
 // FunctionPodLogs : Get logs for a function directly from pod
-func GetFunctionPodLogs(ctx context.Context, client cmd.Client, logFilter LogFilter) (podLogs *bytes.Buffer, err error) {
+func GetFunctionPodLogs(ctx context.Context, client cmd.Client, logFilter LogFilter, podLogs *bytes.Buffer) (err error) {
 
 	f := logFilter.FunctionObject
 
@@ -73,38 +72,49 @@ func GetFunctionPodLogs(ctx context.Context, client cmd.Client, logFilter LogFil
 		LabelSelector: labels.Set(selector).AsSelector().String(),
 	})
 	if err != nil {
-		return podLogs, err
+		return err
+	}
+
+	if len(podList.Items) <= 0 {
+		if logFilter.WarnUser {
+			console.Warn("version<1.18 used fission-function as pod's default namespace. Specify appropriate namespace with --pod-namespace tag.")
+		}
+		return errors.New("no active pods found")
 	}
 
-	// Get the logs for last Pod executed
 	pods := podList.Items
-	sort.Slice(pods, func(i, j int) bool {
+	if logFilter.AllPods {
-		rv1, _ := strconv.ParseInt(pods[i].ObjectMeta.ResourceVersion, 10, 32)
+		for _, pod := range pods {
-		rv2, _ := strconv.ParseInt(pods[j].ObjectMeta.ResourceVersion, 10, 32)
+			// get the pod with highest resource version
-		return rv1 > rv2
+			err = streamContainerLog(ctx, client.KubernetesClient, &pod, logFilter, podLogs)
-	})
+			if err != nil {
-
+				return errors.Wrapf(err, "error getting container logs")
-	if len(pods) <= 0 {
+			}
-		console.Warn("version<1.18 used fission-function as pod's default namespace. Specify appropriate namespace with --pod-namespace tag.")
+		}
-		return podLogs, errors.New("no active pods found")
+	} else {
+		// Get the logs for last Pod executed
+		sort.Slice(pods, func(i, j int) bool {
+			rv1, _ := strconv.ParseInt(pods[i].ObjectMeta.ResourceVersion, 10, 32)
+			rv2, _ := strconv.ParseInt(pods[j].ObjectMeta.ResourceVersion, 10, 32)
+			return rv1 > rv2
+		})
 
+		// get the pod with highest resource version
+		err = streamContainerLog(ctx, client.KubernetesClient, &pods[0], logFilter, podLogs)
+		if err != nil {
+			return errors.Wrapf(err, "error getting container logs")
+		}
 	}
 
-	// get the pod with highest resource version
+	return err
-	podLogs, err = streamContainerLog(ctx, client.KubernetesClient, &pods[0], logFilter)
-	if err != nil {
-		return podLogs, errors.Wrapf(err, "error getting container logs")
-
-	}
-	return podLogs, err
 }
 
-func streamContainerLog(ctx context.Context, kubernetesClient kubernetes.Interface, pod *v1.Pod, logFilter LogFilter) (output *bytes.Buffer, err error) {
+func streamContainerLog(ctx context.Context, kubernetesClient kubernetes.Interface, pod *v1.Pod, logFilter LogFilter, output *bytes.Buffer) (err error) {
-
+	FETCHER := "fetcher"
-	seq := strings.Repeat("=", 35)
-	output = new(bytes.Buffer)
-
 	for _, container := range pod.Spec.Containers {
+		if container.Name == FETCHER {
+			continue
+		}
 		tailLines := int64(logFilter.RecordLimit)
 		sinceTime := metav1.NewTime(logFilter.Since)
 		podLogOpts := v1.PodLogOptions{Container: container.Name, // Only the env container, not fetcher
@@ -116,26 +126,25 @@ func streamContainerLog(ctx context.Context, kubernetesClient kubernetes.Interfa
 
 		podLogs, err := podLogsReq.Stream(ctx)
 		if err != nil {
-			return output, errors.Wrapf(err, "error streaming pod log")
+			return errors.Wrapf(err, "error streaming pod log")
 		}
 
 		if logFilter.Details {
 			fn := logFilter.FunctionObject
-			msg := fmt.Sprintf("\n%v\nFunction: %v\nEnvironment: %v\nNamespace: %v\nPod: %v\nContainer: %v\nNode: %v\n%v\n", seq,
+			msg := fmt.Sprintf("\n=== Function=%s Environment=%s Namespace=%s Pod=%s Container=%s Node=%s\n",
-				fn.ObjectMeta.Name, fn.Spec.Environment.Name, pod.Namespace, pod.Name, container.Name, pod.Spec.NodeName, seq)
+				fn.ObjectMeta.Name, fn.Spec.Environment.Name, pod.Namespace, pod.Name, container.Name, pod.Spec.NodeName)
-
 			if _, err := output.WriteString(msg); err != nil {
-				return output, errors.Wrapf(err, "error copying pod log")
+				return errors.Wrapf(err, "error copying pod log")
 			}
 		}
 
 		_, err = io.Copy(output, podLogs)
 		if err != nil {
-			return output, errors.Wrapf(err, "error copying pod log")
+			return errors.Wrapf(err, "error copying pod log")
 		}
 
 		podLogs.Close()
 	}
 
-	return output, nil
+	return nil
 }

pkg/fission-cli/logdb/logdb.go

@@ -31,7 +31,7 @@ const (
 )
 
 type LogDatabase interface {
-	GetLogs(context.Context, LogFilter) (*bytes.Buffer, error)
+	GetLogs(context.Context, LogFilter, *bytes.Buffer) error
 }
 
 type LogFilter struct {
@@ -44,6 +44,8 @@ type LogFilter struct {
 	RecordLimit    int
 	FunctionObject *v1.Function
 	Details        bool
+	WarnUser       bool
+	AllPods        bool
 }
 
 type LogEntry struct {

test/upgrade_test/fission_objects.sh

@@ -3,7 +3,7 @@ set -eu
 
 ns="fission"
 ROOT=$(pwd)
-PREV_STABLE_VERSION=1.13.1
+PREV_STABLE_VERSION=v1.16.3
 HELM_VARS_PREV_RELEASE="routerServiceType=NodePort,analytics=false"
 HELM_VARS_LATEST_RELEASE="routerServiceType=NodePort,repository=docker.io/library,image=fission-bundle,pullPolicy=IfNotPresent,imageTag=latest,fetcher.image=docker.io/library/fetcher,fetcher.imageTag=latest,postInstallReportImage=reporter,preUpgradeChecks.image=preupgradechecks,preUpgradeChecks.imageTag=latest,analytics=false"