Backport #2683 : Feat: output log with structured tag & add step duration metrics (#2696)

* debug task

(cherry picked from commit 93378eda67)

* metrics

(cherry picked from commit 7366804014)

* trace context

(cherry picked from commit f32105f23b)

* add step_duration metrics

(cherry picked from commit f9fc065e71)

* add readme docs

(cherry picked from commit 69146b468d)

cmd/core/main.go

@@ -44,6 +44,7 @@ import (
 	oamv1alpha2 "github.com/oam-dev/kubevela/pkg/controller/core.oam.dev/v1alpha2"
 	"github.com/oam-dev/kubevela/pkg/controller/utils"
 	"github.com/oam-dev/kubevela/pkg/cue/packages"
+	_ "github.com/oam-dev/kubevela/pkg/monitor/metrics"
 	"github.com/oam-dev/kubevela/pkg/multicluster"
 	"github.com/oam-dev/kubevela/pkg/oam"
 	"github.com/oam-dev/kubevela/pkg/oam/discoverymapper"

go.mod

@@ -43,6 +43,7 @@ require (
 	github.com/opencontainers/runc v1.0.0-rc95 // indirect
 	github.com/openkruise/kruise-api v0.9.0
 	github.com/pkg/errors v0.9.1
+	github.com/prometheus/client_golang v1.11.0
 	github.com/sirupsen/logrus v1.8.1
 	github.com/spf13/cobra v1.2.1
 	github.com/spf13/pflag v1.0.5

pkg/controller/core.oam.dev/v1alpha2/application/application_controller.go

@@ -41,11 +41,11 @@ import (
 	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
 	velatypes "github.com/oam-dev/kubevela/apis/types"
 	"github.com/oam-dev/kubevela/pkg/appfile"
-	common2 "github.com/oam-dev/kubevela/pkg/controller/common"
 	core "github.com/oam-dev/kubevela/pkg/controller/core.oam.dev"
 	"github.com/oam-dev/kubevela/pkg/controller/core.oam.dev/v1alpha1/envbinding"
 	"github.com/oam-dev/kubevela/pkg/controller/core.oam.dev/v1alpha2/application/assemble"
 	"github.com/oam-dev/kubevela/pkg/cue/packages"
+	monitorContext "github.com/oam-dev/kubevela/pkg/monitor/context"
 	"github.com/oam-dev/kubevela/pkg/oam"
 	"github.com/oam-dev/kubevela/pkg/oam/discoverymapper"
 	oamutil "github.com/oam-dev/kubevela/pkg/oam/util"
@@ -88,19 +88,24 @@ type Reconciler struct {
 // Reconcile process app event
 // nolint:gocyclo
 func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	ctx, cancel := common2.NewReconcileContext(ctx)
+	ctx, cancel := context.WithTimeout(ctx, time.Minute)
 	defer cancel()
 
-	klog.InfoS("Reconcile application", "application", klog.KRef(req.Namespace, req.Name))
-
+	logCtx := monitorContext.NewTraceContext(ctx, "").AddTag("application", req.String(), "controller", "application")
+	logCtx.Info("Reconcile application")
+	defer logCtx.Commit("Reconcile application")
 	app := new(v1beta1.Application)
 	if err := r.Get(ctx, client.ObjectKey{
 		Name:      req.Name,
 		Namespace: req.Namespace,
 	}, app); err != nil {
+		logCtx.Error(err, "get application")
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
+
+	logCtx.AddTag("resource_version", app.ResourceVersion)
 	ctx = oamutil.SetNamespaceInCtx(ctx, app.Namespace)
+	logCtx.SetContext(ctx)
 	if len(app.GetAnnotations()[oam.AnnotationKubeVelaVersion]) == 0 {
 		oamutil.AddAnnotations(app, map[string]string{
 			oam.AnnotationKubeVelaVersion: version.VelaVersion,
@@ -112,76 +117,76 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 		app:    app,
 		parser: appParser,
 	}
-	endReconcile, err := r.handleFinalizers(ctx, app)
+	endReconcile, err := r.handleFinalizers(logCtx, app)
 	if err != nil {
-		return r.endWithNegativeCondition(ctx, app, condition.ReconcileError(err), common.ApplicationStarting)
+		return r.endWithNegativeCondition(logCtx, app, condition.ReconcileError(err), common.ApplicationStarting)
 	}
 	if endReconcile {
 		return ctrl.Result{}, nil
 	}
 
-	appFile, err := appParser.GenerateAppFile(ctx, app)
+	appFile, err := appParser.GenerateAppFile(logCtx, app)
 	if err != nil {
-		klog.ErrorS(err, "Failed to parse application", "application", klog.KObj(app))
+		logCtx.Error(err, "Failed to parse application")
 		r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedParse, err))
-		return r.endWithNegativeCondition(ctx, app, condition.ErrorCondition("Parsed", err), common.ApplicationRendering)
+		return r.endWithNegativeCondition(logCtx, app, condition.ErrorCondition("Parsed", err), common.ApplicationRendering)
 	}
 	app.Status.SetConditions(condition.ReadyCondition("Parsed"))
 	r.Recorder.Event(app, event.Normal(velatypes.ReasonParsed, velatypes.MessageParsed))
 
-	if err := handler.PrepareCurrentAppRevision(ctx, appFile); err != nil {
-		klog.ErrorS(err, "Failed to prepare app revision", "application", klog.KObj(app))
+	if err := handler.PrepareCurrentAppRevision(logCtx, appFile); err != nil {
+		logCtx.Error(err, "Failed to prepare app revision")
 		r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedRevision, err))
-		return r.endWithNegativeCondition(ctx, app, condition.ErrorCondition("Revision", err), common.ApplicationRendering)
+		return r.endWithNegativeCondition(logCtx, app, condition.ErrorCondition("Revision", err), common.ApplicationRendering)
 	}
-	if err := handler.FinalizeAndApplyAppRevision(ctx); err != nil {
-		klog.ErrorS(err, "Failed to apply app revision", "application", klog.KObj(app))
+	if err := handler.FinalizeAndApplyAppRevision(logCtx); err != nil {
+		logCtx.Error(err, "Failed to apply app revision")
 		r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedRevision, err))
-		return r.endWithNegativeCondition(ctx, app, condition.ErrorCondition("Revision", err), common.ApplicationRendering)
+		return r.endWithNegativeCondition(logCtx, app, condition.ErrorCondition("Revision", err), common.ApplicationRendering)
 	}
-	klog.InfoS("Successfully prepare current app revision", "revisionName", handler.currentAppRev.Name,
+	logCtx.Info("Successfully prepare current app revision", "revisionName", handler.currentAppRev.Name,
 		"revisionHash", handler.currentRevHash, "isNewRevision", handler.isNewRevision)
 	app.Status.SetConditions(condition.ReadyCondition("Revision"))
 	r.Recorder.Event(app, event.Normal(velatypes.ReasonRevisoned, velatypes.MessageRevisioned))
 
-	if err := handler.UpdateAppLatestRevisionStatus(ctx); err != nil {
-		klog.ErrorS(err, "Failed to update application status", "application", klog.KObj(app))
-		return r.endWithNegativeCondition(ctx, app, condition.ReconcileError(err), common.ApplicationRendering)
+	if err := handler.UpdateAppLatestRevisionStatus(logCtx); err != nil {
+		logCtx.Error(err, "Failed to update application status")
+		return r.endWithNegativeCondition(logCtx, app, condition.ReconcileError(err), common.ApplicationRendering)
 	}
-	klog.InfoS("Successfully apply application revision", "application", klog.KObj(app))
+	logCtx.Info("Successfully apply application revision")
 
 	policies, err := appFile.PrepareWorkflowAndPolicy()
 	if err != nil {
-		klog.Error(err, "[Handle PrepareWorkflowAndPolicy]")
+		logCtx.Error(err, "[Handle PrepareWorkflowAndPolicy]")
 		r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedRender, err))
-		return r.endWithNegativeCondition(ctx, app, condition.ErrorCondition("PrepareWorkflowAndPolicy", err), common.ApplicationPolicyGenerating)
+		return r.endWithNegativeCondition(logCtx, app, condition.ErrorCondition("PrepareWorkflowAndPolicy", err), common.ApplicationPolicyGenerating)
 	}
 
 	if len(policies) > 0 {
 		if err := handler.Dispatch(ctx, "", common.PolicyResourceCreator, policies...); err != nil {
-			klog.Error(err, "[Handle ApplyPolicyResources]")
+			logCtx.Error(err, "[Handle ApplyPolicyResources]")
 			r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedApply, err))
-			return r.endWithNegativeCondition(ctx, app, condition.ErrorCondition("ApplyPolices", err), common.ApplicationPolicyGenerating)
+			return r.endWithNegativeCondition(logCtx, app, condition.ErrorCondition("ApplyPolices", err), common.ApplicationPolicyGenerating)
 		}
-		klog.InfoS("Successfully generated application policies", "application", klog.KObj(app))
+		logCtx.Info("Successfully generated application policies")
 	}
 
 	app.Status.SetConditions(condition.ReadyCondition("Render"))
 	r.Recorder.Event(app, event.Normal(velatypes.ReasonRendered, velatypes.MessageRendered))
 
 	if !appWillRollout(app) {
-		steps, err := handler.GenerateApplicationSteps(ctx, app, appParser, appFile, handler.currentAppRev, r.Client, r.dm, r.pd)
+		steps, err := handler.GenerateApplicationSteps(logCtx, app, appParser, appFile, handler.currentAppRev, r.Client, r.dm, r.pd)
 		if err != nil {
-			klog.Error(err, "[handle workflow]")
+			logCtx.Error(err, "[handle workflow]")
 			r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedWorkflow, err))
-			return r.endWithNegativeCondition(ctx, app, condition.ErrorCondition("Workflow", err), common.ApplicationRunningWorkflow)
+			return r.endWithNegativeCondition(logCtx, app, condition.ErrorCondition("Workflow", err), common.ApplicationRunningWorkflow)
 		}
 
