Feat: sync application from CR to data store (#3428)

* Feat: sync application from CR to data store

Signed-off-by: Jianbo Sun <jianbo.sjb@alibaba-inc.com>

* Feature: address comments

Signed-off-by: Jianbo Sun <jianbo.sjb@alibaba-inc.com>

* Feat: add migrate database feature to avoid max 63 charactor in kubeapi storage

Signed-off-by: Jianbo Sun <jianbo.sjb@alibaba-inc.com>

* update the sync data

Signed-off-by: Jianbo Sun <jianbo.sjb@alibaba-inc.com>

pkg/apiserver/sync/cache.go

@@ -0,0 +1,102 @@
+/*
+ Copyright 2022 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 sync
+
+import (
+	"context"
+	"errors"
+	"strconv"
+	"strings"
+
+	"github.com/sirupsen/logrus"
+
+	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
+	"github.com/oam-dev/kubevela/pkg/apiserver/datastore"
+	"github.com/oam-dev/kubevela/pkg/apiserver/model"
+)
+
+type cached struct {
+	generation int64
+	targets    int64
+}
+
+// InitCache will initialize the cache
+func (c *CR2UX) InitCache(ctx context.Context) error {
+	appsRaw, err := c.ds.List(ctx, &model.Application{}, &datastore.ListOptions{})
+	if err != nil {
+		if errors.Is(err, datastore.ErrRecordNotExist) {
+			return nil
+		}
+		return err
+	}
+	for _, appR := range appsRaw {
+		app, ok := appR.(*model.Application)
+		if !ok {
+			continue
+		}
+		gen, ok := app.Labels[model.LabelSyncGeneration]
+		if !ok || gen == "" {
+			continue
+		}
+		namespace := app.Labels[model.LabelSyncNamespace]
+		var key = formatAppComposedName(app.Name, namespace)
+		if strings.HasSuffix(app.Name, namespace) {
+			key = app.Name
+		}
+		generation, _ := strconv.ParseInt(gen, 10, 64)
+
+		// we should check targets if we synced from app status
+		c.updateCache(key, generation, 0)
+	}
+	return nil
+}
+
+func (c *CR2UX) shouldSync(ctx context.Context, targetApp *v1beta1.Application, del bool) bool {
+	key := formatAppComposedName(targetApp.Name, targetApp.Namespace)
+	cachedData, ok := c.cache.Load(key)
+	if ok {
+		cd := cachedData.(*cached)
+
+		// TODO(wonderflow): we should check targets if we sync that, it can avoid missing the status changed for targets updated in multi-cluster deploy, e.g. resumed suspend case.
+
+		if cd.generation == targetApp.Generation && !del {
+			logrus.Infof("app %s/%s with generation(%v) hasn't updated, ignore the sync event..", targetApp.Name, targetApp.Namespace, targetApp.Generation)
+			return false
+		}
+		if del {
+			c.cache.Delete(key)
+		}
+	}
+
+	sot := CheckSoTFromCR(targetApp)
+
+	// This is a double check to make sure the app not be converted and un-deployed
+	sot = CheckSoTFromAppMeta(ctx, c.ds, targetApp.Name, targetApp.Namespace, sot)
+
+	switch sot {
+	case FromUX, FromInner:
+		// we don't sync if the application is not created from CR
+		return false
+	default:
+	}
+	return true
+}
+
+func (c *CR2UX) updateCache(key string, generation, targets int64) {
+	// update cache
+	c.cache.Store(key, &cached{generation: generation, targets: targets})
+}

pkg/apiserver/sync/convert.go

@@ -0,0 +1,230 @@
+/*
+ Copyright 2022 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 sync
+
+import (
+	"context"
+	"strconv"
+	"strings"
+	"time"
+
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/oam-dev/kubevela/apis/core.oam.dev/common"
+	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1alpha1"
+	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
+	"github.com/oam-dev/kubevela/pkg/apiserver/model"
+	"github.com/oam-dev/kubevela/pkg/multicluster"
+	"github.com/oam-dev/kubevela/pkg/oam"
+	"github.com/oam-dev/kubevela/pkg/workflow/step"
+)
+
+// ConvertFromCRComponent concerts Application CR Component object into velaux data store component
+func ConvertFromCRComponent(appPrimaryKey string, component common.ApplicationComponent) (model.ApplicationComponent, error) {
+	bc := model.ApplicationComponent{
+		AppPrimaryKey:    appPrimaryKey,
+		Name:             component.Name,
+		Type:             component.Type,
+		ExternalRevision: component.ExternalRevision,
+		DependsOn:        component.DependsOn,
+		Inputs:           component.Inputs,
+		Outputs:          component.Outputs,
+		Scopes:           component.Scopes,
+		Creator:          model.AutoGenComp,
+	}
+	if component.Properties != nil {
+		properties, err := model.NewJSONStruct(component.Properties)
+		if err != nil {
+			return bc, err
+		}
+		bc.Properties = properties
+	}
+	for _, trait := range component.Traits {
+		properties, err := model.NewJSONStruct(trait.Properties)
+		if err != nil {
+			return bc, err
+		}
+		bc.Traits = append(bc.Traits, model.ApplicationTrait{CreateTime: time.Now(), UpdateTime: time.Now(), Properties: properties, Type: trait.Type, Alias: trait.Type, Description: "auto gen"})
+	}
+	return bc, nil
+}
+
+// ConvertFromCRPolicy converts Application CR Policy object into velaux data store policy
+func ConvertFromCRPolicy(appPrimaryKey string, policyCR v1beta1.AppPolicy, creator string) (model.ApplicationPolicy, error) {
+	plc := model.ApplicationPolicy{
+		AppPrimaryKey: appPrimaryKey,
+		Name:          policyCR.Name,
+		Type:          policyCR.Type,
+		Creator:       creator,
+	}
+	if policyCR.Properties != nil {
+		properties, err := model.NewJSONStruct(policyCR.Properties)
+		if err != nil {
+			return plc, err
+		}
+		plc.Properties = properties
+	}
+	return plc, nil
+}
+
+// ConvertFromCRWorkflow converts Application CR Workflow section into velaux data store workflow
+func ConvertFromCRWorkflow(ctx context.Context, cli client.Client, appPrimaryKey string, app *v1beta1.Application) (model.Workflow, []v1beta1.WorkflowStep, error) {
+	dataWf := model.Workflow{
+		AppPrimaryKey: appPrimaryKey,
+		// every namespace has a synced env
+		EnvName: model.AutoGenEnvNamePrefix + app.Namespace,
+		// every application has a synced workflow
+		Name:  model.AutoGenWorkflowNamePrefix + app.Name,
+		Alias: "Synced",
+	}
+	if app.Spec.Workflow == nil {
+		return dataWf, nil, nil
+	}
+	var steps []v1beta1.WorkflowStep
+	if app.Spec.Workflow.Ref != "" {
+		dataWf.Name = app.Spec.Workflow.Ref
+		wf := &v1alpha1.Workflow{}
+		if err := cli.Get(ctx, types.NamespacedName{Namespace: app.GetNamespace(), Name: app.Spec.Workflow.Ref}, wf); err != nil {
+			return dataWf, nil, err
+		}
+		steps = wf.Steps
+	} else {
+		steps = app.Spec.Workflow.Steps
+	}
+	for _, s := range steps {
+		if s.Properties == nil {
+			continue
+		}
+		properties, err := model.NewJSONStruct(s.Properties)
+		if err != nil {
+			return dataWf, nil, err
+		}
+		dataWf.Steps = append(dataWf.Steps, model.WorkflowStep{
+			Name:       s.Name,
+			Type:       s.Type,
+			Inputs:     s.Inputs,
+			Outputs:    s.Outputs,
+			DependsOn:  s.DependsOn,
+			Properties: properties,
+		})
+	}
+	return dataWf, steps, nil
+}
+
+// ConvertFromCRTargets converts deployed Cluster/Namespace from Application CR Status into velaux data store
+func ConvertFromCRTargets(targetApp *v1beta1.Application) []*model.Target {
+	var targets []*model.Target
+	nc := make(map[string]struct{})
+	for _, v := range targetApp.Status.AppliedResources {
+		var cluster = v.Cluster
+		if cluster == "" {
+			cluster = multicluster.ClusterLocalName
+		}
+		name := model.AutoGenTargetNamePrefix + cluster + "-" + v.Namespace
+		if _, ok := nc[name]; ok {
+			continue
+		}
+		nc[name] = struct{}{}
+		targets = append(targets, &model.Target{
+			Name: name,
+			Cluster: &model.ClusterTarget{
+				ClusterName: cluster,
+				Namespace:   v.Namespace,
+			},
+		})
+	}
+	return targets
+}
+
+// ConvertApp2DatastoreApp will convert Application CR to datastore application related resources
+func (c *CR2UX) ConvertApp2DatastoreApp(ctx context.Context, targetApp *v1beta1.Application) (*model.DataStoreApp, error) {
+	cli := c.cli
+
+	appName := c.getAppMetaName(ctx, targetApp.Name, targetApp.Namespace)
+
+	project := model.DefaultInitName
+	if _, ok := targetApp.Labels[oam.LabelAddonName]; ok && strings.HasPrefix(targetApp.Name, "addon-") {
+		project = model.DefaultAddonProject
+	}
+	appMeta := &model.Application{
+		Name:        appName,
+		Description: model.AutoGenDesc,
+		Alias:       targetApp.Name,
+		Project:     project,
+		Labels: map[string]string{
+			model.LabelSyncNamespace:  targetApp.Namespace,
+			model.LabelSyncGeneration: strconv.FormatInt(targetApp.Generation, 10),
+		},
+	}
+
+	// 1. convert app meta and env
+	dsApp := &model.DataStoreApp{
+		AppMeta: appMeta,
+		Env: &model.Env{
+			Name:        model.AutoGenEnvNamePrefix + targetApp.Namespace,
+			Namespace:   targetApp.Namespace,
+			Description: model.AutoGenDesc,
+			Project:     project,
+			Alias:       "Synced",
+		},
+		Eb: &model.EnvBinding{
+			AppPrimaryKey: appMeta.PrimaryKey(),
+			Name:          model.AutoGenEnvNamePrefix + targetApp.Namespace,
+			AppDeployName: appMeta.GetAppNameForSynced(),
+		},
+	}
+
+	// 2. convert component and trait
+	for _, cmp := range targetApp.Spec.Components {
+		compModel, err := ConvertFromCRComponent(appMeta.PrimaryKey(), cmp)
+		if err != nil {
+			return nil, err
+		}
+		dsApp.Comps = append(dsApp.Comps, &compModel)
+	}
+
+	// 3. convert workflow
+	wf, steps, err := ConvertFromCRWorkflow(ctx, cli, appMeta.PrimaryKey(), targetApp)
+	if err != nil {
+		return nil, err
+	}
+	dsApp.Workflow = &wf
+
+	// 4. convert policy, some policies are references in workflow step, we need to sync all the outside policy to make that work
+	outsidePLC, err := step.LoadExternalPoliciesForWorkflow(ctx, cli, targetApp.Namespace, steps, targetApp.Spec.Policies)
+	if err != nil {
+		return nil, err
+	}
+	for _, plc := range outsidePLC {
+		plcModel, err := ConvertFromCRPolicy(appMeta.PrimaryKey(), plc, model.AutoGenRefPolicy)
+		if err != nil {
+			return nil, err
+		}
+		dsApp.Policies = append(dsApp.Policies, &plcModel)
+	}
+
+	// TODO(wonderflow): we don't sync targets as it can't be judged well in velaux env
+	// if we want to sync, we can extract targets from status, like below:
+	/*
+		dsApp.Targets = ConvertFromCRTargets(targetApp)
+		for _, t := range dsApp.Targets {
+			dsApp.Env.Targets = append(dsApp.Env.Targets, t.Name)
+		}
+	*/
+	return dsApp, nil
+}

