From 354e92cfbdc7de7851f5aecd65e3e999a5f5f8a8 Mon Sep 17 00:00:00 2001 From: "Jian.Li" Date: Wed, 19 Jan 2022 16:07:19 +0800 Subject: [PATCH] Fix: workflow skip executing all steps occasionally (#3025) * fix asi Signed-off-by: Jian.Li * fix lint Signed-off-by: Jian.Li * add trace tag * add args for this feature Signed-off-by: Jian.Li * enable-asi-compatibility Signed-off-by: Jian.Li --- .../core.oam.dev/v1beta1/application_types.go | 36 ++++++++++++++ cmd/core/main.go | 1 + .../core.oam.dev/oamruntime_controller.go | 3 ++ .../application/application_controller.go | 48 ++++++++++++++----- .../v1alpha2/application/revision.go | 4 +- .../v1alpha2/application/suite_test.go | 13 ++--- 6 files changed, 84 insertions(+), 21 deletions(-) diff --git a/apis/core.oam.dev/v1beta1/application_types.go b/apis/core.oam.dev/v1beta1/application_types.go index b8a5ae050..a8c277f87 100644 --- a/apis/core.oam.dev/v1beta1/application_types.go +++ b/apis/core.oam.dev/v1beta1/application_types.go @@ -17,7 +17,10 @@ package v1beta1 import ( + "encoding/json" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime" "github.com/oam-dev/kubevela/apis/core.oam.dev/common" @@ -138,3 +141,36 @@ func (app *Application) GetComponent(workloadType string) *common.ApplicationCom } return nil } + +// Unstructured convert application to unstructured.Unstructured. +func (app *Application) Unstructured() (*unstructured.Unstructured, error) { + var obj = &unstructured.Unstructured{} + app.SetGroupVersionKind(ApplicationKindVersionKind) + bt, err := json.Marshal(app) + if err != nil { + return nil, err + } + if err := obj.UnmarshalJSON(bt); err != nil { + return nil, err + } + + if app.Status.Services == nil { + if err := unstructured.SetNestedSlice(obj.Object, []interface{}{}, "status", "services"); err != nil { + return nil, err + } + } + + if app.Status.AppliedResources == nil { + if err := unstructured.SetNestedSlice(obj.Object, []interface{}{}, "status", "appliedResources"); err != nil { + return nil, err + } + } + + if wfStatus := app.Status.Workflow; wfStatus != nil && wfStatus.Steps == nil { + if err := unstructured.SetNestedSlice(obj.Object, []interface{}{}, "status", "workflow", "steps"); err != nil { + return nil, err + } + } + + return obj, nil +} diff --git a/cmd/core/main.go b/cmd/core/main.go index 3730b1950..b5e977459 100644 --- a/cmd/core/main.go +++ b/cmd/core/main.go @@ -129,6 +129,7 @@ func main() { flag.DurationVar(&retryPeriod, "leader-election-retry-period", 2*time.Second, "The duration the LeaderElector clients should wait between tries of actions") flag.BoolVar(&enableClusterGateway, "enable-cluster-gateway", false, "Enable cluster-gateway to use multicluster, disabled by default.") + flag.BoolVar(&controllerArgs.EnableCompatibility, "enable-asi-compatibility", false, "enable compatibility for asi") standardcontroller.AddOptimizeFlags() flag.Parse() diff --git a/pkg/controller/core.oam.dev/oamruntime_controller.go b/pkg/controller/core.oam.dev/oamruntime_controller.go index cc6ee18e3..0d5da3feb 100644 --- a/pkg/controller/core.oam.dev/oamruntime_controller.go +++ b/pkg/controller/core.oam.dev/oamruntime_controller.go @@ -80,4 +80,7 @@ type Args struct { // OAMSpecVer is the oam spec version controller want to setup OAMSpecVer string + + // EnableCompatibility indicates that will change some functions of controller to adapt to multiple platforms, such as asi. + EnableCompatibility bool } 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 4795cf1fa..af1080a3d 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/application_controller.go +++ b/pkg/controller/core.oam.dev/v1alpha2/application/application_controller.go @@ -81,12 +81,17 @@ var ( // Reconciler reconciles an Application object type Reconciler struct { client.Client - dm discoverymapper.DiscoveryMapper - pd *packages.PackageDiscover - Scheme *runtime.Scheme - Recorder event.Recorder + dm discoverymapper.DiscoveryMapper + pd *packages.PackageDiscover + Scheme *runtime.Scheme + Recorder event.Recorder + options +} + +type options struct { appRevisionLimit int concurrentReconciles int + disableStatusUpdate bool } // +kubebuilder:rbac:groups=core.oam.dev,resources=applications,verbs=get;list;watch;create;update;patch;delete @@ -121,6 +126,8 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu if annotations := app.GetAnnotations(); annotations == nil || annotations[oam.AnnotationKubeVelaVersion] == "" { metav1.SetMetaDataAnnotation(&app.ObjectMeta, oam.AnnotationKubeVelaVersion, version.VelaVersion) } + logCtx.AddTag("publish_version", app.GetAnnotations()[oam.AnnotationKubeVelaVersion]) + appParser := appfile.NewApplicationParser(r.Client, r.dm, r.pd) handler, err := NewAppHandler(logCtx, r, app, appParser) if err != nil { @@ -286,7 +293,7 @@ func (r *Reconciler) gcResourceTrackers(logCtx monitorContext.Context, handler * return r.endWithNegativeCondition(logCtx, handler.app, condition.ReconcileError(err), phase) } if !finished { - logCtx.Info("GarbageCollecting resourcetrackers") + logCtx.Info("GarbageCollecting resourcetrackers unfinished") cond := condition.Deleting() if len(waiting) > 0 { cond.Message = fmt.Sprintf("Waiting for %s to delete. (At least %d resources are deleting.)", waiting[0].DisplayName(), len(waiting)) @@ -388,7 +395,15 @@ func (r *Reconciler) patchStatus(ctx context.Context, app *v1beta1.Application, func (r *Reconciler) updateStatus(ctx context.Context, app *v1beta1.Application, phase common.ApplicationPhase) error { app.Status.Phase = phase updateObservedGeneration(app) - return r.Status().Update(ctx, app) + + if !r.disableStatusUpdate { + return r.Status().Update(ctx, app) + } + obj, err := app.Unstructured() + if err != nil { + return err + } + return r.Status().Update(ctx, obj) } func (r *Reconciler) doWorkflowFinish(app *v1beta1.Application, wf workflow.Workflow) error { @@ -501,13 +516,12 @@ func (r *Reconciler) SetupWithManager(mgr ctrl.Manager) error { // Setup adds a controller that reconciles AppRollout. func Setup(mgr ctrl.Manager, args core.Args) error { reconciler := Reconciler{ - Client: mgr.GetClient(), - Scheme: mgr.GetScheme(), - Recorder: event.NewAPIRecorder(mgr.GetEventRecorderFor("Application")), - dm: args.DiscoveryMapper, - pd: args.PackageDiscover, - appRevisionLimit: args.AppRevisionLimit, - concurrentReconciles: args.ConcurrentReconciles, + Client: mgr.GetClient(), + Scheme: mgr.GetScheme(), + Recorder: event.NewAPIRecorder(mgr.GetEventRecorderFor("Application")), + dm: args.DiscoveryMapper, + pd: args.PackageDiscover, + options: parseOptions(args), } return reconciler.SetupWithManager(mgr) } @@ -557,3 +571,11 @@ func timeReconcile(app *v1beta1.Application) func() { metrics.ApplicationReconcileTimeHistogram.WithLabelValues(beginPhase, string(app.Status.Phase)).Observe(v) } } + +func parseOptions(args core.Args) options { + return options{ + disableStatusUpdate: args.EnableCompatibility, + appRevisionLimit: args.AppRevisionLimit, + concurrentReconciles: args.ConcurrentReconciles, + } +} diff --git a/pkg/controller/core.oam.dev/v1alpha2/application/revision.go b/pkg/controller/core.oam.dev/v1alpha2/application/revision.go index 90910c625..705957305 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/revision.go +++ b/pkg/controller/core.oam.dev/v1alpha2/application/revision.go @@ -854,7 +854,7 @@ func cleanUpWorkflowComponentRevision(ctx context.Context, h *AppHandler) error } // collect component revision in use compRevisionInUse := map[string]map[string]struct{}{} - for _, resource := range h.app.Status.AppliedResources { + for i, resource := range h.app.Status.AppliedResources { compName := resource.Name ns := resource.Namespace r := &unstructured.Unstructured{} @@ -863,7 +863,7 @@ func cleanUpWorkflowComponentRevision(ctx context.Context, h *AppHandler) error err := h.r.Get(_ctx, ktypes.NamespacedName{Name: compName, Namespace: ns}, r) notFound := apierrors.IsNotFound(err) if err != nil && !notFound { - return err + return errors.WithMessagef(err, "get applied resource index=%d", i) } if compRevisionInUse[compName] == nil { compRevisionInUse[compName] = map[string]struct{}{} diff --git a/pkg/controller/core.oam.dev/v1alpha2/application/suite_test.go b/pkg/controller/core.oam.dev/v1alpha2/application/suite_test.go index 35cb9244b..1f295d210 100644 --- a/pkg/controller/core.oam.dev/v1alpha2/application/suite_test.go +++ b/pkg/controller/core.oam.dev/v1alpha2/application/suite_test.go @@ -137,13 +137,14 @@ var _ = BeforeSuite(func(done Done) { appParser = appfile.NewApplicationParser(k8sClient, dm, pd) reconciler = &Reconciler{ - Client: k8sClient, - Scheme: testScheme, - dm: dm, - pd: pd, - Recorder: event.NewAPIRecorder(recorder), - appRevisionLimit: appRevisionLimit, + Client: k8sClient, + Scheme: testScheme, + dm: dm, + pd: pd, + Recorder: event.NewAPIRecorder(recorder), } + + reconciler.appRevisionLimit = appRevisionLimit // setup the controller manager since we need the component handler to run in the background mgr, err = ctrl.NewManager(cfg, ctrl.Options{ Scheme: testScheme,