Fix: workflow skip executing all steps occasionally (#3025)

* fix asi

Signed-off-by: Jian.Li <lj176172@alibaba-inc.com>

* fix lint

Signed-off-by: Jian.Li <lj176172@alibaba-inc.com>

* add trace tag

* add args for this feature

Signed-off-by: Jian.Li <lj176172@alibaba-inc.com>

* enable-asi-compatibility

Signed-off-by: Jian.Li <lj176172@alibaba-inc.com>

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
+}

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()

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
 }

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,
+	}
+}

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{}{}

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,