Feat: pipeline API for apiserver (#4840)

2026-08-18 20:17:04 +00:00 · 2022-10-17 20:11:06 +08:00
parent 51c803cc12
commit dc7d791127
9 changed files with 2851 additions and 56 deletions
@@ -0,0 +1,61 @@
+/*
+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 model
+
+import "fmt"
+
+func init() {
+	RegisterModel(&PipelineContext{})
+}
+
+// Value is a k-v pair
+type Value struct {
+	Key   string `json:"key"`
+	Value string `json:"value"`
+}
+
+// PipelineContext is pipeline's context groups
+type PipelineContext struct {
+	BaseModel
+	PipelineName string             `json:"pipelineName"`
+	ProjectName  string             `json:"projectName"`
+	Contexts     map[string][]Value `json:"contexts"`
+}
+
+// TableName return custom table name
+func (c *PipelineContext) TableName() string {
+	return tableNamePrefix + "pipeline_context"
+}
+
+// ShortTableName is the compressed version of table name for kubeapi storage and others
+func (c *PipelineContext) ShortTableName() string {
+	return "pp-ctx"
+}
+
+// PrimaryKey return custom primary key
+func (c *PipelineContext) PrimaryKey() string {
+	return fmt.Sprintf("%s-%s", c.ProjectName, c.PipelineName)
+}
+
+// Index return custom index
+func (c *PipelineContext) Index() map[string]string {
+	index := make(map[string]string)
+	if c.ProjectName != "" {
+		index["project_name"] = c.ProjectName
+	}
+	return index
+}
@@ -0,0 +1,463 @@
+/*
+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"
+	"fmt"
+	"strings"
+
+	"github.com/pkg/errors"
+
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kubevela/workflow/api/v1alpha1"
+	wfTypes "github.com/kubevela/workflow/pkg/types"
+
+	"github.com/oam-dev/kubevela/pkg/apiserver/domain/model"
+	"github.com/oam-dev/kubevela/pkg/apiserver/infrastructure/datastore"
+	apis "github.com/oam-dev/kubevela/pkg/apiserver/interfaces/api/dto/v1"
+	"github.com/oam-dev/kubevela/pkg/apiserver/utils"
+	"github.com/oam-dev/kubevela/pkg/apiserver/utils/bcode"
+	"github.com/oam-dev/kubevela/pkg/apiserver/utils/log"
+	"github.com/oam-dev/kubevela/pkg/utils/apply"
+)
+
+const (
+	labelDescription = "pipeline.velaux.oam.dev/description"
+	labelAlias       = "pipeline.velaux.oam.dev/alias"
+)
+
+const (
+	labelContextName = "context.velaux.oam.dev/name"
+)
+
+// PipelineService is the interface for pipeline service
+type PipelineService interface {
+	CreatePipeline(ctx context.Context, req apis.CreatePipelineRequest) (*apis.PipelineBase, error)
+	ListPipelines(ctx context.Context, req apis.ListPipelineRequest) (*apis.ListPipelineResponse, error)
+	GetPipeline(ctx context.Context, name, project string) (*apis.GetPipelineResponse, error)
+	UpdatePipeline(ctx context.Context, name, project string, req apis.UpdatePipelineRequest) (*apis.PipelineBase, error)
+	DeletePipeline(ctx context.Context, base apis.PipelineBase) error
+	RunPipeline(ctx context.Context, pipeline apis.PipelineBase, req apis.RunPipelineRequest) error
+}
+
+type pipelineServiceImpl struct {
+	ContextService ContextService   `inject:""`
+	KubeClient     client.Client    `inject:"kubeClient"`
+	KubeConfig     *rest.Config     `inject:"kubeConfig"`
+	Apply          apply.Applicator `inject:"apply"`
+}
+
+// PipelineRunService is the interface for pipelineRun service
+type PipelineRunService interface {
+	GetPipelineRun(ctx context.Context, meta apis.PipelineRunMeta) (apis.PipelineRun, error)
+	ListPipelineRuns(ctx context.Context, base apis.PipelineBase) (apis.ListPipelineRunResponse, error)
+	DeletePipelineRun(ctx context.Context, meta apis.PipelineRunMeta) error
+	StopPipelineRun(ctx context.Context, pipeline apis.PipelineRunBase) error
+}
+
+type pipelineRunServiceImpl struct {
+	KubeClient     client.Client    `inject:"kubeClient"`
+	KubeConfig     *rest.Config     `inject:"kubeConfig"`
+	Apply          apply.Applicator `inject:"apply"`
+	ContextService ContextService   `inject:""`
+}
+
+// ContextService is the interface for context service
+type ContextService interface {
+	GetContext(ctx context.Context, projectName, pipelineName string, name string) (*apis.Context, error)
+	CreateContext(ctx context.Context, projectName, pipelineName string, context apis.Context) (*model.PipelineContext, error)
+	UpdateContext(ctx context.Context, projectName, pipelineName string, context apis.Context) (*model.PipelineContext, error)
+	ListContexts(ctx context.Context, projectName, pipelineName string) (*apis.ListContextValueResponse, error)
+	DeleteContext(ctx context.Context, projectName, pipelineName, name string) error
+}
+
+type contextServiceImpl struct {
+	Store datastore.DataStore `inject:"datastore"`
+}
+
+// NewPipelineService new pipeline service
+func NewPipelineService() PipelineService {
+	return &pipelineServiceImpl{}
+}
+
+// NewPipelineRunService new pipelineRun service
+func NewPipelineRunService() PipelineRunService {
+	return &pipelineRunServiceImpl{}
+}
+
+// NewContextService new context service
+func NewContextService() ContextService {
+	return &contextServiceImpl{}
+}
+
+// CreatePipeline will create a pipeline
+func (p pipelineServiceImpl) CreatePipeline(ctx context.Context, req apis.CreatePipelineRequest) (*apis.PipelineBase, error) {
+	wf := v1alpha1.Workflow{}
+	wf.SetName(req.Name)
+	wf.SetNamespace(nsForProj(req.Project))
+	wf.WorkflowSpec = req.Spec
+	wf.SetLabels(map[string]string{
+		labelDescription: req.Description,
+		labelAlias:       req.Alias})
+	if err := p.KubeClient.Create(ctx, &wf); err != nil {
+		return nil, err
+	}
+	return &apis.PipelineBase{
+		PipelineMeta: apis.PipelineMeta{
+			Name:        req.Name,
+			Alias:       req.Alias,
+			Project:     req.Project,
+			Description: req.Description,
+		},
+		Spec: wf.WorkflowSpec,
+	}, nil
+}
+
+// ListPipelines will list all pipelines
+func (p pipelineServiceImpl) ListPipelines(ctx context.Context, req apis.ListPipelineRequest) (*apis.ListPipelineResponse, error) {
+	wfs := v1alpha1.WorkflowList{}
+	nsOption := make([]client.ListOption, 0)
+	for _, ns := range req.Projects {
+		nsOption = append(nsOption, client.InNamespace(nsForProj(ns)))
+	}
+	if err := p.KubeClient.List(ctx, &wfs, nsOption...); err != nil {
+		return nil, err
+	}
+	res := apis.ListPipelineResponse{}
+	for _, wf := range wfs.Items {
+		if fuzzyMatch(wf, req.Query) {
+			item := apis.PipelineListItem{
+				PipelineMeta: workflow2PipelineBase(wf).PipelineMeta,
+				// todo info
+			}
+			res.Pipelines = append(res.Pipelines, item)
+		}
+	}
+	return &res, nil
+}
+
+// GetPipeline will get a pipeline
+func (p pipelineServiceImpl) GetPipeline(ctx context.Context, name, project string) (*apis.GetPipelineResponse, error) {
+	wf := v1alpha1.Workflow{}
+	if err := p.KubeClient.Get(ctx, client.ObjectKey{Name: name, Namespace: nsForProj(project)}, &wf); err != nil {
+		return nil, err
+	}
+	return &apis.GetPipelineResponse{
+		PipelineBase: *workflow2PipelineBase(wf),
+		// todo info
+	}, nil
+}
+
+// UpdatePipeline will update a pipeline
+func (p pipelineServiceImpl) UpdatePipeline(ctx context.Context, name, project string, req apis.UpdatePipelineRequest) (*apis.PipelineBase, error) {
+	wf := v1alpha1.Workflow{}
+	if err := p.KubeClient.Get(ctx, client.ObjectKey{Name: name, Namespace: nsForProj(project)}, &wf); err != nil {
+		return nil, err
+	}
+	wf.WorkflowSpec = req.Spec
+	wf.SetLabels(map[string]string{
+		labelDescription: req.Description,
+		labelAlias:       req.Alias})
+	if err := p.KubeClient.Update(ctx, &wf); err != nil {
+		return nil, err
+	}
+	return workflow2PipelineBase(wf), nil
+}
+
+// DeletePipeline will delete a pipeline
+func (p pipelineServiceImpl) DeletePipeline(ctx context.Context, pl apis.PipelineBase) error {
+	wf := v1alpha1.Workflow{}
+	if err := p.KubeClient.Get(ctx, client.ObjectKey{Name: pl.Name, Namespace: nsForProj(pl.Project)}, &wf); err != nil {
+		return err
+	}
+	return p.KubeClient.Delete(ctx, &wf)
+}
+
+// StopPipelineRun will stop a pipelineRun
+func (p pipelineRunServiceImpl) StopPipelineRun(ctx context.Context, pipelineRun apis.PipelineRunBase) error {
+	run, err := p.checkRecordRunning(ctx, pipelineRun)
+	if err != nil {
+		return err
+	}
+	if err := p.terminatePipelineRun(ctx, run); err != nil {
+		return err
+	}
+	return nil
+}
+
+// RunPipeline will run a pipeline
+func (p pipelineServiceImpl) RunPipeline(ctx context.Context, pipeline apis.PipelineBase, req apis.RunPipelineRequest) error {
+	// todo get context and set to workflowRun
+	run := v1alpha1.WorkflowRun{}
+	version := utils.GenerateVersion("")
+	name := fmt.Sprintf("%s-%s", pipeline.Name, version)
+	run.Name = name
+	run.Namespace = fmt.Sprintf("%s-project", pipeline.Project)
+	run.Spec.WorkflowRef = pipeline.Name
+	run.Spec.Mode = &req.Mode
+
+	return p.KubeClient.Create(ctx, &run)
+}
+
+// GetPipelineRun will get a pipeline run
+func (p pipelineRunServiceImpl) GetPipelineRun(ctx context.Context, meta apis.PipelineRunMeta) (apis.PipelineRun, error) {
+	namespacedName := client.ObjectKey{Name: meta.PipelineName, Namespace: nsForProj(meta.Project)}
+	run := v1alpha1.WorkflowRun{}
+	if err := p.KubeClient.Get(ctx, namespacedName, &run); err != nil {
+		return apis.PipelineRun{}, err
+	}
+	return workflowRun2PipelineRun(run), nil
+}
+
+// ListPipelineRuns will list all pipeline runs
+func (p pipelineRunServiceImpl) ListPipelineRuns(ctx context.Context, base apis.PipelineBase) (apis.ListPipelineRunResponse, error) {
+	wfrs := v1alpha1.WorkflowRunList{}
+	if err := p.KubeClient.List(ctx, &wfrs, client.InNamespace(nsForProj(base.Project))); err != nil {
+		return apis.ListPipelineRunResponse{}, err
+	}
+	res := apis.ListPipelineRunResponse{}
+	for _, wfr := range wfrs.Items {
+		if wfr.Spec.WorkflowRef == base.Name {
+			res.Runs = append(res.Runs, p.workflowRun2runBriefing(ctx, wfr))
+		}
+	}
+	return res, nil
+}
+
+// DeletePipelineRun will delete a pipeline run
+func (p pipelineRunServiceImpl) DeletePipelineRun(ctx context.Context, meta apis.PipelineRunMeta) error {
+	namespacedName := client.ObjectKey{Name: meta.PipelineName, Namespace: nsForProj(meta.Project)}
+	run := v1alpha1.WorkflowRun{}
+	if err := p.KubeClient.Get(ctx, namespacedName, &run); err != nil {
+		return err
+	}
+	return p.KubeClient.Delete(ctx, &run)
+}
+
+// GetContext will get a context
+func (c contextServiceImpl) GetContext(ctx context.Context, projectName, pipelineName, name string) (*apis.Context, error) {
+	modelCtx := model.PipelineContext{
+		ProjectName:  projectName,
+		PipelineName: pipelineName,
+	}
+	if err := c.Store.Get(ctx, &modelCtx); err != nil {
+		return nil, err
+	}
+	vals, ok := modelCtx.Contexts[name]
+	if !ok {
+		return nil, errors.New("context not found")
+	}
+	return &apis.Context{Name: name, Values: vals}, nil
+}
+
+// CreateContext will create a context
+func (c contextServiceImpl) CreateContext(ctx context.Context, projectName, pipelineName string, context apis.Context) (*model.PipelineContext, error) {
+	modelCtx := model.PipelineContext{
+		ProjectName:  projectName,
+		PipelineName: pipelineName,
+	}
+	if err := c.Store.Get(ctx, &modelCtx); err != nil {
+		return nil, err
+	}
+	if _, ok := modelCtx.Contexts[context.Name]; ok {
+		log.Logger.Errorf("context %s already exists", context.Name)
+		return nil, bcode.ErrContextAlreadyExist
+	}
+	modelCtx.Contexts[context.Name] = context.Values
+	if err := c.Store.Put(ctx, &modelCtx); err != nil {
+		return nil, err
+	}
+	return &modelCtx, nil
+}
+
+// UpdateContext will update a context
+func (c contextServiceImpl) UpdateContext(ctx context.Context, projectName, pipelineName string, context apis.Context) (*model.PipelineContext, error) {
+	modelCtx := model.PipelineContext{
+		ProjectName:  projectName,
+		PipelineName: pipelineName,
+	}
+	if err := c.Store.Get(ctx, &modelCtx); err != nil {
+		return nil, err
+	}
+	modelCtx.Contexts[context.Name] = context.Values
+	if err := c.Store.Put(ctx, &modelCtx); err != nil {
+		return nil, err
+	}
+	return &modelCtx, nil
+}
+
+// ListContexts will list all contexts
+func (c contextServiceImpl) ListContexts(ctx context.Context, projectName, pipelineName string) (*apis.ListContextValueResponse, error) {
+	modelCtx := model.PipelineContext{
+		ProjectName:  projectName,
+		PipelineName: pipelineName,
+	}
+	if err := c.Store.Get(ctx, &modelCtx); err != nil {
+		return nil, err
+	}
+	return &apis.ListContextValueResponse{
+		Total:    len(modelCtx.Contexts),
+		Contexts: modelCtx.Contexts,
+	}, nil
+}
+
+// DeleteContext will delete a context
+func (c contextServiceImpl) DeleteContext(ctx context.Context, projectName, pipelineName, name string) error {
+	modelCtx := model.PipelineContext{
+		ProjectName:  projectName,
+		PipelineName: pipelineName,
+	}
+	if err := c.Store.Get(ctx, &modelCtx); err != nil {
+		return err
+	}
+	delete(modelCtx.Contexts, name)
+	if err := c.Store.Put(ctx, &modelCtx); err != nil {
+		return err
+	}
+	return nil
+}
+
+func nsForProj(proj string) string {
+	return fmt.Sprintf("project-%s", proj)
+}
+
+func getWfDescription(wf v1alpha1.Workflow) string {
+	if wf.Labels == nil {
+		return ""
+	}
+	return wf.Labels[labelDescription]
+}
+
+func getWfAlias(wf v1alpha1.Workflow) string {
+	if wf.Labels == nil {
+		return ""
+	}
+	return wf.Labels[labelAlias]
+}
+
+func fuzzyMatch(wf v1alpha1.Workflow, q string) bool {
+	if strings.Contains(wf.Name, q) {
+		return true
+	}
+	if strings.Contains(getWfAlias(wf), q) {
+		return true
+	}
+	if strings.Contains(getWfDescription(wf), q) {
+		return true
+	}
+	return false
+}
+
+func workflow2PipelineBase(wf v1alpha1.Workflow) *apis.PipelineBase {
+	project := strings.TrimRight(wf.Namespace, "-project")
+	return &apis.PipelineBase{
+		PipelineMeta: apis.PipelineMeta{
+			Name:        wf.Name,
+			Project:     project,
+			Description: getWfDescription(wf),
+			Alias:       getWfAlias(wf),
+		},
+		Spec: wf.WorkflowSpec,
+	}
+}
+
+func workflowRun2PipelineRun(run v1alpha1.WorkflowRun) apis.PipelineRun {
+	return apis.PipelineRun{
+		PipelineRunBase: apis.PipelineRunBase{
+			PipelineRunMeta: apis.PipelineRunMeta{
+				PipelineName:    run.Spec.WorkflowRef,
+				Project:         strings.TrimRight(run.Namespace, "-project"),
+				PipelineRunName: run.Name,
+			},
+		},
+		Status: run.Status,
+	}
+}
+
+func (p pipelineRunServiceImpl) workflowRun2runBriefing(ctx context.Context, run v1alpha1.WorkflowRun) apis.PipelineRunBriefing {
+	contextName := run.Labels[labelContextName]
+	project := strings.TrimRight(run.Namespace, "-project")
+	apiContext, err := p.ContextService.GetContext(ctx, project, run.Spec.WorkflowRef, contextName)
+	if err != nil {
+		log.Logger.Warnf("failed to get pipeline run context %s/%s/%s: %v", project, run.Spec.WorkflowRef, contextName, err)
+		apiContext = nil
+	}
+
+	return apis.PipelineRunBriefing{
+		PipelineRunName: run.Name,
+		Finished:        run.Status.Finished,
+		Phase:           run.Status.Phase,
+		Message:         run.Status.Message,
+		StartTime:       run.Status.StartTime,
+		EndTime:         run.Status.EndTime,
+		ContextName:     apiContext.Name,
+		ContextValues:   apiContext.Values,
+	}
+}
+func (p pipelineRunServiceImpl) checkRecordRunning(ctx context.Context, pipelineRun apis.PipelineRunBase) (*v1alpha1.WorkflowRun, error) {
+	run := v1alpha1.WorkflowRun{}
+	if err := p.KubeClient.Get(ctx, types.NamespacedName{
+		Namespace: nsForProj(pipelineRun.Project),
+		Name:      pipelineRun.PipelineRunName,
+	}, &run); err != nil {
+		return nil, err
+	}
+	if !run.Status.Suspend && !run.Status.Terminated && !run.Status.Finished {
+		return nil, fmt.Errorf("workflow is still running, can not operate a running workflow")
+	}
+	return &run, nil
+}
+
+func (p pipelineRunServiceImpl) terminatePipelineRun(ctx context.Context, run *v1alpha1.WorkflowRun) error {
+	run.Status.Terminated = true
+	run.Status.Suspend = false
+	steps := run.Status.Steps
+	for i, step := range steps {
+		switch step.Phase {
+		case v1alpha1.WorkflowStepPhaseFailed:
+			if step.Reason != wfTypes.StatusReasonFailedAfterRetries && step.Reason != wfTypes.StatusReasonTimeout {
+				steps[i].Reason = wfTypes.StatusReasonTerminate
+			}
+		case v1alpha1.WorkflowStepPhaseRunning:
+			steps[i].Phase = v1alpha1.WorkflowStepPhaseFailed
+			steps[i].Reason = wfTypes.StatusReasonTerminate
+		default:
+		}
+		for j, sub := range step.SubStepsStatus {
+			switch sub.Phase {
+			case v1alpha1.WorkflowStepPhaseFailed:
+				if sub.Reason != wfTypes.StatusReasonFailedAfterRetries && sub.Reason != wfTypes.StatusReasonTimeout {
+					steps[i].SubStepsStatus[j].Phase = wfTypes.StatusReasonTerminate
+				}
+			case v1alpha1.WorkflowStepPhaseRunning:
+				steps[i].SubStepsStatus[j].Phase = v1alpha1.WorkflowStepPhaseFailed
+				steps[i].SubStepsStatus[j].Reason = wfTypes.StatusReasonTerminate
+			default:
+			}
+		}
+	}
+
+	if err := p.KubeClient.Status().Patch(ctx, run, client.Merge); err != nil {
+		return err
+	}
+	return nil
+
+}
@@ -46,11 +46,15 @@ func InitServiceBean(c config.Config) []interface{} {
 	configService := NewConfigService()
 	applicationService := NewApplicationService()
 	webhookService := NewWebhookService()
+	pipelineService := NewPipelineService()
+	pipelineRunService := NewPipelineRunService()
+	contextService := NewContextService()
 	needInitData = []DataInit{clusterService, userService, rbacService, projectService, targetService, systemInfoService}
 	return []interface{}{
 		clusterService, rbacService, projectService, envService, targetService, workflowService, oamApplicationService,
 		velaQLService, definitionService, addonService, envBindingService, systemInfoService, helmService, userService,
-		authenticationService, configService, applicationService, webhookService, NewImageService(), NewCloudShellService(),
+		authenticationService, configService, applicationService, webhookService, pipelineService, pipelineRunService,
+		contextService, NewImageService(), NewCloudShellService(),
 	}
 }

@@ -19,6 +19,8 @@ package v1
 import (
 	"time"

+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
 	"github.com/getkin/kin-openapi/openapi3"
 	registryv1 "github.com/google/go-containerregistry/pkg/v1"
 	workflowv1alpha1 "github.com/kubevela/workflow/api/v1alpha1"
@@ -50,6 +52,12 @@ var (
 	CtxKeyUser = "user"
 	// 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"
+	// CtxKeyPipelineRun request context key of pipeline run
+	CtxKeyPipelineRun = "pipeline-run"
 )

 // AddonPhase defines the phase of an addon
@@ -1527,3 +1535,205 @@ type CreateConfigDistributionRequest struct {
 type ListConfigDistributionResponse struct {
 	Distributions []*config.Distribution `json:"distributions"`
 }
+
+/********************/
+/* Pipeline Structs */
+/********************/
+
+// 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"`
+}
+
+// PipelineBase is the base info of pipeline
+type PipelineBase struct {
+	PipelineMeta `json:",inline"`
+	Spec         workflowv1alpha1.WorkflowSpec `json:"spec"`
+}
+
+// RunStatInfo is the pipeline run statistics info
+type RunStatInfo struct {
+	Total   int `json:"total"`
+	Success int `json:"success"`
+	Fail    int `json:"fail"`
+}
+
+// RunStat is the statistics of the pipeline in seven days
+type RunStat struct {
+	ActiveNum int           `json:"activeNum"`
+	Total     RunStatInfo   `json:"total"`
+	Week      []RunStatInfo `json:"week"`
+}
+
+// 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"`
+}
+
+// PipelineMetaResponse is the response body contains PipelineMeta
+type PipelineMetaResponse struct {
+	PipelineMeta `json:",inline"`
+}
+
+// ListPipelineRequest is the request body of listing pipeline
+type ListPipelineRequest struct {
+	Projects []string `json:"projects"`
+	Query    string   `json:"query"`
+}
+
+// ListPipelineResponse is the response body of listing pipeline
+type ListPipelineResponse struct {
+	Total     int                `json:"total"`
+	Pipelines []PipelineListItem `json:"pipelines"`
+}
+
+// PipelineListItem is the item of pipeline list
+type PipelineListItem struct {
+	PipelineMeta `json:",inline"`
+	Info         PipelineInfo `json:"info"`
+}
+
+// UpdatePipelineRequest is the request body of updating pipeline
+type UpdatePipelineRequest struct {
+	Alias       string                        `json:"alias" validate:"checkalias" optional:"true"`
+	Description string                        `json:"description" optional:"true"`
+	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"`
+	PipelineInfo `json:"info"`
+}
+
+// PipelineInfo is the info of pipeline
+type PipelineInfo struct {
+	RelatedApps   []ApplicationBase                  `json:"relatedApps"`
+	LastRunStatus workflowv1alpha1.WorkflowRunStatus `json:"lastRunStatus"`
+	RunStat       RunStat                            `json:"runStat"`
+}
+
+/***********************/
+/* PipelineRun Structs */
+/***********************/
+
+// PipelineRunBriefing is the brief info of the pipeline run, contains run name and brief status
+type PipelineRunBriefing struct {
+	PipelineRunName string                            `json:"pipelineRunName"`
+	Finished        bool                              `json:"finished"`
+	Phase           workflowv1alpha1.WorkflowRunPhase `json:"phase"`
+	Message         string                            `json:"message"`
+	StartTime       metav1.Time                       `json:"startTime"`
+	EndTime         metav1.Time                       `json:"endTime"`
+	ContextName     string                            `json:"contextName"`
+	ContextValues   []model.Value                     `json:"contextValues"`
+}
+
+// PipelineRunMeta is the metadata of pipeline run
+type PipelineRunMeta struct {
+	PipelineName    string `json:"pipelineName"`
+	Project         string `json:"project"`
+	PipelineRunName string `json:"pipelineRunName"`
+}
+
+// PipelineRun is the info of pipeline run
+type PipelineRun struct {
+	PipelineRunBase `json:",inline"`
+	Status          workflowv1alpha1.WorkflowRunStatus `json:"status"`
+}
+
+// PipelineRunBase is the base info of pipeline run
+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"`
+}
+
+// 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`
+	Mode        workflowv1alpha1.WorkflowExecuteMode `json:"mode" optional:"true"`
+	ContextName string                               `json:"contextName"`
+}
+
+// ListPipelineRunResponse is the response body of listing pipeline run
+type ListPipelineRunResponse struct {
+	Total int64                 `json:"total"`
+	Runs  []PipelineRunBriefing `json:"runs"`
+}
+
+// GetPipelineRunLogResponse is the response body of getting pipeline run log
+type GetPipelineRunLogResponse struct {
+	Log []Log `json:"log"`
+}
+
+// GetPipelineRunOutputResponse is the response body of getting pipeline run output
+type GetPipelineRunOutputResponse struct {
+	Output []Output `json:"output"`
+}
+
+// StepBase is the base info of step
+type StepBase struct {
+	ID    string `json:"id"`
+	Name  string `json:"name"`
+	Type  string `json:"type"`
+	Phase string `json:"phase"`
+}
+
+// Log is the log of step
+type Log struct {
+	StepBase `json:",inline"`
+	Log      string `json:"log"`
+}
+
+// Output is the output of step
+type Output struct {
+	StepBase `json:",inline"`
+	Vars     map[string]string `json:"vars"`
+}
+
+/*******************/
+/* Context Structs */
+/*******************/
+
+// Context is an internal struct for the context
+type Context struct {
+	Name   string        `json:"name"`
+	Values []model.Value `json:"values"`
+}
+
+// CreateContextValuesRequest is the request body of creating context values
+type CreateContextValuesRequest struct {
+	Name   string        `json:"name"`
+	Values []model.Value `json:"values"`
+}
+
+// UpdateContextValuesRequest is the request body of updating context values
+type UpdateContextValuesRequest struct {
+	Values []model.Value `json:"values"`
+}
+
+// ContextNameResponse is the response body of getting context name
+type ContextNameResponse struct {
+	Name string `json:"name"`
+}
+
+// ListContextValueResponse is the response body of listing context values
+type ListContextValueResponse struct {
+	Total    int                      `json:"total"`
+	Contexts map[string][]model.Value `json:"contexts"`
+}
@@ -82,6 +82,7 @@ func InitAPIBean() []interface{} {
 	RegisterAPIInterface(NewWebhookAPIInterface())
 	RegisterAPIInterface(NewRepositoryAPIInterface())
 	RegisterAPIInterface(NewCloudShellAPIInterface())
+	RegisterAPIInterface(NewPipelineAPIInterface())

 	// Authentication
 	RegisterAPIInterface(NewAuthenticationAPIInterface())
@@ -23,5 +23,5 @@ import (
 )

 func TestInitAPIBean(t *testing.T) {
-	assert.Equal(t, len(InitAPIBean()), 23)
+	assert.Equal(t, len(InitAPIBean()), 24)
 }
@@ -0,0 +1,501 @@
+/*
+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 api
+
+import (
+	"context"
+
+	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/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"
+)
+
+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"
+	// Pipeline is the pipeline name of query param
+	Pipeline pipelinePathParamKey = "pipelineName"
+	// PipelineRun is the pipeline run name of query param
+	PipelineRun pipelinePathParamKey = "runName"
+	// ContextName is the context name of query param
+	ContextName pipelinePathParamKey = "contextName"
+)
+
+// GetWebServiceRoute is the implementation of pipeline Interface
+func (p *pipelineAPIInterface) GetWebServiceRoute() *restful.WebService {
+
+	ws := new(restful.WebService)
+	tags := []string{"pipeline"}
+
+	projParam := func(builder *restful.RouteBuilder) {
+		builder.Param(ws.QueryParameter(string(Project), "project name").Required(true))
+	}
+	pipelineParam := func(builder *restful.RouteBuilder) {
+		builder.Param(ws.PathParameter(string(Pipeline), "pipeline name").Required(true))
+		builder.Filter(p.pipelineCheckFilter)
+	}
+	ctxParam := func(builder *restful.RouteBuilder) {
+		builder.Param(ws.PathParameter(string(ContextName), "pipeline context name").Required(true))
+		builder.Filter(p.pipelineContextCheckFilter)
+	}
+	runParam := func(builder *restful.RouteBuilder) {
+		builder.Param(ws.PathParameter(string(PipelineRun), "pipeline run name").Required(true))
+		builder.Filter(p.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).
+		Doc("create pipeline").
+		Reads(apis.CreatePipelineRequest{}).
+		Returns(200, "OK", apis.PipelineBase{}).
+		Returns(400, "Bad Request", bcode.Bcode{}).
+		Writes(apis.PipelineBase{}).Do(meta))
+
+	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).
+		Doc("get pipeline").
+		Reads(apis.GetPipelineRequest{}).
+		Returns(200, "OK", apis.GetPipelineResponse{}).
+		Returns(400, "Bad Request", bcode.Bcode{}).
+		Writes(apis.GetPipelineResponse{}).Do(meta, projParam, pipelineParam))
+
+	ws.Route(ws.PUT("/{pipelineName}").To(p.updatePipeline).
+		Doc("update pipeline").
+		Reads(apis.UpdatePipelineRequest{}).
+		Returns(200, "OK", apis.PipelineBase{}).
+		Returns(400, "Bad Request", bcode.Bcode{}).
+		Writes(apis.PipelineBase{}).Do(meta, projParam, pipelineParam))
+
+	ws.Route(ws.DELETE("/{pipelineName}").To(p.deletePipeline).
+		Doc("delete pipeline").
+		Returns(200, "OK", apis.PipelineMetaResponse{}).
+		Returns(400, "Bad Request", bcode.Bcode{}).
+		Writes(apis.PipelineMetaResponse{}).Do(meta, projParam, pipelineParam))
+
+	ws.Route(ws.POST("/{pipelineName}/contexts").To(p.createContextValue).
+		Doc("create pipeline context values").
+		Reads(apis.CreateContextValuesRequest{}).
+		Returns(200, "OK", apis.Context{}).
+		Returns(400, "Bad Request", bcode.Bcode{}).
+		Writes(apis.Context{}).Do(meta, projParam, pipelineParam))
+
+	ws.Route(ws.GET("/{pipelineName}/contexts").To(p.listContextValues).
+		Doc("list pipeline context values").
+		Returns(200, "OK", apis.ListContextValueResponse{}).
+		Returns(400, "Bad Request", bcode.Bcode{}).
+		Writes(apis.ListContextValueResponse{}).Do(meta, projParam, pipelineParam))
+
+	ws.Route(ws.PUT("/{pipelineName}/contexts/{contextName}").To(p.updateContextValue).
+		Doc("update pipeline context value").
+		Reads(apis.UpdateContextValuesRequest{}).
+		Returns(200, "OK", apis.Context{}).
+		Returns(400, "Bad Request", bcode.Bcode{}).
+		Writes(apis.Context{}).Do(meta, projParam, pipelineParam, ctxParam))
+
+	ws.Route(ws.DELETE("/{pipelineName}/contexts/{contextName}").To(p.deleteContextValue).
+		Doc("delete pipeline context value").
+		Returns(200, "OK", apis.ContextNameResponse{}).
+		Returns(400, "Bad Request", bcode.Bcode{}).
+		Writes(apis.ContextNameResponse{}).Do(meta, projParam, pipelineParam, ctxParam))
+
+	ws.Route(ws.POST("/{pipelineName}/run").To(p.runPipeline).
+		Doc("run pipeline").
+		Reads(apis.RunPipelineRequest{}).
+		Returns(200, "OK", apis.PipelineRunMeta{}).
+		Returns(400, "Bad Request", bcode.Bcode{}).
+		Writes(apis.PipelineRunMeta{}).Do(meta, projParam, pipelineParam))
+
+	ws.Route(ws.GET("/{pipelineName}/runs").To(p.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{}).
+		Writes(apis.ListPipelineRunResponse{}).Do(meta, projParam, pipelineParam))
+
+	ws.Route(ws.POST("/{pipelineName}/runs/{runName}/stop").To(p.stopPipeline).
+		Doc("stop pipeline run").
+		Returns(200, "OK", apis.PipelineRunMeta{}).
+		Returns(400, "Bad Request", bcode.Bcode{}).
+		Writes(apis.PipelineRunMeta{}).Do(meta, projParam, pipelineParam, runParam))
+
+	ws.Route(ws.GET("/{pipelineName}/runs/{runName}").To(p.getPipelineRun).
+		Doc("get pipeline run").
+		Returns(200, "OK", apis.PipelineRunBase{}).
+		Returns(400, "Bad Request", bcode.Bcode{}).
+		Writes(apis.PipelineRunBase{}).Do(meta, projParam, pipelineParam, runParam))
+
+	ws.Route(ws.DELETE("/{pipelineName}/runs/{runName}").To(p.deletePipelineRun).
+		Doc("delete pipeline run").
+		Returns(200, "OK", apis.PipelineRunMeta{}).
+		Returns(400, "Bad Request", bcode.Bcode{}).
+		Writes(apis.PipelineRunMeta{}).Do(meta, projParam, pipelineParam, runParam))
+
+	// get pipeline run status
+	ws.Route(ws.GET("/{pipelineName}/runs/{runName}/status").To(p.getPipelineRunStatus).
+		Doc("get pipeline run status").
+		Returns(200, "OK", workflowv1alpha1.WorkflowRunStatus{}).
+		Returns(400, "Bad Request", bcode.Bcode{}).
+		Writes(workflowv1alpha1.WorkflowRunStatus{}).Do(meta, projParam, pipelineParam, runParam))
+
+	// get pipeline run log
+	ws.Route(ws.GET("/{pipelineName}/runs/{runName}/log").To(p.getPipelineRunLog).
+		Doc("get pipeline run log").
+		Param(ws.QueryParameter("step", "query by specific id").DataType("string")).
+		Returns(200, "OK", apis.GetPipelineRunLogResponse{}).
+		Returns(400, "Bad Request", bcode.Bcode{}).
+		Writes(apis.GetPipelineRunLogResponse{}).Do(meta, projParam, pipelineParam, runParam))
+
+	// get pipeline run output
+	ws.Route(ws.GET("/{pipelineName}/runs/{runName}/output").To(p.getPipelineRunOutput).
+		Doc("get pipeline run output").
+		Param(ws.QueryParameter("step", "query by specific id").DataType("string")).
+		Returns(200, "OK", apis.GetPipelineRunOutputResponse{}).
+		Returns(400, "Bad Request", bcode.Bcode{}).
+		Writes(apis.GetPipelineRunOutputResponse{}).Do(meta, projParam, pipelineParam, runParam))
+
+	ws.Filter(authCheckFilter)
+	return ws
+}
+
+// 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"))
+	}
+	pipelines, err := p.PipelineService.ListPipelines(req.Request.Context(), apis.ListPipelineRequest{
+		Projects: projetNames,
+		Query:    req.QueryParameter("query"),
+	})
+	if err != nil {
+		log.Logger.Errorf("list pipeline failure %s", err.Error())
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	if err := res.WriteEntity(pipelines); err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+}
+
+func (p *pipelineAPIInterface) getPipeline(req *restful.Request, res *restful.Response) {
+	pipeline := req.Request.Context().Value(apis.CtxKeyPipeline).(apis.PipelineBase)
+	if err := res.WriteEntity(pipeline); err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+}
+
+func (p *pipelineAPIInterface) createPipeline(req *restful.Request, res *restful.Response) {
+	var createReq apis.CreatePipelineRequest
+	if err := req.ReadEntity(&createReq); err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	if err := validate.Struct(&createReq); err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	pipelineBase, err := p.PipelineService.CreatePipeline(req.Request.Context(), createReq)
+	if err != nil {
+		log.Logger.Errorf("create pipeline 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) {
+	var updateReq apis.UpdatePipelineRequest
+	if err := req.ReadEntity(&updateReq); err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	if err := validate.Struct(&updateReq); err != nil {
+		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)
+	if err != nil {
+		log.Logger.Errorf("update pipeline 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) deletePipeline(req *restful.Request, res *restful.Response) {
+	pipeline := req.Request.Context().Value(&apis.CtxKeyPipeline).(apis.PipelineBase)
+	err := p.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 {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+}
+
+func (p *pipelineAPIInterface) 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 {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+
+	err := p.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 {
+		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)
+	if err != nil {
+		log.Logger.Errorf("stop pipeline failure %s", err.Error())
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	if err := res.WriteEntity(pipelineRun.PipelineRunMeta); err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+}
+
+func (p *pipelineAPIInterface) 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)
+	if err != nil {
+		log.Logger.Errorf("list pipeline runs failure %s", err.Error())
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	if err := res.WriteEntity(pipelineRuns); err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+}
+
+func (p *pipelineAPIInterface) 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)
+	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 (p *pipelineAPIInterface) getPipelineRunOutput(req *restful.Request, res *restful.Response) {
+
+}
+
+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)
+	if err != nil {
+		log.Logger.Errorf("delete pipeline run failure %s", err.Error())
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	if err := res.WriteEntity(pipelineRun.PipelineRunMeta); err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+}
+
+func (p *pipelineAPIInterface) 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)
+	if err != nil {
+		log.Logger.Errorf("list context values failure %s", err.Error())
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	if err := res.WriteEntity(contextValues); err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+}
+
+func (p *pipelineAPIInterface) 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 {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	if err := validate.Struct(&createReq); err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+
+	pipelineCtx := apis.Context(createReq)
+	_, err := p.ContextService.CreateContext(req.Request.Context(), pipeline.Project, pipeline.Name, pipelineCtx)
+	if err != nil {
+		log.Logger.Errorf("create context failure %s", err.Error())
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	if err := res.WriteEntity(pipelineCtx); err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+}
+
+func (p *pipelineAPIInterface) updateContextValue(req *restful.Request, res *restful.Response) {
+	plCtx := req.Request.Context().Value(&apis.CtxKeyPipelineContex).(apis.Context)
+	pipeline := req.Request.Context().Value(&apis.CtxKeyPipeline).(apis.PipelineBase)
+	var updateReq apis.UpdateContextValuesRequest
+	if err := req.ReadEntity(&updateReq); err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	if err := validate.Struct(&updateReq); err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	pipelineCtx := apis.Context{Name: plCtx.Name, Values: updateReq.Values}
+	_, err := p.ContextService.UpdateContext(req.Request.Context(), pipeline.Project, pipeline.Name, pipelineCtx)
+	if err != nil {
+		log.Logger.Errorf("update context failure %s", err.Error())
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	if err := res.WriteEntity(pipelineCtx); err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+}
+
+func (p *pipelineAPIInterface) deleteContextValue(req *restful.Request, res *restful.Response) {
+	plCtx := req.Request.Context().Value(&apis.CtxKeyPipelineContex).(apis.Context)
+	pipeline := req.Request.Context().Value(&apis.CtxKeyPipeline).(apis.PipelineBase)
+	err := p.ContextService.DeleteContext(req.Request.Context(), pipeline.Project, pipeline.Name, plCtx.Name)
+	if err != nil {
+		log.Logger.Errorf("delete context failure %s", err.Error())
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	if err := res.WriteEntity(apis.ContextNameResponse{Name: plCtx.Name}); err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+}
+
+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"))
+	if err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	req.Request = req.Request.WithContext(context.WithValue(req.Request.Context(), &apis.CtxKeyPipeline, pipeline.PipelineBase))
+	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"))
+	if err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	contextName := req.PathParameter("contextName")
+	contextValue, ok := contexts.Contexts[contextName]
+	if !ok {
+		bcode.ReturnError(req, res, bcode.ErrContextNotFound)
+		return
+	}
+	req.Request = req.Request.WithContext(context.WithValue(req.Request.Context(), &apis.CtxKeyPipelineContex, apis.Context{
+		Name:   contextName,
+		Values: contextValue,
+	}))
+	chain.ProcessFilter(req, res)
+}
+
+func (p *pipelineAPIInterface) 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)),
+	}
+	run, err := p.PipelineRunService.GetPipelineRun(req.Request.Context(), meta)
+	if err != nil {
+		bcode.ReturnError(req, res, err)
+		return
+	}
+	req.Request = req.Request.WithContext(context.WithValue(req.Request.Context(), &apis.CtxKeyPipelineRun, run))
+
+	chain.ProcessFilter(req, res)
+}
@@ -0,0 +1,24 @@
+/*
+Copyright 2022 The KubeVela Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package bcode
+
+var (
+	// ErrContextNotFound means the certain context is not found
+	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")
+)