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 + } + + 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 { - 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 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") + } } - // 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") - - } - return podLogs, err + return 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) - +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 { 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"