diff --git a/cmd/apiserver/main.go b/cmd/apiserver/main.go index 4e9714ddf..6b5f253cf 100644 --- a/cmd/apiserver/main.go +++ b/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) diff --git a/go.mod b/go.mod index 61d4e26d7..41b2d9728 100644 --- a/go.mod +++ b/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 diff --git a/pkg/apiserver/model/application.go b/pkg/apiserver/model/application.go index bf37fadfd..4a9d88af9 100644 --- a/pkg/apiserver/model/application.go +++ b/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 diff --git a/pkg/apiserver/rest/rest_server.go b/pkg/apiserver/rest/rest_server.go index 8ee324af0..2c74cc86d 100644 --- a/pkg/apiserver/rest/rest_server.go +++ b/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) diff --git a/pkg/apiserver/rest/usecase/workflow.go b/pkg/apiserver/rest/usecase/workflow.go index c411ffff2..3a876e348 100644 --- a/pkg/apiserver/rest/usecase/workflow.go +++ b/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, diff --git a/pkg/apiserver/rest/usecase/workflow_test.go b/pkg/apiserver/rest/usecase/workflow_test.go index 1137ad143..08e645c3f 100644 --- a/pkg/apiserver/rest/usecase/workflow_test.go +++ b/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 diff --git a/pkg/workflow/workflow.go b/pkg/workflow/workflow.go index 2b862591d..fa83b2f7c 100644 --- a/pkg/workflow/workflow.go +++ b/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