diff --git a/cmd/core/main.go b/cmd/core/main.go index ad4f05818..deb898698 100644 --- a/cmd/core/main.go +++ b/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" diff --git a/go.mod b/go.mod index 2cb1d5126..4ee163f02 100644 --- a/go.mod +++ b/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 diff --git a/pkg/controller/core.oam.dev/v1alpha2/application/application_controller.go b/pkg/controller/core.oam.dev/v1alpha2/application/application_controller.go index e5b5fa23f..9596ccb3b 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/application_controller.go +++ b/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 }) } diff --git a/pkg/monitor/README.md b/pkg/monitor/README.md new file mode 100644 index 000000000..a8db58985 --- /dev/null +++ b/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 diff --git a/pkg/monitor/context/context.go b/pkg/monitor/context/context.go new file mode 100644 index 000000000..afddfb191 --- /dev/null +++ b/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)) + } +} diff --git a/pkg/monitor/context/context_test.go b/pkg/monitor/context/context_test.go new file mode 100644 index 000000000..f5a599855 --- /dev/null +++ b/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") + +} diff --git a/pkg/monitor/metrics/workflow.go b/pkg/monitor/metrics/workflow.go new file mode 100644 index 000000000..cc56a042c --- /dev/null +++ b/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) + } +} diff --git a/pkg/workflow/interface.go b/pkg/workflow/interface.go index 234351b22..34febb1af 100644 --- a/pkg/workflow/interface.go +++ b/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" ) diff --git a/pkg/workflow/tasks/custom/task.go b/pkg/workflow/tasks/custom/task.go index bcbb207f5..57009fdd6 100644 --- a/pkg/workflow/tasks/custom/task.go +++ b/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") } diff --git a/pkg/workflow/types/types.go b/pkg/workflow/types/types.go index ce069ee64..7dc8e1984 100644 --- a/pkg/workflow/types/types.go +++ b/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) } diff --git a/pkg/workflow/workflow.go b/pkg/workflow/workflow.go index 3a8394a46..561db1013 100644 --- a/pkg/workflow/workflow.go +++ b/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 { diff --git a/pkg/workflow/workflow_test.go b/pkg/workflow/workflow_test.go index 351014920..9993ee912 100644 --- a/pkg/workflow/workflow_test.go +++ b/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))