pkg/apiserver/sync/cr2ux.go

@@ -0,0 +1,198 @@
+/*
+ Copyright 2022 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 sync
+
+import (
+	"context"
+	"sync"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
+	"github.com/oam-dev/kubevela/pkg/apiserver/datastore"
+	"github.com/oam-dev/kubevela/pkg/apiserver/log"
+	"github.com/oam-dev/kubevela/pkg/apiserver/model"
+	"github.com/oam-dev/kubevela/pkg/oam"
+)
+
+const (
+	// FromCR means the data source of truth is from k8s CR
+	FromCR = "from-CR"
+	// FromUX means the data source of truth is from velaux data store
+	FromUX = "from-UX"
+	// FromInner means the data source of truth is from KubeVela inner usage, such as addon or configuration that don't want to be synced
+	FromInner = "from-inner"
+
+	// SoT means the source of Truth from
+	SoT = "SourceOfTruth"
+)
+
+// CheckSoTFromCR will check the source of truth of the application
+func CheckSoTFromCR(targetApp *v1beta1.Application) string {
+
+	if _, innerUse := targetApp.Annotations[oam.AnnotationSOTFromInner]; innerUse {
+		return FromInner
+	}
+	if _, appName := targetApp.Annotations[oam.AnnotationAppName]; appName {
+		return FromUX
+	}
+	return FromCR
+}
+
+// CheckSoTFromAppMeta will check the source of truth marked in datastore
+func CheckSoTFromAppMeta(ctx context.Context, ds datastore.DataStore, appName, namespace string, sotFromCR string) string {
+
+	app := &model.Application{Name: formatAppComposedName(appName, namespace)}
+	err := ds.Get(ctx, app)
+	if err != nil {
+		app = &model.Application{Name: appName}
+		err = ds.Get(ctx, app)
+		if err != nil {
+			return sotFromCR
+		}
+	}
+	if app.Labels == nil || app.Labels[SoT] == "" {
+		return sotFromCR
+	}
+	return app.Labels[SoT]
+}
+
+// CR2UX provides the Add/Update/Delete method
+type CR2UX struct {
+	ds    datastore.DataStore
+	cli   client.Client
+	cache sync.Map
+}
+
+func formatAppComposedName(name, namespace string) string {
+	return name + "-" + namespace
+}
+
+// we need to prevent the case that one app is deleted ant it's name is pure appName, then other app with namespace suffix will be mixed
+func (c *CR2UX) getAppMetaName(ctx context.Context, name, namespace string) string {
+	alreadyCreated := &model.Application{Name: formatAppComposedName(name, namespace)}
+	err := c.ds.Get(ctx, alreadyCreated)
+	if err == nil {
+		return formatAppComposedName(name, namespace)
+	}
+
+	// check if it's created the first in database
+	existApp := &model.Application{Name: name}
+	err = c.ds.Get(ctx, existApp)
+	if err == nil {
+		en := existApp.Labels[model.LabelSyncNamespace]
+		if en != namespace {
+			return formatAppComposedName(name, namespace)
+		}
+	}
+	return name
+}
+
+// AddOrUpdate will sync application CR to storage of VelaUX automatically
+func (c *CR2UX) AddOrUpdate(ctx context.Context, targetApp *v1beta1.Application) error {
+	ds := c.ds
+
+	if !c.shouldSync(ctx, targetApp, false) {
+		return nil
+	}
+
+	dsApp, err := c.ConvertApp2DatastoreApp(ctx, targetApp)
+	if err != nil {
+		log.Logger.Errorf("Convert App to data store err %v", err)
+		return err
+	}
+
+	if err = StoreProject(ctx, dsApp.AppMeta.Project, ds); err != nil {
+		log.Logger.Errorf("get or create project for sync process err %v", err)
+		return err
+	}
+
+	if err = StoreEnv(ctx, dsApp, ds); err != nil {
+		log.Logger.Errorf("Store Env Metadata to data store err %v", err)
+		return err
+	}
+	if err = StoreEnvBinding(ctx, dsApp.Eb, ds); err != nil {
+		log.Logger.Errorf("Store EnvBinding Metadata to data store err %v", err)
+		return err
+	}
+	if err = StoreComponents(ctx, dsApp.AppMeta.Name, dsApp.Comps, ds); err != nil {
+		log.Logger.Errorf("Store Components Metadata to data store err %v", err)
+		return err
+	}
+	if err = StorePolicy(ctx, dsApp.AppMeta.Name, dsApp.Policies, ds); err != nil {
+		log.Logger.Errorf("Store Policy Metadata to data store err %v", err)
+		return err
+	}
+	if err = StoreWorkflow(ctx, dsApp, ds); err != nil {
+		log.Logger.Errorf("Store Workflow Metadata to data store err %v", err)
+		return err
+	}
+	/*
+		if err = StoreTargets(ctx, dsApp, ds); err != nil {
+			log.Logger.Errorf("Store targets to data store err %v", err)
+			return err
+		}
+
+	*/
+
+	if err = StoreAppMeta(ctx, dsApp, ds); err != nil {
+		log.Logger.Errorf("Store App Metadata to data store err %v", err)
+		return err
+	}
+
+	// update cache
+	c.updateCache(dsApp.AppMeta.PrimaryKey(), targetApp.Generation, int64(len(dsApp.Targets)))
+	return nil
+}
+
+// DeleteApp will delete the application as the CR was deleted
+func (c *CR2UX) DeleteApp(ctx context.Context, targetApp *v1beta1.Application) error {
+	ds := c.ds
+
+	if !c.shouldSync(ctx, targetApp, true) {
+		return nil
+	}
+	appName := c.getAppMetaName(ctx, targetApp.Name, targetApp.Namespace)
+
+	_ = ds.Delete(ctx, &model.Application{Name: appName})
+
+	cmps, err := ds.List(ctx, &model.ApplicationComponent{AppPrimaryKey: appName}, &datastore.ListOptions{})
+	if err != nil {
+		return err
+	}
+	for _, entity := range cmps {
+		comp := entity.(*model.ApplicationComponent)
+		if comp.Creator == model.AutoGenComp {
+			_ = ds.Delete(ctx, comp)
+		}
+	}
+
+	plcs, err := ds.List(ctx, &model.ApplicationPolicy{AppPrimaryKey: appName}, &datastore.ListOptions{})
+	if err != nil {
+		return err
+	}
+	for _, entity := range plcs {
+		comp := entity.(*model.ApplicationPolicy)
+		if comp.Creator == model.AutoGenPolicy {
+			_ = ds.Delete(ctx, comp)
+		}
+	}
+
+	_ = ds.Delete(ctx, &model.Workflow{Name: model.AutoGenWorkflowNamePrefix + appName})
+
+	return nil
+}

