Rename PipelineSyncPlugin -> StagePlugin (missed to rename) (#5593)

t-kikuc · web-flow · commit ad00a56f0974 · 2025-02-20T16:47:16.000+09:00
Signed-off-by: t-kikuc &lt;tkikuchi07f@gmail.com&gt;
diff --git a/pkg/app/pipedv1/plugin/example/main.go b/pkg/app/pipedv1/plugin/example/main.go
@@ -21,7 +21,7 @@ import (
 )
 
 func main() {
-	sdk.RegisterPipelineSyncPlugin(&plugin{})
+	sdk.RegisterStagePlugin(&plugin{})
 
 	if err := sdk.Run(); err != nil {
 		log.Fatalln(err)
diff --git a/pkg/app/pipedv1/plugin/example/plugin.go b/pkg/app/pipedv1/plugin/example/plugin.go
@@ -34,17 +34,17 @@ func (p *plugin) Version() string {
 	return "0.0.1"
 }
 
-// BuildPipelineSyncStages implements sdk.PipelineSyncPlugin.
+// BuildPipelineSyncStages implements sdk.StagePlugin.
 func (p *plugin) BuildPipelineSyncStages(context.Context, *config, *sdk.BuildPipelineSyncStagesInput) (*sdk.BuildPipelineSyncStagesResponse, error) {
 	return &sdk.BuildPipelineSyncStagesResponse{}, nil
 }
 
-// ExecuteStage implements sdk.PipelineSyncPlugin.
+// ExecuteStage implements sdk.StagePlugin.
 func (p *plugin) ExecuteStage(context.Context, *config, sdk.DeployTargetsNone, *sdk.ExecuteStageInput) (*sdk.ExecuteStageResponse, error) {
 	return &sdk.ExecuteStageResponse{}, nil
 }
 
-// FetchDefinedStages implements sdk.PipelineSyncPlugin.
+// FetchDefinedStages implements sdk.StagePlugin.
 func (p *plugin) FetchDefinedStages() []string {
 	return []string{"EXAMPLE_PLAN", "EXAMPLE_APPLY"}
 }
diff --git a/pkg/app/pipedv1/plugin/kubernetes_multicluster/main.go b/pkg/app/pipedv1/plugin/kubernetes_multicluster/main.go
@@ -21,7 +21,7 @@ import (
 )
 
 func main() {
-	sdk.RegisterPipelineSyncPlugin(&plugin{})
+	sdk.RegisterStagePlugin(&plugin{})
 
 	if err := sdk.Run(); err != nil {
 		log.Fatalln(err)
diff --git a/pkg/app/pipedv1/plugin/kubernetes_multicluster/plugin.go b/pkg/app/pipedv1/plugin/kubernetes_multicluster/plugin.go
@@ -34,17 +34,17 @@ func (p *plugin) Version() string {
 	return "0.0.1"
 }
 
-// BuildPipelineSyncStages implements sdk.PipelineSyncPlugin.
+// BuildPipelineSyncStages implements sdk.StagePlugin.
 func (p *plugin) BuildPipelineSyncStages(context.Context, *config, *sdk.BuildPipelineSyncStagesInput) (*sdk.BuildPipelineSyncStagesResponse, error) {
 	return &sdk.BuildPipelineSyncStagesResponse{}, nil
 }
 
-// ExecuteStage implements sdk.PipelineSyncPlugin.
+// ExecuteStage implements sdk.StagePlugin.
 func (p *plugin) ExecuteStage(context.Context, *config, sdk.DeployTargetsNone, *sdk.ExecuteStageInput) (*sdk.ExecuteStageResponse, error) {
 	return &sdk.ExecuteStageResponse{}, nil
 }
 
-// FetchDefinedStages implements sdk.PipelineSyncPlugin.
+// FetchDefinedStages implements sdk.StagePlugin.
 func (p *plugin) FetchDefinedStages() []string {
 	return []string{"K8S_SYNC"}
 }
diff --git a/pkg/app/pipedv1/plugin/wait/main.go b/pkg/app/pipedv1/plugin/wait/main.go
@@ -21,7 +21,7 @@ import (
 )
 
 func main() {
-	sdk.RegisterPipelineSyncPlugin(&plugin{})
+	sdk.RegisterStagePlugin(&plugin{})
 
 	if err := sdk.Run(); err != nil {
 		log.Fatalln(err)
diff --git a/pkg/app/pipedv1/plugin/wait/plugin.go b/pkg/app/pipedv1/plugin/wait/plugin.go
@@ -36,7 +36,7 @@ func (p *plugin) Version() string {
 	return "0.0.1" // TODO
 }
 
-// BuildPipelineSyncStages implements sdk.PipelineSyncPlugin.
+// BuildPipelineSyncStages implements sdk.StagePlugin.
 func (p *plugin) BuildPipelineSyncStages(ctx context.Context, _ sdk.ConfigNone, input *sdk.BuildPipelineSyncStagesInput) (*sdk.BuildPipelineSyncStagesResponse, error) {
 	stages := make([]sdk.PipelineStage, 0, len(input.Request.Stages))
 	for _, rs := range input.Request.Stages {
@@ -55,15 +55,15 @@ func (p *plugin) BuildPipelineSyncStages(ctx context.Context, _ sdk.ConfigNone,
 	}, nil
 }
 
-// ExecuteStage implements sdk.PipelineSyncPlugin.
+// ExecuteStage implements sdk.StagePlugin.
 func (p *plugin) ExecuteStage(ctx context.Context, _ sdk.ConfigNone, _ sdk.DeployTargetsNone, input *sdk.ExecuteStageInput) (*sdk.ExecuteStageResponse, error) {
 	status := p.executeWait(ctx, input)
 	return &sdk.ExecuteStageResponse{
 		Status: status,
 	}, nil
 }
 
-// FetchDefinedStages implements sdk.PipelineSyncPlugin.
+// FetchDefinedStages implements sdk.StagePlugin.
 func (p *plugin) FetchDefinedStages() []string {
 	return []string{stageWait}
 }
diff --git a/pkg/plugin/sdk/deployment.go b/pkg/plugin/sdk/deployment.go
@@ -98,10 +98,10 @@ func RegisterDeploymentPlugin[Config, DeployTargetConfig any](plugin DeploymentP
 	deploymentServiceServer = &DeploymentPluginServiceServer[Config, DeployTargetConfig]{base: plugin}
 }
 
-// RegisterPipelineSyncPlugin registers the given pipeline sync plugin.
+// RegisterStagePlugin registers the given stage plugin.
 // It will be used when running the piped.
-func RegisterPipelineSyncPlugin[Config, DeployTargetConfig any](plugin StagePlugin[Config, DeployTargetConfig]) {
-	deploymentServiceServer = &PipelineSyncPluginServiceServer[Config, DeployTargetConfig]{base: plugin}
+func RegisterStagePlugin[Config, DeployTargetConfig any](plugin StagePlugin[Config, DeployTargetConfig]) {
+	deploymentServiceServer = &StagePluginServiceServer[Config, DeployTargetConfig]{base: plugin}
 }
 
 type logPersister interface {
@@ -193,8 +193,8 @@ func (s *DeploymentPluginServiceServer[Config, DeployTargetConfig]) ExecuteStage
 	return executeStage(ctx, s.base, &s.config, nil, client, request, s.logger) // TODO: pass the deployTargets
 }
 
-// PipelineSyncPluginServiceServer is the gRPC server that handles requests from the piped.
-type PipelineSyncPluginServiceServer[Config, DeployTargetConfig any] struct {
+// StagePluginServiceServer is the gRPC server that handles requests from the piped.
+type StagePluginServiceServer[Config, DeployTargetConfig any] struct {
 	deployment.UnimplementedDeploymentServiceServer
 	commonFields
 
@@ -203,25 +203,25 @@ type PipelineSyncPluginServiceServer[Config, DeployTargetConfig any] struct {
 }
 
 // Name returns the name of the plugin.
-func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) Name() string {
+func (s *StagePluginServiceServer[Config, DeployTargetConfig]) Name() string {
 	return s.base.Name()
 }
 
 // Version returns the version of the plugin.
-func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) Version() string {
+func (s *StagePluginServiceServer[Config, DeployTargetConfig]) Version() string {
 	return s.base.Version()
 }
 
 // Register registers the server to the given gRPC server.
-func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) Register(server *grpc.Server) {
+func (s *StagePluginServiceServer[Config, DeployTargetConfig]) Register(server *grpc.Server) {
 	deployment.RegisterDeploymentServiceServer(server, s)
 }
 
-func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) setCommonFields(fields commonFields) {
+func (s *StagePluginServiceServer[Config, DeployTargetConfig]) setCommonFields(fields commonFields) {
 	s.commonFields = fields
 }
 
-func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) setConfig(bytes []byte) error {
+func (s *StagePluginServiceServer[Config, DeployTargetConfig]) setConfig(bytes []byte) error {
 	if bytes == nil {
 		return nil
 	}
@@ -231,27 +231,27 @@ func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) setConfig(
 	return nil
 }
 
-func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) FetchDefinedStages(context.Context, *deployment.FetchDefinedStagesRequest) (*deployment.FetchDefinedStagesResponse, error) {
+func (s *StagePluginServiceServer[Config, DeployTargetConfig]) FetchDefinedStages(context.Context, *deployment.FetchDefinedStagesRequest) (*deployment.FetchDefinedStagesResponse, error) {
 	return &deployment.FetchDefinedStagesResponse{Stages: s.base.FetchDefinedStages()}, nil
 }
-func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) DetermineVersions(context.Context, *deployment.DetermineVersionsRequest) (*deployment.DetermineVersionsResponse, error) {
+func (s *StagePluginServiceServer[Config, DeployTargetConfig]) DetermineVersions(context.Context, *deployment.DetermineVersionsRequest) (*deployment.DetermineVersionsResponse, error) {
 	return &deployment.DetermineVersionsResponse{}, nil
 }
-func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) DetermineStrategy(context.Context, *deployment.DetermineStrategyRequest) (*deployment.DetermineStrategyResponse, error) {
+func (s *StagePluginServiceServer[Config, DeployTargetConfig]) DetermineStrategy(context.Context, *deployment.DetermineStrategyRequest) (*deployment.DetermineStrategyResponse, error) {
 	return &deployment.DetermineStrategyResponse{Unsupported: true}, nil
 }
-func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) BuildPipelineSyncStages(ctx context.Context, request *deployment.BuildPipelineSyncStagesRequest) (*deployment.BuildPipelineSyncStagesResponse, error) {
+func (s *StagePluginServiceServer[Config, DeployTargetConfig]) BuildPipelineSyncStages(ctx context.Context, request *deployment.BuildPipelineSyncStagesRequest) (*deployment.BuildPipelineSyncStagesResponse, error) {
 	client := &Client{
 		base:       s.client,
 		pluginName: s.Name(),
 	}
 
 	return buildPipelineSyncStages(ctx, s.base, &s.config, client, request, s.logger) // TODO: pass the real client
 }
-func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) BuildQuickSyncStages(context.Context, *deployment.BuildQuickSyncStagesRequest) (*deployment.BuildQuickSyncStagesResponse, error) {
+func (s *StagePluginServiceServer[Config, DeployTargetConfig]) BuildQuickSyncStages(context.Context, *deployment.BuildQuickSyncStagesRequest) (*deployment.BuildQuickSyncStagesResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method BuildQuickSyncStages not implemented")
 }
-func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) ExecuteStage(ctx context.Context, request *deployment.ExecuteStageRequest) (response *deployment.ExecuteStageResponse, _ error) {
+func (s *StagePluginServiceServer[Config, DeployTargetConfig]) ExecuteStage(ctx context.Context, request *deployment.ExecuteStageRequest) (response *deployment.ExecuteStageResponse, _ error) {
 	lp := s.logPersister.StageLogPersister(request.GetInput().GetDeployment().GetId(), request.GetInput().GetStage().GetId())
 	defer func() {
 		// When termination signal received and the stage is not completed yet, we should not mark the log persister as completed.

Original file line number	Diff line number	Diff line change
`@@ -21,7 +21,7 @@ import (`
`21`	`21`	`)`
`22`	`22`
`23`	`23`	`func main() {`
`24`		`- sdk.RegisterPipelineSyncPlugin(&plugin{})`
	`24`	`+ sdk.RegisterStagePlugin(&plugin{})`
`25`	`25`
`26`	`26`	`if err := sdk.Run(); err != nil {`
`27`	`27`	`log.Fatalln(err)`
Original file line number	Diff line number	Diff line change
`@@ -34,17 +34,17 @@ func (p *plugin) Version() string {`
`34`	`34`	`return "0.0.1"`
`35`	`35`	`}`
`36`	`36`
`37`		`-// BuildPipelineSyncStages implements sdk.PipelineSyncPlugin.`
	`37`	`+// BuildPipelineSyncStages implements sdk.StagePlugin.`
`38`	`38`	`func (p plugin) BuildPipelineSyncStages(context.Context, config, sdk.BuildPipelineSyncStagesInput) (sdk.BuildPipelineSyncStagesResponse, error) {`
`39`	`39`	`return &sdk.BuildPipelineSyncStagesResponse{}, nil`
`40`	`40`	`}`
`41`	`41`
`42`		`-// ExecuteStage implements sdk.PipelineSyncPlugin.`
	`42`	`+// ExecuteStage implements sdk.StagePlugin.`
`43`	`43`	`func (p plugin) ExecuteStage(context.Context, config, sdk.DeployTargetsNone, sdk.ExecuteStageInput) (sdk.ExecuteStageResponse, error) {`
`44`	`44`	`return &sdk.ExecuteStageResponse{}, nil`
`45`	`45`	`}`
`46`	`46`
`47`		`-// FetchDefinedStages implements sdk.PipelineSyncPlugin.`
	`47`	`+// FetchDefinedStages implements sdk.StagePlugin.`
`48`	`48`	`func (p *plugin) FetchDefinedStages() []string {`
`49`	`49`	`return []string{"EXAMPLE_PLAN", "EXAMPLE_APPLY"}`
`50`	`50`	`}`
Original file line number	Diff line number	Diff line change
`@@ -36,7 +36,7 @@ func (p *plugin) Version() string {`
`36`	`36`	`return "0.0.1" // TODO`
`37`	`37`	`}`
`38`	`38`
`39`		`-// BuildPipelineSyncStages implements sdk.PipelineSyncPlugin.`
	`39`	`+// BuildPipelineSyncStages implements sdk.StagePlugin.`
`40`	`40`	`func (p plugin) BuildPipelineSyncStages(ctx context.Context, _ sdk.ConfigNone, input sdk.BuildPipelineSyncStagesInput) (*sdk.BuildPipelineSyncStagesResponse, error) {`
`41`	`41`	`stages := make([]sdk.PipelineStage, 0, len(input.Request.Stages))`
`42`	`42`	`for _, rs := range input.Request.Stages {`
`@@ -55,15 +55,15 @@ func (p *plugin) BuildPipelineSyncStages(ctx context.Context, _ sdk.ConfigNone,`
`55`	`55`	`}, nil`
`56`	`56`	`}`
`57`	`57`
`58`		`-// ExecuteStage implements sdk.PipelineSyncPlugin.`
	`58`	`+// ExecuteStage implements sdk.StagePlugin.`
`59`	`59`	`func (p plugin) ExecuteStage(ctx context.Context, _ sdk.ConfigNone, _ sdk.DeployTargetsNone, input sdk.ExecuteStageInput) (*sdk.ExecuteStageResponse, error) {`
`60`	`60`	`status := p.executeWait(ctx, input)`
`61`	`61`	`return &sdk.ExecuteStageResponse{`
`62`	`62`	`Status: status,`
`63`	`63`	`}, nil`
`64`	`64`	`}`
`65`	`65`
`66`		`-// FetchDefinedStages implements sdk.PipelineSyncPlugin.`
	`66`	`+// FetchDefinedStages implements sdk.StagePlugin.`
`67`	`67`	`func (p *plugin) FetchDefinedStages() []string {`
`68`	`68`	`return []string{stageWait}`
`69`	`69`	`}`
Original file line number	Diff line number	Diff line change
`@@ -98,10 +98,10 @@ func RegisterDeploymentPlugin[Config, DeployTargetConfig any](plugin DeploymentP`
`98`	`98`	`deploymentServiceServer = &DeploymentPluginServiceServer[Config, DeployTargetConfig]{base: plugin}`
`99`	`99`	`}`
`100`	`100`
`101`		`-// RegisterPipelineSyncPlugin registers the given pipeline sync plugin.`
	`101`	`+// RegisterStagePlugin registers the given stage plugin.`
`102`	`102`	`// It will be used when running the piped.`
`103`		`-func RegisterPipelineSyncPlugin[Config, DeployTargetConfig any](plugin StagePlugin[Config, DeployTargetConfig]) {`
`104`		`- deploymentServiceServer = &PipelineSyncPluginServiceServer[Config, DeployTargetConfig]{base: plugin}`
	`103`	`+func RegisterStagePlugin[Config, DeployTargetConfig any](plugin StagePlugin[Config, DeployTargetConfig]) {`
	`104`	`+ deploymentServiceServer = &StagePluginServiceServer[Config, DeployTargetConfig]{base: plugin}`
`105`	`105`	`}`
`106`	`106`
`107`	`107`	`type logPersister interface {`
`@@ -193,8 +193,8 @@ func (s *DeploymentPluginServiceServer[Config, DeployTargetConfig]) ExecuteStage`
`193`	`193`	`return executeStage(ctx, s.base, &s.config, nil, client, request, s.logger) // TODO: pass the deployTargets`
`194`	`194`	`}`
`195`	`195`
`196`		`-// PipelineSyncPluginServiceServer is the gRPC server that handles requests from the piped.`
`197`		`-type PipelineSyncPluginServiceServer[Config, DeployTargetConfig any] struct {`
	`196`	`+// StagePluginServiceServer is the gRPC server that handles requests from the piped.`
	`197`	`+type StagePluginServiceServer[Config, DeployTargetConfig any] struct {`
`198`	`198`	`deployment.UnimplementedDeploymentServiceServer`
`199`	`199`	`commonFields`
`200`	`200`
`@@ -203,25 +203,25 @@ type PipelineSyncPluginServiceServer[Config, DeployTargetConfig any] struct {`
`203`	`203`	`}`
`204`	`204`
`205`	`205`	`// Name returns the name of the plugin.`
`206`		`-func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) Name() string {`
	`206`	`+func (s *StagePluginServiceServer[Config, DeployTargetConfig]) Name() string {`
`207`	`207`	`return s.base.Name()`
`208`	`208`	`}`
`209`	`209`
`210`	`210`	`// Version returns the version of the plugin.`
`211`		`-func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) Version() string {`
	`211`	`+func (s *StagePluginServiceServer[Config, DeployTargetConfig]) Version() string {`
`212`	`212`	`return s.base.Version()`
`213`	`213`	`}`
`214`	`214`
`215`	`215`	`// Register registers the server to the given gRPC server.`
`216`		`-func (s PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) Register(server grpc.Server) {`
	`216`	`+func (s StagePluginServiceServer[Config, DeployTargetConfig]) Register(server grpc.Server) {`
`217`	`217`	`deployment.RegisterDeploymentServiceServer(server, s)`
`218`	`218`	`}`
`219`	`219`
`220`		`-func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) setCommonFields(fields commonFields) {`
	`220`	`+func (s *StagePluginServiceServer[Config, DeployTargetConfig]) setCommonFields(fields commonFields) {`
`221`	`221`	`s.commonFields = fields`
`222`	`222`	`}`
`223`	`223`
`224`		`-func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) setConfig(bytes []byte) error {`
	`224`	`+func (s *StagePluginServiceServer[Config, DeployTargetConfig]) setConfig(bytes []byte) error {`
`225`	`225`	`if bytes == nil {`
`226`	`226`	`return nil`
`227`	`227`	`}`
`@@ -231,27 +231,27 @@ func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) setConfig(`
`231`	`231`	`return nil`
`232`	`232`	`}`
`233`	`233`
`234`		`-func (s PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) FetchDefinedStages(context.Context, deployment.FetchDefinedStagesRequest) (*deployment.FetchDefinedStagesResponse, error) {`
	`234`	`+func (s StagePluginServiceServer[Config, DeployTargetConfig]) FetchDefinedStages(context.Context, deployment.FetchDefinedStagesRequest) (*deployment.FetchDefinedStagesResponse, error) {`
`235`	`235`	`return &deployment.FetchDefinedStagesResponse{Stages: s.base.FetchDefinedStages()}, nil`
`236`	`236`	`}`
`237`		`-func (s PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) DetermineVersions(context.Context, deployment.DetermineVersionsRequest) (*deployment.DetermineVersionsResponse, error) {`
	`237`	`+func (s StagePluginServiceServer[Config, DeployTargetConfig]) DetermineVersions(context.Context, deployment.DetermineVersionsRequest) (*deployment.DetermineVersionsResponse, error) {`
`238`	`238`	`return &deployment.DetermineVersionsResponse{}, nil`
`239`	`239`	`}`
`240`		`-func (s PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) DetermineStrategy(context.Context, deployment.DetermineStrategyRequest) (*deployment.DetermineStrategyResponse, error) {`
	`240`	`+func (s StagePluginServiceServer[Config, DeployTargetConfig]) DetermineStrategy(context.Context, deployment.DetermineStrategyRequest) (*deployment.DetermineStrategyResponse, error) {`
`241`	`241`	`return &deployment.DetermineStrategyResponse{Unsupported: true}, nil`
`242`	`242`	`}`
`243`		`-func (s PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) BuildPipelineSyncStages(ctx context.Context, request deployment.BuildPipelineSyncStagesRequest) (*deployment.BuildPipelineSyncStagesResponse, error) {`
	`243`	`+func (s StagePluginServiceServer[Config, DeployTargetConfig]) BuildPipelineSyncStages(ctx context.Context, request deployment.BuildPipelineSyncStagesRequest) (*deployment.BuildPipelineSyncStagesResponse, error) {`
`244`	`244`	`client := &Client{`
`245`	`245`	`base: s.client,`
`246`	`246`	`pluginName: s.Name(),`
`247`	`247`	`}`
`248`	`248`
`249`	`249`	`return buildPipelineSyncStages(ctx, s.base, &s.config, client, request, s.logger) // TODO: pass the real client`
`250`	`250`	`}`
`251`		`-func (s PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) BuildQuickSyncStages(context.Context, deployment.BuildQuickSyncStagesRequest) (*deployment.BuildQuickSyncStagesResponse, error) {`
	`251`	`+func (s StagePluginServiceServer[Config, DeployTargetConfig]) BuildQuickSyncStages(context.Context, deployment.BuildQuickSyncStagesRequest) (*deployment.BuildQuickSyncStagesResponse, error) {`
`252`	`252`	`return nil, status.Errorf(codes.Unimplemented, "method BuildQuickSyncStages not implemented")`
`253`	`253`	`}`
`254`		`-func (s PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) ExecuteStage(ctx context.Context, request deployment.ExecuteStageRequest) (response *deployment.ExecuteStageResponse, _ error) {`
	`254`	`+func (s StagePluginServiceServer[Config, DeployTargetConfig]) ExecuteStage(ctx context.Context, request deployment.ExecuteStageRequest) (response *deployment.ExecuteStageResponse, _ error) {`
`255`	`255`	`lp := s.logPersister.StageLogPersister(request.GetInput().GetDeployment().GetId(), request.GetInput().GetStage().GetId())`
`256`	`256`	`defer func() {`
`257`	`257`	`// When termination signal received and the stage is not completed yet, we should not mark the log persister as completed.`