-		workflowState, err := workflow.NewWorkflow(app, r.Client, appFile.WorkflowMode).ExecuteSteps(ctx, handler.currentAppRev, steps)
+		workflowState, err := workflow.NewWorkflow(app, r.Client, appFile.WorkflowMode).ExecuteSteps(logCtx.Fork("workflow"), handler.currentAppRev, steps)
 		if err != nil {
-			klog.Error(err, "[handle workflow]")
+			logCtx.Error(err, "[handle workflow]")
 			r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedWorkflow, err))
-			return r.endWithNegativeCondition(ctx, app, condition.ErrorCondition("Workflow", err), common.ApplicationRunningWorkflow)
+			return r.endWithNegativeCondition(logCtx, app, condition.ErrorCondition("Workflow", err), common.ApplicationRunningWorkflow)
 		}
 
 		handler.addServiceStatus(false, app.Status.Services...)
@@ -189,9 +194,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 		app.Status.AppliedResources = handler.appliedResources
 		switch workflowState {
 		case common.WorkflowStateSuspended:
-			return ctrl.Result{}, r.patchStatusWithRetryOnConflict(ctx, app, common.ApplicationWorkflowSuspending)
+			logCtx.Info("Workflow return state=Suspend")
+			return ctrl.Result{}, r.patchStatusWithRetryOnConflict(logCtx, app, common.ApplicationWorkflowSuspending)
 		case common.WorkflowStateTerminated:
-			return ctrl.Result{}, r.patchStatusWithRetryOnConflict(ctx, app, common.ApplicationWorkflowTerminated)
+			return ctrl.Result{}, r.patchStatusWithRetryOnConflict(logCtx, app, common.ApplicationWorkflowTerminated)
 		case common.WorkflowStateExecuting:
 			return reconcile.Result{RequeueAfter: baseWorkflowBackoffWaitTime}, r.patchStatusWithRetryOnConflict(ctx, app, common.ApplicationRunningWorkflow)
 		case common.WorkflowStateFinished:
@@ -205,10 +211,9 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 					})
 				}
 				if err != nil {
-					klog.ErrorS(err, "Failed to gc after workflow",
-						"application", klog.KObj(app))
+					logCtx.Error(err, "Failed to gc after workflow")
 					r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedGC, err))
