Repository: fission/fission
Pull request #2634 (2022-11-23 06:05:48)

Handle logs from all pods in function and error condition in fission fn log command
<!--  Thanks for sending a pull request! We request you provide detailed description as much as possible. -->

## Description
- Handle error while fetching logs
- use %s instead of %v while printing data to CLI
- use correct time to fetch updated logs 
- add `--all-pods` flag-
        - all-pods= true - fetches logs from all the Pods which are running for the Fission function. 
        - all-pods= false(**default value**) -  fetches logs for latest pod which was running for the function
eg of `fn log` command with `--all-pods` flag,
![log-all-pod](https://user-images.githubusercontent.com/23420056/203545665-71517759-1b9b-4ff8-80f5-7114e7582626.png)
 

## Which issue(s) this PR fixes:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
-->
Fixes #

## Testing
<!--- Please describe in detail how you tested your changes. -->

## Checklist:
<!-- Please tick following checkboxes as per your understanding. -->
- [ ] I ran tests as well as code linting locally to verify my changes. 
- [ ] I have done manual verification of my changes, changes working as expected.
- [ ] I have added new tests to cover my changes.
- [ ] My changes follow contributing guidelines of Fission.
- [ ] I have signed all of my commits.



Patch:
diff --git a/pkg/fission-cli/cmd/function/command.go b/pkg/fission-cli/cmd/function/command.go
index c4a29489..2ce67353 100644
--- a/pkg/fission-cli/cmd/function/command.go
+++ b/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{
diff --git a/pkg/fission-cli/cmd/function/log.go b/pkg/fission-cli/cmd/function/log.go
index 75808a7a..ee040655 100644
--- a/pkg/fission-cli/cmd/function/log.go
+++ b/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
 				}
diff --git a/pkg/fission-cli/flag/flag.go b/pkg/fission-cli/flag/flag.go
index 38a086df..5150539d 100644
--- a/pkg/fission-cli/flag/flag.go
+++ b/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}
 
diff --git a/pkg/fission-cli/flag/key/key.go b/pkg/fission-cli/flag/key/key.go
index e1116eb6..647a606b 100644
--- a/pkg/fission-cli/flag/key/key.go
+++ b/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"
diff --git a/pkg/fission-cli/logdb/influxdb.go b/pkg/fission-cli/logdb/influxdb.go
index bff9bea9..15df33fc 100644
--- a/pkg/fission-cli/logdb/influxdb.go
+++ b/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) {
diff --git a/pkg/fission-cli/logdb/kubernetes_log.go b/pkg/fission-cli/logdb/kubernetes_log.go
index 107d6fd5..1d208906 100644
--- a/pkg/fission-cli/logdb/kubernetes_log.go
+++ b/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) {
-	podLogs, err = GetFunctionPodLogs(ctx, k.client, logFilter)
-	return podLogs, err
+func (k kubernetesLogs) GetLogs(ctx context.Context, logFilter LogFilter, podLogs *bytes.Buffer) (err error) {
+	err = GetFunctionPodLogs(ctx, k.client, logFilter, podLogs)
+	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
 	}
 
-	// Get the logs for last Pod executed
-	pods := podList.Items
-	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
-	})
-
-	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")
-
+	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 pod with highest resource version
-	podLogs, err = streamContainerLog(ctx, client.KubernetesClient, &pods[0], logFilter)
-	if err != nil {
-		return podLogs, errors.Wrapf(err, "error getting container logs")
-
+	pods := podList.Items
+	if logFilter.AllPods {
+		for _, pod := range pods {
+			// get the pod with highest resource version
+			err = streamContainerLog(ctx, client.KubernetesClient, &pod, logFilter, podLogs)
+			if err != nil {
+				return errors.Wrapf(err, "error getting container logs")
+			}
+		}
+	} 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")
+		}
 	}
-	return podLogs, err
-}
 
-func streamContainerLog(ctx context.Context, kubernetesClient kubernetes.Interface, pod *v1.Pod, logFilter LogFilter) (output *bytes.Buffer, err error) {
-
-	seq := strings.Repeat("=", 35)
-	output = new(bytes.Buffer)
+	return err
+}
 
+func streamContainerLog(ctx context.Context, kubernetesClient kubernetes.Interface, pod *v1.Pod, logFilter LogFilter, output *bytes.Buffer) (err error) {
+	FETCHER := "fetcher"
 	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,
-				fn.ObjectMeta.Name, fn.Spec.Environment.Name, pod.Namespace, pod.Name, container.Name, pod.Spec.NodeName, 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)
 			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
 }
diff --git a/pkg/fission-cli/logdb/logdb.go b/pkg/fission-cli/logdb/logdb.go
index e2c6a65b..90faaa3f 100644
--- a/pkg/fission-cli/logdb/logdb.go
+++ b/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 patch:
diff --git a/test/upgrade_test/fission_objects.sh b/test/upgrade_test/fission_objects.sh
index 08be3d2a..be364300 100644
--- a/test/upgrade_test/fission_objects.sh
+++ b/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"