Feat: add record status sync (#2627)

* Feat: add record status sync

* fix typo

* optimize the code

* fix the port

* fix go mod

* fix rebase

cmd/apiserver/main.go

@@ -24,10 +24,12 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
 
 	restfulspec "github.com/emicklei/go-restful-openapi/v2"
 	"github.com/go-openapi/spec"
 
+	"github.com/google/uuid"
 	"github.com/oam-dev/kubevela/pkg/apiserver/log"
 	"github.com/oam-dev/kubevela/pkg/apiserver/rest"
 	"github.com/oam-dev/kubevela/version"
@@ -40,6 +42,9 @@ func main() {
 	flag.StringVar(&s.restCfg.Datastore.Type, "datastore-type", "kubeapi", "Metadata storage driver type, support kubeapi and mongodb")
 	flag.StringVar(&s.restCfg.Datastore.Database, "datastore-database", "kubevela", "Metadata storage database name, takes effect when the storage driver is mongodb.")
 	flag.StringVar(&s.restCfg.Datastore.URL, "datastore-url", "", "Metadata storage database url,takes effect when the storage driver is mongodb.")
+	flag.StringVar(&s.restCfg.LeaderConfig.ID, "id", uuid.New().String(), "the holder identity name")
+	flag.StringVar(&s.restCfg.LeaderConfig.LockName, "lock-name", "apiserver-lock", "the lease lock resource name")
+	flag.DurationVar(&s.restCfg.LeaderConfig.Duration, "duration", time.Second*5, "the lease lock resource name")
 	flag.Parse()
 
 	if len(os.Args) > 2 && os.Args[1] == "build-swagger" {
@@ -71,9 +76,10 @@ func main() {
 	}
 
 	srvc := make(chan struct{})
+	ctx, cancel := context.WithCancel(context.Background())
 
 	go func() {
-		if err := s.run(); err != nil {
+		if err := s.run(ctx); err != nil {
 			log.Logger.Errorf("failed to run apiserver: %v", err)
 		}
 		close(srvc)
@@ -84,7 +90,9 @@ func main() {
 	select {
 	case <-term:
 		log.Logger.Infof("Received SIGTERM, exiting gracefully...")
+		cancel()
 	case <-srvc:
+		cancel()
 		os.Exit(1)
 	}
 	log.Logger.Infof("See you next time!")
@@ -95,11 +103,9 @@ type Server struct {
 	restCfg rest.Config
 }
 
-func (s *Server) run() error {
+func (s *Server) run(ctx context.Context) error {
 	log.Logger.Infof("KubeVela information: version: %v, gitRevision: %v", version.VelaVersion, version.GitRevision)
 
-	ctx := context.Background()
-
 	server, err := rest.New(s.restCfg)
 	if err != nil {
 		return fmt.Errorf("create apiserver failed : %w ", err)

go.mod

@@ -31,6 +31,7 @@ require (
 	github.com/go-playground/validator/v10 v10.9.0
 	github.com/google/go-cmp v0.5.6
 	github.com/google/go-github/v32 v32.1.0
+	github.com/google/uuid v1.1.2
 	github.com/gosuri/uitable v0.0.4
 	github.com/hashicorp/hcl/v2 v2.9.1
 	github.com/hinshun/vt10x v0.0.0-20180616224451-1954e6464174

pkg/apiserver/model/application.go

@@ -188,6 +188,9 @@ var RevisionStatusComplete = "complete"
 // RevisionStatusFail event status failure
 var RevisionStatusFail = "failure"
 
+// DeployEventTerminated event status terminated
+var DeployEventTerminated = "terminated"
+
 // ApplicationRevision be created when an application initiates deployment and describes the phased version of the application.
 type ApplicationRevision struct {
 	Model

pkg/apiserver/rest/rest_server.go

@@ -20,15 +20,23 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"os"
+	"time"
 
 	restfulspec "github.com/emicklei/go-restful-openapi/v2"
 	"github.com/emicklei/go-restful/v3"
 	"github.com/go-openapi/spec"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+	"k8s.io/klog/v2"
+	ctrl "sigs.k8s.io/controller-runtime"
 
+	"github.com/oam-dev/kubevela/apis/types"
 	"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/apiserver/log"
+	"github.com/oam-dev/kubevela/pkg/apiserver/rest/usecase"
 	"github.com/oam-dev/kubevela/pkg/apiserver/rest/webservice"
 )
 
@@ -43,6 +51,15 @@ type Config struct {
 
 	// Datastore config
 	Datastore datastore.Config
+
+	// LeaderConfig for leader election
+	LeaderConfig leaderConfig
+}
+
+type leaderConfig struct {
+	ID       string
+	LockName string
+	Duration time.Duration
 }
 
 // APIServer interface for call api server
@@ -85,9 +102,72 @@ func New(cfg Config) (a APIServer, err error) {
 
 func (s *restServer) Run(ctx context.Context) error {
 	s.RegisterServices()
+
+	l, err := s.setupLeaderElection()
+	if err != nil {
+		return err
+	}
+
+	go func() {
+		leaderelection.RunOrDie(ctx, *l)
+	}()
+
 	return s.startHTTP(ctx)
 }
 
+func (s *restServer) setupLeaderElection() (*leaderelection.LeaderElectionConfig, error) {
+	restCfg := ctrl.GetConfigOrDie()
+
+	rl, err := resourcelock.NewFromKubeconfig(resourcelock.LeasesResourceLock, types.DefaultKubeVelaNS, s.cfg.LeaderConfig.LockName, resourcelock.ResourceLockConfig{
+		Identity: s.cfg.LeaderConfig.ID,
+	}, restCfg, time.Second*10)
+	if err != nil {
+		klog.ErrorS(err, "Unable to setup the resource lock")
+		return nil, err
+	}
+
+	return &leaderelection.LeaderElectionConfig{
+		Lock:          rl,
+		LeaseDuration: time.Second * 15,
+		RenewDeadline: time.Second * 10,
+		RetryPeriod:   time.Second * 2,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(ctx context.Context) {
+				s.runLeader(ctx, s.cfg.LeaderConfig.Duration)
+			},
+			OnStoppedLeading: func() {
+				klog.Infof("leader lost: %s", s.cfg.LeaderConfig.ID)
+				os.Exit(0)
+			},
+			OnNewLeader: func(identity string) {
+				if identity == s.cfg.LeaderConfig.ID {
+					return
+				}
+				klog.Infof("new leader elected: %s", identity)
+			},
+		},
+		ReleaseOnCancel: true,
+	}, nil
+}
+
+func (s restServer) runLeader(ctx context.Context, duration time.Duration) {
+	w := usecase.NewWorkflowUsecase(s.dataStore)
+
+	t := time.NewTicker(duration)
+	defer t.Stop()
+
+	for {
+		select {
+		case <-t.C:
+			if err := w.SyncWorkflowRecord(ctx); err != nil {
+				klog.ErrorS(err, "syncWorkflowRecordError")
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
 // RegisterServices register web service
 func (s *restServer) RegisterServices() restfulspec.Config {
 	webservice.Init(s.dataStore)

pkg/apiserver/rest/usecase/workflow.go

@@ -23,7 +23,12 @@ import (
 	"strings"
 
 	appsv1 "k8s.io/api/apps/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"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"
 	"github.com/oam-dev/kubevela/pkg/apiserver/log"
 	"github.com/oam-dev/kubevela/pkg/apiserver/model"
@@ -31,6 +36,11 @@ import (
 	"github.com/oam-dev/kubevela/pkg/apiserver/rest/utils/bcode"
 	"github.com/oam-dev/kubevela/pkg/oam"
 	"github.com/oam-dev/kubevela/pkg/oam/util"
+	"github.com/oam-dev/kubevela/pkg/workflow/recorder"
+)
+
+const (
+	labelControllerRevisionSync = "apiserver.oam.dev/cr-sync"
 )
 
 // WorkflowUsecase workflow manage api
@@ -44,15 +54,24 @@ type WorkflowUsecase interface {
 	UpdateWorkflow(ctx context.Context, workflow *model.Workflow, req apisv1.UpdateWorkflowRequest) (*apisv1.DetailWorkflowResponse, error)
 	ListWorkflowRecords(ctx context.Context, workflowName string, page, pageSize int) (*apisv1.ListWorkflowRecordsResponse, error)
 	DetailWorkflowRecord(ctx context.Context, workflowName, recordName string) (*apisv1.DetailWorkflowRecordResponse, error)
+	SyncWorkflowRecord(ctx context.Context) error
 }
 
 // NewWorkflowUsecase new workflow usecase
 func NewWorkflowUsecase(ds datastore.DataStore) WorkflowUsecase {
-	return &workflowUsecaseImpl{ds: ds}
+	kubecli, err := clients.GetKubeClient()
+	if err != nil {
+		log.Logger.Fatalf("get kubeclient failure %s", err.Error())
+	}
+	return &workflowUsecaseImpl{
+		ds:         ds,
+		kubeClient: kubecli,
+	}
 }
 
 type workflowUsecaseImpl struct {
-	ds datastore.DataStore
+	ds         datastore.DataStore
+	kubeClient client.Client
 }
 
 // DeleteWorkflow delete application workflow
@@ -261,11 +280,80 @@ func (w *workflowUsecaseImpl) DetailWorkflowRecord(ctx context.Context, workflow
 	}, nil
 }
 
-func (w *workflowUsecaseImpl) createWorkflowRecord(ctx context.Context, revision *appsv1.ControllerRevision) error {
-	app, err := util.RawExtension2Application(revision.Data)
+func (w *workflowUsecaseImpl) SyncWorkflowRecord(ctx context.Context) error {
+	crList := &appsv1.ControllerRevisionList{}
+	matchLabels := metav1.LabelSelector{
+		MatchExpressions: []metav1.LabelSelectorRequirement{
+			{
+				Key:      labelControllerRevisionSync,
+				Operator: metav1.LabelSelectorOpDoesNotExist,
+			},
+			{
+				Key:      recorder.LabelRecordVersion,
+				Operator: metav1.LabelSelectorOpExists,
+			},
+		},
+	}
+	selector, err := metav1.LabelSelectorAsSelector(&matchLabels)
 	if err != nil {
 		return err
 	}
+	if err := w.kubeClient.List(ctx, crList, &client.ListOptions{
+		LabelSelector: selector,
+	}); err != nil {
+		return err
+	}
+
+	for i, cr := range crList.Items {
+		app, err := util.RawExtension2Application(cr.Data)
+		if err != nil {
+			klog.ErrorS(err, "failed to get app data", "controller revision name", cr.Name)
+			continue
+		}
+
+		if err := w.createWorkflowRecord(ctx, app, strings.TrimPrefix(cr.Name, "record-")); err != nil && !errors.Is(err, datastore.ErrRecordExist) {
+			klog.ErrorS(err, "failed to create workflow record", "controller revision name", cr.Name)
+			continue
+		}
+
+		err = w.updateRecordApplicationRevisionStatus(ctx, app.Name, strings.TrimPrefix(cr.Name, fmt.Sprintf("record-%s-", app.Name)), app.Status.Workflow.Terminated)
+		if err != nil && !errors.Is(err, datastore.ErrRecordNotExist) {
+			klog.ErrorS(err, "failed to update deploy event status", "controller revision name", cr.Name)
+			continue
+		}
+
+		crList.Items[i].Labels[labelControllerRevisionSync] = "true"
+		if err := w.kubeClient.Update(ctx, &crList.Items[i]); err != nil {
+			klog.ErrorS(err, "failed to update annotation", "controller revision name", cr.Name)
+			continue
+		}
+	}
+
+	return nil
+}
+
+func (w *workflowUsecaseImpl) updateRecordApplicationRevisionStatus(ctx context.Context, appPrimaryKey, version string, terminated bool) error {
+	var applicationRevision = &model.ApplicationRevision{
+		AppPrimaryKey: appPrimaryKey,
+		Version:       version,
+	}
+	if err := w.ds.Get(ctx, applicationRevision); err != nil {
+		return err
+	}
+	if terminated {
+		applicationRevision.Status = model.DeployEventTerminated
+	} else {
+		applicationRevision.Status = model.DeployEventComplete
+	}
+
+	if err := w.ds.Put(ctx, applicationRevision); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (w *workflowUsecaseImpl) createWorkflowRecord(ctx context.Context, app *v1beta1.Application, revisionName string) error {
 	if app.Annotations == nil || app.Annotations[oam.AnnotationWorkflowName] == "" {
 		return fmt.Errorf("missing workflow name")
 	}
@@ -274,8 +362,8 @@ func (w *workflowUsecaseImpl) createWorkflowRecord(ctx context.Context, revision
 	return w.ds.Add(ctx, &model.WorkflowRecord{
 		WorkflowPrimaryKey: app.Annotations[oam.AnnotationWorkflowName],
 		AppPrimaryKey:      app.Name,
-		Name:               strings.TrimPrefix(revision.Name, "record-"),
-		Namespace:          revision.Namespace,
+		Name:               strings.TrimPrefix(revisionName, "record-"),
+		Namespace:          app.Namespace,
 		StartTime:          status.StartTime.Time,
 		Suspend:            status.Suspend,
 		Terminated:         status.Terminated,

pkg/apiserver/rest/usecase/workflow_test.go

@@ -18,6 +18,7 @@ package usecase
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 
 	"github.com/google/go-cmp/cmp"
@@ -28,6 +29,8 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"sigs.k8s.io/yaml"
 
+	"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"
 	apisv1 "github.com/oam-dev/kubevela/pkg/apiserver/rest/apis/v1"
 )
@@ -37,7 +40,7 @@ var _ = Describe("Test workflow usecase functions", func() {
 		workflowUsecase *workflowUsecaseImpl
 	)
 	BeforeEach(func() {
-		workflowUsecase = &workflowUsecaseImpl{ds: ds}
+		workflowUsecase = &workflowUsecaseImpl{ds: ds, kubeClient: k8sClient}
 	})
 	It("Test CreateWorkflow function", func() {
 		req := apisv1.CreateWorkflowRequest{
@@ -75,16 +78,11 @@ var _ = Describe("Test workflow usecase functions", func() {
 		By("create some controller revisions to test list workflow records")
 		raw, err := yaml.YAMLToJSON([]byte(yamlStr))
 		Expect(err).Should(BeNil())
+		app := &v1beta1.Application{}
+		err = json.Unmarshal(raw, app)
+		Expect(err).Should(BeNil())
 		for i := 0; i < 3; i++ {
-			err := workflowUsecase.createWorkflowRecord(context.TODO(), &appsv1.ControllerRevision{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      fmt.Sprintf("record-test-%v", i),
-					Namespace: "default",
-				},
-				Data: runtime.RawExtension{
-					Raw: raw,
-				},
-			})
+			err := workflowUsecase.createWorkflowRecord(context.TODO(), app, fmt.Sprintf("record-test-%v", i))
 			Expect(err).Should(BeNil())
 		}
 
@@ -97,15 +95,10 @@ var _ = Describe("Test workflow usecase functions", func() {
 		By("create one controller revision to test detail workflow record")
 		raw, err := yaml.YAMLToJSON([]byte(yamlStr))
 		Expect(err).Should(BeNil())
-		err = workflowUsecase.createWorkflowRecord(context.TODO(), &appsv1.ControllerRevision{
-			ObjectMeta: metav1.ObjectMeta{
-				Name:      "record-test-123",
-				Namespace: "default",
-			},
-			Data: runtime.RawExtension{
-				Raw: raw,
-			},
-		})
+		app := &v1beta1.Application{}
+		err = json.Unmarshal(raw, app)
+		Expect(err).Should(BeNil())
+		err = workflowUsecase.createWorkflowRecord(context.TODO(), app, "record-test-123")
 		Expect(err).Should(BeNil())
 
 		var deployEvent = &model.ApplicationRevision{
@@ -126,6 +119,50 @@ var _ = Describe("Test workflow usecase functions", func() {
 		Expect(detail.WorkflowRecord.Name).Should(Equal("test-123"))
 		Expect(detail.DeployUser).Should(Equal("test-user"))
 	})
+
+	It("Test SyncWorkflowRecord function", func() {
+		By("create one controller revision to test sync workflow record")
+		ctx := context.Background()
+		raw, err := yaml.YAMLToJSON([]byte(yamlStr))
+		Expect(err).Should(BeNil())
+		cr := &appsv1.ControllerRevision{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "record-test-1234",
+				Namespace: "default",
+				Labels:    map[string]string{"vela.io/wf-revision": "1234"},
+			},
+			Data: runtime.RawExtension{Raw: raw},
+		}
+		err = workflowUsecase.kubeClient.Create(ctx, cr)
+		Expect(err).Should(BeNil())
+
+		By("create one deploy event to test sync workflow record")
+		var deployEvent = &model.ApplicationRevision{
+			AppPrimaryKey: "test",
+			Version:       "1234",
+			Status:        model.DeployEventInit,
+			DeployUser:    "test-user",
+			WorkflowName:  "test-workflow-name",
+		}
+
+		err = workflowUsecase.createTestApplicationRevision(context.TODO(), deployEvent)
+		Expect(err).Should(BeNil())
+
+		err = workflowUsecase.SyncWorkflowRecord(ctx)
+		Expect(err).Should(BeNil())
+
+		By("check the record")
+		app := &v1beta1.Application{}
+		err = json.Unmarshal(raw, app)
+		Expect(err).Should(BeNil())
+		err = workflowUsecase.createWorkflowRecord(context.TODO(), app, "test-1234")
+		Expect(err).Should(Equal(datastore.ErrRecordExist))
+
+		By("check the deploy event")
+		err = workflowUsecase.ds.Get(ctx, deployEvent)
+		Expect(err).Should(BeNil())
+		Expect(deployEvent.Status).Should(Equal(model.DeployEventComplete))
+	})
 })
 
 var yamlStr = `apiVersion: core.oam.dev/v1beta1

pkg/workflow/workflow.go

@@ -30,6 +30,7 @@ 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"
+	"github.com/oam-dev/kubevela/pkg/oam"
 	"github.com/oam-dev/kubevela/pkg/oam/util"
 	wfContext "github.com/oam-dev/kubevela/pkg/workflow/context"
 	"github.com/oam-dev/kubevela/pkg/workflow/recorder"
@@ -131,6 +132,12 @@ func (w *workflow) ExecuteSteps(ctx context.Context, appRev *oamcore.Application
 
 // Trace record the workflow execute history.
 func (w *workflow) Trace() error {
+	// add annotation for apiserver sync
+	if w.app.Annotations == nil {
+		w.app.Annotations = make(map[string]string)
+	}
+	w.app.Annotations[oam.AnnotationWorkflowName] = w.app.Name
+
 	data, err := json.Marshal(w.app)
 	if err != nil {
 		return err