github/kubevela
1
0
Fork 0
mirror of https://github.com/kubevela/kubevela.git synced 2026-08-27 16:17:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Feat: add workflow record actions (#2733)

Browse Source 
* Feat: add application revision actions

* refactor workflow record and application revision

* generate doc

* fix rebase

* fix rebase

* delete comment

* fix comment

* delete suspend status

* use apply instead of update

* find latest comlete revision if the revision is not specified

* delete name

* fix primary key
This commit is contained in:
Tianxin Dong
2021-11-20 13:05:52 +08:00
committed by GitHub
parent 54eb662959
commit ff405cd62a
15 changed files with 1099 additions and 321 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/apidoc/swagger.json
+434 -100
View File
@@ -898,6 +898,42 @@
}
},
"/api/v1/applications/{name}/envs": {
"get": {
"consumes": [
"application/xml",
"application/json"
],
"produces": [
"application/json",
"application/xml"
],
"tags": [
"application"
],
"summary": "list policy for application",
"operationId": "listApplicationEnvs",
"parameters": [
{
"type": "string",
"description": "identifier of the application ",
"name": "name",
"in": "path",
"required": true
}
],
"responses": {
"200": {
"schema": {
"$ref": "#/definitions/v1.ListApplicationEnvBinding"
}
},
"400": {
"schema": {
"$ref": "#/definitions/bcode.Bcode"
}
}
}
},
"post": {
"consumes": [
"application/xml",
@@ -957,7 +993,7 @@
"application"
],
"summary": "set application differences in the specified environment",
"operationId": "updateApplicationEnvBinding",
"operationId": "updateApplicationEnv",
"parameters": [
{
"type": "string",
@@ -968,7 +1004,7 @@
},
{
"type": "string",
"description": "identifier of the application ",
"description": "identifier of the envBinding ",
"name": "envName",
"in": "path",
"required": true
@@ -1019,7 +1055,7 @@
},
{
"type": "string",
"description": "identifier of the application envbinding",
"description": "identifier of the envBinding ",
"name": "envName",
"in": "path",
"required": true
@@ -1946,6 +1982,52 @@
}
}
},
"/api/v1/clusters/{clusterName}/namespaces": {
"post": {
"consumes": [
"application/xml",
"application/json"
],
"produces": [
"application/json",
"application/xml"
],
"tags": [
"cluster"
],
"summary": "create namespace in cluster",
"operationId": "createNamespace",
"parameters": [
{
"type": "string",
"description": "name of the target cluster",
"name": "clusterName",
"in": "path",
"required": true
},
{
"name": "body",
"in": "body",
"required": true,
"schema": {
"$ref": "#/definitions/v1.CreateClusterNamespaceRequest"
}
}
],
"responses": {
"200": {
"schema": {
"$ref": "#/definitions/v1.CreateClusterNamespaceResponse"
}
},
"400": {
"schema": {
"$ref": "#/definitions/bcode.Bcode"
}
}
}
}
},
"/api/v1/definitions": {
"get": {
"consumes": [
@@ -1964,9 +2046,9 @@
"parameters": [
{
"enum": [
"workflowstep",
"component",
"trait"
"trait",
"workflowstep"
],
"type": "string",
"description": "query the definition type",
@@ -2652,6 +2734,135 @@
}
}
},
"/api/v1/workflows/{name}/records/{record}/resume": {
"get": {
"consumes": [
"application/xml",
"application/json"
],
"produces": [
"application/json",
"application/xml"
],
"tags": [
"workflow"
],
"summary": "resume suspend workflow record",
"operationId": "resumeWorkflowRecord",
"parameters": [
{
"type": "string",
"description": "identifier of the workflow",
"name": "name",
"in": "path",
"required": true
},
{
"type": "string",
"description": "identifier of the workflow record",
"name": "record",
"in": "path",
"required": true
}
],
"responses": {
"200": {},
"400": {
"schema": {
"$ref": "#/definitions/bcode.Bcode"
}
}
}
}
},
"/api/v1/workflows/{name}/records/{record}/rollback": {
"get": {
"consumes": [
"application/xml",
"application/json"
],
"produces": [
"application/json",
"application/xml"
],
"tags": [
"workflow"
],
"summary": "rollback suspend application record",
"operationId": "rollbackWorkflowRecord",
"parameters": [
{
"type": "string",
"description": "identifier of the workflow",
"name": "name",
"in": "path",
"required": true
},
{
"type": "string",
"description": "identifier of the workflow record",
"name": "record",
"in": "path",
"required": true
},
{
"type": "string",
"description": "identifier of the rollback revision",
"name": "rollbackVersion",
"in": "query"
}
],
"responses": {
"200": {},
"400": {
"schema": {
"$ref": "#/definitions/bcode.Bcode"
}
}
}
}
},
"/api/v1/workflows/{name}/records/{record}/terminate": {
"get": {
"consumes": [
"application/xml",
"application/json"
],
"produces": [
"application/json",
"application/xml"
],
"tags": [
"workflow"
],
"summary": "terminate suspend workflow record",
"operationId": "terminateWorkflowRecord",
"parameters": [
{
"type": "string",
"description": "identifier of the workflow",
"name": "name",
"in": "path",
"required": true
},
{
"type": "string",
"description": "identifier of the workflow record",
"name": "record",
"in": "path",
"required": true
}
],
"responses": {
"200": {},
"400": {
"schema": {
"$ref": "#/definitions/bcode.Bcode"
}
}
}
}
},
"/v1/namespaces/{namespace}/applications/{appname}": {
"get": {
"consumes": [
@@ -2809,6 +3020,9 @@
"name",
"type",
"zone",
"zoneID",
"regionID",
"vpcID",
"labels",
"status",
"apiServerURL",
@@ -2836,22 +3050,31 @@
"provider": {
"type": "string"
},
"regionID": {
"type": "string"
},
"status": {
"type": "string"
},
"type": {
"type": "string"
},
"vpcID": {
"type": "string"
},
"zone": {
"type": "string"
},
"zoneID": {
"type": "string"
}
}
},
"common.AppRolloutStatus": {
"required": [
"rollingState",
"batchRollingState",
"currentBatch",
"batchRollingState",
"upgradedReplicas",
"upgradedReadyReplicas",
"lastTargetAppRevision"
@@ -3427,8 +3650,8 @@
},
"model.ApplicationRevision": {
"required": [
"createTime",
"updateTime",
"createTime",
"appPrimaryKey",
"version",
"status",
@@ -3484,12 +3707,18 @@
"required": [
"alias",
"description",
"type"
"type",
"createTime",
"updateTime"
],
"properties": {
"alias": {
"type": "string"
},
"createTime": {
"type": "string",
"format": "date-time"
},
"description": {
"type": "string"
},
@@ -3498,13 +3727,16 @@
},
"type": {
"type": "string"
},
"updateTime": {
"type": "string",
"format": "date-time"
}
}
},
"model.Cluster": {
"required": [
"createTime",
"updateTime",
"model",
"name",
"alias",
"description",
@@ -3525,10 +3757,6 @@
"apiServerURL": {
"type": "string"
},
"createTime": {
"type": "string",
"format": "date-time"
},
"dashboardURL": {
"type": "string"
},
@@ -3550,6 +3778,9 @@
"type": "string"
}
},
"model": {
"$ref": "#/definitions/model.Model"
},
"name": {
"type": "string"
},
@@ -3561,10 +3792,6 @@
},
"status": {
"type": "string"
},
"updateTime": {
"type": "string",
"format": "date-time"
}
}
},
@@ -3590,13 +3817,14 @@
"model.ProviderInfo": {
"required": [
"provider",
"name",
"id",
"zone",
"clusterID",
"labels"
],
"properties": {
"id": {
"clusterID": {
"type": "string"
},
"clusterName": {
"type": "string"
},
"labels": {
@@ -3605,14 +3833,20 @@
"type": "string"
}
},
"name": {
"type": "string"
},
"provider": {
"type": "string"
},
"regionID": {
"type": "string"
},
"vpcID": {
"type": "string"
},
"zone": {
"type": "string"
},
"zoneID": {
"type": "string"
}
}
},
@@ -3855,9 +4089,16 @@
},
"v1.AddonStatusResponse": {
"required": [
"phase"
"phase",
"args"
],
"properties": {
"args": {
"type": "object",
"additionalProperties": {
"type": "string"
}
},
"enabling_progress": {
"$ref": "#/definitions/v1.EnablingProgress"
},
@@ -3887,12 +4128,6 @@
"description": {
"type": "string"
},
"envBinding": {
"type": "array",
"items": {
"$ref": "#/definitions/v1.EnvBinding"
}
},
"icon": {
"type": "string"
},
@@ -3938,13 +4173,14 @@
},
"v1.ApplicationDeployResponse": {
"required": [
"createTime",
"name",
"version",
"envName",
"createTime",
"status",
"reason",
"deployUser",
"note",
"envName",
"triggerType"
],
"properties": {
@@ -3958,6 +4194,9 @@
"envName": {
"type": "string"
},
"name": {
"type": "string"
},
"note": {
"type": "string"
},
@@ -4033,6 +4272,7 @@
"v1.ApplicationRevisionBase": {
"required": [
"createTime",
"name",
"version",
"status",
"reason",
@@ -4052,6 +4292,9 @@
"envName": {
"type": "string"
},
"name": {
"type": "string"
},
"note": {
"type": "string"
},
@@ -4141,12 +4384,18 @@
"required": [
"name",
"type",
"properties"
"properties",
"createTime",
"updateTime"
],
"properties": {
"alias": {
"type": "string"
},
"createTime": {
"type": "string",
"format": "date-time"
},
"description": {
"type": "string"
},
@@ -4158,6 +4407,10 @@
},
"type": {
"type": "string"
},
"updateTime": {
"type": "string",
"format": "date-time"
}
}
},
@@ -4420,8 +4673,8 @@
},
"v1.CreateApplicationEnvRequest": {
"required": [
"targetNames",
"name"
"name",
"targetNames"
],
"properties": {
"alias": {
@@ -4564,6 +4817,7 @@
},
"v1.CreateCloudClusterResponse": {
"required": [
"clusterName",
"clusterID",
"status"
],
@@ -4571,11 +4825,34 @@
"clusterID": {
"type": "string"
},
"clusterName": {
"type": "string"
},
"status": {
"type": "string"
}
}
},
"v1.CreateClusterNamespaceRequest": {
"required": [
"namespace"
],
"properties": {
"namespace": {
"type": "string"
}
}
},
"v1.CreateClusterNamespaceResponse": {
"required": [
"exists"
],
"properties": {
"exists": {
"type": "boolean"
}
}
},
"v1.CreateClusterRequest": {
"required": [
"name",
@@ -4721,7 +4998,8 @@
"required": [
"appName",
"name",
"default"
"default",
"envName"
],
"properties": {
"alias": {
@@ -4736,6 +5014,9 @@
"description": {
"type": "string"
},
"envName": {
"type": "string"
},
"name": {
"type": "string"
},
@@ -4857,14 +5138,15 @@
},
"v1.DetailApplicationResponse": {
"required": [
"name",
"alias",
"namespace",
"description",
"createTime",
"updateTime",
"icon",
"alias",
"name",
"policies",
"envBindings",
"status",
"resourceInfo",
"workflowStatus"
@@ -4880,10 +5162,10 @@
"description": {
"type": "string"
},
"envBinding": {
"envBindings": {
"type": "array",
"items": {
"$ref": "#/definitions/v1.EnvBinding"
"type": "string"
}
},
"icon": {
@@ -4927,20 +5209,19 @@
},
"v1.DetailClusterResponse": {
"required": [
"createTime",
"alias",
"icon",
"status",
"reason",
"provider",
"dashboardURL",
"updateTime",
"name",
"labels",
"description",
"apiServerURL",
"model",
"name",
"description",
"icon",
"dashboardURL",
"kubeConfig",
"kubeConfigSecret",
"alias",
"labels",
"status",
"reason",
"resourceInfo"
],
"properties": {
@@ -4950,10 +5231,6 @@
"apiServerURL": {
"type": "string"
},
"createTime": {
"type": "string",
"format": "date-time"
},
"dashboardURL": {
"type": "string"
},
@@ -4975,6 +5252,9 @@
"type": "string"
}
},
"model": {
"$ref": "#/definitions/model.Model"
},
"name": {
"type": "string"
},
@@ -4989,22 +5269,18 @@
},
"status": {
"type": "string"
},
"updateTime": {
"type": "string",
"format": "date-time"
}
}
},
"v1.DetailComponentResponse": {
"required": [
"appPrimaryKey",
"creator",
"name",
"type",
"createTime",
"updateTime",
"alias"
"type",
"creator",
"alias",
"createTime",
"appPrimaryKey",
"name"
],
"properties": {
"alias": {
@@ -5099,10 +5375,10 @@
},
"v1.DetailDeliveryTargetResponse": {
"required": [
"createTime",
"updateTime",
"name",
"namespace",
"createTime"
"namespace"
],
"properties": {
"alias": {
@@ -5135,13 +5411,13 @@
},
"v1.DetailPolicyResponse": {
"required": [
"createTime",
"updateTime",
"name",
"type",
"description",
"creator",
"properties"
"properties",
"createTime",
"updateTime"
],
"properties": {
"createTime": {
@@ -5171,17 +5447,17 @@
},
"v1.DetailRevisionResponse": {
"required": [
"triggerType",
"workflowName",
"createTime",
"appPrimaryKey",
"reason",
"deployUser",
"triggerType",
"workflowName",
"envName",
"createTime",
"version",
"note",
"updateTime",
"version",
"status",
"envName"
"appPrimaryKey",
"status"
],
"properties": {
"appPrimaryKey": {
@@ -5226,10 +5502,9 @@
},
"v1.DetailWorkflowRecordResponse": {
"required": [
"terminated",
"name",
"namespace",
"suspend",
"status",
"deployTime",
"deployUser",
"note",
@@ -5256,18 +5531,15 @@
"type": "string",
"format": "date-time"
},
"status": {
"type": "string"
},
"steps": {
"type": "array",
"items": {
"$ref": "#/definitions/common.WorkflowStepStatus"
}
},
"suspend": {
"type": "boolean"
},
"terminated": {
"type": "boolean"
},
"triggerType": {
"type": "string"
}
@@ -5275,13 +5547,14 @@
},
"v1.DetailWorkflowResponse": {
"required": [
"default",
"createTime",
"updateTime",
"name",
"alias",
"description",
"enable",
"default",
"envName",
"createTime",
"updateTime",
"workflowRecord"
],
"properties": {
@@ -5301,6 +5574,9 @@
"enable": {
"type": "boolean"
},
"envName": {
"type": "string"
},
"name": {
"type": "string"
},
@@ -5372,6 +5648,42 @@
}
}
},
"v1.EnvBindingBase": {
"required": [
"name",
"targetNames",
"createTime",
"updateTime"
],
"properties": {
"alias": {
"type": "string"
},
"componentSelector": {
"$ref": "#/definitions/v1.ComponentSelector"
},
"createTime": {
"type": "string",
"format": "date-time"
},
"description": {
"type": "string"
},
"name": {
"type": "string"
},
"targetNames": {
"type": "array",
"items": {
"type": "string"
}
},
"updateTime": {
"type": "string",
"format": "date-time"
}
}
},
"v1.ListAddonRegistryResponse": {
"required": [
"registrys"
@@ -5398,6 +5710,19 @@
}
}
},
"v1.ListApplicationEnvBinding": {
"required": [
"envBindings"
],
"properties": {
"envBindings": {
"type": "array",
"items": {
"$ref": "#/definitions/v1.EnvBindingBase"
}
}
}
},
"v1.ListApplicationPolicy": {
"required": [
"policies"
@@ -5432,7 +5757,7 @@
"creations": {
"type": "array",
"items": {
"type": "string"
"$ref": "#/definitions/v1.CreateCloudClusterResponse"
}
}
}
@@ -5457,7 +5782,8 @@
},
"v1.ListClusterResponse": {
"required": [
"clusters"
"clusters",
"total"
],
"properties": {
"clusters": {
@@ -5465,6 +5791,10 @@
"items": {
"$ref": "#/definitions/v1.ClusterBase"
}
},
"total": {
"type": "integer",
"format": "int64"
}
}
},
@@ -5813,7 +6143,8 @@
"v1.UpdateWorkflowRequest": {
"required": [
"enable",
"default"
"default",
"envName"
],
"properties": {
"alias": {
@@ -5828,6 +6159,9 @@
"enable": {
"type": "boolean"
},
"envName": {
"type": "string"
},
"steps": {
"type": "array",
"items": {
@@ -5846,6 +6180,7 @@
"description",
"enable",
"default",
"envName",
"createTime",
"updateTime"
],
@@ -5866,6 +6201,9 @@
"enable": {
"type": "boolean"
},
"envName": {
"type": "string"
},
"name": {
"type": "string"
},
@@ -5879,8 +6217,7 @@
"required": [
"name",
"namespace",
"suspend",
"terminated"
"status"
],
"properties": {
"name": {
@@ -5893,17 +6230,14 @@
"type": "string",
"format": "date-time"
},
"status": {
"type": "string"
},
"steps": {
"type": "array",
"items": {
"$ref": "#/definitions/common.WorkflowStepStatus"
}
},
"suspend": {
"type": "boolean"
},
"terminated": {
"type": "boolean"
}
}
},
pkg/apiserver/model/workflow.go
+21 -9
View File
@@ -83,15 +83,15 @@ func (w *Workflow) Index() map[string]string {
// WorkflowRecord is the workflow record database model
type WorkflowRecord struct {
Model
WorkflowPrimaryKey string `json:"workflowPrimaryKey"`
AppPrimaryKey string `json:"appPrimaryKey"`
// name is `appName-version`, which is the same as the primary key of deploy event
Name string `json:"name"`
Namespace string `json:"namespace"`
StartTime time.Time `json:"startTime,omitempty"`
Suspend bool `json:"suspend"`
Terminated bool `json:"terminated"`
Steps []common.WorkflowStepStatus `json:"steps,omitempty"`
WorkflowPrimaryKey string `json:"workflowPrimaryKey"`
AppPrimaryKey string `json:"appPrimaryKey"`
RevisionPrimaryKey string `json:"revisionPrimaryKey"`
Name string `json:"name"`
Namespace string `json:"namespace"`
StartTime time.Time `json:"startTime,omitempty"`
Finished string `json:"finished"`
Steps []common.WorkflowStepStatus `json:"steps,omitempty"`
Status string `json:"status"`
}
// TableName return custom table name
@@ -110,8 +110,20 @@ func (w *WorkflowRecord) Index() map[string]string {
if w.Name != "" {
index["name"] = w.Name
}
if w.Namespace != "" {
index["namespace"] = w.Namespace
}
if w.WorkflowPrimaryKey != "" {
index["workflowPrimaryKey"] = w.WorkflowPrimaryKey
}
if w.AppPrimaryKey != "" {
index["appPrimaryKey"] = w.AppPrimaryKey
}
if w.RevisionPrimaryKey != "" {
index["revisionPrimaryKey"] = w.RevisionPrimaryKey
}
if w.Finished != "" {
index["finished"] = w.Finished
}
return index
}
pkg/apiserver/rest/apis/v1/types.go
+5 -6
View File
@@ -606,12 +606,11 @@ type DetailWorkflowRecordResponse struct {
// WorkflowRecord workflow record
type WorkflowRecord struct {
Name string `json:"name"`
Namespace string `json:"namespace"`
StartTime time.Time `json:"startTime,omitempty"`
Suspend bool `json:"suspend"`
Terminated bool `json:"terminated"`
Steps []common.WorkflowStepStatus `json:"steps,omitempty"`
Name string `json:"name"`
Namespace string `json:"namespace"`
StartTime time.Time `json:"startTime,omitempty"`
Status string `json:"status"`
Steps []common.WorkflowStepStatus `json:"steps,omitempty"`
}
// ApplicationDeployRequest the application deploy or update event request
pkg/apiserver/rest/usecase/application.go
+18 -7
View File
@@ -81,7 +81,7 @@ type ApplicationUsecase interface {
DeleteApplicationTrait(ctx context.Context, app *model.Application, component *model.ApplicationComponent, traitType string) error
UpdateApplicationTrait(ctx context.Context, app *model.Application, component *model.ApplicationComponent, traitType string, req apisv1.UpdateApplicationTraitRequest) (*apisv1.ApplicationTrait, error)
ListRevisions(ctx context.Context, appName, envName, status string, page, pageSize int) (*apisv1.ListRevisionsResponse, error)
DetailRevision(ctx context.Context, appName, revisionName string) (*apisv1.DetailRevisionResponse, error)
DetailRevision(ctx context.Context, appName, revisionVersion string) (*apisv1.DetailRevisionResponse, error)
}
type applicationUsecaseImpl struct {
@@ -570,6 +570,7 @@ func (c *applicationUsecaseImpl) DetailPolicy(ctx context.Context, app *model.Ap
// means to render oam application config and apply to cluster.
// An event record is generated for each deploy.
func (c *applicationUsecaseImpl) Deploy(ctx context.Context, app *model.Application, req apisv1.ApplicationDeployRequest) (*apisv1.ApplicationDeployResponse, error) {
// TODO: rollback to handle all the error case
// step1: Render oam application
version := utils.GenerateVersion("")
oamApp, err := c.renderOAMApplication(ctx, app, req.WorkflowName, version)
@@ -613,7 +614,11 @@ func (c *applicationUsecaseImpl) Deploy(ctx context.Context, app *model.Applicat
if err := c.ds.Add(ctx, appRevision); err != nil {
return nil, err
}
// step3: check and create namespace
// step3: create workflow record
if err := c.workflowUsecase.CreateWorkflowRecord(ctx, oamApp); err != nil {
return nil, err
}
// step4: check and create namespace
var namespace corev1.Namespace
if err := c.kubeClient.Get(ctx, types.NamespacedName{Name: oamApp.Namespace}, &namespace); apierrors.IsNotFound(err) {
namespace.Name = oamApp.Namespace
@@ -622,7 +627,7 @@ func (c *applicationUsecaseImpl) Deploy(ctx context.Context, app *model.Applicat
return nil, bcode.ErrCreateNamespace
}
}
// step4: apply to controller cluster
// step5: apply to controller cluster
err = c.apply.Apply(ctx, oamApp)
if err != nil {
appRevision.Status = model.RevisionStatusFail
@@ -638,7 +643,7 @@ func (c *applicationUsecaseImpl) Deploy(ctx context.Context, app *model.Applicat
log.Logger.Warnf("update deploy event failure %s", err.Error())
}
// step5: update deploy event status
// step6: update deploy event status
return &apisv1.ApplicationDeployResponse{
ApplicationRevisionBase: apisv1.ApplicationRevisionBase{
Version: appRevision.Version,
@@ -663,6 +668,8 @@ func (c *applicationUsecaseImpl) renderOAMApplication(ctx context.Context, appMo
Labels: appModel.Labels,
Annotations: map[string]string{
oam.AnnotationDeployVersion: version,
// publish version is the identifier of workflow record
oam.AnnotationPublishVersion: utils.GenerateVersion(reqWorkflowName),
},
},
}
@@ -1040,7 +1047,11 @@ func (c *applicationUsecaseImpl) ListRevisions(ctx context.Context, appName, env
revision.Status = status
}
revisions, err := c.ds.List(ctx, &revision, &datastore.ListOptions{Page: page, PageSize: pageSize})
revisions, err := c.ds.List(ctx, &revision, &datastore.ListOptions{
Page: page,
PageSize: pageSize,
SortBy: []datastore.SortOption{{Key: "model.createTime", Order: datastore.SortOrderDescending}},
})
if err != nil {
return nil, err
}
@@ -1072,10 +1083,10 @@ func (c *applicationUsecaseImpl) ListRevisions(ctx context.Context, appName, env
return resp, nil
}
func (c *applicationUsecaseImpl) DetailRevision(ctx context.Context, appName, revisionName string) (*apisv1.DetailRevisionResponse, error) {
func (c *applicationUsecaseImpl) DetailRevision(ctx context.Context, appName, revisionVersion string) (*apisv1.DetailRevisionResponse, error) {
var revision = model.ApplicationRevision{
AppPrimaryKey: appName,
Version: revisionName,
Version: revisionVersion,
}
if err := c.ds.Get(ctx, &revision); err != nil {
return nil, err
pkg/apiserver/rest/usecase/application_test.go
+43 -1
View File
@@ -26,12 +26,17 @@ import (
"github.com/google/go-cmp/cmp/cmpopts"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"sigs.k8s.io/controller-runtime/pkg/client"
"github.com/oam-dev/kubevela/apis/core.oam.dev/common"
"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
"github.com/oam-dev/kubevela/pkg/apiserver/model"
v1 "github.com/oam-dev/kubevela/pkg/apiserver/rest/apis/v1"
"github.com/oam-dev/kubevela/pkg/apiserver/rest/utils/bcode"
"github.com/oam-dev/kubevela/pkg/oam"
"github.com/oam-dev/kubevela/pkg/utils/apply"
)
@@ -459,7 +464,7 @@ var _ = Describe("Test application usecase function", func() {
Expect(revisions.Total).Should(Equal(int64(0)))
})
It("Test DetailRevisions function", func() {
It("Test DetailRevision function", func() {
err := workflowUsecase.createTestApplicationRevision(context.TODO(), &model.ApplicationRevision{
AppPrimaryKey: "test-app",
Version: "123",
@@ -472,3 +477,40 @@ var _ = Describe("Test application usecase function", func() {
Expect(revision.DeployUser).Should(Equal("test-user"))
})
})
func createTestSuspendApp(ctx context.Context, appName, revisionVersion, wfName, recordName string, kubeClient client.Client) (*v1beta1.Application, error) {
testapp := &v1beta1.Application{
ObjectMeta: metav1.ObjectMeta{
Name: appName,
Namespace: "default",
Annotations: map[string]string{
oam.AnnotationDeployVersion: revisionVersion,
oam.AnnotationWorkflowName: wfName,
oam.AnnotationPublishVersion: recordName,
},
},
Spec: v1beta1.ApplicationSpec{
Components: []common.ApplicationComponent{{
Name: "test-component",
Type: "worker",
Properties: &runtime.RawExtension{Raw: []byte(`{"test":"test"}`)},
Traits: []common.ApplicationTrait{},
Scopes: map[string]string{},
}},
},
Status: common.AppStatus{
Workflow: &common.WorkflowStatus{
Suspend: true,
},
},
}
if err := kubeClient.Create(ctx, testapp); err != nil {
return nil, err
}
if err := kubeClient.Status().Patch(ctx, testapp, client.Merge); err != nil {
return nil, err
}
return testapp, nil
}
pkg/apiserver/rest/usecase/testdata/ui-schema.yaml
Vendored
+61 -61
View File
@@ -137,41 +137,6 @@
label: ReadinessProbe检测
sort: 13
subParameters:
- description: How often, in seconds, to execute the probe.
jsonKey: periodSeconds
label: PeriodSeconds
sort: 100
uiType: Number
validate:
defaultValue: 10
required: true
- description: Minimum consecutive successes for the probe to be considered successful
after having failed.
jsonKey: successThreshold
label: SuccessThreshold
sort: 100
uiType: Number
validate:
defaultValue: 1
required: true
- description: Instructions for assessing container health by probing a TCP socket.
Either this attribute or the exec attribute or the httpGet attribute MUST be
specified. This attribute is mutually exclusive with both the exec attribute
and the httpGet attribute.
jsonKey: tcpSocket
label: TcpSocket
sort: 100
subParameters:
- description: The TCP socket within the container that should be probed to assess
container health.
jsonKey: port
label: Port
sort: 100
uiType: Number
validate:
required: true
uiType: KV
validate: {}
- description: Number of seconds after which the probe times out.
jsonKey: timeoutSeconds
label: TimeoutSeconds
@@ -265,6 +230,41 @@
validate:
defaultValue: 0
required: true
- description: How often, in seconds, to execute the probe.
jsonKey: periodSeconds
label: PeriodSeconds
sort: 100
uiType: Number
validate:
defaultValue: 10
required: true
- description: Minimum consecutive successes for the probe to be considered successful
after having failed.
jsonKey: successThreshold
label: SuccessThreshold
sort: 100
uiType: Number
validate:
defaultValue: 1
required: true
- description: Instructions for assessing container health by probing a TCP socket.
Either this attribute or the exec attribute or the httpGet attribute MUST be
specified. This attribute is mutually exclusive with both the exec attribute
and the httpGet attribute.
jsonKey: tcpSocket
label: TcpSocket
sort: 100
subParameters:
- description: The TCP socket within the container that should be probed to assess
container health.
jsonKey: port
label: Port
sort: 100
uiType: Number
validate:
required: true
uiType: KV
validate: {}
uiType: Group
validate: {}
- description: Instructions for assessing whether the container is alive.
@@ -272,26 +272,6 @@
label: LivenessProbe检测
sort: 15
subParameters:
- description: Instructions for assessing container health by executing a command.
Either this attribute or the httpGet attribute or the tcpSocket attribute MUST
be specified. This attribute is mutually exclusive with both the httpGet attribute
and the tcpSocket attribute.
jsonKey: exec
label: Exec
sort: 100
subParameters:
- description: A command to be executed inside the container to assess its health.
Each space delimited token of the command is a separate array element. Commands
exiting 0 are considered to be successful probes, whilst all other exit codes
are considered failures.
jsonKey: command
label: Command
sort: 100
uiType: Strings
validate:
required: true
uiType: KV
validate: {}
- description: Number of consecutive failures required to determine the container
is not alive (liveness probe) or not ready (readiness probe).
jsonKey: failureThreshold
@@ -400,6 +380,26 @@
validate:
defaultValue: 1
required: true
- description: Instructions for assessing container health by executing a command.
Either this attribute or the httpGet attribute or the tcpSocket attribute MUST
be specified. This attribute is mutually exclusive with both the httpGet attribute
and the tcpSocket attribute.
jsonKey: exec
label: Exec
sort: 100
subParameters:
- description: A command to be executed inside the container to assess its health.
Each space delimited token of the command is a separate array element. Commands
exiting 0 are considered to be successful probes, whilst all other exit codes
are considered failures.
jsonKey: command
label: Command
sort: 100
uiType: Strings
validate:
required: true
uiType: KV
validate: {}
uiType: Group
validate: {}
- description: Specify image pull policy for your service
@@ -416,12 +416,6 @@
value: Always
- label: 永不更新
value: Never
- description: Specify image pull secrets for your service
jsonKey: imagePullSecrets
label: ImagePullSecrets
sort: 100
uiType: Strings
validate: {}
- description: If addRevisionLabel is true, the appRevision label will be added to
the underlying pods
jsonKey: addRevisionLabel
@@ -431,3 +425,9 @@
validate:
defaultValue: false
required: true
- description: Specify image pull secrets for your service
jsonKey: imagePullSecrets
label: ImagePullSecrets
sort: 100
uiType: Strings
validate: {}
pkg/apiserver/rest/usecase/workflow.go
+231 -82
View File
@@ -20,12 +20,14 @@ import (
"context"
"errors"
"fmt"
"strings"
"strconv"
"helm.sh/helm/v3/pkg/time"
appsv1 "k8s.io/api/apps/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
"k8s.io/klog/v2"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/yaml"
"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
"github.com/oam-dev/kubevela/pkg/apiserver/clients"
@@ -33,10 +35,11 @@ import (
"github.com/oam-dev/kubevela/pkg/apiserver/log"
"github.com/oam-dev/kubevela/pkg/apiserver/model"
apisv1 "github.com/oam-dev/kubevela/pkg/apiserver/rest/apis/v1"
"github.com/oam-dev/kubevela/pkg/apiserver/rest/utils"
"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"
"github.com/oam-dev/kubevela/pkg/utils/apply"
)
const (
@@ -51,10 +54,14 @@ type WorkflowUsecase interface {
GetApplicationDefaultWorkflow(ctx context.Context, app *model.Application) (*model.Workflow, error)
DeleteWorkflow(ctx context.Context, workflowName string) error
CreateWorkflow(ctx context.Context, app *model.Application, req apisv1.CreateWorkflowRequest) (*apisv1.DetailWorkflowResponse, error)
CreateWorkflowRecord(ctx context.Context, app *v1beta1.Application) error
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
ResumeRecord(ctx context.Context, appModel *model.Application, recordName string) error
TerminateRecord(ctx context.Context, appModel *model.Application, recordName string) error
RollbackRecord(ctx context.Context, appModel *model.Application, recordName, revisionName string) error
}
// NewWorkflowUsecase new workflow usecase
@@ -66,12 +73,14 @@ func NewWorkflowUsecase(ds datastore.DataStore) WorkflowUsecase {
return &workflowUsecaseImpl{
ds: ds,
kubeClient: kubecli,
apply: apply.NewAPIApplicator(kubecli),
}
}
type workflowUsecaseImpl struct {
ds datastore.DataStore
kubeClient client.Client
apply apply.Applicator
}
// DeleteWorkflow delete application workflow
@@ -268,10 +277,9 @@ func (w *workflowUsecaseImpl) DetailWorkflowRecord(ctx context.Context, workflow
return nil, err
}
version := strings.TrimPrefix(recordName, fmt.Sprintf("%s-", record.AppPrimaryKey))
var revision = model.ApplicationRevision{
AppPrimaryKey: record.AppPrimaryKey,
Version: version,
Version: record.RevisionPrimaryKey,
}
err = w.ds.Get(ctx, &revision)
if err != nil {
@@ -288,115 +296,256 @@ func (w *workflowUsecaseImpl) DetailWorkflowRecord(ctx context.Context, workflow
}
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,
},
},
var record = model.WorkflowRecord{
Finished: "false",
}
selector, err := metav1.LabelSelectorAsSelector(&matchLabels)
// list all unfinished workflow records
records, err := w.ds.List(ctx, &record, &datastore.ListOptions{})
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)
for _, item := range records {
app := &v1beta1.Application{}
index := item.Index()
appPrimaryKey := index["appPrimaryKey"]
namespace := index["namespace"]
recordName := index["name"]
if err := w.kubeClient.Get(ctx, types.NamespacedName{
Name: appPrimaryKey,
Namespace: namespace,
}, app); err != nil {
klog.ErrorS(err, "failed to get app", "app name", appPrimaryKey)
return err
}
// try to sync the status from the running application
if app.Annotations != nil && app.Annotations[oam.AnnotationPublishVersion] == recordName {
if err := w.syncWorkflowStatus(ctx, app, recordName); err != nil {
klog.ErrorS(err, "failed to sync workflow status", "app name", appPrimaryKey, "workflow record name", recordName)
}
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", appPrimaryKey, recordName),
Namespace: namespace,
}, cr); err != nil {
klog.ErrorS(err, "failed to get controller revision", "app name", appPrimaryKey, "workflow record name", recordName)
continue
}
appInRevision, err := util.RawExtension2Application(cr.Data)
if err != nil {
klog.ErrorS(err, "failed to get app data", "controller revision name", cr.Name)
klog.ErrorS(err, "failed to get app data in controller revision", "controller revision name", cr.Name, "app name", appPrimaryKey, "workflow record name", recordName)
continue
}
if err := w.syncWorkflowStatus(ctx, appInRevision, recordName); err != nil {
klog.ErrorS(err, "failed to sync workflow status", "app name", appPrimaryKey, "workflow record version", recordName)
continue
}
if app.Annotations == nil {
klog.ErrorS(err, "empty application annotation", "controller revision name", cr.Name)
continue
}
return nil
}
func (w *workflowUsecaseImpl) syncWorkflowStatus(ctx context.Context, app *v1beta1.Application, recordName string) error {
var record = &model.WorkflowRecord{
AppPrimaryKey: app.Name,
Name: recordName,
}
if err := w.ds.Get(ctx, record); err != nil {
return err
}
var revision = &model.ApplicationRevision{
AppPrimaryKey: app.Name,
Version: record.RevisionPrimaryKey,
}
if err := w.ds.Get(ctx, revision); err != nil {
return err
}
if app.Status.Workflow != nil {
status := app.Status.Workflow
summaryStatus := model.RevisionStatusRunning
if status.Finished {
summaryStatus = model.RevisionStatusComplete
}
if status.Terminated {
summaryStatus = model.RevisionStatusTerminated
}
if _, ok := app.Annotations[oam.AnnotationWorkflowName]; !ok {
klog.ErrorS(err, "missing application workflow name", "controller revision name", cr.Name)
continue
}
revisionName, ok := app.Annotations[oam.AnnotationDeployVersion]
if !ok {
klog.ErrorS(err, "failed to get application revision name", "controller revision name", cr.Name)
continue
record.Status = summaryStatus
record.Steps = status.Steps
record.Finished = strconv.FormatBool(status.Finished)
if err := w.ds.Put(ctx, record); err != nil {
return err
}
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, revisionName, 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
revision.Status = summaryStatus
if err := w.ds.Put(ctx, revision); err != nil {
return err
}
}
return nil
}
func (w *workflowUsecaseImpl) updateRecordApplicationRevisionStatus(ctx context.Context, appPrimaryKey, version string, terminated bool) error {
var applicationRevision = &model.ApplicationRevision{
AppPrimaryKey: appPrimaryKey,
Version: version,
func (w *workflowUsecaseImpl) CreateWorkflowRecord(ctx context.Context, app *v1beta1.Application) error {
if app.Annotations == nil {
return fmt.Errorf("empty annotations in application")
}
if err := w.ds.Get(ctx, applicationRevision); err != nil {
return err
if app.Annotations[oam.AnnotationWorkflowName] == "" {
return fmt.Errorf("failed to get workflow name from application")
}
if terminated {
applicationRevision.Status = model.RevisionStatusTerminated
} else {
applicationRevision.Status = model.RevisionStatusComplete
if app.Annotations[oam.AnnotationPublishVersion] == "" {
return fmt.Errorf("failed to get record version from application")
}
if err := w.ds.Put(ctx, applicationRevision); err != nil {
return err
if app.Annotations[oam.AnnotationDeployVersion] == "" {
return fmt.Errorf("failed to get deploy version from application")
}
return nil
}
func (w *workflowUsecaseImpl) createWorkflowRecord(ctx context.Context, app *v1beta1.Application, revisionName string) error {
status := app.Status.Workflow
return w.ds.Add(ctx, &model.WorkflowRecord{
WorkflowPrimaryKey: app.Annotations[oam.AnnotationWorkflowName],
AppPrimaryKey: app.Name,
Name: strings.TrimPrefix(revisionName, "record-"),
RevisionPrimaryKey: app.Annotations[oam.AnnotationDeployVersion],
Name: app.Annotations[oam.AnnotationPublishVersion],
Namespace: app.Namespace,
StartTime: status.StartTime.Time,
Suspend: status.Suspend,
Terminated: status.Terminated,
Steps: status.Steps,
Finished: "false",
StartTime: time.Now().Time,
Status: model.RevisionStatusInit,
})
}
func (w *workflowUsecaseImpl) ResumeRecord(ctx context.Context, appModel *model.Application, recordName string) error {
oamApp, err := w.checkRecordRunning(ctx, appModel)
if err != nil {
return err
}
oamApp.Status.Workflow.Suspend = false
if err := w.kubeClient.Status().Patch(ctx, oamApp, client.Merge); err != nil {
return err
}
if err := w.syncWorkflowStatus(ctx, oamApp, recordName); err != nil {
return err
}
return nil
}
func (w *workflowUsecaseImpl) TerminateRecord(ctx context.Context, appModel *model.Application, recordName string) error {
oamApp, err := w.checkRecordRunning(ctx, appModel)
if err != nil {
return err
}
oamApp.Status.Workflow.Terminated = true
if err := w.kubeClient.Status().Patch(ctx, oamApp, client.Merge); err != nil {
return err
}
if err := w.syncWorkflowStatus(ctx, oamApp, recordName); err != nil {
return err
}
return nil
}
func (w *workflowUsecaseImpl) RollbackRecord(ctx context.Context, appModel *model.Application, recordName, revisionVersion string) error {
if revisionVersion == "" {
// find the latest complete revision version
var revision = model.ApplicationRevision{
AppPrimaryKey: appModel.Name,
Status: model.RevisionStatusComplete,
}
revisions, err := w.ds.List(ctx, &revision, &datastore.ListOptions{
Page: 0,
PageSize: 1,
SortBy: []datastore.SortOption{{Key: "model.createTime", Order: datastore.SortOrderDescending}},
})
if err != nil {
return err
}
if len(revisions) == 0 {
fmt.Errorf("there is no complete revision, please specify a revision version")
}
revisionVersion = revisions[0].Index()["version"]
}
oamApp, err := w.checkRecordRunning(ctx, appModel)
if err != nil {
return err
}
var record = &model.WorkflowRecord{
AppPrimaryKey: appModel.Name,
Name: recordName,
}
if err := w.ds.Get(ctx, record); err != nil {
return err
}
var rollbackRevision = model.ApplicationRevision{
AppPrimaryKey: appModel.Name,
Version: revisionVersion,
}
if err := w.ds.Get(ctx, &rollbackRevision); err != nil {
return err
}
rollBackApp := &v1beta1.Application{}
if err := yaml.Unmarshal([]byte(rollbackRevision.ApplyAppConfig), rollBackApp); err != nil {
return err
}
// replace the application spec
oamApp.Spec.Components = rollBackApp.Spec.Components
oamApp.Spec.Policies = rollBackApp.Spec.Policies
if oamApp.Annotations == nil {
oamApp.Annotations = make(map[string]string)
}
newRecordName := utils.GenerateVersion(record.WorkflowPrimaryKey)
oamApp.Annotations[oam.AnnotationDeployVersion] = revisionVersion
oamApp.Annotations[oam.AnnotationPublishVersion] = newRecordName
// create a new workflow record
if err := w.CreateWorkflowRecord(ctx, oamApp); err != nil {
return err
}
if err := w.apply.Apply(ctx, oamApp); err != nil {
// rollback error case
if err := w.ds.Delete(ctx, &model.WorkflowRecord{Name: newRecordName}); err != nil {
klog.Error(err, "failed to delete record", newRecordName)
}
return err
}
return nil
}
func (w *workflowUsecaseImpl) checkRecordRunning(ctx context.Context, appModel *model.Application) (*v1beta1.Application, error) {
oamApp := &v1beta1.Application{}
if err := w.kubeClient.Get(ctx, types.NamespacedName{Name: appModel.Name, Namespace: appModel.Namespace}, oamApp); err != nil {
return nil, err
}
if oamApp.Status.Workflow != nil && !oamApp.Status.Workflow.Suspend && !oamApp.Status.Workflow.Terminated && !oamApp.Status.Workflow.Finished {
return nil, fmt.Errorf("workflow is still running, can not operate a running workflow")
}
oamApp.SetGroupVersionKind(v1beta1.ApplicationKindVersionKind)
return oamApp, nil
}
func convertFromRecordModel(record *model.WorkflowRecord) *apisv1.WorkflowRecord {
return &apisv1.WorkflowRecord{
Name: record.Name,
Namespace: record.Namespace,
StartTime: record.StartTime,
Suspend: record.Suspend,
Terminated: record.Terminated,
Steps: record.Steps,
Name: record.Name,
Namespace: record.Namespace,
StartTime: record.StartTime,
Status: record.Status,
Steps: record.Steps,
}
}
pkg/apiserver/rest/usecase/workflow_test.go
+202 -38
View File
@@ -27,12 +27,14 @@ import (
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"
"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"
"github.com/oam-dev/kubevela/pkg/oam"
"github.com/oam-dev/kubevela/pkg/utils/apply"
)
var _ = Describe("Test workflow usecase functions", func() {
@@ -40,7 +42,7 @@ var _ = Describe("Test workflow usecase functions", func() {
workflowUsecase *workflowUsecaseImpl
)
BeforeEach(func() {
workflowUsecase = &workflowUsecaseImpl{ds: ds, kubeClient: k8sClient}
workflowUsecase = &workflowUsecaseImpl{ds: ds, kubeClient: k8sClient, apply: apply.NewAPIApplicator(k8sClient)}
})
It("Test CreateWorkflow function", func() {
req := apisv1.CreateWorkflowRequest{
@@ -75,35 +77,40 @@ var _ = Describe("Test workflow usecase functions", func() {
})
It("Test ListWorkflowRecords function", func() {
By("create some controller revisions to test list workflow records")
By("create some workflow records 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())
app.Annotations[oam.AnnotationWorkflowName] = "list-workflow-name"
for i := 0; i < 3; i++ {
err := workflowUsecase.createWorkflowRecord(context.TODO(), app, fmt.Sprintf("record-test-%v", i))
app.Annotations[oam.AnnotationPublishVersion] = fmt.Sprintf("list-workflow-name-%d", i)
err := workflowUsecase.CreateWorkflowRecord(context.TODO(), app)
Expect(err).Should(BeNil())
}
resp, err := workflowUsecase.ListWorkflowRecords(context.TODO(), "test-workflow-name", 0, 10)
resp, err := workflowUsecase.ListWorkflowRecords(context.TODO(), "list-workflow-name", 0, 10)
Expect(err).Should(BeNil())
Expect(resp.Total).Should(Equal(int64(3)))
})
It("Test DetailWorkflowRecord function", func() {
By("create one controller revision to test detail workflow record")
By("create one workflow record to test detail workflow record")
raw, err := yaml.YAMLToJSON([]byte(yamlStr))
Expect(err).Should(BeNil())
app := &v1beta1.Application{}
err = json.Unmarshal(raw, app)
Expect(err).Should(BeNil())
err = workflowUsecase.createWorkflowRecord(context.TODO(), app, "record-test-123")
app.Annotations[oam.AnnotationWorkflowName] = "test-workflow-name"
app.Annotations[oam.AnnotationPublishVersion] = "test-workflow-name-123"
app.Annotations[oam.AnnotationDeployVersion] = "1234"
err = workflowUsecase.CreateWorkflowRecord(context.TODO(), app)
Expect(err).Should(BeNil())
var deployEvent = &model.ApplicationRevision{
var revision = &model.ApplicationRevision{
AppPrimaryKey: "test",
Version: "123",
Version: "1234",
Status: model.RevisionStatusInit,
DeployUser: "test-user",
Note: "test-commit",
@@ -111,57 +118,212 @@ var _ = Describe("Test workflow usecase functions", func() {
WorkflowName: "test-workflow-name",
}
err = workflowUsecase.createTestApplicationRevision(context.TODO(), deployEvent)
err = workflowUsecase.createTestApplicationRevision(context.TODO(), revision)
Expect(err).Should(BeNil())
detail, err := workflowUsecase.DetailWorkflowRecord(context.TODO(), "test-workflow-name", "test-123")
detail, err := workflowUsecase.DetailWorkflowRecord(context.TODO(), "test-workflow-name", "test-workflow-name-123")
Expect(err).Should(BeNil())
Expect(detail.WorkflowRecord.Name).Should(Equal("test-123"))
Expect(detail.WorkflowRecord.Name).Should(Equal("test-workflow-name-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()
By("create one workflow record to test sync status from application")
raw, err := yaml.YAMLToJSON([]byte(yamlStr))
Expect(err).Should(BeNil())
app := &v1beta1.Application{}
err = json.Unmarshal(raw, app)
Expect(err).Should(BeNil())
app.Status.Workflow.Finished = false
app.Annotations[oam.AnnotationWorkflowName] = "test-workflow-name"
app.Annotations[oam.AnnotationPublishVersion] = "test-workflow-name-233"
app.Annotations[oam.AnnotationDeployVersion] = "4321"
err = workflowUsecase.CreateWorkflowRecord(context.TODO(), app)
Expect(err).Should(BeNil())
By("create one revision to test sync workflow record")
var revision = &model.ApplicationRevision{
AppPrimaryKey: "test",
Version: "4321",
Status: model.RevisionStatusInit,
DeployUser: "test-user",
WorkflowName: "test-workflow-name",
}
err = workflowUsecase.createTestApplicationRevision(context.TODO(), revision)
Expect(err).Should(BeNil())
By("create the application to sync")
ctx := context.Background()
app.Status.Workflow.Finished = true
err = workflowUsecase.kubeClient.Create(ctx, app)
Expect(err).Should(BeNil())
err = workflowUsecase.kubeClient.Status().Patch(ctx, app, client.Merge)
err = workflowUsecase.SyncWorkflowRecord(ctx)
Expect(err).Should(BeNil())
By("check the record")
record, err := workflowUsecase.DetailWorkflowRecord(context.TODO(), "test-workflow-name", "test-workflow-name-233")
Expect(err).Should(BeNil())
Expect(record.Status).Should(Equal(model.RevisionStatusComplete))
By("check the application revision")
err = workflowUsecase.ds.Get(ctx, revision)
Expect(err).Should(BeNil())
Expect(revision.Status).Should(Equal(model.RevisionStatusComplete))
By("create another workflow record to test sync status from controller revision")
app.Status.Workflow.Finished = false
app.Annotations[oam.AnnotationPublishVersion] = "test-workflow-name-111"
app.Annotations[oam.AnnotationDeployVersion] = "1111"
err = workflowUsecase.CreateWorkflowRecord(context.TODO(), app)
Expect(err).Should(BeNil())
By("create another revision to test sync workflow record")
var anotherRevision = &model.ApplicationRevision{
AppPrimaryKey: "test",
Version: "1111",
Status: model.RevisionStatusInit,
DeployUser: "test-user",
WorkflowName: "test-workflow-name",
}
err = workflowUsecase.createTestApplicationRevision(context.TODO(), anotherRevision)
Expect(err).Should(BeNil())
By("create one controller revision to test sync workflow record")
Expect(err).Should(BeNil())
cr := &appsv1.ControllerRevision{
ObjectMeta: metav1.ObjectMeta{
Name: "record-test-1234",
Name: "record-test-test-workflow-name-111",
Namespace: "default",
Labels: map[string]string{"vela.io/wf-revision": "1234"},
Labels: map[string]string{"vela.io/wf-revision": "test-workflow-name-111"},
},
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.RevisionStatusInit,
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)
anotherRecord, err := workflowUsecase.DetailWorkflowRecord(context.TODO(), "test-workflow-name", "test-workflow-name-111")
Expect(err).Should(BeNil())
err = workflowUsecase.createWorkflowRecord(context.TODO(), app, "test-1234")
Expect(err).Should(Equal(datastore.ErrRecordExist))
Expect(anotherRecord.Status).Should(Equal(model.RevisionStatusComplete))
By("check the deploy event")
err = workflowUsecase.ds.Get(ctx, deployEvent)
By("check the application revision")
err = workflowUsecase.ds.Get(ctx, anotherRevision)
Expect(err).Should(BeNil())
Expect(deployEvent.Status).Should(Equal(model.RevisionStatusComplete))
Expect(anotherRevision.Status).Should(Equal(model.RevisionStatusComplete))
})
It("Test ResumeRecord function", func() {
ctx := context.TODO()
app, err := createTestSuspendApp(ctx, "resume-app", "revision-resume1", "workflow-resume", "workflow-resume-1", workflowUsecase.kubeClient)
Expect(err).Should(BeNil())
err = workflowUsecase.CreateWorkflowRecord(context.TODO(), app)
Expect(err).Should(BeNil())
err = workflowUsecase.createTestApplicationRevision(ctx, &model.ApplicationRevision{
AppPrimaryKey: "resume-app",
Version: "revision-resume1",
Status: model.RevisionStatusRunning,
})
Expect(err).Should(BeNil())
err = workflowUsecase.ResumeRecord(ctx, &model.Application{
Name: "resume-app",
Namespace: "default",
}, "workflow-resume-1")
Expect(err).Should(BeNil())
record, err := workflowUsecase.DetailWorkflowRecord(ctx, "workflow-resume", "workflow-resume-1")
Expect(err).Should(BeNil())
Expect(record.Status).Should(Equal(model.RevisionStatusRunning))
})
It("Test TerminateRecord function", func() {
ctx := context.TODO()
app, err := createTestSuspendApp(ctx, "terminate-app", "revision-terminate1", "workflow-terminate", "workflow-terminate-1", workflowUsecase.kubeClient)
Expect(err).Should(BeNil())
err = workflowUsecase.CreateWorkflowRecord(context.TODO(), app)
Expect(err).Should(BeNil())
err = workflowUsecase.createTestApplicationRevision(ctx, &model.ApplicationRevision{
AppPrimaryKey: "terminate-app",
Version: "revision-terminate1",
Status: model.RevisionStatusRunning,
})
Expect(err).Should(BeNil())
err = workflowUsecase.TerminateRecord(ctx, &model.Application{
Name: "terminate-app",
Namespace: "default",
}, "workflow-terminate-1")
Expect(err).Should(BeNil())
record, err := workflowUsecase.DetailWorkflowRecord(ctx, "workflow-terminate", "workflow-terminate-1")
Expect(err).Should(BeNil())
Expect(record.Status).Should(Equal(model.RevisionStatusTerminated))
})
It("Test RollbackRecord function", func() {
ctx := context.TODO()
app, err := createTestSuspendApp(ctx, "rollback-app", "revision-rollback1", "workflow-rollback", "workflow-rollback-1", workflowUsecase.kubeClient)
Expect(err).Should(BeNil())
err = workflowUsecase.CreateWorkflowRecord(context.TODO(), app)
Expect(err).Should(BeNil())
err = workflowUsecase.createTestApplicationRevision(ctx, &model.ApplicationRevision{
AppPrimaryKey: "rollback-app",
Version: "revision-rollback1",
Status: model.RevisionStatusRunning,
})
Expect(err).Should(BeNil())
err = workflowUsecase.createTestApplicationRevision(ctx, &model.ApplicationRevision{
AppPrimaryKey: "rollback-app",
Version: "revision-rollback0",
ApplyAppConfig: `{"apiVersion":"core.oam.dev/v1beta1","kind":"Application","metadata":{"annotations":{"app.oam.dev/workflowName":"workflow-rollback","app.oam.dev/deployVersion":"revision-rollback1","vela.io/publish-version":"workflow-rollback1"},"name":"first-vela-app","namespace":"default"},"spec":{"components":[{"name":"express-server","properties":{"image":"crccheck/hello-world","port":8000},"traits":[{"properties":{"domain":"testsvc.example.com","http":{"/":8000}},"type":"ingress-1-20"}],"type":"webservice"}]}}`,
Status: model.RevisionStatusComplete,
})
Expect(err).Should(BeNil())
err = workflowUsecase.RollbackRecord(ctx, &model.Application{
Name: "rollback-app",
Namespace: "default",
}, "workflow-rollback-1", "revision-rollback0")
Expect(err).Should(BeNil())
recordsNum, err := workflowUsecase.ds.Count(ctx, &model.WorkflowRecord{
AppPrimaryKey: "rollback-app",
WorkflowPrimaryKey: "workflow-rollback",
RevisionPrimaryKey: "revision-rollback0",
}, nil)
Expect(err).Should(BeNil())
Expect(recordsNum).Should(Equal(int64(1)))
By("rollback application without revision version")
app.Annotations[oam.AnnotationPublishVersion] = "workflow-rollback-2"
err = workflowUsecase.CreateWorkflowRecord(context.TODO(), app)
Expect(err).Should(BeNil())
err = workflowUsecase.RollbackRecord(ctx, &model.Application{
Name: "rollback-app",
Namespace: "default",
}, "workflow-rollback-2", "")
Expect(err).Should(BeNil())
recordsNum, err = workflowUsecase.ds.Count(ctx, &model.WorkflowRecord{
AppPrimaryKey: "rollback-app",
WorkflowPrimaryKey: "workflow-rollback",
RevisionPrimaryKey: "revision-rollback0",
}, nil)
Expect(err).Should(BeNil())
Expect(recordsNum).Should(Equal(int64(2)))
})
})
@@ -171,6 +333,7 @@ metadata:
annotations:
app.oam.dev/workflowName: test-workflow-name
app.oam.dev/deployVersion: "1234"
vela.io/publish-version: "test-workflow-name-111"
name: test
namespace: default
spec:
@@ -202,10 +365,11 @@ status:
phase: succeeded
type: apply-component
suspend: false
terminated: false`
terminated: false
finished: true`
func (w *workflowUsecaseImpl) createTestApplicationRevision(ctx context.Context, deployEvent *model.ApplicationRevision) error {
if err := w.ds.Add(ctx, deployEvent); err != nil {
func (w *workflowUsecaseImpl) createTestApplicationRevision(ctx context.Context, revision *model.ApplicationRevision) error {
if err := w.ds.Add(ctx, revision); err != nil {
return err
}
return nil
pkg/apiserver/rest/webservice/workflow.go
+76
View File
@@ -112,6 +112,36 @@ func (w *workflowWebService) GetWebService() *restful.WebService {
Returns(200, "", apis.DetailWorkflowRecordResponse{}).
Writes(apis.DetailWorkflowRecordResponse{}).Do(returns200, returns500))
ws.Route(ws.GET("/{name}/records/{record}/resume").To(w.resumeWorkflowRecord).
Doc("resume suspend workflow record").
Param(ws.PathParameter("name", "identifier of the workflow").DataType("string")).
Param(ws.PathParameter("record", "identifier of the workflow record").DataType("string")).
Metadata(restfulspec.KeyOpenAPITags, tags).
Filter(w.applicationCheckFilter).
Returns(200, "", nil).
Returns(400, "", bcode.Bcode{}).
Writes(apis.DetailWorkflowRecordResponse{}))
ws.Route(ws.GET("/{name}/records/{record}/terminate").To(w.terminateWorkflowRecord).
Doc("terminate suspend workflow record").
Param(ws.PathParameter("name", "identifier of the workflow").DataType("string")).
Param(ws.PathParameter("record", "identifier of the workflow record").DataType("string")).
Metadata(restfulspec.KeyOpenAPITags, tags).
Filter(w.applicationCheckFilter).
Returns(200, "", nil).
Returns(400, "", bcode.Bcode{}).
Writes(apis.DetailWorkflowRecordResponse{}))
ws.Route(ws.GET("/{name}/records/{record}/rollback").To(w.rollbackWorkflowRecord).
Doc("rollback suspend application record").
Param(ws.PathParameter("name", "identifier of the workflow").DataType("string")).
Param(ws.PathParameter("record", "identifier of the workflow record").DataType("string")).
Param(ws.QueryParameter("rollbackVersion", "identifier of the rollback revision").DataType("string")).
Metadata(restfulspec.KeyOpenAPITags, tags).
Filter(w.applicationCheckFilter).
Returns(200, "", nil).
Returns(400, "", bcode.Bcode{}).
Writes(apis.DetailWorkflowRecordResponse{}))
return ws
}
@@ -125,6 +155,22 @@ func (w *workflowWebService) workflowCheckFilter(req *restful.Request, res *rest
chain.ProcessFilter(req, res)
}
func (w *workflowWebService) applicationCheckFilter(req *restful.Request, res *restful.Response, chain *restful.FilterChain) {
workflow, err := w.workflowUsecase.GetWorkflow(req.Request.Context(), req.PathParameter("name"))
if err != nil {
bcode.ReturnError(req, res, err)
return
}
app, err := w.applicationUsecase.GetApplication(req.Request.Context(), workflow.AppPrimaryKey)
if err != nil {
bcode.ReturnError(req, res, err)
return
}
req.Request = req.Request.WithContext(context.WithValue(req.Request.Context(), &apis.CtxKeyApplication, app))
chain.ProcessFilter(req, res)
}
func (w *workflowWebService) listApplicationWorkflows(req *restful.Request, res *restful.Response) {
if req.QueryParameter("appName") == "" {
bcode.ReturnError(req, res, bcode.ErrMustQueryByApp)
@@ -260,3 +306,33 @@ func (w *workflowWebService) detailWorkflowRecord(req *restful.Request, res *res
return
}
}
func (w *workflowWebService) resumeWorkflowRecord(req *restful.Request, res *restful.Response) {
app := req.Request.Context().Value(&apis.CtxKeyApplication).(*model.Application)
err := w.workflowUsecase.ResumeRecord(req.Request.Context(), app, req.PathParameter("record"))
if err != nil {
bcode.ReturnError(req, res, err)
return
}
return
}
func (w *workflowWebService) terminateWorkflowRecord(req *restful.Request, res *restful.Response) {
app := req.Request.Context().Value(&apis.CtxKeyApplication).(*model.Application)
err := w.workflowUsecase.TerminateRecord(req.Request.Context(), app, req.PathParameter("record"))
if err != nil {
bcode.ReturnError(req, res, err)
return
}
return
}
func (w *workflowWebService) rollbackWorkflowRecord(req *restful.Request, res *restful.Response) {
app := req.Request.Context().Value(&apis.CtxKeyApplication).(*model.Application)
err := w.workflowUsecase.RollbackRecord(req.Request.Context(), app, req.PathParameter("record"), req.QueryParameter("rollbackVersion"))
if err != nil {
bcode.ReturnError(req, res, err)
return
}
return
}
pkg/controller/core.oam.dev/v1alpha2/application/application_controller_test.go
+2 -3
View File
@@ -55,7 +55,6 @@ import (
"github.com/oam-dev/kubevela/pkg/oam/testutil"
"github.com/oam-dev/kubevela/pkg/oam/util"
common2 "github.com/oam-dev/kubevela/pkg/utils/common"
wfTypes "github.com/oam-dev/kubevela/pkg/workflow/types"
)
// TODO: Refactor the tests to not copy and paste duplicated code 10 times
@@ -2089,7 +2088,7 @@ var _ = Describe("Test Application Controller", func() {
app := appwithNoTrait.DeepCopy()
app.Name = "vela-test-app-trace"
app.SetNamespace(ns.Name)
app.Annotations = map[string]string{wfTypes.AnnotationPublishVersion: "v134"}
app.Annotations = map[string]string{oam.AnnotationPublishVersion: "v134"}
Expect(k8sClient.Create(ctx, ns)).Should(BeNil())
Expect(k8sClient.Create(ctx, app)).Should(BeNil())
@@ -2116,7 +2115,7 @@ var _ = Describe("Test Application Controller", func() {
web.Spec.Replicas = pointer.Int32(0)
Expect(k8sClient.Update(ctx, web)).Should(BeNil())
checkApp.Annotations[wfTypes.AnnotationPublishVersion] = "v135"
checkApp.Annotations[oam.AnnotationPublishVersion] = "v135"
Expect(k8sClient.Update(ctx, checkApp)).Should(BeNil())
testutil.ReconcileOnceAfterFinalizer(reconciler, reconcile.Request{NamespacedName: appKey})
checkApp = &v1beta1.Application{}
pkg/oam/labels.go
+3
View File
@@ -132,6 +132,9 @@ const (
// AnnotationDeployVersion know the version number of the deployment.
AnnotationDeployVersion = "app.oam.dev/deployVersion"
// AnnotationPublishVersion is annotation that record the application workflow version.
AnnotationPublishVersion = "vela.io/publish-version"
// AnnotationWorkflowName specifies the workflow name for execution.
AnnotationWorkflowName = "app.oam.dev/workflowName"
pkg/velaql/providers/query/collector.go
+1 -2
View File
@@ -39,7 +39,6 @@ import (
"github.com/oam-dev/kubevela/pkg/multicluster"
"github.com/oam-dev/kubevela/pkg/oam"
oamutil "github.com/oam-dev/kubevela/pkg/oam/util"
"github.com/oam-dev/kubevela/pkg/workflow/types"
)
// AppCollector collect resource created by application
@@ -77,7 +76,7 @@ func (c *AppCollector) CollectLatestResourceFromApp() ([]AppResources, error) {
if app.Status.LatestRevision != nil {
revision = app.Status.LatestRevision.Revision
}
publishVersion := app.GetAnnotations()[types.AnnotationPublishVersion]
publishVersion := app.GetAnnotations()[oam.AnnotationPublishVersion]
appRevName := fmt.Sprintf("%s-v%d", app.Name, revision)
comps := make(map[string][]Resource, len(app.Spec.Components))
pkg/workflow/recorder/recorder.go
+1 -3
View File
@@ -66,9 +66,7 @@ func (r *recorder) Save(version string, data []byte) Store {
if version == "" {
wfStatus := r.source.Status.Workflow
if wfStatus != nil {
if !strings.Contains(wfStatus.AppRevision, ":") {
version = wfStatus.AppRevision
}
version = strings.ReplaceAll(wfStatus.AppRevision, ":", "-")
}
}
pkg/workflow/types/types.go
-2
View File
@@ -77,6 +77,4 @@ type Action interface {
const (
// ContextKeyMetadata is key that refer to application metadata.
ContextKeyMetadata = "metadata__"
// AnnotationPublishVersion is annotation that record the application workflow version.
AnnotationPublishVersion = "vela.io/publish-version"
)
pkg/workflow/workflow.go
+1 -7
View File
@@ -132,12 +132,6 @@ 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
@@ -339,7 +333,7 @@ func (e *engine) needStop() bool {
func computeAppRevisionHash(rev string, app *oamcore.Application) (string, error) {
version := ""
if annos := app.Annotations; annos != nil {
version = annos[wfTypes.AnnotationPublishVersion]
version = annos[oam.AnnotationPublishVersion]
}
if version == "" {
specHash, err := utils.ComputeSpecHash(app.Spec)