-					return r.endWithNegativeConditionWithRetry(ctx, app, condition.ErrorCondition("GCAfterWorkflow", err), common.ApplicationRunningWorkflow)
+					return r.endWithNegativeConditionWithRetry(logCtx, app, condition.ErrorCondition("GCAfterWorkflow", err), common.ApplicationRunningWorkflow)
 				}
 				app.Status.ResourceTracker = ref
 			}
@@ -220,33 +225,33 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 		var comps []*velatypes.ComponentManifest
 		comps, err = appFile.GenerateComponentManifests()
 		if err != nil {
-			klog.ErrorS(err, "Failed to render components", "application", klog.KObj(app))
+			logCtx.Error(err, "Failed to render components")
 			r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedRender, err))
-			return r.endWithNegativeConditionWithRetry(ctx, app, condition.ErrorCondition("Render", err), common.ApplicationRendering)
+			return r.endWithNegativeConditionWithRetry(logCtx, app, condition.ErrorCondition("Render", err), common.ApplicationRendering)
 		}
 
 		assemble.HandleCheckManageWorkloadTrait(*handler.currentAppRev, comps)
 
-		if err := handler.HandleComponentsRevision(ctx, comps); err != nil {
-			klog.ErrorS(err, "Failed to handle compoents revision", "application", klog.KObj(app))
+		if err := handler.HandleComponentsRevision(logCtx, comps); err != nil {
+			logCtx.Error(err, "Failed to handle components revision")
 			r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedRevision, err))
-			return r.endWithNegativeConditionWithRetry(ctx, app, condition.ErrorCondition("Render", err), common.ApplicationRendering)
+			return r.endWithNegativeConditionWithRetry(logCtx, app, condition.ErrorCondition("Render", err), common.ApplicationRendering)
 		}
 		klog.Info("Application manifests has prepared and ready for appRollout to handle", "application", klog.KObj(app))
 	}
 	// if inplace is false and rolloutPlan is nil, it means the user will use an outer AppRollout object to rollout the application
 	if handler.app.Spec.RolloutPlan != nil {
-		res, err := handler.handleRollout(ctx)
+		res, err := handler.handleRollout(logCtx)
 		if err != nil {
-			klog.ErrorS(err, "Failed to handle rollout", "application", klog.KObj(app))
+			logCtx.Error(err, "Failed to handle rollout")
 			r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedRollout, err))
-			return r.endWithNegativeCondition(ctx, app, condition.ErrorCondition("Rollout", err), common.ApplicationRollingOut)
+			return r.endWithNegativeCondition(logCtx, app, condition.ErrorCondition("Rollout", err), common.ApplicationRollingOut)
 		}
 		// skip health check and garbage collection if rollout have not finished
 		// start next reconcile immediately
 		if res.Requeue || res.RequeueAfter > 0 {
-			if err := r.patchStatus(ctx, app, common.ApplicationRollingOut); err != nil {
-				return r.endWithNegativeCondition(ctx, app, condition.ReconcileError(err), common.ApplicationRollingOut)
+			if err := r.patchStatus(logCtx, app, common.ApplicationRollingOut); err != nil {
+				return r.endWithNegativeCondition(logCtx, app, condition.ReconcileError(err), common.ApplicationRollingOut)
 			}
 			return res, nil
 		}
@@ -254,7 +259,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 		// there is no need reconcile immediately, that means the rollout operation have finished
 		r.Recorder.Event(app, event.Normal(velatypes.ReasonRollout, velatypes.MessageRollout))
 		app.Status.SetConditions(condition.ReadyCondition("Rollout"))
-		klog.InfoS("Finished rollout ", "application", klog.KObj(app))
+		logCtx.Info("Finished rollout ")
 	}
 	var phase = common.ApplicationRunning
 	if !hasHealthCheckPolicy(appFile.Policies) {
@@ -265,11 +270,11 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 	}
 
 	if err := garbageCollection(ctx, handler); err != nil {
-		klog.ErrorS(err, "Failed to run garbage collection")
+		logCtx.Error(err, "Failed to run garbage collection")
 		r.Recorder.Event(app, event.Warning(velatypes.ReasonFailedGC, err))
-		return r.endWithNegativeCondition(ctx, app, condition.ReconcileError(err), phase)
+		return r.endWithNegativeCondition(logCtx, app, condition.ReconcileError(err), phase)
 	}
-	klog.Info("Successfully garbage collect", "application", klog.KObj(app))
+	logCtx.Info("Successfully garbage collect")
 	app.Status.SetConditions(condition.Condition{
 		Type:               condition.TypeReady,
 		Status:             corev1.ConditionTrue,
@@ -277,17 +282,17 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 		Reason:             condition.ReasonReconcileSuccess,
 	})
 	r.Recorder.Event(app, event.Normal(velatypes.ReasonDeployed, velatypes.MessageDeployed))