pkg/apiserver/sync/store.go

@@ -0,0 +1,230 @@
+/*
+ Copyright 2022 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 sync
+
+import (
+	"context"
+	"errors"
+
+	"github.com/oam-dev/kubevela/pkg/apiserver/datastore"
+	"github.com/oam-dev/kubevela/pkg/apiserver/log"
+	"github.com/oam-dev/kubevela/pkg/apiserver/model"
+	"github.com/oam-dev/kubevela/pkg/utils"
+)
+
+// StoreProject will create project for synced application
+func StoreProject(ctx context.Context, name string, ds datastore.DataStore) error {
+	err := ds.Get(ctx, &model.Project{Name: name})
+	if err == nil {
+		// it means the record already exists, don't need to add anything
+		return nil
+	}
+	if !errors.Is(err, datastore.ErrRecordNotExist) {
+		// other database error, return it
+		return err
+	}
+	proj := &model.Project{
+		Name:        name,
+		Description: model.AutoGenProj,
+	}
+	return ds.Add(ctx, proj)
+}
+
+// StoreAppMeta will sync application metadata from CR to datastore
+func StoreAppMeta(ctx context.Context, app *model.DataStoreApp, ds datastore.DataStore) error {
+	err := ds.Get(ctx, &model.Application{Name: app.AppMeta.Name})
+	if err == nil {
+		// it means the record already exists
+		return ds.Put(ctx, app.AppMeta)
+	}
+	if !errors.Is(err, datastore.ErrRecordNotExist) {
+		// other database error, return it
+		return err
+	}
+	return ds.Add(ctx, app.AppMeta)
+}
+
+// StoreEnv will sync application namespace from CR to datastore env, one namespace belongs to one env
+func StoreEnv(ctx context.Context, app *model.DataStoreApp, ds datastore.DataStore) error {
+	curEnv := &model.Env{Name: app.Env.Name}
+	err := ds.Get(ctx, curEnv)
+	if err == nil {
+		// it means the record already exists, compare the targets
+		if utils.EqualSlice(curEnv.Targets, app.Env.Targets) {
+			return nil
+		}
+		return ds.Put(ctx, app.Env)
+	}
+	if !errors.Is(err, datastore.ErrRecordNotExist) {
+		// other database error, return it
+		return err
+	}
+	return ds.Add(ctx, app.Env)
+}
+
+// StoreEnvBinding will add envbinding for application CR one application one envbinding
+func StoreEnvBinding(ctx context.Context, eb *model.EnvBinding, ds datastore.DataStore) error {
+	err := ds.Get(ctx, eb)
+	if err == nil {
+		// it means the record already exists, don't need to add anything
+		return nil
+	}
+	if !errors.Is(err, datastore.ErrRecordNotExist) {
+		// other database error, return it
+		return err
+	}
+	return ds.Add(ctx, eb)
+}
+
+// StoreComponents will sync application components from CR to datastore
+func StoreComponents(ctx context.Context, appPrimaryKey string, expComps []*model.ApplicationComponent, ds datastore.DataStore) error {
+
+	// list the existing components in datastore
+	originComps, err := ds.List(ctx, &model.ApplicationComponent{AppPrimaryKey: appPrimaryKey}, &datastore.ListOptions{})
+	if err != nil {
+		return err
+	}
+	var originCompNames []string
+	for _, entity := range originComps {
+		comp := entity.(*model.ApplicationComponent)
+		originCompNames = append(originCompNames, comp.Name)
+	}
+
+	var targetCompNames []string
+	for _, comp := range expComps {
+		targetCompNames = append(targetCompNames, comp.Name)
+	}
+
+	_, readyToDelete, readyToAdd := utils.ThreeWaySliceCompare(originCompNames, targetCompNames)
+
+	// delete the components that not belongs to the new app
+	for _, entity := range originComps {
+		comp := entity.(*model.ApplicationComponent)
+		// we only compare for components that automatically generated by sync process.
+		if comp.Creator != model.AutoGenComp {
+			continue
+		}
+		if !utils.StringsContain(readyToDelete, comp.Name) {
+			continue
+		}
+		if err := ds.Delete(ctx, comp); err != nil {
+			if errors.Is(err, datastore.ErrRecordNotExist) {
+				continue
+			}
+			log.Logger.Warnf("delete comp %s for app %s  failure %s", comp.Name, appPrimaryKey, err.Error())
+		}
+	}
+
+	// add or update new app's components for datastore
+	for _, comp := range expComps {
+		if utils.StringsContain(readyToAdd, comp.Name) {
+			err = ds.Add(ctx, comp)
+		} else {
+			err = ds.Put(ctx, comp)
+		}
+		if err != nil {
+			log.Logger.Warnf("convert comp %s for app %s into datastore failure %s", comp.Name, utils.Sanitize(appPrimaryKey), err.Error())
+			return err
+		}
+	}
+	return nil
+}
+
+// StorePolicy will add/update/delete policies, we don't delete ref policy
+func StorePolicy(ctx context.Context, appPrimaryKey string, expPolicies []*model.ApplicationPolicy, ds datastore.DataStore) error {
+	// list the existing policies for this app in datastore
+	originPolicies, err := ds.List(ctx, &model.ApplicationPolicy{AppPrimaryKey: appPrimaryKey}, &datastore.ListOptions{})
+	if err != nil {
+		return err
+	}
+	var originPolicyNames []string
+	for _, entity := range originPolicies {
+		plc := entity.(*model.ApplicationPolicy)
+		originPolicyNames = append(originPolicyNames, plc.Name)
+	}
+
+	var targetPLCNames []string
+	for _, plc := range expPolicies {
+		targetPLCNames = append(targetPLCNames, plc.Name)
+	}
+
+	_, readyToDelete, readyToAdd := utils.ThreeWaySliceCompare(originPolicyNames, targetPLCNames)
+
+	// delete the components that not belongs to the new app
+	for _, entity := range originPolicies {
+		plc := entity.(*model.ApplicationPolicy)
+		// we only compare for policies that automatically generated by sync process, and the policy should not be ref ones.
+		if plc.Creator != model.AutoGenPolicy {
+			continue
+		}
+		if !utils.StringsContain(readyToDelete, plc.Name) {
+			continue
+		}
+		if err := ds.Delete(ctx, plc); err != nil {
+			if errors.Is(err, datastore.ErrRecordNotExist) {
+				continue
+			}
+			log.Logger.Warnf("delete policy %s for app %s failure %s", plc.Name, appPrimaryKey, err.Error())
+		}
+	}
+
+	// add or update new app's policies for datastore
+	for _, plc := range expPolicies {
+		if utils.StringsContain(readyToAdd, plc.Name) {
+			err = ds.Add(ctx, plc)
+		} else {
+			err = ds.Put(ctx, plc)
+		}
+		if err != nil {
+			log.Logger.Warnf("convert policy %s for app %s into datastore failure %s", plc.Name, utils.Sanitize(appPrimaryKey), err.Error())
+			return err
+		}
+	}
+	return nil
+}
+
+// StoreWorkflow will sync workflow application CR to datastore, it updates the only one workflow from the application with specified name
+func StoreWorkflow(ctx context.Context, dsApp *model.DataStoreApp, ds datastore.DataStore) error {
+	err := ds.Get(ctx, &model.Workflow{AppPrimaryKey: dsApp.AppMeta.Name, Name: dsApp.Workflow.Name})
+	if err == nil {
+		// it means the record already exists, update it
+		return ds.Put(ctx, dsApp.Workflow)
+	}
+	if !errors.Is(err, datastore.ErrRecordNotExist) {
+		// other database error, return it
+		return err
+	}
+	return ds.Add(ctx, dsApp.Workflow)
+}
+
+// StoreTargets will sync targets from application CR to datastore
+func StoreTargets(ctx context.Context, dsApp *model.DataStoreApp, ds datastore.DataStore) error {
+	for _, t := range dsApp.Targets {
+		err := ds.Get(ctx, t)
+		if err == nil {
+			continue
+		}
+		if !errors.Is(err, datastore.ErrRecordNotExist) {
+			// other database error, return it
+			return err
+		}
+		if err = ds.Add(ctx, t); err != nil {
+			return err
+		}
+	}
+	return nil
+}

pkg/apiserver/sync/suit_test.go

@@ -0,0 +1,100 @@
+/*
+ Copyright 2022 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 sync
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+	"testing"
+	"time"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"k8s.io/client-go/rest"
+	"k8s.io/utils/pointer"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
+
+	"github.com/oam-dev/kubevela/pkg/apiserver/clients"
+	"github.com/oam-dev/kubevela/pkg/apiserver/datastore"
+	"github.com/oam-dev/kubevela/pkg/apiserver/datastore/kubeapi"
+	"github.com/oam-dev/kubevela/pkg/apiserver/datastore/mongodb"
+	"github.com/oam-dev/kubevela/pkg/utils/common"
+)
+
+var cfg *rest.Config
+var k8sClient client.Client
+var testEnv *envtest.Environment
+
+func TestSync(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Sync Suite Test")
+}
+
+var _ = BeforeSuite(func(done Done) {
+	rand.Seed(time.Now().UnixNano())
+	By("bootstrapping Sync test environment")
+
+	testEnv = &envtest.Environment{
+		ControlPlaneStartTimeout: time.Minute * 3,
+		ControlPlaneStopTimeout:  time.Minute,
+		UseExistingCluster:       pointer.BoolPtr(false),
+		CRDDirectoryPaths:        []string{"../../../charts/vela-core/crds"},
+	}
+
+	By("start kube test env")
+	var err error
+	cfg, err = testEnv.Start()
+	Expect(err).ShouldNot(HaveOccurred())
+	Expect(cfg).ToNot(BeNil())
+
+	By("new kube client")
+	cfg.Timeout = time.Minute * 2
+	k8sClient, err = client.New(cfg, client.Options{Scheme: common.Scheme})
+	Expect(err).Should(BeNil())
+	Expect(k8sClient).ToNot(BeNil())
+	clients.SetKubeClient(k8sClient)
+	By("new kube client success")
+	clients.SetKubeClient(k8sClient)
+	Expect(err).Should(BeNil())
+	close(done)
+}, 240)
+
+var _ = AfterSuite(func() {
+	By("tearing down the test environment")
+	err := testEnv.Stop()
+	Expect(err).ToNot(HaveOccurred())
+})
+
+func NewDatastore(cfg datastore.Config) (ds datastore.DataStore, err error) {
+	switch cfg.Type {
+	case "mongodb":
+		ds, err = mongodb.New(context.Background(), cfg)
+		if err != nil {
+			return nil, fmt.Errorf("create mongodb datastore instance failure %w", err)
+		}
+	case "kubeapi":
+		ds, err = kubeapi.New(context.Background(), cfg)
+		if err != nil {
+			return nil, fmt.Errorf("create mongodb datastore instance failure %w", err)
+		}
+	default:
+		return nil, fmt.Errorf("not support datastore type %s", cfg.Type)
+	}
+	return ds, nil
+}

pkg/apiserver/sync/testdata/test-app1.yaml

@@ -0,0 +1,37 @@
+apiVersion: core.oam.dev/v1beta1
+kind: Application
+metadata:
+  name: example
+spec:
+  components:
+    - name: nginx
+      type: webservice
+      properties:
+        image: nginx
+      traits:
+        - type: gateway
+          properties:
+            domain: testsvc.example.com
+            http:
+              "/": 8000
+    - name: nginx2
+      type: webservice
+      properties:
+        image: nginx2
+  policies:
+    - name: topology-beijing-demo
+      type: topology
+      properties:
+        clusterLabelSelector:
+          region: beijing
+        namespace: demo
+    - name: topology-local
+      type: topology
+      properties:
+        targets: ["local/demo", "local/ackone-demo"]
+  workflow:
+    steps:
+      - type: deploy
+        name: deploy-local
+        properties:
+          policies: ["topology-local", "topology-beijing-demo"]

pkg/apiserver/sync/testdata/test-app2.yaml

@@ -0,0 +1,10 @@
+apiVersion: core.oam.dev/v1beta1
+kind: Application
+metadata:
+  name: example
+spec:
+  components:
+  - name: blog
+    type: webservice
+    properties:
+      image: wordpress

pkg/apiserver/sync/testdata/test-app3.yaml

@@ -0,0 +1,33 @@
+apiVersion: core.oam.dev/v1beta1
+kind: Application
+metadata:
+  name: example
+spec:
+  components:
+    - name: blog
+      type: webservice
+      properties:
+        image: wordpress
+      traits:
+        - type: gateway
+          properties:
+            domain: testsvc.example.com
+            http:
+              "/": 8000
+    - name: nginx2
+      type: webservice
+      properties:
+        image: nginx
+  policies:
+    - name: topology-beijing-demo
+      type: topology
+      properties:
+        clusterLabelSelector:
+          region: beijing
+        namespace: demo
+  workflow:
+    steps:
+      - type: deploy
+        name: deploy-local
+        properties:
+          policies: ["topology-beijing-demo"]

pkg/apiserver/sync/worker.go

@@ -0,0 +1,103 @@
+/*
+ Copyright 2022 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 sync
+
+import (
+	"context"
+	"encoding/json"
+	"sync"
+
+	"github.com/fatih/color"
+	"github.com/sirupsen/logrus"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/dynamic/dynamicinformer"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/klog/v2"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
+	"github.com/oam-dev/kubevela/pkg/apiserver/clients"
+	"github.com/oam-dev/kubevela/pkg/apiserver/datastore"
+)
+
+// Start prepares watchers and run their controllers, then waits for process termination signals
+func Start(ctx context.Context, ds datastore.DataStore, cfg *rest.Config) {
+	k8sClient, err := clients.GetKubeClient()
+	if err != nil {
+		logrus.Fatal(err)
+	}
+
+	dynamicClient, err := dynamic.NewForConfig(cfg)
+	if err != nil {
+		logrus.Fatal(err)
+	}
+
+	f := dynamicinformer.NewFilteredDynamicSharedInformerFactory(dynamicClient, 0, v1.NamespaceAll, nil)
+
+	startAppSyncing(ctx, f, ds, k8sClient)
+}
+
+func startAppSyncing(ctx context.Context, factory dynamicinformer.DynamicSharedInformerFactory, ds datastore.DataStore, cli client.Client) {
+	var err error
+	informer := factory.ForResource(v1beta1.SchemeGroupVersion.WithResource("applications")).Informer()
+	getApp := func(obj interface{}) *v1beta1.Application {
+		app := &v1beta1.Application{}
+		bs, _ := json.Marshal(obj)
+		_ = json.Unmarshal(bs, app)
+		return app
+	}
+	cu := &CR2UX{
+		ds:    ds,
+		cli:   cli,
+		cache: sync.Map{},
+	}
+	if err = cu.InitCache(ctx); err != nil {
+		klog.Fatal("sync app init err", err)
+	}
+
+	handlers := cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			app := getApp(obj)
+			klog.Infof("watched add app event, namespace: %s, name: %s", app.Namespace, app.Name)
+			err = cu.AddOrUpdate(ctx, app)
+			if err != nil {
+				logrus.Errorf("Application %-30s Create Sync to db err %v", color.WhiteString(app.Namespace+"/"+app.Name), err)
+			}
+		},
+		UpdateFunc: func(oldObj, obj interface{}) {
+			app := getApp(obj)
+			klog.Infof("watched update app event, namespace: %s, name: %s", app.Namespace, app.Name)
+			err = cu.AddOrUpdate(ctx, app)
+			if err != nil {
+				klog.Errorf("Application %-30s Update Sync to db err %v", color.WhiteString(app.Namespace+"/"+app.Name), err)
+			}
+		},
+		DeleteFunc: func(obj interface{}) {
+			app := getApp(obj)
+			klog.Infof("watched delete app event, namespace: %s, name: %s", app.Namespace, app.Name)
+			err = cu.DeleteApp(ctx, app)
+			if err != nil {
+				klog.Errorf("Application %-30s Deleted Sync to db err %v", color.WhiteString(app.Namespace+"/"+app.Name), err)
+			}
+		},
+	}
+	informer.AddEventHandler(handlers)
+	klog.Info("app syncing started")
+	informer.Run(ctx.Done())
+}

pkg/apiserver/sync/worker_test.go

@@ -0,0 +1,127 @@
+/*
+ Copyright 2022 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 sync
+
+import (
+	"context"
+	"time"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	corev1 "k8s.io/api/core/v1"
+
+	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
+	"github.com/oam-dev/kubevela/pkg/apiserver/datastore"
+	"github.com/oam-dev/kubevela/pkg/apiserver/model"
+	"github.com/oam-dev/kubevela/pkg/oam/util"
+	common2 "github.com/oam-dev/kubevela/pkg/utils/common"
+)
+
+var _ = Describe("Test Worker CR sync to datastore", func() {
+	BeforeEach(func() {
+	})
+
+	It("Test app sync test-app1", func() {
+
+		By("Preparing database")
+		dbNamespace := "sync-db-ns1-test"
+		appNS1 := "sync-worker-test-ns1"
+		appNS2 := "sync-worker-test-ns2"
+		ds, err := NewDatastore(datastore.Config{Type: "kubeapi", Database: "sync-test-db1"})
+		Expect(ds).ToNot(BeNil())
+		Expect(err).Should(BeNil())
+		var ns = corev1.Namespace{}
+		ns.Name = dbNamespace
+		err = k8sClient.Create(context.TODO(), &ns)
+		Expect(err).Should(SatisfyAny(BeNil(), &util.AlreadyExistMatcher{}))
+		ns.Name = appNS1
+		ns.ResourceVersion = ""
+		err = k8sClient.Create(context.TODO(), &ns)
+		Expect(err).Should(SatisfyAny(BeNil(), &util.AlreadyExistMatcher{}))
+		ns.Name = appNS2
+		ns.ResourceVersion = ""
+		err = k8sClient.Create(context.TODO(), &ns)
+		Expect(err).Should(SatisfyAny(BeNil(), &util.AlreadyExistMatcher{}))
+
+		By("Start syncing")
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		go Start(ctx, ds, cfg)
+
+		By("create test app1 and check the syncing results")
+		app1 := &v1beta1.Application{}
+		Expect(common2.ReadYamlToObject("testdata/test-app1.yaml", app1)).Should(BeNil())
+		app1.Namespace = appNS1
+		Expect(k8sClient.Create(context.TODO(), app1)).Should(BeNil())
+
+		Eventually(func() error {
+			appm := model.Application{Name: app1.Name}
+			return ds.Get(ctx, &appm)
+		}, 10*time.Second, 100*time.Millisecond).Should(BeNil())
+
+		comp1 := model.ApplicationComponent{AppPrimaryKey: app1.Name, Name: "nginx"}
+		Expect(ds.Get(ctx, &comp1)).Should(BeNil())
+		Expect(comp1.Properties).Should(BeEquivalentTo(&model.JSONStruct{"image": "nginx"}))
+
+		comp2 := model.ApplicationComponent{AppPrimaryKey: app1.Name, Name: "nginx2"}
+		Expect(ds.Get(ctx, &comp2)).Should(BeNil())
+		Expect(comp2.Properties).Should(BeEquivalentTo(&model.JSONStruct{"image": "nginx2"}))
+
+		appPlc1 := model.ApplicationPolicy{AppPrimaryKey: app1.Name, Name: "topology-beijing-demo"}
+		Expect(ds.Get(ctx, &appPlc1)).Should(BeNil())
+		Expect(appPlc1.Properties).Should(BeEquivalentTo(&model.JSONStruct{"namespace": "demo", "clusterLabelSelector": map[string]interface{}{"region": "beijing"}}))
+
+		appPlc2 := model.ApplicationPolicy{AppPrimaryKey: app1.Name, Name: "topology-local"}
+		Expect(ds.Get(ctx, &appPlc2)).Should(BeNil())
+		Expect(appPlc2.Properties).Should(BeEquivalentTo(&model.JSONStruct{"targets": []interface{}{"local/demo", "local/ackone-demo"}}))
+
+		appwf1 := model.Workflow{AppPrimaryKey: app1.Name, Name: model.AutoGenWorkflowNamePrefix + app1.Name}
+		Expect(ds.Get(ctx, &appwf1)).Should(BeNil())
+
+		By("create test app2 and check the syncing results")
+		app2 := &v1beta1.Application{}
+		Expect(common2.ReadYamlToObject("testdata/test-app2.yaml", app2)).Should(BeNil())
+		app2.Namespace = appNS2
+		Expect(k8sClient.Create(context.TODO(), app2)).Should(BeNil())
+
+		Eventually(func() error {
+			appm := model.Application{Name: formatAppComposedName(app2.Name, app2.Namespace)}
+			return ds.Get(ctx, &appm)
+		}, 10*time.Second, 100*time.Millisecond).Should(BeNil())
+
+		By("delete test app1 and check the syncing results")
+		Expect(k8sClient.Delete(context.TODO(), app1)).Should(BeNil())
+		Eventually(func() error {
+			appm := model.Application{Name: app1.Name}
+			return ds.Get(ctx, &appm)
+		}, 10*time.Second, 100*time.Millisecond).Should(BeEquivalentTo(datastore.ErrRecordNotExist))
+
+		By("update test app2 and check the syncing results")
+		newapp2 := &v1beta1.Application{}
+		Expect(common2.ReadYamlToObject("testdata/test-app3.yaml", newapp2)).Should(BeNil())
+		app2.Spec = newapp2.Spec
+		Expect(k8sClient.Update(context.TODO(), app2)).Should(BeNil())
+
+		Eventually(func() error {
+			appm := model.ApplicationComponent{AppPrimaryKey: formatAppComposedName(app2.Name, app2.Namespace), Name: "nginx2"}
+			return ds.Get(ctx, &appm)
+		}, 10*time.Second, 100*time.Millisecond).Should(BeNil())
+
+	})
+
+})