From 96ece000dc5eefe9f7793c8957af5e6143bb49aa Mon Sep 17 00:00:00 2001 From: barnettZQG Date: Fri, 22 Jul 2022 16:14:38 +0800 Subject: [PATCH] Feat: support to sync the revision and the workflow status (#4419) * Feat: support to sync the revision and the workflow status Signed-off-by: barnettZQG * Fix: pass the unit test and e2e test Signed-off-by: barnettZQG * Fix: e2e test case Signed-off-by: barnettZQG * Fix: update the component pod view Signed-off-by: barnettZQG * Fix: the pod struct does not match Signed-off-by: barnettZQG * Fix: optimize the e2e test case Signed-off-by: barnettZQG --- pkg/apiserver/domain/model/application.go | 18 +-- pkg/apiserver/domain/model/whole.go | 2 + pkg/apiserver/domain/model/workflow.go | 7 +- pkg/apiserver/domain/service/application.go | 77 ++++++---- pkg/apiserver/domain/service/image.go | 7 +- pkg/apiserver/domain/service/workflow.go | 133 ++++++++++++---- pkg/apiserver/domain/service/workflow_test.go | 48 +++--- pkg/apiserver/event/event.go | 2 +- pkg/apiserver/event/sync/cache.go | 5 +- pkg/apiserver/event/sync/convert.go | 11 +- pkg/apiserver/event/sync/convert/convert.go | 61 ++++++++ pkg/apiserver/event/sync/cr2ux.go | 10 ++ pkg/apiserver/event/sync/store.go | 32 ++++ pkg/apiserver/event/sync/worker.go | 24 +-- pkg/apiserver/event/sync/worker_test.go | 2 +- .../interfaces/api/assembler/v1/dto2do.go | 4 +- .../application_sync_test.go | 99 ++++++++++++ test/e2e-apiserver-test/suite_test.go | 36 +++-- .../testdata/component-pod-view.yaml | 144 ++++++++++++------ .../testdata/example-app.yaml | 86 ++++------- test/e2e-apiserver-test/velaql_test.go | 81 +++++----- 21 files changed, 633 insertions(+), 256 deletions(-) create mode 100644 test/e2e-apiserver-test/application_sync_test.go diff --git a/pkg/apiserver/domain/model/application.go b/pkg/apiserver/domain/model/application.go index bb9b2edfa..4d37a7d4d 100644 --- a/pkg/apiserver/domain/model/application.go +++ b/pkg/apiserver/domain/model/application.go @@ -18,7 +18,6 @@ package model import ( "fmt" - "strings" "time" "github.com/oam-dev/kubevela/apis/core.oam.dev/common" @@ -69,18 +68,6 @@ func (a *Application) Index() map[string]string { return index } -// GetAppNameForSynced will trim namespace suffix for synced CR -func (a *Application) GetAppNameForSynced() string { - if a.Labels == nil { - return a.Name - } - namespace := a.Labels[LabelSyncNamespace] - if namespace == "" { - return a.Name - } - return strings.TrimSuffix(a.Name, "-"+namespace) -} - // GetAppNamespaceForSynced will return the namespace of synced CR func (a *Application) GetAppNamespaceForSynced() string { if a.Labels == nil { @@ -267,6 +254,9 @@ type ApplicationRevision struct { // ApplyAppConfig Stores the application configuration during the current deploy. ApplyAppConfig string `json:"applyAppConfig,omitempty"` + // RevisionCRName This is associated with the application revision in the cluster. + RevisionCRName string `json:"revisionCRName"` + // Deploy event status Status string `json:"status"` Reason string `json:"reason"` @@ -276,7 +266,7 @@ type ApplicationRevision struct { // Information that users can note. Note string `json:"note"` - // TriggerType the event trigger source, Web or API + // TriggerType the event trigger source, Web、API、SyncFromCR TriggerType string `json:"triggerType"` // WorkflowName deploy controller by workflow diff --git a/pkg/apiserver/domain/model/whole.go b/pkg/apiserver/domain/model/whole.go index f0184a5e5..839784892 100644 --- a/pkg/apiserver/domain/model/whole.go +++ b/pkg/apiserver/domain/model/whole.go @@ -65,6 +65,8 @@ type DataStoreApp struct { Policies []*ApplicationPolicy Workflow *Workflow Targets []*Target + Record *WorkflowRecord + Revision *ApplicationRevision } const ( diff --git a/pkg/apiserver/domain/model/workflow.go b/pkg/apiserver/domain/model/workflow.go index 2e0b8f0e2..e37673572 100644 --- a/pkg/apiserver/domain/model/workflow.go +++ b/pkg/apiserver/domain/model/workflow.go @@ -99,9 +99,10 @@ func (w *Workflow) Index() map[string]string { // WorkflowRecord is the workflow record database model type WorkflowRecord struct { BaseModel - WorkflowName string `json:"workflowName"` - WorkflowAlias string `json:"workflowAlias"` - AppPrimaryKey string `json:"appPrimaryKey"` + WorkflowName string `json:"workflowName"` + WorkflowAlias string `json:"workflowAlias"` + AppPrimaryKey string `json:"appPrimaryKey"` + // RevisionPrimaryKey: should be assigned the version(PublishVersion) RevisionPrimaryKey string `json:"revisionPrimaryKey"` Name string `json:"name"` Namespace string `json:"namespace"` diff --git a/pkg/apiserver/domain/service/application.go b/pkg/apiserver/domain/service/application.go index 00c585eb0..4c814453f 100644 --- a/pkg/apiserver/domain/service/application.go +++ b/pkg/apiserver/domain/service/application.go @@ -30,8 +30,6 @@ import ( corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/labels" - "k8s.io/apimachinery/pkg/selection" "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/rest" "k8s.io/klog/v2" @@ -264,7 +262,6 @@ func (c *applicationServiceImpl) DetailApplication(ctx context.Context, app *mod for _, e := range envBindings { envBindingNames = append(envBindingNames, e.Name) } - var detail = &apisv1.DetailApplicationResponse{ ApplicationBase: *base, Policies: policyNames, @@ -283,7 +280,11 @@ func (c *applicationServiceImpl) GetApplicationStatus(ctx context.Context, appmo if err != nil { return nil, err } - err = c.KubeClient.Get(ctx, types.NamespacedName{Namespace: env.Namespace, Name: appmodel.GetAppNameForSynced()}, &app) + envBinding, err := c.EnvBindingService.GetEnvBinding(ctx, appmodel, envName) + if err != nil { + return nil, err + } + err = c.KubeClient.Get(ctx, types.NamespacedName{Namespace: env.Namespace, Name: envBinding.AppDeployName}, &app) if err != nil { if apierrors.IsNotFound(err) { return nil, nil @@ -303,7 +304,11 @@ func (c *applicationServiceImpl) GetApplicationCRInEnv(ctx context.Context, appm if err != nil { return nil, err } - err = c.KubeClient.Get(ctx, types.NamespacedName{Namespace: env.Namespace, Name: appmodel.GetAppNameForSynced()}, &app) + envBinding, err := c.EnvBindingService.GetEnvBinding(ctx, appmodel, envName) + if err != nil { + return nil, err + } + err = c.KubeClient.Get(ctx, types.NamespacedName{Namespace: env.Namespace, Name: envBinding.AppDeployName}, &app) if err != nil { if apierrors.IsNotFound(err) { return nil, nil @@ -314,35 +319,23 @@ func (c *applicationServiceImpl) GetApplicationCRInEnv(ctx context.Context, appm } // GetApplicationCR get application CR in cluster -func (c *applicationServiceImpl) GetApplicationCR(ctx context.Context, appModel *model.Application) (*v1beta1.ApplicationList, error) { - var apps v1beta1.ApplicationList - if appModel.IsSynced() { +func (c *applicationServiceImpl) GetApplicationCR(ctx context.Context, appModel *model.Application) ([]v1beta1.Application, error) { + var apps []v1beta1.Application + envbindings, err := c.EnvBindingService.GetEnvBindings(ctx, appModel) + if err != nil { + return nil, err + } + for _, env := range envbindings { var app v1beta1.Application - err := c.KubeClient.Get(ctx, types.NamespacedName{Namespace: appModel.GetAppNamespaceForSynced(), Name: appModel.GetAppNameForSynced()}, &app) + err := c.KubeClient.Get(ctx, types.NamespacedName{Namespace: env.AppDeployNamespace, Name: env.AppDeployName}, &app) if err != nil && !apierrors.IsNotFound(err) { return nil, err } if err == nil { - apps.Items = append(apps.Items, app) - return &apps, nil + apps = append(apps, app) } } - selector := labels.NewSelector() - re, err := labels.NewRequirement(oam.AnnotationAppName, selection.Equals, []string{appModel.GetAppNameForSynced()}) - if err != nil { - return nil, err - } - selector = selector.Add(*re) - err = c.KubeClient.List(ctx, &apps, &client.ListOptions{ - LabelSelector: selector, - }) - if err != nil { - if apierrors.IsNotFound(err) { - return &apps, nil - } - return nil, err - } - return &apps, nil + return apps, nil } // PublishApplicationTemplate publish app template @@ -512,6 +505,21 @@ func (c *applicationServiceImpl) UpdateApplication(ctx context.Context, app *mod } app.Alias = req.Alias app.Description = req.Description + + // Some built-in labels can not be updated + if app.Labels != nil && req.Labels != nil { + if _, exist := app.Labels[model.LabelSyncNamespace]; exist { + req.Labels[model.LabelSyncNamespace] = app.Labels[model.LabelSyncNamespace] + } + + if _, exist := app.Labels[model.LabelSourceOfTruth]; exist { + req.Labels[model.LabelSourceOfTruth] = app.Labels[model.LabelSourceOfTruth] + } + + if _, exist := app.Labels[model.LabelSyncGeneration]; exist { + req.Labels[model.LabelSyncGeneration] = app.Labels[model.LabelSyncGeneration] + } + } app.Labels = req.Labels app.Icon = req.Icon if err := c.Store.Put(ctx, app); err != nil { @@ -541,7 +549,6 @@ func (c *applicationServiceImpl) ListRecords(ctx context.Context, appName string return nil, err } } - resp := &apisv1.ListWorkflowRecordsResponse{ Records: []apisv1.WorkflowRecord{}, } @@ -583,7 +590,6 @@ func (c *applicationServiceImpl) ListComponents(ctx context.Context, app *model. } // DetailComponent detail app component -// TODO: Add status data about the component. func (c *applicationServiceImpl) DetailComponent(ctx context.Context, app *model.Application, compName string) (*apisv1.DetailComponentResponse, error) { var component = model.ApplicationComponent{ AppPrimaryKey: app.PrimaryKey(), @@ -618,7 +624,6 @@ func (c *applicationServiceImpl) ListPolicies(ctx context.Context, app *model.Ap } // DetailPolicy detail app policy -// TODO: Add status data about the policy. func (c *applicationServiceImpl) DetailPolicy(ctx context.Context, app *model.Application, policyName string) (*apisv1.DetailPolicyResponse, error) { var policy = model.ApplicationPolicy{ AppPrimaryKey: app.PrimaryKey(), @@ -704,6 +709,7 @@ func (c *applicationServiceImpl) Deploy(ctx context.Context, app *model.Applicat var appRevision = &model.ApplicationRevision{ AppPrimaryKey: app.PrimaryKey(), Version: version, + RevisionCRName: version, ApplyAppConfig: string(configByte), Status: model.RevisionStatusInit, DeployUser: userName, @@ -750,6 +756,15 @@ func (c *applicationServiceImpl) Deploy(ctx context.Context, app *model.Applicat log.Logger.Warnf("update app revision failure %s", err.Error()) } + // step7: change the source of trust + if app.Labels == nil { + app.Labels = make(map[string]string) + } + app.Labels[model.LabelSourceOfTruth] = model.FromUX + if err := c.Store.Put(ctx, app); err != nil { + log.Logger.Warnf("failed to update app %s", err.Error()) + } + return &apisv1.ApplicationDeployResponse{ ApplicationRevisionBase: c.convertRevisionModelToBase(ctx, appRevision), }, nil @@ -957,7 +972,7 @@ func (c *applicationServiceImpl) DeleteApplication(ctx context.Context, app *mod if err != nil { return err } - if len(crs.Items) > 0 { + if len(crs) > 0 { return bcode.ErrApplicationRefusedDelete } // query all components to deleted diff --git a/pkg/apiserver/domain/service/image.go b/pkg/apiserver/domain/service/image.go index b26e4f87b..fbf892623 100644 --- a/pkg/apiserver/domain/service/image.go +++ b/pkg/apiserver/domain/service/image.go @@ -40,6 +40,9 @@ import ( "github.com/oam-dev/kubevela/pkg/apiserver/utils/log" ) +// True - +const True = "true" + // NewImageService create a image service instance func NewImageService() ImageService { return &imageImpl{} @@ -148,8 +151,8 @@ func (i *imageImpl) GetImageInfo(ctx context.Context, project, secretName, image func getAccountFromSecret(secret corev1.Secret, registryDomain string) (insecure, useHTTP bool, username, password string) { if secret.Data != nil { // If users use the self-signed certificate, enable the insecure-skip-verify - insecure = string(secret.Data["insecure-skip-verify"]) == "true" - useHTTP = string(secret.Data["protocol-use-http"]) == "true" + insecure = string(secret.Data["insecure-skip-verify"]) == True + useHTTP = string(secret.Data["protocol-use-http"]) == True conf := secret.Data[".dockerconfigjson"] if len(conf) > 0 { var authConfig map[string]map[string]map[string]string diff --git a/pkg/apiserver/domain/service/workflow.go b/pkg/apiserver/domain/service/workflow.go index 36414296b..a55bfce51 100644 --- a/pkg/apiserver/domain/service/workflow.go +++ b/pkg/apiserver/domain/service/workflow.go @@ -21,9 +21,10 @@ import ( "errors" "fmt" "strconv" + "strings" "helm.sh/helm/v3/pkg/time" - appsv1 "k8s.io/api/apps/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/types" "k8s.io/klog/v2" "sigs.k8s.io/controller-runtime/pkg/client" @@ -40,7 +41,6 @@ import ( "github.com/oam-dev/kubevela/pkg/apiserver/utils/bcode" "github.com/oam-dev/kubevela/pkg/apiserver/utils/log" "github.com/oam-dev/kubevela/pkg/oam" - "github.com/oam-dev/kubevela/pkg/oam/util" pkgUtils "github.com/oam-dev/kubevela/pkg/utils" "github.com/oam-dev/kubevela/pkg/utils/apply" wfTypes "github.com/oam-dev/kubevela/pkg/workflow/types" @@ -358,45 +358,79 @@ func (w *workflowServiceImpl) SyncWorkflowRecord(ctx context.Context) error { Name: appName, Namespace: record.Namespace, }, app); err != nil { + if apierrors.IsNotFound(err) { + if err := w.setRecordToTerminated(ctx, record.AppPrimaryKey, record.Name); err != nil { + log.Logger.Errorf("failed to set the record status to terminated %s", err.Error()) + } + continue + } klog.ErrorS(err, "failed to get app", "oam app name", appName, "workflow name", record.WorkflowName, "record name", record.Name) continue } + // there is a ":" in the default app revision + recordName := strings.Replace(app.Status.Workflow.AppRevision, ":", "-", 1) + // try to sync the status from the running application - if app.Annotations != nil && app.Status.Workflow != nil && app.Status.Workflow.AppRevision == record.Name { - if err := w.syncWorkflowStatus(ctx, app, record.Name, app.Name); err != nil { + if app.Annotations != nil && app.Status.Workflow != nil && recordName == record.Name { + if err := w.syncWorkflowStatus(ctx, record.AppPrimaryKey, app, record.Name, app.Name); err != nil { klog.ErrorS(err, "failed to sync workflow status", "oam app name", appName, "workflow name", record.WorkflowName, "record name", record.Name) } continue } // try to sync the status from the controller revision - cr := &appsv1.ControllerRevision{} - if err := w.KubeClient.Get(ctx, types.NamespacedName{ - Name: fmt.Sprintf("record-%s-%s", appName, record.Name), - Namespace: record.Namespace, - }, cr); err != nil { - klog.ErrorS(err, "failed to get controller revision", "oam app name", appName, "workflow name", record.WorkflowName, "record name", record.Name) - continue - } - appInRevision, err := util.RawExtension2Application(cr.Data) - if err != nil { - klog.ErrorS(err, "failed to get app data in controller revision", "controller revision name", cr.Name, "app name", appName, "workflow name", record.WorkflowName, "record name", record.Name) - continue - } - if err := w.syncWorkflowStatus(ctx, appInRevision, record.Name, cr.Name); err != nil { - klog.ErrorS(err, "failed to sync workflow status", "oam app name", appName, "workflow name", record.WorkflowName, "record name", record.Name) + var revision = &model.ApplicationRevision{AppPrimaryKey: record.AppPrimaryKey, Version: record.RevisionPrimaryKey} + if err := w.Store.Get(ctx, revision); err != nil { + if errors.Is(err, datastore.ErrRecordNotExist) { + // If the application revision is not exist, the record do not need be synced + var record = &model.WorkflowRecord{ + AppPrimaryKey: record.AppPrimaryKey, + Name: recordName, + } + if err := w.Store.Get(ctx, record); err == nil { + record.Finished = "true" + record.Status = model.RevisionStatusFail + err := w.Store.Put(ctx, record) + if err != nil { + log.Logger.Errorf("failed to set the workflow status is failure %s", err.Error()) + } + continue + } + } + log.Logger.Errorf("failed to get the application revision from database %s", err.Error()) continue } + var appRevision v1beta1.ApplicationRevision + if err := w.KubeClient.Get(ctx, types.NamespacedName{Namespace: app.Namespace, Name: revision.RevisionCRName}, &appRevision); err != nil { + if apierrors.IsNotFound(err) { + if err := w.setRecordToTerminated(ctx, record.AppPrimaryKey, record.Name); err != nil { + log.Logger.Errorf("failed to set the record status to terminated %s", err.Error()) + } + continue + } + log.Logger.Warnf("failed to get the application revision %s", err.Error()) + continue + } + + appRevision.Spec.Application.Status.Workflow = appRevision.Status.Workflow + if !appRevision.Spec.Application.Status.Workflow.Finished { + appRevision.Spec.Application.Status.Workflow.Finished = true + appRevision.Spec.Application.Status.Workflow.Terminated = true + } + if err := w.syncWorkflowStatus(ctx, record.AppPrimaryKey, &appRevision.Spec.Application, record.Name, revision.RevisionCRName); err != nil { + klog.ErrorS(err, "failed to sync workflow status", "oam app name", appName, "workflow name", record.WorkflowName, "record name", record.Name) + continue + } } return nil } -func (w *workflowServiceImpl) syncWorkflowStatus(ctx context.Context, app *v1beta1.Application, recordName, source string) error { +func (w *workflowServiceImpl) setRecordToTerminated(ctx context.Context, appPrimaryKey, recordName string) error { var record = &model.WorkflowRecord{ - AppPrimaryKey: app.Annotations[oam.AnnotationAppName], + AppPrimaryKey: appPrimaryKey, Name: recordName, } if err := w.Store.Get(ctx, record); err != nil { @@ -405,11 +439,40 @@ func (w *workflowServiceImpl) syncWorkflowStatus(ctx context.Context, app *v1bet } return err } - var revision = &model.ApplicationRevision{ - AppPrimaryKey: app.Annotations[oam.AnnotationAppName], - Version: record.RevisionPrimaryKey, + var revision = &model.ApplicationRevision{AppPrimaryKey: appPrimaryKey, Version: record.RevisionPrimaryKey} + if err := w.Store.Get(ctx, revision); err != nil { + if errors.Is(err, datastore.ErrRecordNotExist) { + return bcode.ErrApplicationRevisionNotExist + } + return err + } + record.Status = model.RevisionStatusTerminated + record.Finished = "true" + + revision.Status = model.RevisionStatusTerminated + + if err := w.Store.Put(ctx, record); err != nil { + return err } + if err := w.Store.Put(ctx, revision); err != nil { + return err + } + return nil +} + +func (w *workflowServiceImpl) syncWorkflowStatus(ctx context.Context, appPrimaryKey string, app *v1beta1.Application, recordName, source string) error { + var record = &model.WorkflowRecord{ + AppPrimaryKey: appPrimaryKey, + Name: recordName, + } + if err := w.Store.Get(ctx, record); err != nil { + if errors.Is(err, datastore.ErrRecordNotExist) { + return bcode.ErrWorkflowRecordNotExist + } + return err + } + var revision = &model.ApplicationRevision{AppPrimaryKey: appPrimaryKey, Version: record.RevisionPrimaryKey} if err := w.Store.Get(ctx, revision); err != nil { if errors.Is(err, datastore.ErrRecordNotExist) { return bcode.ErrApplicationRevisionNotExist @@ -420,10 +483,10 @@ func (w *workflowServiceImpl) syncWorkflowStatus(ctx context.Context, app *v1bet if app.Status.Workflow != nil { status := app.Status.Workflow summaryStatus := model.RevisionStatusRunning - if status.Finished { + switch { + case status.Finished: summaryStatus = model.RevisionStatusComplete - } - if status.Terminated { + case status.Terminated: summaryStatus = model.RevisionStatusTerminated } @@ -575,7 +638,8 @@ func (w *workflowServiceImpl) ResumeRecord(ctx context.Context, appModel *model. if err := ResumeWorkflow(ctx, w.KubeClient, oamApp); err != nil { return err } - if err := w.syncWorkflowStatus(ctx, oamApp, recordName, oamApp.Name); err != nil { + + if err := w.syncWorkflowStatus(ctx, appModel.PrimaryKey(), oamApp, recordName, oamApp.Name); err != nil { return err } @@ -587,11 +651,10 @@ func (w *workflowServiceImpl) TerminateRecord(ctx context.Context, appModel *mod if err != nil { return err } - if err := TerminateWorkflow(ctx, w.KubeClient, oamApp); err != nil { return err } - if err := w.syncWorkflowStatus(ctx, oamApp, recordName, oamApp.Name); err != nil { + if err := w.syncWorkflowStatus(ctx, appModel.PrimaryKey(), oamApp, recordName, oamApp.Name); err != nil { return err } @@ -752,7 +815,15 @@ func (w *workflowServiceImpl) checkRecordRunning(ctx context.Context, appModel * if err != nil { return nil, err } - if err := w.KubeClient.Get(ctx, types.NamespacedName{Name: appModel.Name, Namespace: env.Namespace}, oamApp); err != nil { + envbinding, err := w.EnvBindingService.GetEnvBinding(ctx, appModel, envName) + if err != nil { + return nil, err + } + name := envbinding.AppDeployName + if name == "" { + name = appModel.Name + } + if err := w.KubeClient.Get(ctx, types.NamespacedName{Name: name, Namespace: env.Namespace}, oamApp); err != nil { return nil, err } if oamApp.Status.Workflow != nil && !oamApp.Status.Workflow.Suspend && !oamApp.Status.Workflow.Terminated && !oamApp.Status.Workflow.Finished { diff --git a/pkg/apiserver/domain/service/workflow_test.go b/pkg/apiserver/domain/service/workflow_test.go index c0222bf4d..7532a2372 100644 --- a/pkg/apiserver/domain/service/workflow_test.go +++ b/pkg/apiserver/domain/service/workflow_test.go @@ -26,9 +26,7 @@ import ( "github.com/google/go-cmp/cmp" . "github.com/onsi/ginkgo" . "github.com/onsi/gomega" - appsv1 "k8s.io/api/apps/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/yaml" @@ -48,6 +46,7 @@ var _ = Describe("Test workflow service functions", func() { appService *applicationServiceImpl projectService *projectServiceImpl envService *envServiceImpl + envBinding *envBindingServiceImpl testProject = "workflow-project" ds datastore.DataStore ) @@ -60,7 +59,7 @@ var _ = Describe("Test workflow service functions", func() { rbacService := &rbacServiceImpl{Store: ds} projectService = &projectServiceImpl{Store: ds, RbacService: rbacService} envService = &envServiceImpl{Store: ds, KubeClient: k8sClient, ProjectService: projectService} - envBinding := &envBindingServiceImpl{ + envBinding = &envBindingServiceImpl{ Store: ds, WorkflowService: workflowService, EnvService: envService, @@ -278,28 +277,36 @@ var _ = Describe("Test workflow service functions", func() { By("create another revision to test sync workflow record") var anotherRevision = &model.ApplicationRevision{ - AppPrimaryKey: appName, - Version: "1111", - Status: model.RevisionStatusInit, - DeployUser: "test-user", - WorkflowName: "test-workflow-2", + AppPrimaryKey: appName, + Version: "1111", + Status: model.RevisionStatusInit, + DeployUser: "test-user", + WorkflowName: "test-workflow-2", + RevisionCRName: "1111-v1", } err = workflowService.createTestApplicationRevision(context.TODO(), anotherRevision) Expect(err).Should(BeNil()) - By("create one controller revision to test sync workflow record") + By("create one application revision to test sync workflow record") + + appWithRevision := &v1beta1.Application{} + err = json.Unmarshal(raw, appWithRevision) Expect(err).Should(BeNil()) - cr := &appsv1.ControllerRevision{ + var appRevision = &v1beta1.ApplicationRevision{ ObjectMeta: metav1.ObjectMeta{ - Name: "record-app-workflow-test-workflow-2-111", + Name: "1111-v1", Namespace: "default", Labels: map[string]string{"vela.io/wf-revision": "test-workflow-2-111"}, }, - Data: runtime.RawExtension{Raw: raw}, + Spec: v1beta1.ApplicationRevisionSpec{ + Application: *appWithRevision, + }, } - err = workflowService.KubeClient.Create(ctx, cr) + err = workflowService.KubeClient.Create(ctx, appRevision) + Expect(err).Should(BeNil()) + appRevision.Status.Workflow = appWithRevision.Status.Workflow + err = workflowService.KubeClient.Status().Update(ctx, appRevision) Expect(err).Should(BeNil()) - err = workflowService.SyncWorkflowRecord(ctx) Expect(err).Should(BeNil()) @@ -348,6 +355,8 @@ var _ = Describe("Test workflow service functions", func() { _, err := envService.CreateEnv(context.TODO(), apisv1.CreateEnvRequest{Name: "resume"}) Expect(err).Should(BeNil()) + _, err = envBinding.CreateEnvBinding(context.TODO(), &model.Application{Name: appName}, apisv1.CreateApplicationEnvbindingRequest{EnvBinding: apisv1.EnvBinding{Name: "resume"}}) + Expect(err).Should(BeNil()) ResumeWorkflow := "resume-workflow" req := apisv1.CreateWorkflowRequest{ Name: ResumeWorkflow, @@ -370,9 +379,10 @@ var _ = Describe("Test workflow service functions", func() { Expect(err).Should(BeNil()) err = workflowService.createTestApplicationRevision(ctx, &model.ApplicationRevision{ - AppPrimaryKey: appName, - Version: "revision-resume1", - Status: model.RevisionStatusRunning, + AppPrimaryKey: appName, + Version: "revision-resume1", + RevisionCRName: "revision-resume1", + Status: model.RevisionStatusRunning, }) Expect(err).Should(BeNil()) @@ -391,6 +401,8 @@ var _ = Describe("Test workflow service functions", func() { _, err := envService.CreateEnv(context.TODO(), apisv1.CreateEnvRequest{Name: "terminate"}) Expect(err).Should(BeNil()) + _, err = envBinding.CreateEnvBinding(context.TODO(), &model.Application{Name: appName}, apisv1.CreateApplicationEnvbindingRequest{EnvBinding: apisv1.EnvBinding{Name: "terminate"}}) + Expect(err).Should(BeNil()) workflowName := "terminate-workflow" req := apisv1.CreateWorkflowRequest{ Name: workflowName, @@ -433,6 +445,8 @@ var _ = Describe("Test workflow service functions", func() { ctx := context.TODO() _, err := envService.CreateEnv(context.TODO(), apisv1.CreateEnvRequest{Name: "rollback"}) Expect(err).Should(BeNil()) + _, err = envBinding.CreateEnvBinding(context.TODO(), &model.Application{Name: appName}, apisv1.CreateApplicationEnvbindingRequest{EnvBinding: apisv1.EnvBinding{Name: "rollback"}}) + Expect(err).Should(BeNil()) workflowName := "rollback-workflow" req := apisv1.CreateWorkflowRequest{ Name: workflowName, diff --git a/pkg/apiserver/event/event.go b/pkg/apiserver/event/event.go index 74c27d1cf..25f53e10e 100644 --- a/pkg/apiserver/event/event.go +++ b/pkg/apiserver/event/event.go @@ -39,7 +39,7 @@ func InitEvent(cfg config.Config) []interface{} { Duration: cfg.LeaderConfig.Duration, } application := &sync.ApplicationSync{ - Queue: workqueue.New(), + Queue: workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()), } collect := &collect.InfoCalculateCronJob{} workers = append(workers, workflow, application, collect) diff --git a/pkg/apiserver/event/sync/cache.go b/pkg/apiserver/event/sync/cache.go index de5903a32..ca9c12179 100644 --- a/pkg/apiserver/event/sync/cache.go +++ b/pkg/apiserver/event/sync/cache.go @@ -22,11 +22,10 @@ import ( "strconv" "strings" - "github.com/sirupsen/logrus" - "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1" "github.com/oam-dev/kubevela/pkg/apiserver/domain/model" "github.com/oam-dev/kubevela/pkg/apiserver/infrastructure/datastore" + "github.com/oam-dev/kubevela/pkg/apiserver/utils/log" "github.com/oam-dev/kubevela/pkg/oam" ) @@ -87,7 +86,7 @@ func (c *CR2UX) shouldSync(ctx context.Context, targetApp *v1beta1.Application, if del || err != nil { c.cache.Delete(key) } else if cd.generation == targetApp.Generation { - logrus.Infof("app %s/%s with generation(%v) hasn't updated, ignore the sync event..", targetApp.Name, targetApp.Namespace, targetApp.Generation) + log.Logger.Infof("app %s/%s with generation(%v) hasn't updated, ignore the sync event..", targetApp.Name, targetApp.Namespace, targetApp.Generation) return false } } diff --git a/pkg/apiserver/event/sync/convert.go b/pkg/apiserver/event/sync/convert.go index 749600ef7..d9b5d7c8f 100644 --- a/pkg/apiserver/event/sync/convert.go +++ b/pkg/apiserver/event/sync/convert.go @@ -123,7 +123,7 @@ func (c *CR2UX) ConvertApp2DatastoreApp(ctx context.Context, targetApp *v1beta1. dsApp.Eb = &model.EnvBinding{ AppPrimaryKey: appMeta.PrimaryKey(), Name: dsApp.Env.Name, - AppDeployName: appMeta.GetAppNameForSynced(), + AppDeployName: targetApp.Name, } for i := range dsApp.Targets { @@ -167,5 +167,14 @@ func (c *CR2UX) ConvertApp2DatastoreApp(ctx context.Context, targetApp *v1beta1. plcModel.EnvName = dsApp.Env.Name dsApp.Policies = append(dsApp.Policies, &plcModel) } + + // 7. convert the revision + if revision := convert.FromCRApplicationRevision(ctx, cli, targetApp, *dsApp.Workflow, dsApp.Env.Name); revision != nil { + dsApp.Revision = revision + } + // 8. convert the workflow record + if record := convert.FromCRWorkflowRecord(targetApp, *dsApp.Workflow, dsApp.Revision); record != nil { + dsApp.Record = record + } return dsApp, nil } diff --git a/pkg/apiserver/event/sync/convert/convert.go b/pkg/apiserver/event/sync/convert/convert.go index bae930e75..7409e4329 100644 --- a/pkg/apiserver/event/sync/convert/convert.go +++ b/pkg/apiserver/event/sync/convert/convert.go @@ -19,10 +19,12 @@ package convert import ( "context" "fmt" + "strings" "time" "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/yaml" "github.com/oam-dev/kubevela/apis/core.oam.dev/common" "github.com/oam-dev/kubevela/apis/core.oam.dev/v1alpha1" @@ -85,6 +87,7 @@ func FromCRPolicy(appPrimaryKey string, policyCR v1beta1.AppPolicy, creator stri // FromCRWorkflow converts Application CR Workflow section into velaux data store workflow func FromCRWorkflow(ctx context.Context, cli client.Client, appPrimaryKey string, app *v1beta1.Application) (model.Workflow, []v1beta1.WorkflowStep, error) { + var defaultWorkflow = true dataWf := model.Workflow{ AppPrimaryKey: appPrimaryKey, // every namespace has a synced env @@ -93,6 +96,7 @@ func FromCRWorkflow(ctx context.Context, cli client.Client, appPrimaryKey string Name: model.AutoGenWorkflowNamePrefix + appPrimaryKey, Alias: model.AutoGenWorkflowNamePrefix + app.Name, Description: model.AutoGenDesc, + Default: &defaultWorkflow, } if app.Spec.Workflow == nil { return dataWf, nil, nil @@ -175,3 +179,60 @@ func FromCRTargets(ctx context.Context, cli client.Client, targetApp *v1beta1.Ap } return targets, targetNames } + +// FromCRWorkflowRecord convert the workflow status to workflow record +func FromCRWorkflowRecord(app *v1beta1.Application, workflow model.Workflow, revision *model.ApplicationRevision) *model.WorkflowRecord { + if app.Status.Workflow == nil || app.Status.Workflow.AppRevision == "" || revision == nil { + return nil + } + steps := make([]model.WorkflowStepStatus, len(workflow.Steps)) + for i, step := range workflow.Steps { + steps[i] = model.WorkflowStepStatus{ + Name: step.Name, + Alias: step.Alias, + Type: step.Type, + } + } + return &model.WorkflowRecord{ + WorkflowName: workflow.Name, + WorkflowAlias: workflow.Alias, + AppPrimaryKey: workflow.AppPrimaryKey, + Name: strings.Replace(app.Status.Workflow.AppRevision, ":", "-", 1), + Namespace: app.Namespace, + Finished: model.UnFinished, + RevisionPrimaryKey: revision.Version, + Steps: steps, + Status: model.RevisionStatusRunning, + } +} + +// FromCRApplicationRevision convert the application revision to the revision in the data store +func FromCRApplicationRevision(ctx context.Context, cli client.Client, app *v1beta1.Application, workflow model.Workflow, envName string) *model.ApplicationRevision { + if app.Status.Workflow == nil || app.Status.Workflow.AppRevision == "" { + return nil + } + versions := strings.Split(app.Status.Workflow.AppRevision, ":") + versionName := versions[0] + var appRevision v1beta1.ApplicationRevision + ctxTimeout, cancel := context.WithTimeout(ctx, time.Second*20) + defer cancel() + if err := cli.Get(ctxTimeout, types.NamespacedName{Namespace: app.Namespace, Name: versionName}, &appRevision); err != nil { + log.Logger.Errorf("failed to get the application revision %s", err.Error()) + return nil + } + configByte, _ := yaml.Marshal(appRevision.Spec.Application) + return &model.ApplicationRevision{ + BaseModel: model.BaseModel{ + CreateTime: appRevision.CreationTimestamp.Time, + UpdateTime: time.Now(), + }, + AppPrimaryKey: workflow.AppPrimaryKey, + RevisionCRName: appRevision.Name, + WorkflowName: workflow.Name, + Version: versionName, + ApplyAppConfig: string(configByte), + TriggerType: "SyncFromCR", + EnvName: envName, + Status: model.RevisionStatusRunning, + } +} diff --git a/pkg/apiserver/event/sync/cr2ux.go b/pkg/apiserver/event/sync/cr2ux.go index da50ca8a9..155ac54cf 100644 --- a/pkg/apiserver/event/sync/cr2ux.go +++ b/pkg/apiserver/event/sync/cr2ux.go @@ -142,6 +142,16 @@ func (c *CR2UX) AddOrUpdate(ctx context.Context, targetApp *v1beta1.Application) return err } + if err = StoreApplicationRevision(ctx, dsApp, ds); err != nil { + log.Logger.Errorf("Store application revision to data store err %v", err) + return err + } + + if err = StoreWorkflowRecord(ctx, dsApp, ds); err != nil { + log.Logger.Errorf("Store Workflow Record 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 diff --git a/pkg/apiserver/event/sync/store.go b/pkg/apiserver/event/sync/store.go index 345c88469..00cfba5a7 100644 --- a/pkg/apiserver/event/sync/store.go +++ b/pkg/apiserver/event/sync/store.go @@ -233,6 +233,38 @@ func StoreWorkflow(ctx context.Context, dsApp *model.DataStoreApp, ds datastore. return ds.Add(ctx, dsApp.Workflow) } +// StoreWorkflowRecord will sync workflow status to datastore. +func StoreWorkflowRecord(ctx context.Context, dsApp *model.DataStoreApp, ds datastore.DataStore) error { + if dsApp.Record == nil { + return nil + } + records, err := ds.List(ctx, &model.WorkflowRecord{AppPrimaryKey: dsApp.AppMeta.Name, Name: dsApp.Record.Name}, nil) + if err == nil && len(records) > 0 { + return nil + } + if err != nil { + // other database error, return it + return err + } + return ds.Add(ctx, dsApp.Record) +} + +// StoreApplicationRevision will sync the application revision to datastore. +func StoreApplicationRevision(ctx context.Context, dsApp *model.DataStoreApp, ds datastore.DataStore) error { + if dsApp.Revision == nil { + return nil + } + err := ds.Get(ctx, &model.ApplicationRevision{AppPrimaryKey: dsApp.AppMeta.Name, Version: dsApp.Revision.Version}) + if err == nil { + return ds.Put(ctx, dsApp.Revision) + } + if !errors.Is(err, datastore.ErrRecordNotExist) { + // other database error, return it + return err + } + return ds.Add(ctx, dsApp.Revision) +} + // StoreTargets will sync targets from application CR to datastore func StoreTargets(ctx context.Context, dsApp *model.DataStoreApp, ds datastore.DataStore, targetService service.TargetService) error { for _, t := range dsApp.Targets { diff --git a/pkg/apiserver/event/sync/worker.go b/pkg/apiserver/event/sync/worker.go index ddb0c6fb2..b356582b2 100644 --- a/pkg/apiserver/event/sync/worker.go +++ b/pkg/apiserver/event/sync/worker.go @@ -28,7 +28,6 @@ import ( "k8s.io/client-go/rest" "k8s.io/client-go/tools/cache" "k8s.io/client-go/util/workqueue" - "k8s.io/klog/v2" "sigs.k8s.io/controller-runtime/pkg/client" "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1" @@ -46,7 +45,7 @@ type ApplicationSync struct { ApplicationService service.ApplicationService `inject:""` TargetService service.TargetService `inject:""` EnvService service.EnvService `inject:""` - Queue workqueue.Interface + Queue workqueue.RateLimitingInterface } // Start prepares watchers and run their controllers, then waits for process termination signals @@ -93,24 +92,31 @@ func (a *ApplicationSync) Start(ctx context.Context, errorChan chan error) { } }() + addOrUpdateHandler := func(obj interface{}) { + app := getApp(obj) + if app.DeletionTimestamp == nil { + a.Queue.Add(app) + log.Logger.Infof("watched update/add app event, namespace: %s, name: %s", app.Namespace, app.Name) + } + } + handlers := cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { - app := getApp(obj) - klog.Infof("watched add app event, namespace: %s, name: %s", app.Namespace, app.Name) - a.Queue.Add(app) + addOrUpdateHandler(obj) }, UpdateFunc: func(oldObj, obj interface{}) { - app := getApp(obj) - klog.Infof("watched update app event, namespace: %s, name: %s", app.Namespace, app.Name) - a.Queue.Add(app) + addOrUpdateHandler(obj) }, DeleteFunc: func(obj interface{}) { app := getApp(obj) - klog.Infof("watched delete app event, namespace: %s, name: %s", app.Namespace, app.Name) + log.Logger.Infof("watched delete app event, namespace: %s, name: %s", app.Namespace, app.Name) + a.Queue.Forget(app) + a.Queue.Done(app) err = cu.DeleteApp(ctx, app) if err != nil { log.Logger.Errorf("Application %-30s Deleted Sync to db err %v", color.WhiteString(app.Namespace+"/"+app.Name), err) } + log.Logger.Infof("delete the application (%s/%s) metadata successfully", app.Namespace, app.Name) }, } informer.AddEventHandler(handlers) diff --git a/pkg/apiserver/event/sync/worker_test.go b/pkg/apiserver/event/sync/worker_test.go index 68bca5a25..50fc390e4 100644 --- a/pkg/apiserver/event/sync/worker_test.go +++ b/pkg/apiserver/event/sync/worker_test.go @@ -67,7 +67,7 @@ var _ = Describe("Test Worker CR sync to datastore", func() { KubeClient: k8sClient, KubeConfig: cfg, Store: ds, - Queue: workqueue.New(), + Queue: workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()), ProjectService: crux.projectService, ApplicationService: crux.applicationService, TargetService: crux.targetService, diff --git a/pkg/apiserver/interfaces/api/assembler/v1/dto2do.go b/pkg/apiserver/interfaces/api/assembler/v1/dto2do.go index 56e4baf67..79597896f 100644 --- a/pkg/apiserver/interfaces/api/assembler/v1/dto2do.go +++ b/pkg/apiserver/interfaces/api/assembler/v1/dto2do.go @@ -28,7 +28,7 @@ func CreateEnvBindingModel(app *model.Application, req apisv1.CreateApplicationE envBinding := model.EnvBinding{ AppPrimaryKey: app.Name, Name: req.Name, - AppDeployName: app.GetAppNameForSynced(), + AppDeployName: app.Name, } return envBinding } @@ -38,7 +38,7 @@ func ConvertToEnvBindingModel(app *model.Application, envBind apisv1.EnvBinding) re := model.EnvBinding{ AppPrimaryKey: app.Name, Name: envBind.Name, - AppDeployName: app.GetAppNameForSynced(), + AppDeployName: app.Name, } return &re } diff --git a/test/e2e-apiserver-test/application_sync_test.go b/test/e2e-apiserver-test/application_sync_test.go new file mode 100644 index 000000000..c304ead16 --- /dev/null +++ b/test/e2e-apiserver-test/application_sync_test.go @@ -0,0 +1,99 @@ +/* +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 e2e_apiserver_test + +import ( + "fmt" + "time" + + "github.com/google/go-cmp/cmp" + . "github.com/onsi/ginkgo" + . "github.com/onsi/gomega" + + "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1" + apisv1 "github.com/oam-dev/kubevela/pkg/apiserver/interfaces/api/dto/v1" + "github.com/oam-dev/kubevela/pkg/utils/common" +) + +var _ = Describe("Test the application synchronizing", func() { + var appName = "test-synchronizing" + It("Test create an application", func() { + var app v1beta1.Application + Expect(common.ReadYamlToObject("./testdata/example-app.yaml", &app)).Should(BeNil()) + app.Spec.Components[0].Name = appName + app.Name = appName + req := apisv1.ApplicationRequest{ + Components: app.Spec.Components, + Policies: app.Spec.Policies, + Workflow: app.Spec.Workflow, + } + res := post(fmt.Sprintf("/v1/namespaces/%s/applications/%s", "default", appName), req) + Expect(res).ShouldNot(BeNil()) + Expect(cmp.Diff(res.StatusCode, 200)).Should(BeEmpty()) + Expect(res.Body).ShouldNot(BeNil()) + defer res.Body.Close() + }) + + It("Test get the synchronizing application", func() { + for retry := 0; retry < 5; retry++ { + // Sleep 5 seconds to wait for the sync completed + time.Sleep(time.Second * 5) + res := get(fmt.Sprintf("/applications/%s", appName)) + Expect(res).ShouldNot(BeNil()) + if res.StatusCode == 404 { + continue + } + var detail apisv1.DetailApplicationResponse + Expect(decodeResponseBody(res, &detail)).Should(Succeed()) + Expect(cmp.Diff(len(detail.Policies), 3)).Should(BeEmpty()) + Expect(cmp.Diff(len(detail.EnvBindings), 1)).Should(BeEmpty()) + Expect(cmp.Diff(detail.ResourceInfo.ComponentNum, int64(2))).Should(BeEmpty()) + break + } + }) + + It("Test get the synchronizing application revision", func() { + res := get(fmt.Sprintf("/applications/%s/revisions", appName)) + Expect(res).ShouldNot(BeNil()) + var list apisv1.ListRevisionsResponse + Expect(decodeResponseBody(res, &list)).Should(Succeed()) + Expect(cmp.Diff(len(list.Revisions), 1)).Should(BeEmpty()) + }) + + It("Test get the synchronizing workflow record", func() { + res := get(fmt.Sprintf("/applications/%s/records", appName)) + Expect(res).ShouldNot(BeNil()) + var list apisv1.ListWorkflowRecordsResponse + Expect(decodeResponseBody(res, &list)).Should(Succeed()) + Expect(cmp.Diff(len(list.Records), 1)).Should(BeEmpty()) + Expect(cmp.Diff(len(list.Records[0].Steps), 3)).Should(BeEmpty()) + }) + + It("Test delete the application", func() { + res := delete(fmt.Sprintf("/v1/namespaces/%s/applications/%s", "default", appName)) + Expect(res).ShouldNot(BeNil()) + Expect(cmp.Diff(res.StatusCode, 200)).Should(BeEmpty()) + }) + + It("Test get the application", func() { + // Sleep 5 seconds to wait for the sync completed + time.Sleep(time.Second * 5) + res := get(fmt.Sprintf("/applications/%s", appName)) + Expect(res).ShouldNot(BeNil()) + Expect(res.StatusCode).Should(Equal(404)) + }) +}) diff --git a/test/e2e-apiserver-test/suite_test.go b/test/e2e-apiserver-test/suite_test.go index 4adddc560..6581ef8d5 100644 --- a/test/e2e-apiserver-test/suite_test.go +++ b/test/e2e-apiserver-test/suite_test.go @@ -21,7 +21,9 @@ import ( "context" "encoding/json" "fmt" + "io/ioutil" "net/http" + "os" "strings" "testing" "time" @@ -34,6 +36,7 @@ import ( "github.com/oam-dev/kubevela/pkg/apiserver" "github.com/oam-dev/kubevela/pkg/apiserver/config" + "github.com/oam-dev/kubevela/pkg/apiserver/domain/service" "github.com/oam-dev/kubevela/pkg/apiserver/infrastructure/clients" "github.com/oam-dev/kubevela/pkg/apiserver/infrastructure/datastore" apisv1 "github.com/oam-dev/kubevela/pkg/apiserver/interfaces/api/dto/v1" @@ -82,9 +85,13 @@ var _ = BeforeSuite(func() { By("wait for api server to start") Eventually( func() error { + password := os.Getenv("VELA_UX_PASSWORD") + if password == "" { + password = service.InitAdminPassword + } var req = apisv1.LoginRequest{ Username: "admin", - Password: "VelaUX12345", + Password: password, } bodyByte, err := json.Marshal(req) Expect(err).Should(BeNil()) @@ -118,11 +125,13 @@ var _ = BeforeSuite(func() { var _ = AfterSuite(func() { By("tearing down the test environment") var nsList v1.NamespaceList - err := k8sClient.List(context.TODO(), &nsList) - Expect(err).ToNot(HaveOccurred()) - for _, ns := range nsList.Items { - if strings.HasPrefix(ns.Name, testNSprefix) { - _ = k8sClient.Delete(context.TODO(), &ns) + if k8sClient != nil { + err := k8sClient.List(context.TODO(), &nsList) + Expect(err).ToNot(HaveOccurred()) + for _, ns := range nsList.Items { + if strings.HasPrefix(ns.Name, testNSprefix) { + _ = k8sClient.Delete(context.TODO(), &ns) + } } } }) @@ -203,10 +212,17 @@ func decodeResponseBody(resp *http.Response, dst interface{}) error { if resp.Body == nil { return fmt.Errorf("response body is nil") } + defer resp.Body.Close() if dst != nil { - err := json.NewDecoder(resp.Body).Decode(dst) - Expect(err).Should(BeNil()) - return resp.Body.Close() + body, err := ioutil.ReadAll(resp.Body) + if err != nil { + return err + } + err = json.Unmarshal(body, dst) + if err != nil { + return err + } + return nil } - return resp.Body.Close() + return nil } diff --git a/test/e2e-apiserver-test/testdata/component-pod-view.yaml b/test/e2e-apiserver-test/testdata/component-pod-view.yaml index c6d7b2894..c6b774707 100644 --- a/test/e2e-apiserver-test/testdata/component-pod-view.yaml +++ b/test/e2e-apiserver-test/testdata/component-pod-view.yaml @@ -8,6 +8,7 @@ data: import ( "vela/ql" "vela/op" + "strings" ) parameter: { @@ -18,7 +19,51 @@ data: clusterNs?: string } - application: ql.#ListResourcesInApp & { + annotationDeployVersion: "app.oam.dev/deployVersion" + annotationPublishVersion: "app.oam.dev/publishVersion" + labelComponentName: "app.oam.dev/component" + + ignoreCollectPodKindMap: { + "ConfigMap": true + "Endpoints": true + "LimitRange": true + "Namespace": true + "Node": true + "PersistentVolumeClaim": true + "PersistentVolume": true + "ReplicationController": true + "ResourceQuota": true + "ServiceAccount": true + "Service": true + "Event": true + "Ingress": true + "StorageClass": true + "NetworkPolicy": true + "PodDisruptionBudget": true + "PodSecurityPolicy": true + "PriorityClass": true + "CustomResourceDefinition": true + "HorizontalPodAutoscaler": true + "CertificateSigningRequest": true + "ManagedCluster": true + "ManagedClusterSetBinding": true + "ManagedClusterSet": true + "ApplicationRevision": true + "ComponentDefinition": true + "DefinitionRevision": true + "EnvBinding": true + "PolicyDefinition": true + "ResourceTracker": true + "ScopeDefinition": true + "TraitDefinition": true + "WorkflowStepDefinition": true + "WorkloadDefinition": true + "GitRepository": true + "HelmRepository": true + "ComponentStatus": true + } + + resources: ql.#ListResourcesInApp & { app: { name: parameter.appName namespace: parameter.appNs @@ -36,63 +81,76 @@ data: } } - if application.err != _|_ { - status: error: application.err - } - - if application.err == _|_ { - resources: application.list - - podsMap: op.#Steps & { - for i, resource in resources { + if resources.err == _|_ { + collectedPods: op.#Steps & { + for i, resource in resources.list if ignoreCollectPodKindMap[resource.object.kind] == _|_ { "\(i)": ql.#CollectPods & { value: resource.object cluster: resource.cluster } } } - - podsWithCluster: [ for i, pods in podsMap if pods.list != null for podObj in pods.list { + podsWithCluster: [ for pods in collectedPods if pods.list != _|_ && pods.list != null for podObj in pods.list { cluster: pods.cluster obj: podObj + workload: { + apiVersion: pods.value.apiVersion + kind: pods.value.kind + name: pods.value.metadata.name + namespace: pods.value.metadata.namespace + } + if pods.value.metadata.labels[labelComponentName] != _|_ { + component: pods.value.metadata.labels[labelComponentName] + } + if pods.value.metadata.annotations[annotationPublishVersion] != _|_ { + publishVersion: pods.value.metadata.annotations[annotationPublishVersion] + } + if pods.value.metadata.annotations[annotationDeployVersion] != _|_ { + deployVersion: pods.value.metadata.annotations[annotationDeployVersion] + } }] - - podStatus: op.#Steps & { - for i, pod in podsWithCluster { - "\(i)": op.#Steps & { - name: pod.obj.metadata.name - containers: {for container in pod.obj.status.containerStatuses { - "\(container.name)": { - image: container.image - state: container.state - } - }} - events: ql.#SearchEvents & { - value: pod.obj - cluster: pod.cluster - } - metrics: ql.#Read & { - cluster: pod.cluster - value: { - apiVersion: "metrics.k8s.io/v1beta1" - kind: "PodMetrics" - metadata: { - name: pod.obj.metadata.name - namespace: pod.obj.metadata.namespace + podsError: [ for pods in collectedPods if pods.err != _|_ {pods.err}] + status: { + if len(podsError) == 0 && podsWithCluster != _|_ { + podList: [ for pod in podsWithCluster { + cluster: pod.cluster + workload: pod.workload + component: pod.component + metadata: { + name: pod.obj.metadata.name + namespace: pod.obj.metadata.namespace + creationTime: pod.obj.metadata.creationTimestamp + version: { + if pod.publishVersion != _|_ { + publishVersion: pod.publishVersion + } + if pod.deployVersion != _|_ { + deployVersion: pod.deployVersion } } } - } + status: { + phase: pod.obj.status.phase + // refer to https://kubernetes.io/docs/concepts/workloads/pods/pod-lifecycle/#pod-phase + if phase != "Pending" && phase != "Unknown" { + podIP: pod.obj.status.podIP + hostIP: pod.obj.status.hostIP + nodeName: pod.obj.spec.nodeName + } + } + }] + } + if len(podsError) != 0 { + error: strings.Join(podsError, ",") + } + if podsWithCluster == _|_ { + podList: [] } } + } + if resources.err != _|_ { status: { - podList: [ for podInfo in podStatus { - name: podInfo.name - containers: [ for containerName, container in podInfo.containers { - containerName - }] - events: podInfo.events.list - }] + error: resources.err } } diff --git a/test/e2e-apiserver-test/testdata/example-app.yaml b/test/e2e-apiserver-test/testdata/example-app.yaml index 4e77d8bf3..6acc759b5 100644 --- a/test/e2e-apiserver-test/testdata/example-app.yaml +++ b/test/e2e-apiserver-test/testdata/example-app.yaml @@ -2,14 +2,15 @@ apiVersion: core.oam.dev/v1beta1 kind: Application metadata: name: example-app - namespace: default spec: components: - - name: hello-world-server + - name: express-server type: webservice properties: - image: crccheck/hello-world - port: 8000 + image: oamdev/hello-world + ports: + - port: 8000 + expose: true traits: - type: scaler properties: @@ -22,57 +23,36 @@ spec: - sleep - '1000000' policies: - - name: example-multi-env-policy - type: env-binding + - name: target-default + type: topology properties: - envs: - - name: test - placement: # selecting the namespace (in local cluster) to deploy to - namespaceSelector: - name: TEST_NAMESPACE - selector: - components: - - data-worker - - - name: staging - placement: # selecting the cluster to deploy to - clusterSelector: - name: cluster-worker - - - name: prod - placement: # selecting both namespace and cluster to deploy to - clusterSelector: - name: cluster-worker - namespaceSelector: - name: PROD_NAMESPACE - patch: # overlay patch on above components - components: - - name: hello-world-server - type: webservice - traits: - - type: scaler - properties: - replicas: 3 - + # The cluster with name local is installed the KubeVela. + clusters: ["local"] + namespace: "default" + - name: target-prod + type: topology + properties: + clusters: ["local"] + # This namespace must be created before deploying. + namespace: "prod" + - name: deploy-ha + type: override + properties: + components: + - type: webservice + traits: + - type: scaler + properties: + replicas: 2 workflow: steps: - # deploy to test env - - name: deploy-test - type: deploy2env + - name: deploy2default + type: deploy properties: - policy: example-multi-env-policy - env: test - - # deploy to staging env - - name: deploy-staging - type: deploy2env + policies: ["target-default"] + - name: manual-approval + type: suspend + - name: deploy2prod + type: deploy properties: - policy: example-multi-env-policy - env: staging - - # deploy to prod env - - name: deploy-prod - type: deploy2env - properties: - policy: example-multi-env-policy - env: prod + policies: ["target-prod", "deploy-ha"] \ No newline at end of file diff --git a/test/e2e-apiserver-test/velaql_test.go b/test/e2e-apiserver-test/velaql_test.go index dfd47b3bb..8df2f3db8 100644 --- a/test/e2e-apiserver-test/velaql_test.go +++ b/test/e2e-apiserver-test/velaql_test.go @@ -42,9 +42,10 @@ import ( ) type PodStatus struct { - Name string `json:"name"` - Containers []string `json:"containers"` - Events interface{} `json:"events"` + Cluster string `json:"cluster"` + Component string `json:"component"` + Metadata map[string]interface{} `json:"metadata"` + Workload map[string]interface{} `json:"workload"` } type Status struct { PodList []PodStatus `json:"podList,omitempty"` @@ -66,6 +67,7 @@ var _ = Describe("Test velaQL rest api", func() { Expect(common.ReadYamlToObject("./testdata/example-app.yaml", &app)).Should(BeNil()) app.Spec.Components[0].Name = component1Name app.Spec.Components[1].Name = component2Name + app.Name = appName req := apiv1.ApplicationRequest{ Components: app.Spec.Components, @@ -79,11 +81,11 @@ var _ = Describe("Test velaQL rest api", func() { if err := k8sClient.Get(context.Background(), client.ObjectKey{Name: appName, Namespace: namespace}, oldApp); err != nil { return err } - if len(oldApp.Status.AppliedResources) != 2 { - return errors.Errorf("expect the applied resources number is %d, but get %d", 2, len(oldApp.Status.AppliedResources)) + if len(oldApp.Status.AppliedResources) != 3 { + return errors.Errorf("expect the applied resources number is %d, but get %d", 3, len(oldApp.Status.AppliedResources)) } return nil - }, 3*time.Second, 300*time.Microsecond).Should(BeNil()) + }, 3*time.Second).WithTimeout(time.Minute * 1).Should(BeNil()) queryRes := get(fmt.Sprintf("/query?velaql=%s{name=%s,namespace=%s}.%s", "read-view", appName, namespace, "output.value.spec")) var appSpec v1beta1.ApplicationSpec @@ -103,6 +105,7 @@ var _ = Describe("Test velaQL rest api", func() { It("Test query application component view", func() { componentView := new(corev1.ConfigMap) Expect(common.ReadYamlToObject("./testdata/component-pod-view.yaml", componentView)).Should(BeNil()) + Expect(k8sClient.Delete(context.Background(), componentView)).Should(SatisfyAny(BeNil(), &util.NotFoundMatcher{})) Expect(k8sClient.Create(context.Background(), componentView)).Should(SatisfyAny(BeNil(), &util.AlreadyExistMatcher{})) oldApp := new(v1beta1.Application) @@ -110,17 +113,19 @@ var _ = Describe("Test velaQL rest api", func() { if err := k8sClient.Get(context.Background(), client.ObjectKey{Name: appName, Namespace: namespace}, oldApp); err != nil { return err } - if len(oldApp.Status.AppliedResources) != 2 { - return errors.Errorf("expect the applied resources number is %d, but get %d", 2, len(oldApp.Status.AppliedResources)) + if len(oldApp.Status.AppliedResources) != 3 { + return errors.Errorf("expect the applied resources number is %d, but get %d", 3, len(oldApp.Status.AppliedResources)) } return nil - }, 3*time.Second, 300*time.Microsecond).Should(BeNil()) + }, 3*time.Second).WithTimeout(time.Minute * 3).Should(BeNil()) - queryRes := get(fmt.Sprintf("/query?velaql=%s{appName=%s,appNs=%s,name=%s}.%s", "test-component-pod-view", appName, namespace, component1Name, "status")) - status := new(Status) - Expect(decodeResponseBody(queryRes, status)).Should(Succeed()) - Expect(len(status.PodList)).Should(Equal(1)) - Expect(status.PodList[0].Containers[0]).Should(Equal(component1Name)) + Eventually(func(g Gomega) { + queryRes := get(fmt.Sprintf("/query?velaql=%s{appName=%s,appNs=%s,name=%s}.%s", "test-component-pod-view", appName, namespace, component1Name, "status")) + status := new(Status) + g.Expect(decodeResponseBody(queryRes, status)).Should(Succeed()) + g.Expect(len(status.PodList)).Should(Equal(1)) + g.Expect(status.PodList[0].Component).Should(Equal(component1Name)) + }, 3*time.Second).WithTimeout(time.Minute * 3).Should(BeNil()) Eventually(func() error { queryRes1 := get(fmt.Sprintf("/query?velaql=%s{appName=%s,appNs=%s,name=%s}.%s", "test-component-pod-view", appName, namespace, component2Name, "status")) @@ -136,11 +141,11 @@ var _ = Describe("Test velaQL rest api", func() { if len(status1.PodList) != 1 { return errors.New("pod number is zero") } - if status1.PodList[0].Containers[0] != component2Name { + if status1.PodList[0].Component != component2Name { return errors.New("container name is not correct") } return nil - }, 10*time.Second, 300*time.Microsecond).Should(BeNil()) + }, 3*time.Second).WithTimeout(time.Minute * 1).Should(BeNil()) }) It("Test collect pod from cronJob", func() { @@ -162,7 +167,7 @@ var _ = Describe("Test velaQL rest api", func() { return err } return nil - }, 10*time.Second, 300*time.Microsecond).Should(BeNil()) + }, 3*time.Second).WithTimeout(time.Minute * 1).Should(BeNil()) newApp := new(v1beta1.Application) Eventually(func() error { @@ -180,12 +185,12 @@ var _ = Describe("Test velaQL rest api", func() { return errors.New("fail to apply cronjob") } return nil - }, 10*time.Second, 300*time.Microsecond).Should(BeNil()) + }, 3*time.Second).WithTimeout(time.Minute).Should(BeNil()) newWorkload := new(batchv1beta1.CronJob) Eventually(func() error { return k8sClient.Get(context.Background(), client.ObjectKey{Name: component2Name, Namespace: namespace}, newWorkload) - }, 10*time.Second, 300*time.Microsecond).Should(BeNil()) + }, 3*time.Second).WithTimeout(time.Minute).Should(BeNil()) Eventually(func() error { queryRes := get(fmt.Sprintf("/query?velaql=%s{appName=%s,appNs=%s}.%s", "test-component-pod-view", appName, namespace, "status")) @@ -202,7 +207,7 @@ var _ = Describe("Test velaQL rest api", func() { return errors.New("pod list is 0") } return nil - }, 2*time.Minute, 3*time.Microsecond).Should(BeNil()) + }, 3*time.Second).WithTimeout(3 * time.Minute).Should(BeNil()) }) PIt("Test collect pod from helmRelease", func() { @@ -211,11 +216,9 @@ var _ = Describe("Test velaQL rest api", func() { req := apiv1.ApplicationRequest{ Components: appWithHelm.Spec.Components, } - Eventually(func(g Gomega) { - res := post(fmt.Sprintf("/v1/namespaces/%s/applications/%s", namespace, appWithHelm.Name), req) - g.Expect(res).ShouldNot(BeNil()) - g.Expect(res.StatusCode).Should(Equal(200)) - }, 1*time.Minute).Should(Succeed()) + res := post(fmt.Sprintf("/v1/namespaces/%s/applications/%s", namespace, appWithHelm.Name), req) + Expect(res).ShouldNot(BeNil()) + Expect(res.StatusCode).Should(Equal(200)) newApp := new(v1beta1.Application) Eventually(func() error { @@ -226,7 +229,7 @@ var _ = Describe("Test velaQL rest api", func() { return errors.New("application is not ready") } return nil - }, 2*time.Minute, 1*time.Second).Should(BeNil()) + }, 3*time.Second).WithTimeout(3 * time.Minute).Should(BeNil()) Eventually(func() error { queryRes := get(fmt.Sprintf("/query?velaql=%s{appName=%s,appNs=%s,name=%s}.%s", "test-component-pod-view", appWithHelm.Name, namespace, "podinfo", "status")) @@ -247,7 +250,7 @@ var _ = Describe("Test velaQL rest api", func() { return errors.New("pod list is 0") } return nil - }, 2*time.Minute, 300*time.Microsecond).Should(BeNil()) + }, 3*time.Second).WithTimeout(3 * time.Minute).Should(BeNil()) }) It("Test collect legacy resources from application", func() { @@ -270,7 +273,7 @@ var _ = Describe("Test velaQL rest api", func() { return errors.New("application is not ready") } return nil - }, 2*time.Minute, 1*time.Second).Should(BeNil()) + }, 3*time.Second).WithTimeout(3 * time.Minute).Should(BeNil()) Eventually(func() error { queryRes := get(fmt.Sprintf("/query?velaql=%s{appName=%s,appNs=%s,name=%s}.%s", "test-component-pod-view", appWithGC.Name, namespace, "express-server", "status")) @@ -291,7 +294,7 @@ var _ = Describe("Test velaQL rest api", func() { return errors.New("pod is not ready") } return nil - }, 2*time.Minute, 300*time.Microsecond).Should(BeNil()) + }, 3*time.Second).WithTimeout(3 * time.Minute).Should(BeNil()) appWithGC.Spec.Components[0].Name = "new-express-server" updateReq := apiv1.ApplicationRequest{ @@ -310,7 +313,7 @@ var _ = Describe("Test velaQL rest api", func() { return errors.New("application is not ready") } return nil - }, 2*time.Minute, 1*time.Second).Should(BeNil()) + }, 3*time.Second).WithTimeout(3 * time.Minute).Should(BeNil()) Eventually(func() error { queryRes := get(fmt.Sprintf("/query?velaql=%s{appName=%s,appNs=%s,name=%s}.%s", "test-component-pod-view", appWithGC.Name, namespace, "express-server", "status")) @@ -331,7 +334,7 @@ var _ = Describe("Test velaQL rest api", func() { return errors.New("pod is not ready") } return nil - }, 2*time.Minute, 300*time.Microsecond).Should(BeNil()) + }, 3*time.Second).WithTimeout(3 * time.Minute).Should(BeNil()) Eventually(func() error { queryRes := get(fmt.Sprintf("/query?velaql=%s{appName=%s,appNs=%s,name=%s}.%s", "test-component-pod-view", appWithGC.Name, namespace, "new-express-server", "status")) @@ -352,7 +355,10 @@ var _ = Describe("Test velaQL rest api", func() { return errors.New("pod is not ready") } return nil - }, 2*time.Minute, 300*time.Microsecond).Should(BeNil()) + }, 3*time.Second).WithTimeout(3 * time.Minute).Should(BeNil()) + + res = delete(fmt.Sprintf("/v1/namespaces/%s/applications/%s", namespace, appWithGC.Name)) + Expect(res).ShouldNot(BeNil()) }) It("Test query logs in pod", func() { @@ -377,7 +383,7 @@ var _ = Describe("Test velaQL rest api", func() { Eventually(func(g Gomega) { g.Expect(k8sClient.Get(context.Background(), types.NamespacedName{Name: podName, Namespace: "default"}, pod)).Should(Succeed()) g.Expect(pod.Status.Phase).Should(Equal(corev1.PodRunning)) - }, 30*time.Second).Should(Succeed()) + }, 3*time.Second).WithTimeout(3 * time.Minute).Should(Succeed()) queryRes := get(fmt.Sprintf("/query?velaql=%s{cluster=%s,namespace=%s,pod=%s,container=%s}.%s", "test-collect-logs", "local", "default", podName, containerName, "status")) status := &struct { Logs string `json:"logs"` @@ -403,7 +409,7 @@ var _ = Describe("Test velaQL rest api", func() { return fmt.Errorf("applied resource velaql error, expect to be 1 but %d", len(status.Resources)) } return nil - }, 30*time.Second, 300*time.Millisecond).Should(BeNil()) + }, 3*time.Second).WithTimeout(3 * time.Minute).Should(BeNil()) // test app resource tree velaql Eventually(func() error { @@ -426,10 +432,15 @@ var _ = Describe("Test velaQL rest api", func() { return fmt.Errorf("replciaset not ready") } return nil - }, 30*time.Second, 300*time.Millisecond).Should(BeNil()) + }, 3*time.Second).WithTimeout(3 * time.Minute).Should(BeNil()) Expect(k8sClient.Delete(ctx, &app)).Should(BeNil()) }) + + It("delete the test application", func() { + res := delete(fmt.Sprintf("/v1/namespaces/%s/applications/%s", namespace, appName)) + Expect(res).ShouldNot(BeNil()) + }) }) var cronJobComponentDefinition = `