github/kubevela
1
0
Fork 0
mirror of https://github.com/kubevela/kubevela.git synced 2026-08-23 22:46:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Feat: support step logs in vela logs (#4845)

Browse Source 
* Feat: support step logs in vela logs

Signed-off-by: FogDong <dongtianxin.tx@alibaba-inc.com>

* resolve comments

Signed-off-by: FogDong <dongtianxin.tx@alibaba-inc.com>

Signed-off-by: FogDong <dongtianxin.tx@alibaba-inc.com>
This commit is contained in:
Tianxin Dong
2022-10-17 19:01:59 +08:00
committed by GitHub
parent 49ed837f97
commit ccf7bdd2d6
3 changed files with 185 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/stdlib/pkgs/util.cue
+13 -1
View File
@@ -20,5 +20,17 @@
#do: "log"
#provider: "util"
data: {...}
data?: {...} | string
level: *3 | int
// note that if you set source in multiple op.#Log, only the latest one will work
source?: close({
url: string
}) | close({
resources?: [...{
name?: string
cluster?: string
namespace?: string
labelSelector?: {...}
}]
})
}
references/cli/cli.go
+9 -2
View File
@@ -25,9 +25,11 @@ import (
gov "github.com/hashicorp/go-version"
"github.com/pkg/errors"
"github.com/spf13/cobra"
"k8s.io/klog"
"k8s.io/klog/v2"
"sigs.k8s.io/controller-runtime/pkg/client/config"
workflowv1alpha1 "github.com/kubevela/workflow/api/v1alpha1"
"github.com/oam-dev/kubevela/apis/types"
velacmd "github.com/oam-dev/kubevela/pkg/cmd"
"github.com/oam-dev/kubevela/pkg/utils/common"
@@ -68,8 +70,13 @@ func NewCommandWithIOStreams(ioStream util.IOStreams) *cobra.Command {
},
}
scheme := common.Scheme
err := workflowv1alpha1.AddToScheme(scheme)
if err != nil {
klog.Fatal(err)
}
commandArgs := common.Args{
Schema: common.Scheme,
Schema: scheme,
}
f := velacmd.NewDeferredFactory(config.GetConfig)
references/cli/logs.go
+163 -25
View File
@@ -20,10 +20,13 @@ import (
"context"
"encoding/json"
"fmt"
"io"
"regexp"
"strings"
"text/template"
"time"
"github.com/AlecAivazis/survey/v2"
"github.com/fatih/color"
"github.com/pkg/errors"
"github.com/spf13/cobra"
@@ -31,6 +34,11 @@ import (
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/selection"
"k8s.io/client-go/kubernetes"
"sigs.k8s.io/controller-runtime/pkg/client"
workflowv1alpha1 "github.com/kubevela/workflow/api/v1alpha1"
wfTypes "github.com/kubevela/workflow/pkg/types"
wfUtils "github.com/kubevela/workflow/pkg/utils"
"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
"github.com/oam-dev/kubevela/apis/types"
@@ -55,12 +63,25 @@ func NewLogsCommand(c common.Args, order string, ioStreams util.IOStreams) *cobr
if err != nil {
return err
}
largs.Name = args[0]
ctx := context.Background()
if largs.StepName != "" {
cli, err := c.GetClient()
if err != nil {
return err
}
ctxName, label, err := getContextFromInstance(ctx, cli, largs.Namespace, largs.Name)
if err != nil {
return err
}
largs.CtxName = ctxName
return largs.printStepLogs(ctx, ioStreams, label)
}
app, err := appfile.LoadApplication(largs.Namespace, args[0], c)
if err != nil {
return err
}
largs.App = app
ctx := context.Background()
if err := largs.Run(ctx, ioStreams); err != nil {
return err
}
@@ -77,6 +98,7 @@ func NewLogsCommand(c common.Args, order string, ioStreams util.IOStreams) *cobr
cmd.Flags().StringVarP(&largs.ClusterName, "cluster", "", "", "filter the pod by the cluster name")
cmd.Flags().StringVarP(&largs.PodName, "pod", "p", "", "specify the pod name")
cmd.Flags().StringVarP(&largs.ContainerName, "container", "", "", "specify the container name")
cmd.Flags().StringVarP(&largs.StepName, "step", "s", "", "specify the step name, note that this flag cannot be used together with the pod, container or component flags")
addNamespaceAndEnvArg(cmd)
return cmd
}
@@ -85,49 +107,100 @@ func NewLogsCommand(c common.Args, order string, ioStreams util.IOStreams) *cobr
type Args struct {
Output string
Args common.Args
Name string
CtxName string
Namespace string
ContainerName string
PodName string
ClusterName string
ComponentName string
StepName string
App *v1beta1.Application
}
// Run refer to the implementation at https://github.com/oam-dev/stern/blob/master/stern/main.go
func (l *Args) Run(ctx context.Context, ioStreams util.IOStreams) error {
pods, err := GetApplicationPods(ctx, l.App.Name, l.App.Namespace, l.Args, Filter{
Component: l.ComponentName,
Cluster: l.ClusterName,
})
func (l *Args) printStepLogs(ctx context.Context, ioStreams util.IOStreams, label map[string]string) error {
cli, err := l.Args.GetClient()
if err != nil {
return err
}
var selectPod *querytypes.PodBase
if l.PodName != "" {
for i, pod := range pods {
if pod.Metadata.Name == l.PodName {
selectPod = &pods[i]
break
logConfig, err := wfUtils.GetLogConfigFromStep(ctx, cli, l.CtxName, l.Name, l.Namespace, l.StepName)
if err != nil {
return err
}
if err := selectStepLogSource(logConfig); err != nil {
return err
}
switch {
case logConfig.Data:
return l.printResourceLogs(ctx, cli, ioStreams, []wfTypes.Resource{{
Namespace: types.DefaultKubeVelaNS,
LabelSelector: label,
}}, []string{fmt.Sprintf(`step_name="%s"`, l.StepName), fmt.Sprintf("%s/%s", l.Namespace, l.Name)})
case logConfig.Source != nil:
if len(logConfig.Source.Resources) > 0 {
return l.printResourceLogs(ctx, cli, ioStreams, logConfig.Source.Resources, nil)
}
if logConfig.Source.URL != "" {
readCloser, err := wfUtils.GetLogsFromURL(ctx, logConfig.Source.URL)
if err != nil {
return err
}
//nolint:errcheck
defer readCloser.Close()
if _, err := io.Copy(ioStreams.Out, readCloser); err != nil {
return err
}
}
if selectPod == nil {
fmt.Println("The Pod you specified does not exist, please select it from the list.")
}
return nil
}
func selectStepLogSource(logConfig *wfTypes.LogConfig) error {
var source string
if logConfig.Data && logConfig.Source != nil {
prompt := &survey.Select{
Message: "Select logs from data or source",
Options: []string{"data", "source"},
}
err := survey.AskOne(prompt, &source, survey.WithValidator(survey.Required))
if err != nil {
return fmt.Errorf("failed to select %s: %w", source, err)
}
if source != "data" {
logConfig.Data = false
}
}
if selectPod == nil {
selectPod, err = AskToChooseOnePod(pods)
return nil
}
func (l *Args) printResourceLogs(ctx context.Context, cli client.Client, ioStreams util.IOStreams, resources []wfTypes.Resource, filters []string) error {
pods, err := wfUtils.GetPodListFromResources(ctx, cli, resources)
if err != nil {
return err
}
podList := make([]querytypes.PodBase, 0)
for _, pod := range pods {
podBase := querytypes.PodBase{}
podBase.Metadata.Name = pod.Name
podBase.Metadata.Namespace = pod.Namespace
podList = append(podList, podBase)
}
if len(pods) == 0 {
return errors.New("no pod found")
}
var selectPod *querytypes.PodBase
if len(pods) > 1 {
selectPod, err = AskToChooseOnePod(podList)
if err != nil {
return err
}
} else {
selectPod = &podList[0]
}
return l.printPodLogs(ctx, ioStreams, selectPod, filters)
}
if selectPod == nil {
return nil
}
if selectPod.Cluster != "" {
ctx = multicluster.ContextWithClusterName(ctx, selectPod.Cluster)
}
func (l *Args) printPodLogs(ctx context.Context, ioStreams util.IOStreams, selectPod *querytypes.PodBase, filters []string) error {
pod, err := regexp.Compile(selectPod.Metadata.Name + ".*")
if err != nil {
return fmt.Errorf("fail to compile '%s' for logs query", selectPod.Metadata.Name+".*")
@@ -171,7 +244,16 @@ func (l *Args) Run(ctx context.Context, ioStreams util.IOStreams) error {
for {
select {
case str := <-logC:
ioStreams.Infonln(str)
show := true
for _, filter := range filters {
if !strings.Contains(str, filter) {
show = false
break
}
}
if show {
ioStreams.Infonln(str)
}
case <-ctx.Done():
return
}
@@ -245,3 +327,59 @@ func (l *Args) Run(ctx context.Context, ioStreams util.IOStreams) error {
return nil
}
// Run refer to the implementation at https://github.com/oam-dev/stern/blob/master/stern/main.go
func (l *Args) Run(ctx context.Context, ioStreams util.IOStreams) error {
pods, err := GetApplicationPods(ctx, l.App.Name, l.App.Namespace, l.Args, Filter{
Component: l.ComponentName,
Cluster: l.ClusterName,
})
if err != nil {
return err
}
var selectPod *querytypes.PodBase
if l.PodName != "" {
for i, pod := range pods {
if pod.Metadata.Name == l.PodName {
selectPod = &pods[i]
break
}
}
if selectPod == nil {
fmt.Println("The Pod you specified does not exist, please select it from the list.")
}
}
if selectPod == nil {
selectPod, err = AskToChooseOnePod(pods)
if err != nil {
return err
}
}
if selectPod == nil {
return nil
}
if selectPod.Cluster != "" {
ctx = multicluster.ContextWithClusterName(ctx, selectPod.Cluster)
}
return l.printPodLogs(ctx, ioStreams, selectPod, nil)
}
func getContextFromInstance(ctx context.Context, cli client.Client, namespace, name string) (string, map[string]string, error) {
app := &v1beta1.Application{}
if err := cli.Get(ctx, client.ObjectKey{Namespace: namespace, Name: name}, app); err == nil {
if app.Status.Workflow != nil && app.Status.Workflow.ContextBackend != nil {
return app.Status.Workflow.ContextBackend.Name, map[string]string{"app.kubernetes.io/name": "vela-core"}, nil
}
return "", nil, fmt.Errorf("no context found in application %s", name)
}
wr := &workflowv1alpha1.WorkflowRun{}
if err := cli.Get(ctx, client.ObjectKey{Namespace: namespace, Name: name}, wr); err == nil {
if wr.Status.ContextBackend != nil {
return wr.Status.ContextBackend.Name, map[string]string{"app.kubernetes.io/name": "vela-workflow"}, nil
}
return "", nil, fmt.Errorf("no context found in workflowrun %s", name)
}
return "", nil, fmt.Errorf("no context found in application %s", name)
}