Feat: implement pipeline APIs (#4908)

* add context when run pipeline Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * Feat: implement pipeline API Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * Extract get log logic and implement getPipelineRunLog API Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * Init and delete pipeline contexts Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * fix panic Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * Allow not specifying context Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * change pipeline to path parameter Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * Add permission check filter Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * project -> projects in route Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * fix route conflict Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * Add project alias Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * Feat: change the list pipeline API Signed-off-by: barnettZQG <barnett.zqg@gmail.com> * Feat: filter the project Signed-off-by: barnettZQG <barnett.zqg@gmail.com> * Fix: the error of the run APi Signed-off-by: barnettZQG <barnett.zqg@gmail.com> * fix log pipeline run API Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * Fix lint, fix the error of log api Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * fix error returning Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * Fix: change the lable to annotation Signed-off-by: barnettZQG <barnett.zqg@gmail.com> * remove log config not found error Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * fix pipeline list api return no context info Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * Fix: create the namespace Signed-off-by: barnettZQG <barnett.zqg@gmail.com> * get pipeline lastrun info Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * allow query single step output Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * organize code in api layer Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * fix project filter, add context value when get pp run, extend lastRun Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * fix get output and implement get input api Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * Fix: change the last run Signed-off-by: barnettZQG <barnett.zqg@gmail.com> * if query sub-step outout, return it directly Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * Fix: change the run stats Signed-off-by: barnettZQG <barnett.zqg@gmail.com> * Fix: change the output Signed-off-by: barnettZQG <barnett.zqg@gmail.com> * flatten the input/output api Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * more info for i/o vars Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * fix nested i/o struct Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * add fromStep in input api Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * add e2e test skeleton Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * add more e2e test Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * use db to store pipeline Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * keep the last 5k lines of log Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * use stern param to keep last lines of logs Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * filter, nil labels, spec check Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * empty res, index, detail param Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * Add e2e test Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * fix e2e test and unit test Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * add context e2e test Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * goimports Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * add more test Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * review Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * remove optional tag in returned value, unify the imports name Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * fix e2e test Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * add stop test Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * more coverage Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * single case selct Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * optimize log color Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * add default permission and role Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * fix permission ut Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * change the log api implementation Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * add color, add container order Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * lint Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * fix filter nil will cut all log Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> * longer timeout and lint Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> Signed-off-by: Qiaozp <qiaozhongpei.qzp@alibaba-inc.com> Signed-off-by: barnettZQG <barnett.zqg@gmail.com> Co-authored-by: barnettZQG <barnett.zqg@gmail.com>
2026-08-18 03:56:36 +00:00 · 2022-10-31 23:59:04 +08:00
parent 8f4ce76543
commit 23cbc9c91f
30 changed files with 3154 additions and 1427 deletions
@@ -176,10 +176,12 @@ jobs:
          make e2e-cleanup
          make e2e-setup-core
          bin/vela addon enable fluxcd
+          bin/vela addon enable vela-workflow
          timeout 600s bash -c -- 'while true; do kubectl get ns flux-system; if [ $? -eq 0 ] ; then break; else sleep 5; fi;done'
          kubectl wait --for=condition=Ready pod -l app.kubernetes.io/name=vela-core,app.kubernetes.io/instance=kubevela -n vela-system --timeout=600s
          kubectl wait --for=condition=Ready pod -l app=source-controller -n flux-system --timeout=600s
          kubectl wait --for=condition=Ready pod -l app=helm-controller -n flux-system --timeout=600s
+          kubectl wait --for=condition=Ready pod -l app.kubernetes.io/name=vela-workflow -n vela-system --timeout=600s

      - name: Run api server e2e test
        run: |
@@ -23,4 +23,15 @@ entries:
      annotations:
        system.vela: ">=1.5.0"
        system.kubernetes: ">=1.30.0"
+  vela-workflow:
+    - annotations:
+        system.vela: '>=v1.6.0-beta.1'
+      created: "2022-10-29T09:11:16.865230605Z"
+      description: vela-workflow provides the capability to run a standalone workflow
+      home: https://github.com/kubevela/workflow
+      icon: https://static.kubevela.net/images/logos/KubeVela%20-03.png
+      name: vela-workflow
+      urls:
+        - http://127.0.0.1:9098/helm/vela-workflow-v0.3.1.tgz
+      version: v0.3.1
 generated: "2022-06-15T13:17:04.733573+08:00"
@@ -22,9 +22,9 @@ import (
 	"fmt"
 	"html/template"
 	"io/fs"
-	"io/ioutil"
 	"log"
 	"net/http"
+	"os"
 	"path"
 	"strings"

@@ -108,24 +108,31 @@ var ossHandler http.HandlerFunc = func(rw http.ResponseWriter, req *http.Request
 var helmHandler http.HandlerFunc = func(rw http.ResponseWriter, req *http.Request) {
 	switch {
 	case strings.Contains(req.URL.Path, "index.yaml"):
-		file, err := ioutil.ReadFile("./e2e/addon/mock/testrepo/helm-repo/index.yaml")
+		file, err := os.ReadFile("./e2e/addon/mock/testrepo/helm-repo/index.yaml")
 		if err != nil {
 			_, _ = rw.Write([]byte(err.Error()))
 		}
 		rw.Write(file)
 	case strings.Contains(req.URL.Path, "fluxcd-test-version-1.0.0.tgz"):
-		file, err := ioutil.ReadFile("./e2e/addon/mock/testrepo/helm-repo/fluxcd-test-version-1.0.0.tgz")
+		file, err := os.ReadFile("./e2e/addon/mock/testrepo/helm-repo/fluxcd-test-version-1.0.0.tgz")
 		if err != nil {
 			_, _ = rw.Write([]byte(err.Error()))
 		}
 		rw.Write(file)
 	case strings.Contains(req.URL.Path, "fluxcd-test-version-2.0.0.tgz"):
-		file, err := ioutil.ReadFile("./e2e/addon/mock/testrepo/helm-repo/fluxcd-test-version-2.0.0.tgz")
+		file, err := os.ReadFile("./e2e/addon/mock/testrepo/helm-repo/fluxcd-test-version-2.0.0.tgz")
+		if err != nil {
+			_, _ = rw.Write([]byte(err.Error()))
+		}
+		rw.Write(file)
+	case strings.Contains(req.URL.Path, "vela-workflow-v0.3.1.tgz"):
+		file, err := os.ReadFile("./e2e/addon/mock/testrepo/helm-repo/vela-workflow-v0.3.1.tgz")
 		if err != nil {
 			_, _ = rw.Write([]byte(err.Error()))
 		}
 		rw.Write(file)
 	}
+
 }

 func init() {
@@ -60,6 +60,7 @@ require (
 	github.com/kubevela/workflow v0.3.1
 	github.com/kyokomi/emoji v2.2.4+incompatible
 	github.com/mitchellh/hashstructure/v2 v2.0.1
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd
 	github.com/oam-dev/cluster-gateway v1.4.0
 	github.com/oam-dev/cluster-register v1.0.4-0.20220928064144-5f76a9d7ca8c
 	github.com/oam-dev/terraform-config-inspect v0.0.0-20210418082552-fc72d929aa28
@@ -239,7 +240,6 @@ require (
 	github.com/moby/locker v1.0.1 // indirect
 	github.com/moby/spdystream v0.2.0 // indirect
 	github.com/moby/term v0.0.0-20210610120745-9d4ed1856297 // indirect
-	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/monochromegane/go-gitignore v0.0.0-20200626010858-205db1a8cc00 // indirect
 	github.com/morikuni/aec v1.0.0 // indirect
@@ -612,7 +612,7 @@ func unmarshalToContent(content []byte) (fileContent *github.RepositoryContent,
 }

 func genAddonAPISchema(addonRes *UIData) error {
-	cueScript := script.CUE([]byte(addonRes.Parameters))
+	cueScript := script.CUE(addonRes.Parameters)
 	schema, err := cueScript.ParsePropertiesToSchema()
 	if err != nil {
 		return err
@@ -16,10 +16,52 @@ limitations under the License.

 package model

-import "fmt"
+import (
+	"fmt"
+
+	"github.com/kubevela/workflow/api/v1alpha1"
+)

 func init() {
 	RegisterModel(&PipelineContext{})
+	RegisterModel(&Pipeline{})
+}
+
+// Pipeline is the model of pipeline
+type Pipeline struct {
+	BaseModel
+	Spec        v1alpha1.WorkflowSpec
+	Name        string `json:"name"`
+	Project     string `json:"project"`
+	Alias       string `json:"alias"`
+	Description string `json:"description"`
+}
+
+// PrimaryKey return custom primary key
+func (p Pipeline) PrimaryKey() string {
+	return fmt.Sprintf("%s-%s", p.Project, p.Name)
+}
+
+// TableName return custom table name
+func (p Pipeline) TableName() string {
+	return tableNamePrefix + "pipeline"
+}
+
+// ShortTableName is the compressed version of table name for kubeapi storage and others
+func (p Pipeline) ShortTableName() string {
+	return "pipeline"
+}
+
+// Index return custom index
+func (p Pipeline) Index() map[string]string {
+	var index = make(map[string]string)
+	if p.Project != "" {
+		index["project"] = p.Project
+	}
+	if p.Name != "" {
+		index["name"] = p.Name
+	}
+	return index
 }

 // Value is a k-v pair
@@ -16,8 +16,6 @@ limitations under the License.

 package model

-import "fmt"
-
 func init() {
 	RegisterModel(&Project{})
 }
@@ -29,11 +27,15 @@ type Project struct {
 	Alias       string `json:"alias"`
 	Owner       string `json:"owner"`
 	Description string `json:"description,omitempty"`
+	Namespace   string `json:"namespace"`
 }

 // GetNamespace get the namespace name of this project.
 func (p *Project) GetNamespace() string {
-	return fmt.Sprintf("project-%s", p.Name)
+	if p.Namespace != "" {
+		return p.Namespace
+	}
+	return p.Name
 }

 // TableName return custom table name
@@ -66,7 +66,8 @@ var _ = Describe("Test cloudshell service function", func() {
 			ProjectService: projectService,
 		}
 		projectService = &projectServiceImpl{
-			Store: ds,
+			Store:     ds,
+			K8sClient: k8sClient,
 			RbacService: &rbacServiceImpl{
 				Store: ds,
 			},
@@ -83,7 +83,7 @@ var _ = Describe("Test helm repo list", func() {
 		pSec = v1.Secret{}
 		gSec = v1.Secret{}
 		Expect(k8sClient.Create(ctx, &v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "vela-system"}})).Should(SatisfyAny(BeNil(), util.AlreadyExistMatcher{}))
-		Expect(k8sClient.Create(ctx, &v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "project-my-project"}})).Should(SatisfyAny(BeNil(), util.AlreadyExistMatcher{}))
+		Expect(k8sClient.Create(ctx, &v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "my-project"}})).Should(SatisfyAny(BeNil(), util.AlreadyExistMatcher{}))
 		Expect(yaml.Unmarshal([]byte(projectSecret), &pSec)).Should(BeNil())
 		Expect(yaml.Unmarshal([]byte(globalSecret), &gSec)).Should(BeNil())
 		Expect(k8sClient.Create(ctx, &pSec)).Should(BeNil())
@@ -390,7 +390,7 @@ apiVersion: v1
 kind: Secret
 metadata:
  name: project-helm-repo
-  namespace: project-my-project
+  namespace: my-project
  labels:
    config.oam.dev/type: helm-repository
    config.oam.dev/catalog: velacore-config
@@ -0,0 +1,143 @@
+/*
+Copyright 2021 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 service
+
+import (
+	"context"
+
+	"github.com/kubevela/workflow/api/v1alpha1"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	"github.com/oam-dev/kubevela/pkg/apiserver/domain/model"
+	"github.com/oam-dev/kubevela/pkg/apiserver/infrastructure/datastore"
+	apisv1 "github.com/oam-dev/kubevela/pkg/apiserver/interfaces/api/dto/v1"
+	"github.com/oam-dev/kubevela/pkg/oam/util"
+)
+
+var (
+	// defaultNamespace   = "project-default-ns1-test"
+	pipelineService    *pipelineServiceImpl
+	pipelineRunService *pipelineRunServiceImpl
+	userService        *userServiceImpl
+	contextService     *contextServiceImpl
+	projectService     *projectServiceImpl
+	ctx                context.Context
+
+	pipelineName = "test-pipeline"
+	projectName  = "test-project"
+)
+var _ = Describe("Test pipeline service functions", func() {
+	It("Init services and project", func() {
+		ds, err := NewDatastore(datastore.Config{Type: "kubeapi", Database: "pipeline-test-kubevela"})
+		Expect(ds).ToNot(BeNil())
+		Expect(err).Should(BeNil())
+		Expect(err).Should(SatisfyAny(BeNil(), &util.AlreadyExistMatcher{}))
+		pipelineService = NewTestPipelineService(ds, k8sClient, cfg).(*pipelineServiceImpl)
+		pipelineRunService = pipelineService.PipelineRunService.(*pipelineRunServiceImpl)
+		contextService = pipelineService.ContextService.(*contextServiceImpl)
+		projectService = pipelineService.ProjectService.(*projectServiceImpl)
+		userService = &userServiceImpl{Store: ds, K8sClient: k8sClient}
+
+		ctx = context.WithValue(context.TODO(), &apisv1.CtxKeyUser, "admin")
+		err = userService.Init(context.TODO())
+		Expect(err).Should(BeNil())
+		_, err = projectService.CreateProject(ctx, apisv1.CreateProjectRequest{
+			Name:  projectName,
+			Owner: "admin",
+		})
+		Expect(err).Should(BeNil())
+		projModel, err := projectService.GetProject(context.TODO(), projectName)
+		Expect(err).Should(BeNil())
+		ctx = context.WithValue(ctx, &apisv1.CtxKeyProject, projModel)
+	})
+
+	It("Test create pipeline", func() {
+		rawProps := []byte(`{"url":"https://api.github.com/repos/kubevela/kubevela"}`)
+		testPipelineSteps := []v1alpha1.WorkflowStep{
+			{
+				SubSteps: []v1alpha1.WorkflowStepBase{
+					{
+						Name: "request",
+						Type: "request",
+						Outputs: v1alpha1.StepOutputs{
+							{
+								ValueFrom: "import \"strconv\"\n\"Current star count: \" + strconv.FormatInt(response[\"stargazers_count\"], 10)\n",
+								Name:      "stars",
+							},
+						},
+						Properties: &runtime.RawExtension{
+							Raw: rawProps,
+						},
+					},
+				},
+				WorkflowStepBase: v1alpha1.WorkflowStepBase{
+					Name: "step-group",
+					Type: "step-group",
+				},
+			},
+		}
+
+		By("create pipeline with sub-steps")
+		pipeline, err := pipelineService.CreatePipeline(ctx, apisv1.CreatePipelineRequest{
+			Name: pipelineName,
+			Spec: v1alpha1.WorkflowSpec{
+				Steps: testPipelineSteps,
+			},
+		})
+		Expect(err).Should(BeNil())
+		Expect(pipeline.Name).Should(Equal(pipelineName))
+		Expect(pipeline.Spec.Steps[0].Name).Should(Equal("step-group"))
+	})
+
+	It("list pipeline", func() {
+		pipelines, err := pipelineService.ListPipelines(ctx, apisv1.ListPipelineRequest{
+			Detailed: true,
+		})
+		Expect(err).Should(BeNil())
+		Expect(pipelines).ShouldNot(BeNil())
+		Expect(pipelines.Total).Should(Equal(1))
+		Expect(len(pipelines.Pipelines)).Should(Equal(1))
+		Expect(pipelines.Pipelines[0].Info).ShouldNot(BeNil())
+	})
+
+	It("get pipeline contexts", func() {
+		By("no context")
+		contexts, err := contextService.ListContexts(ctx, projectName, pipelineName)
+		Expect(err).Should(BeNil())
+		Expect(contexts.Total).Should(Equal(0))
+		Expect(len(contexts.Contexts)).Should(Equal(0))
+
+		By("create context")
+		contextName := "test-context"
+		contextKey := "test-key"
+		contextVal := "test-val"
+		ppCtx := apisv1.Context{
+			Name: contextName,
+			Values: []model.Value{
+				{
+					Key:   contextKey,
+					Value: contextVal,
+				},
+			},
+		}
+		context, err := contextService.CreateContext(ctx, projectName, pipelineName, ppCtx)
+		Expect(err).Should(BeNil())
+		Expect(len(context.Contexts)).Should(Equal(1))
+	})
+})
@@ -22,6 +22,7 @@ import (
 	"fmt"

 	terraformapi "github.com/oam-dev/terraform-controller/api/v1beta1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"sigs.k8s.io/controller-runtime/pkg/client"

 	"github.com/oam-dev/kubevela/apis/types"
@@ -31,6 +32,7 @@ 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/multicluster"
+	"github.com/oam-dev/kubevela/pkg/utils"
 )

 // ProjectService project manage service.
@@ -159,6 +161,13 @@ func (p *projectServiceImpl) GetProject(ctx context.Context, projectName string)
 		}
 		return nil, err
 	}
+	if _, err := utils.GetNamespace(ctx, p.K8sClient, project.GetNamespace()); err != nil {
+		if apierrors.IsNotFound(err) {
+			if err := utils.CreateNamespace(ctx, p.K8sClient, projectName); err != nil && !apierrors.IsAlreadyExists(err) {
+				return nil, bcode.ErrProjectNamespaceFail
+			}
+		}
+	}
 	return project, nil
 }

@@ -319,11 +328,16 @@ func (p *projectServiceImpl) CreateProject(ctx context.Context, req apisv1.Creat
 		}
 	}

+	if err := utils.CreateNamespace(ctx, p.K8sClient, req.Name); err != nil && !apierrors.IsAlreadyExists(err) {
+		return nil, bcode.ErrProjectNamespaceFail
+	}
+
 	newProject := &model.Project{
 		Name:        req.Name,
 		Description: req.Description,
 		Alias:       req.Alias,
 		Owner:       owner,
+		Namespace:   req.Name,
 	}

 	if err := p.Store.Add(ctx, newProject); err != nil {
@@ -526,6 +540,7 @@ func ConvertProjectModel2Base(project *model.Project, owner *model.User) *apisv1
 		CreateTime:  project.CreateTime,
 		UpdateTime:  project.UpdateTime,
 		Owner:       apisv1.NameAlias{Name: project.Owner},
+		Namespace:   project.GetNamespace(),
 	}
 	if owner != nil && owner.Name == project.Owner {
 		base.Owner = apisv1.NameAlias{Name: owner.Name, Alias: owner.Alias}
@@ -59,6 +59,7 @@ var defaultProjectPermissionTemplate = []*model.PermissionTemplate{
 			"project:{projectName}/permission:*",
 			"project:{projectName}/environment:*",
 			"project:{projectName}/application:*/*",
+			"project:{projectName}/pipeline:*/*",
 		},
 		Actions: []string{"detail", "list"},
 		Effect:  "Allow",
@@ -96,6 +97,16 @@ var defaultProjectPermissionTemplate = []*model.PermissionTemplate{
 		Effect:    "Allow",
 		Scope:     "project",
 	},
+	{
+		Name:  "pipeline-management",
+		Alias: "Pipeline Management",
+		Resources: []string{
+			"project:{projectName}/pipeline:*",
+		},
+		Actions: []string{"*"},
+		Effect:  "Allow",
+		Scope:   "project",
+	},
 }

 var defaultPlatformPermission = []*model.PermissionTemplate{
@@ -234,6 +245,17 @@ var ResourceMaps = map[string]resourceMetadata{
 				pathName: "configName",
 			},
 			"provider": {},
+			"pipeline": {
+				pathName: "pipelineName",
+				subResources: map[string]resourceMetadata{
+					"context": {
+						pathName: "contextName",
+					},
+					"pipelineRun": {
+						pathName: "pipelineRunName",
+					},
+				},
+			},
 		},
 		pathName: "projectName",
 	},
@@ -866,7 +888,7 @@ func (p *rbacServiceImpl) InitDefaultRoleAndUsersForProject(ctx context.Context,
 	}, &model.Role{
 		Name:        "project-admin",
 		Alias:       "Project Admin",
-		Permissions: []string{"project-view", "app-management", "env-management", "role-management", "configuration-read"},
+		Permissions: []string{"project-view", "app-management", "env-management", "role-management", "pipeline-management", "configuration-read"},
 		Project:     project.Name,
 	}, &model.Role{
 		Name:        "project-viewer",
@@ -198,7 +198,7 @@ var _ = Describe("Test rbac service", func() {

 		policies, err := rbacService.ListPermissions(context.TODO(), "init-test")
 		Expect(err).Should(BeNil())
-		Expect(len(policies)).Should(BeEquivalentTo(int64(5)))
+		Expect(len(policies)).Should(BeEquivalentTo(int64(6)))
 	})

 	It("Test UpdatePermission", func() {
@@ -25,6 +25,7 @@ import (
 	"testing"
 	"time"

+	"github.com/kubevela/workflow/api/v1alpha1"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 	corev1 "k8s.io/api/core/v1"
@@ -74,7 +75,10 @@ var _ = BeforeSuite(func(done Done) {

 	By("new kube client")
 	cfg.Timeout = time.Minute * 2
-	k8sClient, err = client.New(cfg, client.Options{Scheme: common.Scheme})
+	scheme := common.Scheme
+	err = v1alpha1.AddToScheme(scheme)
+	Expect(err).ShouldNot(HaveOccurred())
+	k8sClient, err = client.New(cfg, client.Options{Scheme: scheme})
 	Expect(err).Should(BeNil())
 	Expect(k8sClient).ToNot(BeNil())
 	By("new kube client success")
@@ -58,7 +58,7 @@ var _ = Describe("Test workflow service functions", func() {
 		Expect(ds).ToNot(BeNil())
 		Expect(err).Should(BeNil())
 		rbacService := &rbacServiceImpl{Store: ds}
-		projectService = &projectServiceImpl{Store: ds, RbacService: rbacService}
+		projectService = &projectServiceImpl{Store: ds, RbacService: rbacService, K8sClient: k8sClient}
 		envService = &envServiceImpl{Store: ds, KubeClient: k8sClient, ProjectService: projectService}
 		envBinding = &envBindingServiceImpl{
 			Store:           ds,
@@ -20,13 +20,13 @@ import (
 	"fmt"

 	pkgmulticluster "github.com/kubevela/pkg/multicluster"
+	"github.com/kubevela/workflow/api/v1alpha1"
+	"github.com/kubevela/workflow/pkg/cue/packages"
 	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/rest"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"

-	"github.com/kubevela/workflow/pkg/cue/packages"
-
 	apiConfig "github.com/oam-dev/kubevela/pkg/apiserver/config"
 	"github.com/oam-dev/kubevela/pkg/auth"
 	"github.com/oam-dev/kubevela/pkg/oam/discoverymapper"
@@ -73,11 +73,13 @@ func GetKubeClient() (client.Client, error) {
 	if kubeConfig == nil {
 		return nil, fmt.Errorf("please call SetKubeConfig first")
 	}
-	var err error
-	kubeClient, err = pkgmulticluster.NewClient(kubeConfig, pkgmulticluster.ClientOptions{
+	err := v1alpha1.AddToScheme(common.Scheme)
+	if err != nil {
+		return nil, err
+	}
+	return pkgmulticluster.NewClient(kubeConfig, pkgmulticluster.ClientOptions{
 		Options: client.Options{Scheme: common.Scheme},
 	})
-	return kubeClient, err
 }

 // GetKubeConfig create/get kube runtime config
@@ -50,12 +50,14 @@ var (
 	CtxKeyApplicationComponent = "component"
 	// CtxKeyUser request context key of user
 	CtxKeyUser = "user"
+	// CtxKeyProject request context key of project
+	CtxKeyProject = "project"
 	// CtxKeyToken request context key of request token
 	CtxKeyToken = "token"
 	// CtxKeyPipeline request context key of pipeline
 	CtxKeyPipeline = "pipeline"
-	// CtxKeyPipelineContex request context key of pipeline context
-	CtxKeyPipelineContex = "pipeline-context"
+	// CtxKeyPipelineContext request context key of pipeline context
+	CtxKeyPipelineContext = "pipeline-context"
 	// CtxKeyPipelineRun request context key of pipeline run
 	CtxKeyPipelineRun = "pipeline-run"
 )
@@ -791,6 +793,7 @@ type ProjectBase struct {
 	CreateTime  time.Time `json:"createTime"`
 	UpdateTime  time.Time `json:"updateTime"`
 	Owner       NameAlias `json:"owner,omitempty"`
+	Namespace   string    `json:"-"`
 }

 // CreateProjectRequest create project request body
@@ -1551,10 +1554,11 @@ type ListConfigDistributionResponse struct {

 // PipelineMeta is metadata of pipeline
 type PipelineMeta struct {
-	Name        string `json:"name" validate:"checkname"`
-	Alias       string `json:"alias" validate:"checkalias" optional:"true"`
-	Project     string `json:"project"`
-	Description string `json:"description" optional:"true"`
+	Name        string    `json:"name"`
+	Alias       string    `json:"alias"`
+	Project     NameAlias `json:"project"`
+	Description string    `json:"description"`
+	CreateTime  time.Time `json:"createTime"`
 }

 // PipelineBase is the base info of pipeline
@@ -1580,7 +1584,6 @@ type RunStat struct {
 // CreatePipelineRequest is the request body of creating pipeline
 type CreatePipelineRequest struct {
 	Name        string                        `json:"name" validate:"checkname"`
-	Project     string                        `json:"project"`
 	Alias       string                        `json:"alias" validate:"checkalias" optional:"true"`
 	Description string                        `json:"description" optional:"true"`
 	Spec        workflowv1alpha1.WorkflowSpec `json:"spec"`
@@ -1593,8 +1596,9 @@ type PipelineMetaResponse struct {

 // ListPipelineRequest is the request body of listing pipeline
 type ListPipelineRequest struct {
-	Projects []string `json:"projects"`
-	Query    string   `json:"query"`
+	Projects []string `json:"projects" optional:"true"`
+	Query    string   `json:"query" optional:"true"`
+	Detailed bool     `json:"detailed" optional:"true"`
 }

 // ListPipelineResponse is the response body of listing pipeline
@@ -1616,11 +1620,6 @@ type UpdatePipelineRequest struct {
 	Spec        workflowv1alpha1.WorkflowSpec `json:"spec" optional:"true"`
 }

-// GetPipelineRequest is the request body of getting pipeline
-type GetPipelineRequest struct {
-	Detailed bool `json:"detailed"`
-}
-
 // GetPipelineResponse is the response body of getting pipeline
 type GetPipelineResponse struct {
 	PipelineBase `json:",inline"`
@@ -1629,9 +1628,8 @@ type GetPipelineResponse struct {

 // PipelineInfo is the info of pipeline
 type PipelineInfo struct {
-	RelatedApps   []ApplicationBase                  `json:"relatedApps"`
-	LastRunStatus workflowv1alpha1.WorkflowRunStatus `json:"lastRunStatus"`
-	RunStat       RunStat                            `json:"runStat"`
+	LastRun *PipelineRun `json:"lastRun"`
+	RunStat RunStat      `json:"runStat"`
 }

 /***********************/
@@ -1652,9 +1650,9 @@ type PipelineRunBriefing struct {

 // PipelineRunMeta is the metadata of pipeline run
 type PipelineRunMeta struct {
-	PipelineName    string `json:"pipelineName"`
-	Project         string `json:"project"`
-	PipelineRunName string `json:"pipelineRunName"`
+	PipelineName    string    `json:"pipelineName"`
+	Project         NameAlias `json:"project"`
+	PipelineRunName string    `json:"pipelineRunName"`
 }

 // PipelineRun is the info of pipeline run
@@ -1667,14 +1665,16 @@ type PipelineRun struct {
 type PipelineRunBase struct {
 	PipelineRunMeta `json:",inline"`
 	// Record marks the run of the pipeline
-	Record      int64                            `json:"record"`
-	ContextName string                           `json:"contextName"`
-	Spec        workflowv1alpha1.WorkflowRunSpec `json:"spec"`
+	Record        int64                            `json:"record"`
+	ContextName   string                           `json:"contextName"`
+	ContextValues []model.Value                    `json:"contextValues"`
+	Spec          workflowv1alpha1.WorkflowRunSpec `json:"spec"`
 }

 // RunPipelineRequest is the request body of running pipeline
 type RunPipelineRequest struct {
 	// Mode is the mode of the pipeline run. Available values are: "StepByStep", "DAG" for both `step` and `subStep`
+	// default: "StepByStep" for `step`, "DAG" for `subStep`
 	Mode        workflowv1alpha1.WorkflowExecuteMode `json:"mode" optional:"true"`
 	ContextName string                               `json:"contextName"`
 }
@@ -1687,12 +1687,18 @@ type ListPipelineRunResponse struct {

 // GetPipelineRunLogResponse is the response body of getting pipeline run log
 type GetPipelineRunLogResponse struct {
-	Log []Log `json:"log"`
+	StepBase `json:",inline"`
+	Log      string `json:"log"`
 }

 // GetPipelineRunOutputResponse is the response body of getting pipeline run output
 type GetPipelineRunOutputResponse struct {
-	Output []Output `json:"output"`
+	StepOutputs []StepOutputBase `json:"outputs"`
+}
+
+// GetPipelineRunInputResponse is the response body of getting pipeline run input
+type GetPipelineRunInputResponse struct {
+	StepInputs []StepInputBase `json:"inputs"`
 }

 // StepBase is the base info of step
@@ -1703,16 +1709,31 @@ type StepBase struct {
 	Phase string `json:"phase"`
 }

-// Log is the log of step
-type Log struct {
+// StepOutputBase is the output of step
+type StepOutputBase struct {
 	StepBase `json:",inline"`
-	Log      string `json:"log"`
+	Values   []OutputVar `json:"values"`
 }

-// Output is the output of step
-type Output struct {
+// StepInputBase is the input of step
+type StepInputBase struct {
 	StepBase `json:",inline"`
-	Vars     map[string]string `json:"vars"`
+	Values   []InputVar `json:"values"`
+}
+
+// OutputVar is one output var
+type OutputVar struct {
+	Name      string `json:"name"`
+	Value     string `json:"value"`
+	ValueFrom string `json:"valueFrom"`
+}
+
+// InputVar is one input var
+type InputVar struct {
+	From         string `json:"from"`
+	FromStep     string `json:"fromStep"`
+	ParameterKey string `json:"parameterKey"`
+	Value        string `json:"value"`
 }

 /*******************/
@@ -62,6 +62,7 @@ func InitAPIBean() []interface{} {
 	RegisterAPIInterface(NewApplicationAPIInterface())
 	RegisterAPIInterface(NewProjectAPIInterface())
 	RegisterAPIInterface(NewEnvAPIInterface())
+	RegisterAPIInterface(NewPipelineAPIInterface())

 	// Extension
 	RegisterAPIInterface(NewDefinitionAPIInterface())
@@ -82,7 +83,6 @@ func InitAPIBean() []interface{} {
 	RegisterAPIInterface(NewWebhookAPIInterface())
 	RegisterAPIInterface(NewRepositoryAPIInterface())
 	RegisterAPIInterface(NewCloudShellAPIInterface())
-	RegisterAPIInterface(NewPipelineAPIInterface())

 	// Authentication
 	RegisterAPIInterface(NewAuthenticationAPIInterface())
@@ -18,199 +18,242 @@ package api

 import (
 	"context"
+	"strconv"

 	restfulspec "github.com/emicklei/go-restful-openapi/v2"
 	"github.com/emicklei/go-restful/v3"
-	workflowv1alpha1 "github.com/kubevela/workflow/api/v1alpha1"
-
-	"github.com/oam-dev/kubevela/pkg/apiserver/utils/log"
+	"github.com/kubevela/workflow/api/v1alpha1"
+	"github.com/pkg/errors"

 	"github.com/oam-dev/kubevela/pkg/apiserver/domain/service"
 	apis "github.com/oam-dev/kubevela/pkg/apiserver/interfaces/api/dto/v1"
 	"github.com/oam-dev/kubevela/pkg/apiserver/utils/bcode"
+	"github.com/oam-dev/kubevela/pkg/apiserver/utils/log"
 )

-type pipelineAPIInterface struct {
-	PipelineService    service.PipelineService    `inject:""`
-	PipelineRunService service.PipelineRunService `inject:""`
-	ContextService     service.ContextService     `inject:""`
-}
-
-type pipelinePathParamKey string
-
 const (
 	// Project is the project name key of query param
-	Project pipelinePathParamKey = "projectName"
+	Project string = "projectName"
 	// Pipeline is the pipeline name of query param
-	Pipeline pipelinePathParamKey = "pipelineName"
+	Pipeline string = "pipelineName"
 	// PipelineRun is the pipeline run name of query param
-	PipelineRun pipelinePathParamKey = "runName"
+	PipelineRun string = "runName"
 	// ContextName is the context name of query param
-	ContextName pipelinePathParamKey = "contextName"
+	ContextName string = "contextName"
 )

-// GetWebServiceRoute is the implementation of pipeline Interface
-func (p *pipelineAPIInterface) GetWebServiceRoute() *restful.WebService {
-
-	ws := new(restful.WebService)
+func initPipelineRoutes(ws *restful.WebService, n *projectAPIInterface) {
 	tags := []string{"pipeline"}
-
 	projParam := func(builder *restful.RouteBuilder) {
-		builder.Param(ws.QueryParameter(string(Project), "project name").Required(true))
+		builder.Param(ws.PathParameter(Project, "project name").Required(true))
+		builder.Filter(n.projectCheckFilter)
 	}
 	pipelineParam := func(builder *restful.RouteBuilder) {
-		builder.Param(ws.PathParameter(string(Pipeline), "pipeline name").Required(true))
-		builder.Filter(p.pipelineCheckFilter)
+		builder.Param(ws.PathParameter(Pipeline, "pipeline name").Required(true))
+		builder.Filter(n.pipelineCheckFilter)
 	}
 	ctxParam := func(builder *restful.RouteBuilder) {
-		builder.Param(ws.PathParameter(string(ContextName), "pipeline context name").Required(true))
-		builder.Filter(p.pipelineContextCheckFilter)
+		builder.Param(ws.PathParameter(ContextName, "pipeline context name").Required(true))
+		builder.Filter(n.pipelineContextCheckFilter)
 	}
 	runParam := func(builder *restful.RouteBuilder) {
-		builder.Param(ws.PathParameter(string(PipelineRun), "pipeline run name").Required(true))
-		builder.Filter(p.pipelineRunCheckFilter)
+		builder.Param(ws.PathParameter(PipelineRun, "pipeline run name").Required(true))
+		builder.Filter(n.pipelineRunCheckFilter)
 	}
 	meta := func(builder *restful.RouteBuilder) {
 		builder.Metadata(restfulspec.KeyOpenAPITags, tags)
 	}

-	ws.Path(versionPrefix+"/pipelines").
-		Consumes(restful.MIME_JSON, restful.MIME_XML).
-		Produces(restful.MIME_JSON, restful.MIME_XML).
-		Doc("api for pipeline manage")
-
-	ws.Route(ws.POST("").To(p.createPipeline).
+	ws.Route(ws.POST("/{projectName}/pipelines").To(n.createPipeline).
 		Doc("create pipeline").
 		Reads(apis.CreatePipelineRequest{}).
 		Returns(200, "OK", apis.PipelineBase{}).
 		Returns(400, "Bad Request", bcode.Bcode{}).
-		Writes(apis.PipelineBase{}).Do(meta))
+		Filter(n.RBACService.CheckPerm("project/pipeline", "create")).
+		Writes(apis.PipelineBase{}).Do(meta, projParam))

-	ws.Route(ws.GET("").To(p.listPipelines).
-		Doc("list pipelines").
-		Param(ws.QueryParameter("query", "Fuzzy search based on name or description").DataType("string")).
-		Returns(200, "OK", apis.ListPipelineResponse{}).
-		Returns(400, "Bad Request", bcode.Bcode{}).
-		Writes(apis.ListPipelineResponse{}).Do(meta, projParam))
-
-	ws.Route(ws.GET("/{pipelineName}").To(p.getPipeline).
+	ws.Route(ws.GET("/{projectName}/pipelines/{pipelineName}").To(n.getPipeline).
 		Doc("get pipeline").
-		Reads(apis.GetPipelineRequest{}).
 		Returns(200, "OK", apis.GetPipelineResponse{}).
 		Returns(400, "Bad Request", bcode.Bcode{}).
-		Writes(apis.GetPipelineResponse{}).Do(meta, projParam, pipelineParam))
+		// use Param instead of pipelineParam to get pipeline information
+		Param(ws.PathParameter(Pipeline, "pipeline name").Required(true)).
+		Filter(n.RBACService.CheckPerm("project/pipeline", "detail")).
+		Writes(apis.GetPipelineResponse{}).Do(meta, projParam))

-	ws.Route(ws.PUT("/{pipelineName}").To(p.updatePipeline).
+	ws.Route(ws.PUT("/{projectName}/pipelines/{pipelineName}").To(n.updatePipeline).
 		Doc("update pipeline").
 		Reads(apis.UpdatePipelineRequest{}).
 		Returns(200, "OK", apis.PipelineBase{}).
 		Returns(400, "Bad Request", bcode.Bcode{}).
+		Filter(n.RBACService.CheckPerm("project/pipeline", "update")).
 		Writes(apis.PipelineBase{}).Do(meta, projParam, pipelineParam))

-	ws.Route(ws.DELETE("/{pipelineName}").To(p.deletePipeline).
+	ws.Route(ws.DELETE("/{projectName}/pipelines/{pipelineName}").To(n.deletePipeline).
 		Doc("delete pipeline").
 		Returns(200, "OK", apis.PipelineMetaResponse{}).
 		Returns(400, "Bad Request", bcode.Bcode{}).
+		Filter(n.RBACService.CheckPerm("project/pipeline", "delete")).
 		Writes(apis.PipelineMetaResponse{}).Do(meta, projParam, pipelineParam))

-	ws.Route(ws.POST("/{pipelineName}/contexts").To(p.createContextValue).
+	ws.Route(ws.POST("/{projectName}/pipelines/{pipelineName}/contexts").To(n.createContextValue).
 		Doc("create pipeline context values").
 		Reads(apis.CreateContextValuesRequest{}).
 		Returns(200, "OK", apis.Context{}).
 		Returns(400, "Bad Request", bcode.Bcode{}).
+		Filter(n.RBACService.CheckPerm("project/pipeline/context", "create")).
 		Writes(apis.Context{}).Do(meta, projParam, pipelineParam))

-	ws.Route(ws.GET("/{pipelineName}/contexts").To(p.listContextValues).
+	ws.Route(ws.GET("/{projectName}/pipelines/{pipelineName}/contexts").To(n.listContextValues).
 		Doc("list pipeline context values").
 		Returns(200, "OK", apis.ListContextValueResponse{}).
 		Returns(400, "Bad Request", bcode.Bcode{}).
+		Filter(n.RBACService.CheckPerm("project/pipeline/context", "list")).
 		Writes(apis.ListContextValueResponse{}).Do(meta, projParam, pipelineParam))

-	ws.Route(ws.PUT("/{pipelineName}/contexts/{contextName}").To(p.updateContextValue).
+	ws.Route(ws.PUT("/{projectName}/pipelines/{pipelineName}/contexts/{contextName}").To(n.updateContextValue).
 		Doc("update pipeline context value").
 		Reads(apis.UpdateContextValuesRequest{}).
 		Returns(200, "OK", apis.Context{}).
 		Returns(400, "Bad Request", bcode.Bcode{}).
+		Filter(n.RBACService.CheckPerm("project/pipeline/context", "update")).
 		Writes(apis.Context{}).Do(meta, projParam, pipelineParam, ctxParam))

-	ws.Route(ws.DELETE("/{pipelineName}/contexts/{contextName}").To(p.deleteContextValue).
+	ws.Route(ws.DELETE("/{projectName}/pipelines/{pipelineName}/contexts/{contextName}").To(n.deleteContextValue).
 		Doc("delete pipeline context value").
 		Returns(200, "OK", apis.ContextNameResponse{}).
 		Returns(400, "Bad Request", bcode.Bcode{}).
+		Filter(n.RBACService.CheckPerm("project/pipeline/context", "delete")).
 		Writes(apis.ContextNameResponse{}).Do(meta, projParam, pipelineParam, ctxParam))

-	ws.Route(ws.POST("/{pipelineName}/run").To(p.runPipeline).
+	ws.Route(ws.POST("/{projectName}/pipelines/{pipelineName}/run").To(n.runPipeline).
 		Doc("run pipeline").
 		Reads(apis.RunPipelineRequest{}).
-		Returns(200, "OK", apis.PipelineRunMeta{}).
+		Returns(200, "OK", apis.PipelineRun{}).
 		Returns(400, "Bad Request", bcode.Bcode{}).
+		Filter(n.RBACService.CheckPerm("project/pipeline", "run")).
 		Writes(apis.PipelineRunMeta{}).Do(meta, projParam, pipelineParam))

-	ws.Route(ws.GET("/{pipelineName}/runs").To(p.listPipelineRuns).
+	ws.Route(ws.GET("/{projectName}/pipelines/{pipelineName}/runs").To(n.listPipelineRuns).
 		Doc("list pipeline runs").
 		Param(ws.QueryParameter("status", "query identifier of the status").DataType("string")).
 		Returns(200, "OK", apis.ListPipelineRunResponse{}).
 		Returns(400, "Bad Request", bcode.Bcode{}).
+		Filter(n.RBACService.CheckPerm("project/pipeline/pipelineRun", "list")).
 		Writes(apis.ListPipelineRunResponse{}).Do(meta, projParam, pipelineParam))

-	ws.Route(ws.POST("/{pipelineName}/runs/{runName}/stop").To(p.stopPipeline).
+	ws.Route(ws.POST("/{projectName}/pipelines/{pipelineName}/runs/{runName}/stop").To(n.stopPipeline).
 		Doc("stop pipeline run").
 		Returns(200, "OK", apis.PipelineRunMeta{}).
 		Returns(400, "Bad Request", bcode.Bcode{}).
+		Filter(n.RBACService.CheckPerm("project/pipeline/pipelineRun", "stop")).
 		Writes(apis.PipelineRunMeta{}).Do(meta, projParam, pipelineParam, runParam))

-	ws.Route(ws.GET("/{pipelineName}/runs/{runName}").To(p.getPipelineRun).
+	ws.Route(ws.GET("/{projectName}/pipelines/{pipelineName}/runs/{runName}").To(n.getPipelineRun).
 		Doc("get pipeline run").
 		Returns(200, "OK", apis.PipelineRunBase{}).
 		Returns(400, "Bad Request", bcode.Bcode{}).
+		Filter(n.RBACService.CheckPerm("project/pipeline/pipelineRun", "get")).
 		Writes(apis.PipelineRunBase{}).Do(meta, projParam, pipelineParam, runParam))

-	ws.Route(ws.DELETE("/{pipelineName}/runs/{runName}").To(p.deletePipelineRun).
+	ws.Route(ws.DELETE("/{projectName}/pipelines/{pipelineName}/runs/{runName}").To(n.deletePipelineRun).
 		Doc("delete pipeline run").
 		Returns(200, "OK", apis.PipelineRunMeta{}).
 		Returns(400, "Bad Request", bcode.Bcode{}).
+		Filter(n.RBACService.CheckPerm("project/pipeline/pipelineRun", "delete")).
 		Writes(apis.PipelineRunMeta{}).Do(meta, projParam, pipelineParam, runParam))

 	// get pipeline run status
-	ws.Route(ws.GET("/{pipelineName}/runs/{runName}/status").To(p.getPipelineRunStatus).
+	ws.Route(ws.GET("/{projectName}/pipelines/{pipelineName}/runs/{runName}/status").To(n.getPipelineRunStatus).
 		Doc("get pipeline run status").
-		Returns(200, "OK", workflowv1alpha1.WorkflowRunStatus{}).
+		Returns(200, "OK", v1alpha1.WorkflowRunStatus{}).
 		Returns(400, "Bad Request", bcode.Bcode{}).
-		Writes(workflowv1alpha1.WorkflowRunStatus{}).Do(meta, projParam, pipelineParam, runParam))
+		Filter(n.RBACService.CheckPerm("project/pipeline/pipelineRun", "detail")).
+		Writes(v1alpha1.WorkflowRunStatus{}).Do(meta, projParam, pipelineParam, runParam))

 	// get pipeline run log
-	ws.Route(ws.GET("/{pipelineName}/runs/{runName}/log").To(p.getPipelineRunLog).
+	ws.Route(ws.GET("/{projectName}/pipelines/{pipelineName}/runs/{runName}/log").To(n.getPipelineRunLog).
 		Doc("get pipeline run log").
-		Param(ws.QueryParameter("step", "query by specific id").DataType("string")).
+		Param(ws.QueryParameter("step", "query by specific step name").DataType("string")).
 		Returns(200, "OK", apis.GetPipelineRunLogResponse{}).
 		Returns(400, "Bad Request", bcode.Bcode{}).
+		Filter(n.RBACService.CheckPerm("project/pipeline/pipelineRun", "detail")).
 		Writes(apis.GetPipelineRunLogResponse{}).Do(meta, projParam, pipelineParam, runParam))

 	// get pipeline run output
-	ws.Route(ws.GET("/{pipelineName}/runs/{runName}/output").To(p.getPipelineRunOutput).
+	ws.Route(ws.GET("/{projectName}/pipelines/{pipelineName}/runs/{runName}/output").To(n.getPipelineRunOutput).
 		Doc("get pipeline run output").
-		Param(ws.QueryParameter("step", "query by specific id").DataType("string")).
+		Param(ws.QueryParameter("step", "query by specific step name").DataType("string").Required(true)).
 		Returns(200, "OK", apis.GetPipelineRunOutputResponse{}).
 		Returns(400, "Bad Request", bcode.Bcode{}).
+		Filter(n.RBACService.CheckPerm("project/pipeline/pipelineRun", "detail")).
 		Writes(apis.GetPipelineRunOutputResponse{}).Do(meta, projParam, pipelineParam, runParam))

+	// get pipeline run input
+	ws.Route(ws.GET("/{projectName}/pipelines/{pipelineName}/runs/{runName}/input").To(n.getPipelineRunInput).
+		Doc("get pipeline run input").
+		Param(ws.QueryParameter("step", "query by specific step name").DataType("string").Required(true)).
+		Returns(200, "OK", apis.GetPipelineRunInputResponse{}).
+		Returns(400, "Bad Request", bcode.Bcode{}).
+		Filter(n.RBACService.CheckPerm("project/pipeline/pipelineRun", "detail")).
+		Writes(apis.GetPipelineRunInputResponse{}).Do(meta, projParam, pipelineParam, runParam))
+
+	ws.Filter(authCheckFilter)
+}
+
+// GetWebServiceRoute is the implementation of pipeline Interface
+func (n *pipelineAPIInterface) GetWebServiceRoute() *restful.WebService {
+	tags := []string{"pipeline"}
+	meta := func(builder *restful.RouteBuilder) {
+		builder.Metadata(restfulspec.KeyOpenAPITags, tags)
+	}
+
+	ws := new(restful.WebService)
+
+	ws.Path(versionPrefix+"/pipelines").
+		Consumes(restful.MIME_XML, restful.MIME_JSON).
+		Produces(restful.MIME_JSON, restful.MIME_XML).
+		Doc("api for project manage")
+
+	ws.Route(ws.GET("").To(n.listPipelines).
+		Doc("list pipelines").
+		Param(ws.QueryParameter("query", "Fuzzy search based on name or description").DataType("string")).
+		Param(ws.QueryParameter("projectName", "query pipelines within a project").DataType("string")).
+		Param(ws.QueryParameter("detailed", "query pipelines with detail").DataType("bool").DefaultValue("true")).
+		Returns(200, "OK", apis.ListPipelineResponse{}).
+		Returns(400, "Bad Request", bcode.Bcode{}).
+		Writes(apis.ListPipelineResponse{}).Do(meta))
+
 	ws.Filter(authCheckFilter)
 	return ws
 }

+type pipelineAPIInterface struct {
+	PipelineService service.PipelineService `inject:""`
+}
+
 // NewPipelineAPIInterface new pipeline manage APIInterface
 func NewPipelineAPIInterface() Interface {
 	return &pipelineAPIInterface{}
 }

-func (p *pipelineAPIInterface) listPipelines(req *restful.Request, res *restful.Response) {
-	var projetNames []string
-	if req.QueryParameter("project") != "" {
-		projetNames = append(projetNames, req.QueryParameter("project"))
+func (n *pipelineAPIInterface) listPipelines(req *restful.Request, res *restful.Response) {
+	var projectNames []string
+	if req.QueryParameter(Project) != "" {
+		projectNames = append(projectNames, req.QueryParameter(Project))
 	}
-	pipelines, err := p.PipelineService.ListPipelines(req.Request.Context(), apis.ListPipelineRequest{
-		Projects: projetNames,
+	_detailed := req.QueryParameter("detailed")
+	if _detailed == "" {
+		_detailed = "true"
+	}
+	detailed, err := strconv.ParseBool(_detailed)
+	if err != nil {
+		bcode.ReturnError(req, res, errors.Wrap(err, "invalid detailed param"))
+	}
+	pipelines, err := n.PipelineService.ListPipelines(req.Request.Context(), apis.ListPipelineRequest{
+		Projects: projectNames,
 		Query:    req.QueryParameter("query"),
+		Detailed: detailed,
 	})
 	if err != nil {
 		log.Logger.Errorf("list pipeline failure %s", err.Error())
@@ -223,15 +266,18 @@ func (p *pipelineAPIInterface) listPipelines(req *restful.Request, res *restful.
 	}
 }

-func (p *pipelineAPIInterface) getPipeline(req *restful.Request, res *restful.Response) {
-	pipeline := req.Request.Context().Value(apis.CtxKeyPipeline).(apis.PipelineBase)
+func (n *projectAPIInterface) getPipeline(req *restful.Request, res *restful.Response) {
+	pipeline, err := n.PipelineService.GetPipeline(req.Request.Context(), req.PathParameter(Pipeline), true)
+	if err != nil {
+		return
+	}
 	if err := res.WriteEntity(pipeline); err != nil {
 		bcode.ReturnError(req, res, err)
 		return
 	}
 }

-func (p *pipelineAPIInterface) createPipeline(req *restful.Request, res *restful.Response) {
+func (n *projectAPIInterface) createPipeline(req *restful.Request, res *restful.Response) {
 	var createReq apis.CreatePipelineRequest
 	if err := req.ReadEntity(&createReq); err != nil {
 		bcode.ReturnError(req, res, err)
@@ -241,19 +287,25 @@ func (p *pipelineAPIInterface) createPipeline(req *restful.Request, res *restful
 		bcode.ReturnError(req, res, err)
 		return
 	}
-	pipelineBase, err := p.PipelineService.CreatePipeline(req.Request.Context(), createReq)
+	pipelineBase, err := n.PipelineService.CreatePipeline(req.Request.Context(), createReq)
 	if err != nil {
 		log.Logger.Errorf("create pipeline failure %s", err.Error())
 		bcode.ReturnError(req, res, err)
 		return
 	}
+	_, err = n.ContextService.InitContext(req.Request.Context(), pipelineBase.Project.Name, pipelineBase.Name)
+	if err != nil {
+		log.Logger.Errorf("init pipeline context failure: %s", err.Error())
+		bcode.ReturnError(req, res, err)
+		return
+	}
 	if err := res.WriteEntity(pipelineBase); err != nil {
 		bcode.ReturnError(req, res, err)
 		return
 	}
 }

-func (p *pipelineAPIInterface) updatePipeline(req *restful.Request, res *restful.Response) {
+func (n *projectAPIInterface) updatePipeline(req *restful.Request, res *restful.Response) {
 	var updateReq apis.UpdatePipelineRequest
 	if err := req.ReadEntity(&updateReq); err != nil {
 		bcode.ReturnError(req, res, err)
@@ -263,8 +315,8 @@ func (p *pipelineAPIInterface) updatePipeline(req *restful.Request, res *restful
 		bcode.ReturnError(req, res, err)
 		return
 	}
-	base := req.Request.Context().Value(apis.CtxKeyPipeline).(apis.PipelineBase)
-	pipelineBase, err := p.PipelineService.UpdatePipeline(req.Request.Context(), base.Name, base.Project, updateReq)
+	pipeline := req.Request.Context().Value(&apis.CtxKeyPipeline).(apis.PipelineBase)
+	pipelineBase, err := n.PipelineService.UpdatePipeline(req.Request.Context(), pipeline.Name, updateReq)
 	if err != nil {
 		log.Logger.Errorf("update pipeline failure %s", err.Error())
 		bcode.ReturnError(req, res, err)
@@ -276,43 +328,43 @@ func (p *pipelineAPIInterface) updatePipeline(req *restful.Request, res *restful
 	}
 }

-func (p *pipelineAPIInterface) deletePipeline(req *restful.Request, res *restful.Response) {
+func (n *projectAPIInterface) deletePipeline(req *restful.Request, res *restful.Response) {
 	pipeline := req.Request.Context().Value(&apis.CtxKeyPipeline).(apis.PipelineBase)
-	err := p.PipelineService.DeletePipeline(req.Request.Context(), pipeline)
+	err := n.PipelineService.DeletePipeline(req.Request.Context(), pipeline)
 	if err != nil {
 		log.Logger.Errorf("delete pipeline failure %s", err.Error())
 		bcode.ReturnError(req, res, err)
 		return
 	}
-	if err := res.WriteEntity(pipeline.PipelineMeta); err != nil {
+	if err := res.WriteEntity(apis.EmptyResponse{}); err != nil {
 		bcode.ReturnError(req, res, err)
 		return
 	}
 }

-func (p *pipelineAPIInterface) runPipeline(req *restful.Request, res *restful.Response) {
+func (n *projectAPIInterface) runPipeline(req *restful.Request, res *restful.Response) {
 	var runReq apis.RunPipelineRequest
 	pipeline := req.Request.Context().Value(&apis.CtxKeyPipeline).(apis.PipelineBase)
-	if err := req.ReadEntity(runReq); err != nil {
+	if err := req.ReadEntity(&runReq); err != nil {
 		bcode.ReturnError(req, res, err)
 		return
 	}

-	err := p.PipelineService.RunPipeline(req.Request.Context(), pipeline, runReq)
+	run, err := n.PipelineService.RunPipeline(req.Request.Context(), pipeline, runReq)
 	if err != nil {
 		log.Logger.Errorf("run pipeline failure %s", err.Error())
 		bcode.ReturnError(req, res, err)
 		return
 	}
-	if err := res.WriteEntity(pipeline.PipelineMeta); err != nil {
+	if err := res.WriteEntity(run); err != nil {
 		bcode.ReturnError(req, res, err)
 		return
 	}
 }

-func (p *pipelineAPIInterface) stopPipeline(req *restful.Request, res *restful.Response) {
-	pipelineRun := req.Request.Context().Value(&apis.CtxKeyPipelineRun).(apis.PipelineRun)
-	err := p.PipelineRunService.StopPipelineRun(req.Request.Context(), pipelineRun.PipelineRunBase)
+func (n *projectAPIInterface) stopPipeline(req *restful.Request, res *restful.Response) {
+	pipelineRun := req.Request.Context().Value(&apis.CtxKeyPipelineRun).(*apis.PipelineRun)
+	err := n.PipelineRunService.StopPipelineRun(req.Request.Context(), pipelineRun.PipelineRunBase)
 	if err != nil {
 		log.Logger.Errorf("stop pipeline failure %s", err.Error())
 		bcode.ReturnError(req, res, err)
@@ -324,9 +376,9 @@ func (p *pipelineAPIInterface) stopPipeline(req *restful.Request, res *restful.R
 	}
 }

-func (p *pipelineAPIInterface) listPipelineRuns(req *restful.Request, res *restful.Response) {
+func (n *projectAPIInterface) listPipelineRuns(req *restful.Request, res *restful.Response) {
 	pipeline := req.Request.Context().Value(&apis.CtxKeyPipeline).(apis.PipelineBase)
-	pipelineRuns, err := p.PipelineRunService.ListPipelineRuns(req.Request.Context(), pipeline)
+	pipelineRuns, err := n.PipelineRunService.ListPipelineRuns(req.Request.Context(), pipeline)
 	if err != nil {
 		log.Logger.Errorf("list pipeline runs failure %s", err.Error())
 		bcode.ReturnError(req, res, err)
@@ -338,33 +390,65 @@ func (p *pipelineAPIInterface) listPipelineRuns(req *restful.Request, res *restf
 	}
 }

-func (p *pipelineAPIInterface) getPipelineRun(req *restful.Request, res *restful.Response) {
-	pipelineRun := req.Request.Context().Value(&apis.CtxKeyPipelineRun).(apis.PipelineRun)
+func (n *projectAPIInterface) getPipelineRun(req *restful.Request, res *restful.Response) {
+	pipelineRun := req.Request.Context().Value(&apis.CtxKeyPipelineRun).(*apis.PipelineRun)
 	if err := res.WriteEntity(pipelineRun.PipelineRunBase); err != nil {
 		bcode.ReturnError(req, res, err)
 		return
 	}
 }

-func (p *pipelineAPIInterface) getPipelineRunStatus(req *restful.Request, res *restful.Response) {
-	pipelineRun := req.Request.Context().Value(&apis.CtxKeyPipelineRun).(apis.PipelineRun)
+func (n *projectAPIInterface) getPipelineRunStatus(req *restful.Request, res *restful.Response) {
+	pipelineRun := req.Request.Context().Value(&apis.CtxKeyPipelineRun).(*apis.PipelineRun)
 	if err := res.WriteEntity(pipelineRun.Status); err != nil {
 		bcode.ReturnError(req, res, err)
 		return
 	}
 }

-func (p *pipelineAPIInterface) getPipelineRunLog(req *restful.Request, res *restful.Response) {
-
+func (n *projectAPIInterface) getPipelineRunLog(req *restful.Request, res *restful.Response) {
+	pipelineRun := req.Request.Context().Value(&apis.CtxKeyPipelineRun).(*apis.PipelineRun)
+	logs, err := n.PipelineRunService.GetPipelineRunLog(req.Request.Context(), *pipelineRun, req.QueryParameter("step"))
+	if err != nil {
+		log.Logger.Errorf("get pipeline run log failure %s", err.Error())
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	if err := res.WriteEntity(logs); err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
 }

-func (p *pipelineAPIInterface) getPipelineRunOutput(req *restful.Request, res *restful.Response) {
-
+func (n *projectAPIInterface) getPipelineRunOutput(req *restful.Request, res *restful.Response) {
+	pipelineRun := req.Request.Context().Value(&apis.CtxKeyPipelineRun).(*apis.PipelineRun)
+	output, err := n.PipelineRunService.GetPipelineRunOutput(req.Request.Context(), *pipelineRun, req.QueryParameter("step"))
+	if err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	if err := res.WriteEntity(output); err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
 }

-func (p *pipelineAPIInterface) deletePipelineRun(req *restful.Request, res *restful.Response) {
-	pipelineRun := req.Request.Context().Value(&apis.CtxKeyPipelineRun).(apis.PipelineRun)
-	err := p.PipelineRunService.DeletePipelineRun(req.Request.Context(), pipelineRun.PipelineRunMeta)
+func (n *projectAPIInterface) getPipelineRunInput(req *restful.Request, res *restful.Response) {
+	pipelineRun := req.Request.Context().Value(&apis.CtxKeyPipelineRun).(*apis.PipelineRun)
+	input, err := n.PipelineRunService.GetPipelineRunInput(req.Request.Context(), *pipelineRun, req.QueryParameter("step"))
+	if err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	if err := res.WriteEntity(input); err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+}
+
+func (n *projectAPIInterface) deletePipelineRun(req *restful.Request, res *restful.Response) {
+	pipelineRun := req.Request.Context().Value(&apis.CtxKeyPipelineRun).(*apis.PipelineRun)
+	err := n.PipelineRunService.DeletePipelineRun(req.Request.Context(), pipelineRun.PipelineRunMeta)
 	if err != nil {
 		log.Logger.Errorf("delete pipeline run failure %s", err.Error())
 		bcode.ReturnError(req, res, err)
@@ -376,11 +460,11 @@ func (p *pipelineAPIInterface) deletePipelineRun(req *restful.Request, res *rest
 	}
 }

-func (p *pipelineAPIInterface) listContextValues(req *restful.Request, res *restful.Response) {
+func (n *projectAPIInterface) listContextValues(req *restful.Request, res *restful.Response) {
 	pipeline := req.Request.Context().Value(&apis.CtxKeyPipeline).(apis.PipelineBase)
-	contextValues, err := p.ContextService.ListContexts(req.Request.Context(), pipeline.Project, pipeline.Name)
+	contextValues, err := n.ContextService.ListContexts(req.Request.Context(), pipeline.Project.Name, pipeline.Name)
 	if err != nil {
-		log.Logger.Errorf("list context values failure %s", err.Error())
+		log.Logger.Errorf("list context values failure: %s", err.Error())
 		bcode.ReturnError(req, res, err)
 		return
 	}
@@ -390,7 +474,7 @@ func (p *pipelineAPIInterface) listContextValues(req *restful.Request, res *rest
 	}
 }

-func (p *pipelineAPIInterface) createContextValue(req *restful.Request, res *restful.Response) {
+func (n *projectAPIInterface) createContextValue(req *restful.Request, res *restful.Response) {
 	pipeline := req.Request.Context().Value(&apis.CtxKeyPipeline).(apis.PipelineBase)
 	var createReq apis.CreateContextValuesRequest
 	if err := req.ReadEntity(&createReq); err != nil {
@@ -403,7 +487,7 @@ func (p *pipelineAPIInterface) createContextValue(req *restful.Request, res *res
 	}

 	pipelineCtx := apis.Context(createReq)
-	_, err := p.ContextService.CreateContext(req.Request.Context(), pipeline.Project, pipeline.Name, pipelineCtx)
+	_, err := n.ContextService.CreateContext(req.Request.Context(), pipeline.Project.Name, pipeline.Name, pipelineCtx)
 	if err != nil {
 		log.Logger.Errorf("create context failure %s", err.Error())
 		bcode.ReturnError(req, res, err)
@@ -415,8 +499,8 @@ func (p *pipelineAPIInterface) createContextValue(req *restful.Request, res *res
 	}
 }

-func (p *pipelineAPIInterface) updateContextValue(req *restful.Request, res *restful.Response) {
-	plCtx := req.Request.Context().Value(&apis.CtxKeyPipelineContex).(apis.Context)
+func (n *projectAPIInterface) updateContextValue(req *restful.Request, res *restful.Response) {
+	plCtx := req.Request.Context().Value(&apis.CtxKeyPipelineContext).(apis.Context)
 	pipeline := req.Request.Context().Value(&apis.CtxKeyPipeline).(apis.PipelineBase)
 	var updateReq apis.UpdateContextValuesRequest
 	if err := req.ReadEntity(&updateReq); err != nil {
@@ -428,7 +512,7 @@ func (p *pipelineAPIInterface) updateContextValue(req *restful.Request, res *res
 		return
 	}
 	pipelineCtx := apis.Context{Name: plCtx.Name, Values: updateReq.Values}
-	_, err := p.ContextService.UpdateContext(req.Request.Context(), pipeline.Project, pipeline.Name, pipelineCtx)
+	_, err := n.ContextService.UpdateContext(req.Request.Context(), pipeline.Project.Name, pipeline.Name, pipelineCtx)
 	if err != nil {
 		log.Logger.Errorf("update context failure %s", err.Error())
 		bcode.ReturnError(req, res, err)
@@ -440,10 +524,10 @@ func (p *pipelineAPIInterface) updateContextValue(req *restful.Request, res *res
 	}
 }

-func (p *pipelineAPIInterface) deleteContextValue(req *restful.Request, res *restful.Response) {
-	plCtx := req.Request.Context().Value(&apis.CtxKeyPipelineContex).(apis.Context)
+func (n *projectAPIInterface) deleteContextValue(req *restful.Request, res *restful.Response) {
+	plCtx := req.Request.Context().Value(&apis.CtxKeyPipelineContext).(apis.Context)
 	pipeline := req.Request.Context().Value(&apis.CtxKeyPipeline).(apis.PipelineBase)
-	err := p.ContextService.DeleteContext(req.Request.Context(), pipeline.Project, pipeline.Name, plCtx.Name)
+	err := n.ContextService.DeleteContext(req.Request.Context(), pipeline.Project.Name, pipeline.Name, plCtx.Name)
 	if err != nil {
 		log.Logger.Errorf("delete context failure %s", err.Error())
 		bcode.ReturnError(req, res, err)
@@ -455,8 +539,18 @@ func (p *pipelineAPIInterface) deleteContextValue(req *restful.Request, res *res
 	}
 }

-func (p *pipelineAPIInterface) pipelineCheckFilter(req *restful.Request, res *restful.Response, chain *restful.FilterChain) {
-	pipeline, err := p.PipelineService.GetPipeline(req.Request.Context(), req.PathParameter("pipelineName"), req.QueryParameter("projectName"))
+func (n *projectAPIInterface) projectCheckFilter(req *restful.Request, res *restful.Response, chain *restful.FilterChain) {
+	project, err := n.ProjectService.GetProject(req.Request.Context(), req.PathParameter(Project))
+	if err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	req.Request = req.Request.WithContext(context.WithValue(req.Request.Context(), &apis.CtxKeyProject, project))
+	chain.ProcessFilter(req, res)
+}
+
+func (n *projectAPIInterface) pipelineCheckFilter(req *restful.Request, res *restful.Response, chain *restful.FilterChain) {
+	pipeline, err := n.PipelineService.GetPipeline(req.Request.Context(), req.PathParameter(Pipeline), false)
 	if err != nil {
 		bcode.ReturnError(req, res, err)
 		return
@@ -465,8 +559,8 @@ func (p *pipelineAPIInterface) pipelineCheckFilter(req *restful.Request, res *re
 	chain.ProcessFilter(req, res)
 }

-func (p *pipelineAPIInterface) pipelineContextCheckFilter(req *restful.Request, res *restful.Response, chain *restful.FilterChain) {
-	contexts, err := p.ContextService.ListContexts(req.Request.Context(), req.PathParameter("pipelineName"), req.QueryParameter("projectName"))
+func (n *projectAPIInterface) pipelineContextCheckFilter(req *restful.Request, res *restful.Response, chain *restful.FilterChain) {
+	contexts, err := n.ContextService.ListContexts(req.Request.Context(), req.PathParameter(Project), req.PathParameter(Pipeline))
 	if err != nil {
 		bcode.ReturnError(req, res, err)
 		return
@@ -477,20 +571,22 @@ func (p *pipelineAPIInterface) pipelineContextCheckFilter(req *restful.Request,
 		bcode.ReturnError(req, res, bcode.ErrContextNotFound)
 		return
 	}
-	req.Request = req.Request.WithContext(context.WithValue(req.Request.Context(), &apis.CtxKeyPipelineContex, apis.Context{
+	req.Request = req.Request.WithContext(context.WithValue(req.Request.Context(), &apis.CtxKeyPipelineContext, apis.Context{
 		Name:   contextName,
 		Values: contextValue,
 	}))
 	chain.ProcessFilter(req, res)
 }

-func (p *pipelineAPIInterface) pipelineRunCheckFilter(req *restful.Request, res *restful.Response, chain *restful.FilterChain) {
+func (n *projectAPIInterface) pipelineRunCheckFilter(req *restful.Request, res *restful.Response, chain *restful.FilterChain) {
 	meta := apis.PipelineRunMeta{
-		PipelineName:    req.PathParameter(string(Pipeline)),
-		Project:         req.QueryParameter(string(Project)),
-		PipelineRunName: req.PathParameter(string(PipelineRun)),
+		PipelineName: req.PathParameter(Pipeline),
+		Project: apis.NameAlias{
+			Name: req.PathParameter(Project),
+		},
+		PipelineRunName: req.PathParameter(PipelineRun),
 	}
-	run, err := p.PipelineRunService.GetPipelineRun(req.Request.Context(), meta)
+	run, err := n.PipelineRunService.GetPipelineRun(req.Request.Context(), meta)
 	if err != nil {
 		bcode.ReturnError(req, res, err)
 		return
@@ -29,10 +29,14 @@ import (
 )

 type projectAPIInterface struct {
-	RbacService    service.RBACService    `inject:""`
-	ProjectService service.ProjectService `inject:""`
-	TargetService  service.TargetService  `inject:""`
-	ConfigService  service.ConfigService  `inject:""`
+	RbacService        service.RBACService        `inject:""`
+	ProjectService     service.ProjectService     `inject:""`
+	TargetService      service.TargetService      `inject:""`
+	ConfigService      service.ConfigService      `inject:""`
+	PipelineService    service.PipelineService    `inject:""`
+	PipelineRunService service.PipelineRunService `inject:""`
+	ContextService     service.ContextService     `inject:""`
+	RBACService        service.RBACService        `inject:""`
 }

 // NewProjectAPIInterface new project APIInterface
@@ -305,6 +309,7 @@ func (n *projectAPIInterface) GetWebServiceRoute() *restful.WebService {
 		Returns(400, "Bad Request", bcode.Bcode{}).
 		Writes(apis.ListTerraformProviderResponse{}))

+	initPipelineRoutes(ws, n)
 	ws.Filter(authCheckFilter)
 	return ws
 }
@@ -21,4 +21,22 @@ var (
 	ErrContextNotFound = NewBcode(400, 17001, "pipeline context is not found")
 	// ErrContextAlreadyExist means the certain context already exists
 	ErrContextAlreadyExist = NewBcode(400, 17002, "pipeline context of pipeline already exist")
+	// ErrGetPipelineInfo means failed to get pipeline info
+	ErrGetPipelineInfo = NewBcode(400, 17003, "get pipeline info failed")
+	// ErrPipelineNotExist means specific pipeline not found
+	ErrPipelineNotExist = NewBcode(404, 17004, "failed to find log pods")
+	// ErrGetPodsLogs means failed to get pods logs
+	ErrGetPodsLogs = NewBcode(500, 17006, "failed to get pods logs")
+	// ErrReadSourceLog means failed to read source log
+	ErrReadSourceLog = NewBcode(500, 17007, "failed to read log from URL source")
+	// ErrGetContextBackendData means failed to get context backend data
+	ErrGetContextBackendData = NewBcode(500, 17008, "failed to get context backend data")
+	// ErrNoSteps means pipeline doesn't have a step
+	ErrNoSteps = NewBcode(400, 17009, "pipeline step number is zero")
+	// ErrPipelineExist means the pipeline is exist
+	ErrPipelineExist = NewBcode(400, 17010, "the pipeline is exist")
+	// ErrPipelineRunFinished means pipeline run is finished
+	ErrPipelineRunFinished = NewBcode(400, 17011, "pipeline run is finished")
+	// ErrWrongMode means the pipeline run mode is wrong
+	ErrWrongMode = NewBcode(400, 17012, "wrong pipeline run mode, only \"DAG\" and \"StepByStep\" are supported")
 )
@@ -21,7 +21,19 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
+	"regexp"
 	"strings"
+	"text/template"
+	"time"
+
+	"github.com/fatih/color"
+	"github.com/pkg/errors"
+	"github.com/wercker/stern/stern"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/selection"
+	"k8s.io/client-go/kubernetes"
+
+	querytypes "github.com/oam-dev/kubevela/pkg/velaql/providers/query/types"

 	authv1 "k8s.io/api/authentication/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -199,3 +211,125 @@ func IsClusterScope(gvk schema.GroupVersionKind, mapper meta.RESTMapper) (bool,
 	isClusterScope := len(mappings) > 0 && mappings[0].Scope.Name() == meta.RESTScopeNameRoot
 	return isClusterScope, err
 }
+
+// GetPodsLogs get logs from pods
+func GetPodsLogs(ctx context.Context, config *rest.Config, containerName string, selectPods []*querytypes.PodBase, tmpl string, logC chan<- string, tailLines *int64) error {
+	if err := verifyPods(selectPods); err != nil {
+		return err
+	}
+	podRegex := getPodRegex(selectPods)
+	pods, err := regexp.Compile(podRegex)
+	if err != nil {
+		return fmt.Errorf("fail to compile '%s' for logs query", podRegex)
+	}
+	container := regexp.MustCompile(".*")
+	if containerName != "" {
+		container = regexp.MustCompile(containerName + ".*")
+	}
+	// These pods are from the same namespace, so we can use the first one to get the namespace
+	namespace := selectPods[0].Metadata.Namespace
+	selector := labels.NewSelector()
+	// Only use the labels to select pod if query one pod's log. It is only used when query vela-core log
+	if len(selectPods) == 1 {
+		for k, v := range selectPods[0].Metadata.Labels {
+			req, _ := labels.NewRequirement(k, selection.Equals, []string{v})
+			if req != nil {
+				selector = selector.Add(*req)
+			}
+		}
+	}
+	clientSet, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		return err
+	}
+	added, removed, err := stern.Watch(ctx,
+		clientSet.CoreV1().Pods(namespace),
+		pods,
+		container,
+		nil,
+		[]stern.ContainerState{stern.RUNNING, stern.TERMINATED},
+		selector,
+	)
+	if err != nil {
+		return err
+	}
+	tails := make(map[string]*stern.Tail)
+
+	funs := map[string]interface{}{
+		"json": func(in interface{}) (string, error) {
+			b, err := json.Marshal(in)
+			if err != nil {
+				return "", err
+			}
+			return string(b), nil
+		},
+		"color": func(color color.Color, text string) string {
+			return color.SprintFunc()(text)
+		},
+	}
+	template, err := template.New("log").Funcs(funs).Parse(tmpl)
+	if err != nil {
+		return errors.Wrap(err, "unable to parse template")
+	}
+
+	go func() {
+		for p := range added {
+			id := p.GetID()
+			if tails[id] != nil {
+				continue
+			}
+			// 48h
+			dur, _ := time.ParseDuration("48h")
+			tail := stern.NewTail(p.Namespace, p.Pod, p.Container, template, &stern.TailOptions{
+				Timestamps:   true,
+				SinceSeconds: int64(dur.Seconds()),
+				Exclude:      nil,
+				Include:      nil,
+				Namespace:    false,
+				TailLines:    tailLines, // default for all logs
+			})
+			tails[id] = tail
+
+			tail.Start(ctx, clientSet.CoreV1().Pods(p.Namespace), logC)
+		}
+	}()
+
+	go func() {
+		for p := range removed {
+			id := p.GetID()
+			if tails[id] == nil {
+				continue
+			}
+			tails[id].Close()
+			delete(tails, id)
+		}
+	}()
+
+	<-ctx.Done()
+	close(logC)
+	return nil
+}
+
+func getPodRegex(pods []*querytypes.PodBase) string {
+	var podNames []string
+	for _, pod := range pods {
+		podNames = append(podNames, fmt.Sprintf("(%s.*)", pod.Metadata.Name))
+	}
+	return strings.Join(podNames, "|")
+}
+
+func verifyPods(pods []*querytypes.PodBase) error {
+	if len(pods) == 0 {
+		return errors.New("no pods selected")
+	}
+	if len(pods) == 1 {
+		return nil
+	}
+	namespace := pods[0].Metadata.Namespace
+	for _, pod := range pods {
+		if pod.Metadata.Namespace != namespace {
+			return errors.New("cannot select pods from different namespaces")
+		}
+	}
+	return nil
+}
@@ -18,24 +18,16 @@ package cli

 import (
 	"context"
-	"encoding/json"
 	"fmt"
-	"regexp"
 	"strings"
-	"text/template"
-	"time"

 	"github.com/fatih/color"
-	"github.com/pkg/errors"
 	"github.com/spf13/cobra"
-	"github.com/wercker/stern/stern"
-	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/selection"
-	"k8s.io/client-go/kubernetes"

 	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
 	"github.com/oam-dev/kubevela/apis/types"
 	"github.com/oam-dev/kubevela/pkg/multicluster"
+	"github.com/oam-dev/kubevela/pkg/utils"
 	"github.com/oam-dev/kubevela/pkg/utils/common"
 	"github.com/oam-dev/kubevela/pkg/utils/util"
 	querytypes "github.com/oam-dev/kubevela/pkg/velaql/providers/query/types"
@@ -99,45 +91,25 @@ type Args struct {
 }

 func (l *Args) printPodLogs(ctx context.Context, ioStreams util.IOStreams, selectPod *querytypes.PodBase, filters []string) error {
-	pod, err := regexp.Compile(selectPod.Metadata.Name + ".*")
-	if err != nil {
-		return fmt.Errorf("fail to compile '%s' for logs query", selectPod.Metadata.Name+".*")
-	}
-	container := regexp.MustCompile(".*")
-	if l.ContainerName != "" {
-		container = regexp.MustCompile(l.ContainerName + ".*")
-	}
-	namespace := selectPod.Metadata.Namespace
-	selector := labels.NewSelector()
-	for k, v := range selectPod.Metadata.Labels {
-		req, _ := labels.NewRequirement(k, selection.Equals, []string{v})
-		if req != nil {
-			selector = selector.Add(*req)
-		}
-	}
-
 	config, err := l.Args.GetConfig()
 	if err != nil {
 		return err
 	}
-	clientSet, err := kubernetes.NewForConfig(config)
-	if err != nil {
-		return err
-	}
-	added, removed, err := stern.Watch(ctx,
-		clientSet.CoreV1().Pods(namespace),
-		pod,
-		container,
-		nil,
-		[]stern.ContainerState{stern.RUNNING, stern.TERMINATED},
-		selector,
-	)
-	if err != nil {
-		return err
-	}
-	tails := make(map[string]*stern.Tail)
 	logC := make(chan string, 1024)

+	var t string
+	switch l.Output {
+	case "default":
+		if color.NoColor {
+			t = "{{.ContainerName}} {{.Message}}"
+		} else {
+			t = "{{color .ContainerColor .ContainerName}} {{.Message}}"
+		}
+	case "raw":
+		t = "{{.Message}}"
+	case "json":
+		t = "{{json .}}\n"
+	}
 	go func() {
 		for {
 			select {
@@ -158,71 +130,11 @@ func (l *Args) printPodLogs(ctx context.Context, ioStreams util.IOStreams, selec
 		}
 	}()

-	var t string
-	switch l.Output {
-	case "default":
-		if color.NoColor {
-			t = "{{.ContainerName}} {{.Message}}"
-		} else {
-			t = "{{color .ContainerColor .ContainerName}} {{.Message}}"
-		}
-	case "raw":
-		t = "{{.Message}}"
-	case "json":
-		t = "{{json .}}\n"
-	}
-	funs := map[string]interface{}{
-		"json": func(in interface{}) (string, error) {
-			b, err := json.Marshal(in)
-			if err != nil {
-				return "", err
-			}
-			return string(b), nil
-		},
-		"color": func(color color.Color, text string) string {
-			return color.SprintFunc()(text)
-		},
-	}
-	template, err := template.New("log").Funcs(funs).Parse(t)
+	err = utils.GetPodsLogs(ctx, config, l.ContainerName, []*querytypes.PodBase{selectPod}, t, logC, nil)
 	if err != nil {
-		return errors.Wrap(err, "unable to parse template")
+		return err
 	}

-	go func() {
-		for p := range added {
-			id := p.GetID()
-			if tails[id] != nil {
-				continue
-			}
-			// 48h
-			dur, _ := time.ParseDuration("48h")
-			tail := stern.NewTail(p.Namespace, p.Pod, p.Container, template, &stern.TailOptions{
-				Timestamps:   true,
-				SinceSeconds: int64(dur.Seconds()),
-				Exclude:      nil,
-				Include:      nil,
-				Namespace:    false,
-				TailLines:    nil, // default for all logs
-			})
-			tails[id] = tail
-
-			tail.Start(ctx, clientSet.CoreV1().Pods(p.Namespace), logC)
-		}
-	}()
-
-	go func() {
-		for p := range removed {
-			id := p.GetID()
-			if tails[id] == nil {
-				continue
-			}
-			tails[id].Close()
-			delete(tails, id)
-		}
-	}()
-
-	<-ctx.Done()
-
 	return nil
 }

@@ -0,0 +1,363 @@
+/*
+Copyright 2021 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 (
+	"context"
+	"net/http"
+	"strconv"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/kubevela/workflow/api/v1alpha1"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
+	"github.com/oam-dev/kubevela/pkg/apiserver/domain/model"
+	apisv1 "github.com/oam-dev/kubevela/pkg/apiserver/interfaces/api/dto/v1"
+	"github.com/oam-dev/kubevela/pkg/oam/util"
+	"github.com/oam-dev/kubevela/pkg/utils/common"
+)
+
+var testPipelineSteps []v1alpha1.WorkflowStep
+
+func init() {
+	rawProps := []byte(`{"url":"https://api.github.com/repos/kubevela/kubevela"}`)
+	testPipelineSteps = []v1alpha1.WorkflowStep{
+		{
+			WorkflowStepBase: v1alpha1.WorkflowStepBase{
+				Name: "request",
+				Type: "request",
+				Outputs: v1alpha1.StepOutputs{
+					{
+						ValueFrom: "import \"strconv\"\n\"Current star count: \" + strconv.FormatInt(response[\"stargazers_count\"], 10)\n",
+						Name:      "stars",
+					},
+				},
+				Properties: &runtime.RawExtension{
+					Raw: rawProps,
+				},
+			},
+		},
+	}
+}
+
+var _ = Describe("Test the rest api about the pipeline", func() {
+	var (
+		projectName1    = testNSprefix + strconv.FormatInt(time.Now().UnixNano(), 10)
+		pipelineName    = "test-pipeline"
+		description     = "amazing pipeline"
+		contextName     = "test-context"
+		contextKey      = "test-key"
+		contextVal      = "test-val"
+		pipelineRunName string
+	)
+	defer GinkgoRecover()
+	It("create project and apply definitions", func() {
+		defer GinkgoRecover()
+		var req = apisv1.CreateProjectRequest{
+			Name:        projectName1,
+			Description: "KubeVela Project",
+		}
+		res := post("/projects", req)
+		var projectBase apisv1.ProjectBase
+		Expect(decodeResponseBody(res, &projectBase)).Should(Succeed())
+		Expect(cmp.Diff(projectBase.Name, req.Name)).Should(BeEmpty())
+		Expect(cmp.Diff(projectBase.Description, req.Description)).Should(BeEmpty())
+
+		def1 := new(v1beta1.WorkflowStepDefinition)
+		def2 := new(v1beta1.WorkflowStepDefinition)
+		Expect(common.ReadYamlToObject("./testdata/request.yaml", def1)).Should(BeNil())
+		Expect(k8sClient.Create(context.Background(), def1)).Should(SatisfyAny(BeNil(), &util.AlreadyExistMatcher{}))
+		Expect(common.ReadYamlToObject("./testdata/log.yaml", def2)).Should(BeNil())
+		Expect(k8sClient.Create(context.Background(), def2)).Should(SatisfyAny(BeNil(), &util.AlreadyExistMatcher{}))
+	})
+
+	It("create pipeline", func() {
+		var req = apisv1.CreatePipelineRequest{
+			Name:        pipelineName,
+			Description: description,
+			Spec: v1alpha1.WorkflowSpec{
+				Steps: testPipelineSteps,
+			},
+		}
+		res := post("/projects/"+projectName1+"/pipelines", req)
+		var pipeline apisv1.PipelineBase
+		Expect(decodeResponseBody(res, &pipeline)).Should(Succeed())
+		Expect(cmp.Diff(pipeline.Name, req.Name)).Should(BeEmpty())
+		Expect(len(pipeline.Spec.Steps)).Should(Equal(len(req.Spec.Steps)))
+	})
+
+	It("create context", func() {
+		var req = apisv1.CreateContextValuesRequest{
+			Name: contextName,
+			Values: []model.Value{
+				{
+					Key:   contextKey,
+					Value: contextVal,
+				},
+			},
+		}
+		res := post("/projects/"+projectName1+"/pipelines/"+pipelineName+"/contexts", req)
+		var context apisv1.Context
+		Expect(decodeResponseBody(res, &context)).Should(Succeed())
+		Expect(cmp.Diff(context.Name, req.Name)).Should(BeEmpty())
+		Expect(cmp.Diff(context.Values, req.Values)).Should(BeEmpty())
+	})
+
+	It("get contexts", func() {
+		res := get("/projects/" + projectName1 + "/pipelines/" + pipelineName + "/contexts")
+		var contexs apisv1.ListContextValueResponse
+		Expect(decodeResponseBody(res, &contexs)).Should(Succeed())
+		Expect(len(contexs.Contexts)).Should(Equal(1))
+		ctx, ok := contexs.Contexts[contextName]
+		Expect(ok).Should(BeTrue())
+		Expect(len(ctx)).Should(Equal(1))
+	})
+
+	It("update context", func() {
+		var req = apisv1.UpdateContextValuesRequest{
+			Values: []model.Value{
+				{
+					Key:   contextKey,
+					Value: "new-val",
+				},
+			},
+		}
+		res := put("/projects/"+projectName1+"/pipelines/"+pipelineName+"/contexts/"+contextName, req)
+		var context apisv1.Context
+		Expect(res.StatusCode).Should(Equal(http.StatusOK))
+		Expect(decodeResponseBody(res, &context)).Should(Succeed())
+
+		By("check the context value")
+		Expect(cmp.Diff(context.Values[0].Value, "new-val")).Should(BeEmpty())
+	})
+
+	It("update pipeline", func() {
+		rawProps := []byte(`{"url":"https://api.github.com/repos/kubevela/kubevela"}`)
+		newSteps := make([]v1alpha1.WorkflowStep, 0)
+		newSteps = append(newSteps, v1alpha1.WorkflowStep{
+			SubSteps: []v1alpha1.WorkflowStepBase{
+				{
+					Name: "request1",
+					Type: "request",
+					Outputs: v1alpha1.StepOutputs{
+						{
+							ValueFrom: "import \"strconv\"\n\"Current star count: \" + strconv.FormatInt(response[\"stargazers_count\"], 10)\n",
+							Name:      "stars",
+						},
+					},
+					Properties: &runtime.RawExtension{
+						Raw: rawProps,
+					},
+				},
+				{
+					Name: "request2",
+					Type: "request",
+					Outputs: v1alpha1.StepOutputs{
+						{
+							ValueFrom: "import \"strconv\"\n\"Current star count: \" + strconv.FormatInt(response[\"stargazers_count\"], 10)\n",
+							Name:      "stars-copy",
+						},
+					},
+					Properties: &runtime.RawExtension{
+						Raw: rawProps,
+					},
+				},
+			},
+			WorkflowStepBase: v1alpha1.WorkflowStepBase{
+				Name: "request-group",
+				Type: "step-group",
+			},
+		})
+		newSteps = append(newSteps, v1alpha1.WorkflowStep{
+			WorkflowStepBase: v1alpha1.WorkflowStepBase{
+				Name: "log",
+				Type: "log",
+				Inputs: v1alpha1.StepInputs{
+					{
+						ParameterKey: "data",
+						From:         "stars",
+					},
+				},
+			},
+		})
+		var req = apisv1.UpdatePipelineRequest{
+			Description: description,
+			Spec: v1alpha1.WorkflowSpec{
+				Steps: newSteps,
+			},
+		}
+		res := put("/projects/"+projectName1+"/pipelines/"+pipelineName, req)
+		var pipeline apisv1.PipelineBase
+		Expect(decodeResponseBody(res, &pipeline)).Should(Succeed())
+		Expect(len(pipeline.Spec.Steps)).Should(Equal(len(req.Spec.Steps)))
+	})
+
+	It("run pipeline", func() {
+		var req = apisv1.RunPipelineRequest{
+			Mode: v1alpha1.WorkflowExecuteMode{
+				Steps:    "StepByStep",
+				SubSteps: "DAG",
+			},
+			ContextName: contextName,
+		}
+		res := post("/projects/"+projectName1+"/pipelines/"+pipelineName+"/run", req)
+		var run apisv1.PipelineRun
+		Expect(decodeResponseBody(res, &run)).Should(Succeed())
+		Expect(run.PipelineRunName).ShouldNot(BeEmpty())
+		pipelineRunName = run.PipelineRunName
+	})
+
+	It("list pipeline", func() {
+		res := get("/pipelines?query=amazing")
+		var pipelines apisv1.ListPipelineResponse
+		Expect(decodeResponseBody(res, &pipelines)).Should(Succeed())
+		Expect(pipelines.Total).Should(BeNumerically("==", 1))
+		Expect(pipelines.Pipelines[0].Name).Should(Equal(pipelineName))
+	})
+
+	It("get pipeline", func() {
+		Eventually(func(g Gomega) {
+			res := get("/projects/" + projectName1 + "/pipelines/" + pipelineName)
+			var pipeline apisv1.GetPipelineResponse
+			g.Expect(decodeResponseBody(res, &pipeline)).Should(Succeed())
+			g.Expect(pipeline.Name).Should(Equal(pipelineName))
+			g.Expect(pipeline.Description).Should(Equal(description))
+			g.Expect(pipeline.PipelineInfo.LastRun).ShouldNot(BeNil())
+			g.Expect(pipeline.PipelineInfo.RunStat.Total).Should(Equal(apisv1.RunStatInfo{Total: 1, Success: 1}))
+			g.Expect(len(pipeline.PipelineInfo.RunStat.Week)).Should(Equal(7))
+		}, 10*time.Second, 1*time.Second).Should(Succeed())
+	})
+
+	It("list pipeline runs", func() {
+		res := get("/projects/" + projectName1 + "/pipelines/" + pipelineName + "/runs")
+		var runs apisv1.ListPipelineRunResponse
+		Expect(decodeResponseBody(res, &runs)).Should(Succeed())
+		Expect(runs.Total).Should(BeNumerically("==", 1))
+	})
+
+	It("get pipeline run", func() {
+		res := get("/projects/" + projectName1 + "/pipelines/" + pipelineName + "/runs/" + pipelineRunName)
+		var run apisv1.PipelineRunBase
+		Expect(decodeResponseBody(res, &run)).Should(Succeed())
+		Expect(run.PipelineRunName).Should(Equal(pipelineRunName))
+	})
+
+	It("get pipeline run status", func() {
+		Eventually(func(g Gomega) {
+			res := get("/projects/" + projectName1 + "/pipelines/" + pipelineName + "/runs/" + pipelineRunName + "/status")
+			var status v1alpha1.WorkflowRunStatus
+			g.Expect(decodeResponseBody(res, &status)).Should(Succeed())
+			g.Expect(status.Finished).Should(Equal(true))
+			g.Expect(status.Phase).Should(Equal(v1alpha1.WorkflowStateSucceeded))
+			g.Expect(status.Message).Should(BeEmpty())
+		}, 100*time.Second, 1*time.Second).Should(Succeed())
+	})
+
+	It("get pipeline run output", func() {
+		outputStep := "request1"
+		res := get("/projects/" + projectName1 + "/pipelines/" + pipelineName + "/runs/" + pipelineRunName + "/output?step=" + outputStep)
+		var output apisv1.GetPipelineRunOutputResponse
+		Expect(decodeResponseBody(res, &output)).Should(Succeed())
+		Expect(output.StepOutputs).Should(HaveLen(1))
+		Expect(output.StepOutputs[0].Name).Should(Equal(outputStep))
+		Expect(output.StepOutputs[0].Values).Should(HaveLen(1))
+		Expect(output.StepOutputs[0].Values[0].Value).ShouldNot(BeEmpty())
+	})
+
+	It("get pipeline run input", func() {
+		inputStep := "log"
+		res := get("/projects/" + projectName1 + "/pipelines/" + pipelineName + "/runs/" + pipelineRunName + "/input?step=" + inputStep)
+		var input apisv1.GetPipelineRunInputResponse
+		Expect(decodeResponseBody(res, &input)).Should(Succeed())
+		Expect(input.StepInputs).Should(HaveLen(1))
+		Expect(input.StepInputs[0].Name).Should(Equal(inputStep))
+		Expect(input.StepInputs[0].Values).Should(HaveLen(1))
+		Expect(input.StepInputs[0].Values[0].Value).ShouldNot(BeEmpty())
+	})
+
+	It("get pipeline run logs", func() {
+		logStep := "log"
+		res := get("/projects/" + projectName1 + "/pipelines/" + pipelineName + "/runs/" + pipelineRunName + "/log?step=" + logStep)
+		var logs apisv1.GetPipelineRunLogResponse
+		Expect(decodeResponseBody(res, &logs)).Should(Succeed())
+		Expect(logs.Name).Should(Equal(logStep))
+		Expect(logs.Log).ShouldNot(BeEmpty())
+	})
+
+	It("delete pipeline run", func() {
+		res := delete("/projects/" + projectName1 + "/pipelines/" + pipelineName + "/runs/" + pipelineRunName)
+		Expect(res.StatusCode).Should(Equal(http.StatusOK))
+	})
+
+	It("stop pipeline", func() {
+		By("update pipeline so that it will run for a while")
+		var req = apisv1.UpdatePipelineRequest{
+			Spec: v1alpha1.WorkflowSpec{
+				Steps: []v1alpha1.WorkflowStep{
+					{
+						WorkflowStepBase: v1alpha1.WorkflowStepBase{
+							Name:      "request",
+							Type:      "request",
+							Timeout:   "20s",
+							DependsOn: []string{"not-exist-step"},
+						},
+					},
+				},
+			},
+		}
+		res := put("/projects/"+projectName1+"/pipelines/"+pipelineName, req)
+		Expect(res.StatusCode).Should(Equal(http.StatusOK))
+
+		By("run the pipeline")
+		var run apisv1.PipelineRun
+		res = post("/projects/"+projectName1+"/pipelines/"+pipelineName+"/run", apisv1.RunPipelineRequest{})
+		Expect(res.StatusCode).Should(Equal(http.StatusOK))
+		Expect(decodeResponseBody(res, &run)).Should(Succeed())
+		pipelineRunName = run.PipelineRunName
+
+		By("stop the pipeline")
+		var meta apisv1.PipelineRunMeta
+		res = post("/projects/"+projectName1+"/pipelines/"+pipelineName+"/runs/"+pipelineRunName+"/stop", nil)
+		Expect(res.StatusCode).Should(Equal(http.StatusOK))
+		Expect(decodeResponseBody(res, &meta)).Should(Succeed())
+		Expect(meta.PipelineRunName).Should(Equal(pipelineRunName))
+
+		By("delete pipeline run")
+		res = delete("/projects/" + projectName1 + "/pipelines/" + pipelineName + "/runs/" + pipelineRunName)
+		Expect(res.StatusCode).Should(Equal(http.StatusOK))
+	})
+
+	It("delete context", func() {
+		res := delete("/projects/" + projectName1 + "/pipelines/" + pipelineName + "/contexts/" + contextName)
+		Expect(res.StatusCode).Should(Equal(http.StatusOK))
+	})
+
+	It("delete pipeline", func() {
+		res := delete("/projects/" + projectName1 + "/pipelines/" + pipelineName)
+		Expect(res.StatusCode).Should(Equal(http.StatusOK))
+	})
+
+	It("delete project", func() {
+		res := delete("/projects/" + projectName1)
+		Expect(res.StatusCode).Should(Equal(http.StatusOK))
+	})
+
+})
@@ -59,6 +59,7 @@ func TestE2eApiserverTest(t *testing.T) {

 // Suite test in e2e-apiserver-test relies on the pre-setup kubernetes environment
 var _ = BeforeSuite(func() {
+	defer GinkgoRecover()

 	ctx := context.Background()

@@ -115,7 +116,7 @@ var _ = BeforeSuite(func() {
 			err = json.NewDecoder(resp.Body).Decode(code)
 			Expect(err).Should(BeNil())
 			return fmt.Errorf("rest service not ready code:%d message:%s", resp.StatusCode, code.Message)
-		}, time.Second*10, time.Millisecond*200).Should(BeNil())
+		}, time.Second*20, time.Millisecond*200).Should(BeNil())
 	var err error
 	k8sClient, err = clients.GetKubeClient()
 	Expect(err).ShouldNot(HaveOccurred())
@@ -0,0 +1,35 @@
+apiVersion: core.oam.dev/v1beta1
+kind: WorkflowStepDefinition
+metadata:
+  annotations:
+    definition.oam.dev/alias: ""
+    definition.oam.dev/description: Apply raw kubernetes objects for your workflow steps
+  labels:
+    custom.definition.oam.dev/ui-hidden: "true"
+  name: log
+  namespace: vela-system
+spec:
+  schematic:
+    cue:
+      template: |
+        import (
+                "vela/op"
+        )
+
+        apply: op.#Log & {
+                parameter
+        }
+        parameter: {
+                data?:   string
+                level:   *3 | int
+                source?: close({
+                        url: string
+                }) | close({
+                        resources?: [...{
+                                name?:      string
+                                cluster?:   string
+                                namespace?: string
+                                labelSelector?: {...}
+                        }]
+                })
+        }
@@ -0,0 +1,43 @@
+apiVersion: core.oam.dev/v1beta1
+kind: WorkflowStepDefinition
+metadata:
+  annotations:
+    definition.oam.dev/alias: ""
+    definition.oam.dev/description: Send request to the url
+  name: request
+  namespace: vela-system
+spec:
+  schematic:
+    cue:
+      template: |
+        import (
+                "vela/op"
+                "encoding/json"
+        )
+
+        http: op.http.#Do & {
+                method: parameter.method
+                url:    parameter.url
+                request: {
+                        if parameter.body != _|_ {
+                                body: json.Marshal(parameter.body)
+                        }
+                        if parameter.header != _|_ {
+                                header: parameter.header
+                        }
+                }
+        }
+        fail: op.#Steps & {
+                if http.response.statusCode > 400 {
+                        requestFail: op.#Fail & {
+                                message: "request of \(parameter.url) is fail: \(http.response.statusCode)"
+                        }
+                }
+        }
+        response: json.Unmarshal(http.response.body)
+        parameter: {
+                url:    string
+                method: *"GET" | "POST" | "PUT" | "DELETE"
+                body?: {...}
+                header?: [string]: string
+        }