-	return ctrl.Result{}, r.patchStatus(ctx, app, phase)
+	return ctrl.Result{}, r.patchStatus(logCtx, app, phase)
 }
 
 // NOTE Because resource tracker is cluster-scoped resources, we cannot garbage collect them
 // by setting application(namespace-scoped) as their owners.
 // We must delete all resource trackers related to an application through finalizer logic.
-func (r *Reconciler) handleFinalizers(ctx context.Context, app *v1beta1.Application) (bool, error) {
+func (r *Reconciler) handleFinalizers(ctx monitorContext.Context, app *v1beta1.Application) (bool, error) {
 	if app.ObjectMeta.DeletionTimestamp.IsZero() {
 		if !meta.FinalizerExists(app, resourceTrackerFinalizer) {
 			meta.AddFinalizer(app, resourceTrackerFinalizer)
-			klog.InfoS("Register new finalizer for application", "application", klog.KObj(app), "finalizer", resourceTrackerFinalizer)
+			ctx.Info("Register new finalizer for application", "finalizer", resourceTrackerFinalizer)
 			return true, errors.Wrap(r.Client.Update(ctx, app), errUpdateApplicationFinalizer)
 		}
 	} else {
@@ -297,7 +302,7 @@ func (r *Reconciler) handleFinalizers(ctx context.Context, app *v1beta1.Applicat
 			rt := &v1beta1.ResourceTracker{}
 			rt.SetName(fmt.Sprintf("%s-%s", app.Namespace, app.Name))
 			if err := r.Client.Delete(ctx, rt); err != nil && !kerrors.IsNotFound(err) {
-				klog.ErrorS(err, "Failed to delete legacy resource tracker", "name", rt.Name)
+				ctx.Error(err, "Failed to delete legacy resource tracker", "name", rt.Name)
 				return true, errors.WithMessage(err, "cannot remove finalizer")
 			}
 			meta.RemoveFinalizer(app, legacyResourceTrackerFinalizer)
@@ -311,12 +316,12 @@ func (r *Reconciler) handleFinalizers(ctx context.Context, app *v1beta1.Applicat
 				}}
 			rtList := &v1beta1.ResourceTrackerList{}
 			if err := r.Client.List(ctx, rtList, listOpts...); err != nil {
-				klog.ErrorS(err, "Failed to list resource tracker of app", "name", app.Name)
+				ctx.Error(err, "Failed to list resource tracker of app", "name", app.Name)
 				return true, errors.WithMessage(err, "cannot remove finalizer")
 			}
 			for _, rt := range rtList.Items {
 				if err := r.Client.Delete(ctx, rt.DeepCopy()); err != nil && !kerrors.IsNotFound(err) {
-					klog.ErrorS(err, "Failed to delete resource tracker", "name", rt.Name)
+					ctx.Error(err, "Failed to delete resource tracker", "name", rt.Name)
 					return true, errors.WithMessage(err, "cannot remove finalizer")
 				}
 			}
@@ -364,15 +369,14 @@ func (r *Reconciler) patchStatusWithRetryOnConflict(ctx context.Context, app *v1
 	return retry.RetryOnConflict(retry.DefaultRetry, func() error {
 		status := app.Status.DeepCopy()
 		if err := r.Client.Get(ctx, client.ObjectKeyFromObject(app), app); err != nil {
-			klog.ErrorS(err, "failed to get application while patching status", "application", klog.KObj(app))
-			return err
+			return errors.WithMessage(err, "failed to get application while patching status")
 		}
 		app.Status = *status
 		err := r.Client.Status().Patch(ctx, app, client.Merge)
 		if err != nil {
-			klog.ErrorS(err, "failed to re-patch status", "application", klog.KObj(app))
+			return errors.WithMessage(err, "failed to re-patch status")
 		}
-		return err
+		return nil
 	})
 }
 

pkg/monitor/README.md

@@ -0,0 +1,46 @@
+# Package Usage
+
+## Context
+First, this context is compatible with built-in context interface.
+Also it supports fork and commit like trace span.
+
+### Fork
+`Fork` will generate a sub context that inherit the parent's tags. When new tags are added to the `sub-context`, the `parent-context` will not be affected.
+
+### Commit
+`Commit` will log the context duration, and export metrics or other execution information.
+
+### usage
+```
+tracerCtx:=context.NewTraceContext(stdCtx,"$id") 
+defer tracerCtx.Commit("success")
+
+// Execute sub-code logic
+subCtx:=tracerCtx.Fork("sub-id")
+...
+subCtx.Commit("step is executed")
+
+```
+
+## Metrics
+First, you need register `metricVec` in package `pkg/monitor/metrics`, like below:
+```
+StepDurationSummary = prometheus.NewSummaryVec(prometheus.SummaryOpts{
+		Name:        "step_duration_ms",
+		Help:        "step latency distributions.",
+		Objectives:  map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
+		ConstLabels: prometheus.Labels{},
+	}, []string{"application", "workflow_revision", "step_name", "step_type"})
+```
+
+Now, you can export metrics by context,for example
+```
+subCtx:=tracerCtx.Fork("sub-id",DurationMetric(func(v float64) {
+					metrics.StepDurationSummary.WithLabelValues(e.app.Name, e.status.AppRevision, stepStatus.Name, stepStatus.Type).Observe(v)
+				})
+subCtx.Commit("export") // At this time, it will export the StepDurationSummary metrics. 			
+
+```
+
+Context only support `DurationMetric` exporter. you can submit pr to support more exporters.
+If metrics have nothing to do with context, there is no need to extend it through context exporter

pkg/monitor/context/context.go

@@ -0,0 +1,170 @@
+/*
+ Copyright 2021. The KubeVela 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 context
+
+import (
+	stdctx "context"
+	"fmt"
+	"time"
+
+	"github.com/oam-dev/kubevela/pkg/utils"
+
+	"k8s.io/klog/v2"
+)
+
+const (
+	// spanTagID is the tag name of span ID.
+	spanTagID = "spanID"
+)
+
+// Context keep the trace info
+type Context interface {
+	stdctx.Context
+	Logger
+	GetContext() stdctx.Context
+	SetContext(ctx stdctx.Context)
+	AddTag(keysAndValues ...interface{}) Context
+	Fork(name string, exporters ...Exporter) Context
+	Commit(msg string)
+}
+
+// Logger represents the ability to log messages, both errors and not.
+type Logger interface {
+	InfoDepth(depth int, msg string, keysAndValues ...interface{})
+	Info(msg string, keysAndValues ...interface{})
+	Error(err error, msg string, keysAndValues ...interface{})
+	ErrorDepth(depth int, err error, msg string, keysAndValues ...interface{})
+	Printf(format string, args ...interface{})
+	V(level int)
+}
+
+type traceContext struct {
+	stdctx.Context
+
+	id             string
+	beginTimestamp time.Time
+	logLevel       int
+
+	tags      []interface{}
+	exporters []Exporter
+	parent    *traceContext
+}
+
+// Fork a child Context extends parent Context
+func (t *traceContext) Fork(id string, exporters ...Exporter) Context {
+	if id == "" {
+		id = t.id
+	} else {
+		id = t.id + "." + id
+	}
+
+	return &traceContext{
+		Context:        t.Context,
+		id:             id,
+		tags:           copySlice(t.tags),
+		logLevel:       t.logLevel,
+		parent:         t,
+		beginTimestamp: time.Now(),
+		exporters:      exporters,
+	}
+}
+
+// Commit finish the span record
+func (t *traceContext) Commit(msg string) {
+	msg = fmt.Sprintf("[Finished]: %s(%s)", t.id, msg)
+	duration := time.Since(t.beginTimestamp)
+	for _, export := range t.exporters {
+		export(t, duration.Microseconds())
+	}
+	klog.InfoSDepth(1, msg, t.getTagsWith("duration", duration.String())...)
+}
+
+func (t *traceContext) getTagsWith(keysAndValues ...interface{}) []interface{} {
+	tags := append(t.tags, keysAndValues...)
+	return append(tags, spanTagID, t.id)
+}
+
+// Info logs a non-error message with the given key/value pairs as context.
+func (t *traceContext) Info(msg string, keysAndValues ...interface{}) {
+	klog.InfoSDepth(1, msg, t.getTagsWith(keysAndValues...)...)
+}
+
+// GetContext get raw context.
+func (t *traceContext) GetContext() stdctx.Context {
+	return t.Context
+}
+
+// SetContext set raw context.
+func (t *traceContext) SetContext(ctx stdctx.Context) {
+	t.Context = ctx
+}
+
+// InfoDepth acts as Info but uses depth to determine which call frame to log.
+func (t *traceContext) InfoDepth(depth int, msg string, keysAndValues ...interface{}) {
+	klog.InfoSDepth(depth+1, msg, t.getTagsWith(keysAndValues...)...)
+}
+
+// Error logs an error, with the given message and key/value pairs as context.
+func (t *traceContext) Error(err error, msg string, keysAndValues ...interface{}) {
+	klog.ErrorSDepth(1, err, msg, t.getTagsWith(keysAndValues...)...)
+}
+
+// ErrorDepth acts as Error but uses depth to determine which call frame to log.
+func (t *traceContext) ErrorDepth(depth int, err error, msg string, keysAndValues ...interface{}) {
+	klog.ErrorSDepth(depth+1, err, msg, t.getTagsWith(keysAndValues...)...)
+}
+
+// Printf formats according to a format specifier and logs.
+func (t *traceContext) Printf(format string, args ...interface{}) {
+	klog.InfoSDepth(1, fmt.Sprintf(format, args...), t.getTagsWith()...)
+}
+
+// V reports whether verbosity at the call site is at least the requested level.
+func (t *traceContext) V(level int) {
+	t.logLevel = level
+}
+
+// AddTag adds some key-value pairs of context to a logger.
+func (t *traceContext) AddTag(keysAndValues ...interface{}) Context {
+	t.tags = append(t.tags, keysAndValues...)
+	return t
+}
+
+// NewTraceContext new a TraceContext
+func NewTraceContext(ctx stdctx.Context, id string) Context {
+	if id == "" {
+		id = "i-" + utils.RandomString(8)
+	}
+	return &traceContext{
+		Context:        ctx,
+		id:             id,
+		beginTimestamp: time.Now(),
+	}
+}
+
+func copySlice(in []interface{}) []interface{} {
+	out := make([]interface{}, len(in))
+	copy(out, in)
+	return out
+}
+
+// Exporter export context info.
+type Exporter func(t *traceContext, duration int64)
+
+// DurationMetric export context duration metric.
+func DurationMetric(h func(v float64)) Exporter {
+	return func(t *traceContext, duration int64) {
+		h(float64(duration / 1000))
+	}
+}

pkg/monitor/context/context_test.go

@@ -0,0 +1,45 @@
+/*
+ Copyright 2021. The KubeVela 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 context
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/pkg/errors"
+	"k8s.io/apimachinery/pkg/types"
+)
+
+func TestLog(t *testing.T) {
+	ctx := NewTraceContext(context.Background(), types.NamespacedName{
+		Namespace: "default",
+		Name:      "test-app",
+	}.String())
+
+	ctx.AddTag("controller", "application")
+	ctx.Info("init")
+	ctx.InfoDepth(1, "init")
+	defer ctx.Commit("close")
+	spanCtx := ctx.Fork("child1", DurationMetric(func(v float64) {
+		fmt.Println(v)
+	}))
+	time.Sleep(time.Millisecond * 30)
+	err := errors.New("mock error")
+	ctx.Error(err, "test case", "generated", "test_log")
+	ctx.ErrorDepth(1, err, "test case", "generated", "test_log")
+	spanCtx.Commit("finished")
+
+}

pkg/monitor/metrics/workflow.go

@@ -0,0 +1,36 @@
+/*
+ Copyright 2021. The KubeVela 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 metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"k8s.io/klog/v2"
+	"sigs.k8s.io/controller-runtime/pkg/metrics"
+)
+
+var (
+	// StepDurationSummary report the step execution duration summary.
+	StepDurationSummary = prometheus.NewSummaryVec(prometheus.SummaryOpts{
+		Name:        "step_duration_ms",
+		Help:        "step latency distributions.",
+		Objectives:  map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
+		ConstLabels: prometheus.Labels{},
+	}, []string{"application", "workflow_revision", "step_name", "step_type"})
+)
+
+func init() {
+	if err := metrics.Registry.Register(StepDurationSummary); err != nil {
+		klog.Error(err)
+	}
+}

pkg/workflow/interface.go

@@ -16,10 +16,9 @@ limitations under the License.
 package workflow
 
 import (
-	"context"
-
 	"github.com/oam-dev/kubevela/apis/core.oam.dev/common"
 	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
+	"github.com/oam-dev/kubevela/pkg/monitor/context"
 	"github.com/oam-dev/kubevela/pkg/workflow/types"
 )
 

pkg/workflow/tasks/custom/task.go

@@ -22,6 +22,8 @@ import (
 	"fmt"
 	"strings"
 
+	monitorContext "github.com/oam-dev/kubevela/pkg/monitor/context"
+
 	"github.com/oam-dev/kubevela/pkg/workflow/hooks"
 
 	"cuelang.org/go/cue"
@@ -148,16 +150,28 @@ func (t *TaskLoader) makeTaskGenerator(templ string) (wfTypes.TaskGenerator, err
 			return false
 		}
 		tRunner.run = func(ctx wfContext.Context, options *wfTypes.TaskRunOptions) (common.WorkflowStepStatus, *wfTypes.Operation, error) {
+			if options.GetTracer == nil {
+				options.GetTracer = func(id string, step v1beta1.WorkflowStep) monitorContext.Context {
+					return monitorContext.NewTraceContext(context.Background(), "")
+				}
+			}
+			tracer := options.GetTracer(exec.wfStatus.ID, wfStep).AddTag("step_name", wfStep.Name, "step_type", wfStep.Type)
+			defer func() {
+				tracer.Commit(string(exec.status().Phase))
+			}()
+
 			if t.runOptionsProcess != nil {
 				t.runOptionsProcess(options)
 			}
 			paramsValue, err := ctx.MakeParameter(params)
 			if err != nil {
+				tracer.Error(err, "make parameter")
 				return common.WorkflowStepStatus{}, nil, errors.WithMessage(err, "make parameter")
 			}
 
 			for _, hook := range options.PreStartHooks {
 				if err := hook(ctx, paramsValue, wfStep); err != nil {
+					tracer.Error(err, "do preStartHook")
 					return common.WorkflowStepStatus{}, nil, errors.WithMessage(err, "do preStartHook")
 				}
 			}
@@ -176,13 +190,19 @@ func (t *TaskLoader) makeTaskGenerator(templ string) (wfTypes.TaskGenerator, err
 				paramFile = fmt.Sprintf(model.ParameterFieldName+": {%s}\n", ps)
 			}
 
-			taskv, err := t.makeValue(ctx, strings.Join([]string{templ, paramFile}, "\n"), genOpt.ID)
+			taskv, err := t.makeValue(ctx, strings.Join([]string{templ, paramFile}, "\n"), exec.wfStatus.ID)
 			if err != nil {
 				exec.err(err, StatusReasonRendering)
 				return exec.status(), exec.operation(), nil
 			}
 
+			exec.tracer = tracer
+			if isDebugMode(taskv) {
+				exec.printStep("workflowStepStart", "workflow", "", taskv)
+				defer exec.printStep("workflowStepEnd", "workflow", "", taskv)
+			}
 			if err := exec.doSteps(ctx, taskv); err != nil {
+				tracer.Error(err, "do steps")
 				exec.err(err, StatusReasonExecute)
 				return exec.status(), exec.operation(), nil
 			}
@@ -221,6 +241,8 @@ type executor struct {
 	suspend    bool
 	terminated bool
 	wait       bool
+
+	tracer monitorContext.Context
 }
 
 // Suspend let workflow pause.
@@ -264,8 +286,17 @@ func (exec *executor) status() common.WorkflowStepStatus {
 	return exec.wfStatus
 }
 
+func (exec *executor) printStep(phase string, provider string, do string, v *value.Value) {
+	msg, _ := v.String()
+	exec.tracer.Info("cue eval: "+msg, "phase", phase, "provider", provider, "do", do)
+}
+
 // Handle process task-step value by provider and do.
 func (exec *executor) Handle(ctx wfContext.Context, provider string, do string, v *value.Value) error {
+	if isDebugMode(v) {
+		exec.printStep("stepStart", provider, do, v)
+		defer exec.printStep("stepEnd", provider, do, v)
+	}
 	h, exist := exec.handlers.GetHandler(provider, do)
 	if !exist {
 		return errors.Errorf("handler not found")
@@ -336,6 +367,11 @@ func isStepList(fieldName string) bool {
 	return strings.HasPrefix(fieldName, "#up_")
 }
 
+func isDebugMode(v *value.Value) bool {
+	debug, _ := v.CueValue().LookupDef("#debug").Bool()
+	return debug
+}
+
 func opTpy(v *value.Value) string {
 	return getLabel(v, "#do")
 }

pkg/workflow/types/types.go

@@ -22,6 +22,7 @@ import (
 	"github.com/oam-dev/kubevela/apis/core.oam.dev/common"
 	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
 	"github.com/oam-dev/kubevela/pkg/cue/model/value"
+	monitorCtx "github.com/oam-dev/kubevela/pkg/monitor/context"
 	wfContext "github.com/oam-dev/kubevela/pkg/workflow/context"
 )
 
@@ -42,6 +43,7 @@ type TaskRunOptions struct {
 	Data          *value.Value
 	PreStartHooks []TaskPreStartHook
 	PostStopHooks []TaskPostStopHook
+	GetTracer     func(id string, step v1beta1.WorkflowStep) monitorCtx.Context
 	RunSteps      func(isDag bool, runners ...TaskRunner) (*common.WorkflowStatus, error)
 }
 

pkg/workflow/workflow.go

@@ -17,7 +17,6 @@ limitations under the License.
 package workflow
 
 import (
-	"context"
 	"fmt"
 
 	"github.com/pkg/errors"
@@ -27,6 +26,8 @@ import (
 	oamcore "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
 	"github.com/oam-dev/kubevela/pkg/controller/utils"
 	"github.com/oam-dev/kubevela/pkg/cue/model/value"
+	monitorContext "github.com/oam-dev/kubevela/pkg/monitor/context"
+	"github.com/oam-dev/kubevela/pkg/monitor/metrics"
 	"github.com/oam-dev/kubevela/pkg/oam/util"
 	wfContext "github.com/oam-dev/kubevela/pkg/workflow/context"
 	wfTypes "github.com/oam-dev/kubevela/pkg/workflow/types"
@@ -52,16 +53,18 @@ func NewWorkflow(app *oamcore.Application, cli client.Client, mode common.Workfl
 }
 
 // ExecuteSteps process workflow step in order.
-func (w *workflow) ExecuteSteps(ctx context.Context, appRev *oamcore.ApplicationRevision, taskRunners []wfTypes.TaskRunner) (common.WorkflowState, error) {
+func (w *workflow) ExecuteSteps(ctx monitorContext.Context, appRev *oamcore.ApplicationRevision, taskRunners []wfTypes.TaskRunner) (common.WorkflowState, error) {
 	revAndSpecHash, err := computeAppRevisionHash(appRev.Name, w.app)
 	if err != nil {
 		return common.WorkflowStateExecuting, err
 	}
+	ctx.AddTag("workflow_version", revAndSpecHash)
 	if len(taskRunners) == 0 {
 		return common.WorkflowStateFinished, nil
 	}
 
 	if w.app.Status.Workflow == nil || w.app.Status.Workflow.AppRevision != revAndSpecHash {
+		ctx.Info("Restart Workflow")
 		w.app.Status.Workflow = &common.WorkflowStatus{
 			AppRevision: revAndSpecHash,
 			Mode:        common.WorkflowModeStep,
@@ -87,22 +90,22 @@ func (w *workflow) ExecuteSteps(ctx context.Context, appRev *oamcore.Application
 		return common.WorkflowStateFinished, nil
 	}
 
-	var (
-		wfCtx wfContext.Context
-	)
-
-	wfCtx, err = w.makeContext(w.app.Name)
+	wfCtx, err := w.makeContext(w.app.Name)
 	if err != nil {
+		ctx.Error(err, "make context")
 		return common.WorkflowStateExecuting, err
 	}
 
 	e := &engine{
-		status:  wfStatus,
-		dagMode: w.dagMode,
+		status:     wfStatus,
+		dagMode:    w.dagMode,
+		monitorCtx: ctx,
+		app:        w.app,
 	}
 
 	err = e.run(wfCtx, taskRunners)
 	if err != nil {
+		ctx.Error(err, "run steps")
 		return common.WorkflowStateExecuting, err
 	}
 	if wfStatus.Terminated {
@@ -242,7 +245,13 @@ func (e *engine) todoByIndex(taskRunners []wfTypes.TaskRunner) []wfTypes.TaskRun
 
 func (e *engine) steps(wfCtx wfContext.Context, taskRunners []wfTypes.TaskRunner) error {
 	for _, runner := range taskRunners {
-		status, operation, err := runner.Run(wfCtx, &wfTypes.TaskRunOptions{})
+		status, operation, err := runner.Run(wfCtx, &wfTypes.TaskRunOptions{
+			GetTracer: func(id string, stepStatus oamcore.WorkflowStep) monitorContext.Context {
+				return e.monitorCtx.Fork(id, monitorContext.DurationMetric(func(v float64) {
+					metrics.StepDurationSummary.WithLabelValues(e.app.Namespace+"/"+e.app.Name, e.status.AppRevision, stepStatus.Name, stepStatus.Type).Observe(v)
+				}))
+			},
+		})
 		if err != nil {
 			return err
 		}
@@ -269,8 +278,10 @@ func (e *engine) steps(wfCtx wfContext.Context, taskRunners []wfTypes.TaskRunner
 }
 
 type engine struct {
-	dagMode bool
-	status  *common.WorkflowStatus
+	dagMode    bool
+	status     *common.WorkflowStatus
+	monitorCtx monitorContext.Context
+	app        *oamcore.Application
 }
 
 func (e *engine) isDag() bool {

pkg/workflow/workflow_test.go

@@ -20,6 +20,8 @@ import (
 	"context"
 	"encoding/json"
 
+	monitorContext "github.com/oam-dev/kubevela/pkg/monitor/context"
+
 	"github.com/oam-dev/kubevela/pkg/cue/model/value"
 
 	. "github.com/onsi/ginkgo"
@@ -67,8 +69,9 @@ var _ = Describe("Test Workflow", func() {
 				Type: "success",
 			},
 		})
+		ctx := monitorContext.NewTraceContext(context.Background(), "test-app")
 		wf := NewWorkflow(app, k8sClient, common.WorkflowModeStep)
-		state, err := wf.ExecuteSteps(context.Background(), revision, runners)
+		state, err := wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateExecuting))
 		workflowStatus := app.Status.Workflow
@@ -105,7 +108,7 @@ var _ = Describe("Test Workflow", func() {
 
 		app.Status.Workflow = workflowStatus
 		wf = NewWorkflow(app, k8sClient, common.WorkflowModeStep)
-		state, err = wf.ExecuteSteps(context.Background(), revision, runners)
+		state, err = wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateFinished))
 		app.Status.Workflow.ContextBackend = nil
@@ -144,8 +147,9 @@ var _ = Describe("Test Workflow", func() {
 				Type: "success",
 			},
 		})
+		ctx := monitorContext.NewTraceContext(context.Background(), "test-app")
 		wf := NewWorkflow(app, k8sClient, common.WorkflowModeStep)
-		state, err := wf.ExecuteSteps(context.Background(), revision, runners)
+		state, err := wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateSuspended))
 		wfStatus := *app.Status.Workflow
@@ -166,7 +170,7 @@ var _ = Describe("Test Workflow", func() {
 		})).Should(BeEquivalentTo(""))
 
 		// check suspend...
-		state, err = wf.ExecuteSteps(context.Background(), revision, runners)
+		state, err = wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateSuspended))
 
@@ -174,7 +178,7 @@ var _ = Describe("Test Workflow", func() {
 		app.Status.Workflow.Suspend = false
 		// check app meta changed
 		app.Labels = map[string]string{"for-test": "changed"}
-		state, err = wf.ExecuteSteps(context.Background(), revision, runners)
+		state, err = wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateFinished))
 		app.Status.Workflow.ContextBackend = nil
@@ -196,7 +200,7 @@ var _ = Describe("Test Workflow", func() {
 			}},
 		})).Should(BeEquivalentTo(""))
 
-		state, err = wf.ExecuteSteps(context.Background(), revision, runners)
+		state, err = wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateFinished))
 	})
@@ -212,8 +216,9 @@ var _ = Describe("Test Workflow", func() {
 				Type: "terminate",
 			},
 		})
+		ctx := monitorContext.NewTraceContext(context.Background(), "test-app")
 		wf := NewWorkflow(app, k8sClient, common.WorkflowModeStep)
-		state, err := wf.ExecuteSteps(context.Background(), revision, runners)
+		state, err := wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateTerminated))
 		app.Status.Workflow.ContextBackend = nil
@@ -232,7 +237,7 @@ var _ = Describe("Test Workflow", func() {
 			}},
 		})).Should(BeEquivalentTo(""))
 
-		state, err = wf.ExecuteSteps(context.Background(), revision, runners)
+		state, err = wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateTerminated))
 	})
@@ -248,8 +253,9 @@ var _ = Describe("Test Workflow", func() {
 				Type: "error",
 			},
 		})
+		ctx := monitorContext.NewTraceContext(context.Background(), "test-app")
 		wf := NewWorkflow(app, k8sClient, common.WorkflowModeStep)
-		state, err := wf.ExecuteSteps(context.Background(), revision, runners)
+		state, err := wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).To(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateExecuting))
 		app.Status.Workflow.ContextBackend = nil
@@ -266,8 +272,9 @@ var _ = Describe("Test Workflow", func() {
 
 	It("skip workflow", func() {
 		app, runners := makeTestCase([]oamcore.WorkflowStep{})
+		ctx := monitorContext.NewTraceContext(context.Background(), "test-app")
 		wf := NewWorkflow(app, k8sClient, common.WorkflowModeStep)
-		state, err := wf.ExecuteSteps(context.Background(), revision, runners)
+		state, err := wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateFinished))
 	})
@@ -289,7 +296,8 @@ var _ = Describe("Test Workflow", func() {
 		})
 		pending = true
 		wf := NewWorkflow(app, k8sClient, common.WorkflowModeDAG)
-		state, err := wf.ExecuteSteps(context.Background(), revision, runners)
+		ctx := monitorContext.NewTraceContext(context.Background(), "test-app")
+		state, err := wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateExecuting))
 		app.Status.Workflow.ContextBackend = nil
@@ -307,12 +315,12 @@ var _ = Describe("Test Workflow", func() {
 			}},
 		})).Should(BeEquivalentTo(""))
 
-		state, err = wf.ExecuteSteps(context.Background(), revision, runners)
+		state, err = wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateExecuting))
 
 		pending = false
-		state, err = wf.ExecuteSteps(context.Background(), revision, runners)
+		state, err = wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateFinished))
 		app.Status.Workflow.ContextBackend = nil
@@ -346,8 +354,9 @@ var _ = Describe("Test Workflow", func() {
 				Type: "success",
 			},
 		})
+		ctx := monitorContext.NewTraceContext(context.Background(), "test-app")
 		wf := NewWorkflow(app, k8sClient, common.WorkflowModeStep)
-		state, err := wf.ExecuteSteps(context.Background(), revision, runners)
+		state, err := wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateExecuting))
 		Expect(app.Status.Workflow.Steps[0].Phase).Should(BeEquivalentTo(common.WorkflowStepPhaseRunning))