Add InChainDeploymentPlannable RPC

pkg/app/api/grpcapi/piped_api.go

@@ -1105,6 +1105,56 @@ func (a *PipedAPI) CreateDeploymentChain(ctx context.Context, req *pipedservice.
 	return &pipedservice.CreateDeploymentChainResponse{}, nil
 }
 
+// InChainDeploymentPlannable hecks the completion and status of the previous block in the deployment chain.
+// An in chain deployment is treated as plannable in case:
+// - It's the first deployment of its deployment chain.
+// - All deployments of its previous block in chain are at DEPLOYMENT_SUCCESS state.
+func (a *PipedAPI) InChainDeploymentPlannable(ctx context.Context, req *pipedservice.InChainDeploymentPlannableRequest) (*pipedservice.InChainDeploymentPlannableResponse, error) {
+	_, pipedID, _, err := rpcauth.ExtractPipedToken(ctx)
+	if err != nil {
+		return nil, err
+	}
+	if err := a.validateDeploymentBelongsToPiped(ctx, req.Deployment.Id, pipedID); err != nil {
+		return nil, err
+	}
+
+	dc, err := a.deploymentChainStore.GetDeploymentChain(ctx, req.Deployment.DeploymentChainId)
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, "unable to find the deployment chain which this deployment belongs to")
+	}
+
+	// Deployment of blocks[0] in the chain means it's the first deployment of the chain;
+	// hence it should be processed without any lock.
+	if req.Deployment.DeploymentChainBlockIndex == 0 {
+		return &pipedservice.InChainDeploymentPlannableResponse{
+			Plannable: true,
+		}, nil
+	}
+
+	if req.Deployment.DeploymentChainBlockIndex >= int32(len(dc.Blocks)) {
+		return nil, status.Error(codes.InvalidArgument, "invalid deployment with chain block index provided")
+	}
+
+	prevBlock := dc.Blocks[req.Deployment.DeploymentChainBlockIndex-1]
+	plannable := true
+	for _, node := range prevBlock.Nodes {
+		// TODO: Consider add deployment status to the deployment ref in the deployment chain model
+		// instead of fetching deployment model here.
+		dp, err := a.deploymentStore.GetDeployment(ctx, node.DeploymentRef.DeploymentId)
+		if err != nil {
+			return nil, status.Error(codes.Internal, "unable to process previous block nodes in deployment chain")
+		}
+		if model.IsCompletedSuccessfullyDeployment(dp.Status) {
+			plannable = false
+			break
+		}
+	}
+
+	return &pipedservice.InChainDeploymentPlannableResponse{
+		Plannable: plannable,
+	}, nil
+}
+
 // validateAppBelongsToPiped checks if the given application belongs to the given piped.
 // It gives back an error unless the application belongs to the piped.
 func (a *PipedAPI) validateAppBelongsToPiped(ctx context.Context, appID, pipedID string) error {

pkg/app/api/service/pipedservice/service.proto

@@ -173,6 +173,10 @@ service PipedService {
     // CreateDeploymentChain creates a new deployment chain object and all required commands to
     // trigger deployment for applications in the chain.
     rpc CreateDeploymentChain(CreateDeploymentChainRequest) returns (CreateDeploymentChainResponse) {}
+    // DeploymentPlannable checks the completion and status of the previous block in the deployment chain,
+    // only when all the nodes of the previous block are completed with a success status,
+    // the nodes of the next block will be treat as processable.
+    rpc InChainDeploymentPlannable(InChainDeploymentPlannableRequest) returns (InChainDeploymentPlannableResponse) {}
 }
 
 enum ListOrder {
@@ -495,3 +499,11 @@ message CreateDeploymentChainRequest {
 
 message CreateDeploymentChainResponse {
 }
+
+message InChainDeploymentPlannableRequest {
+    pipe.model.Deployment deployment = 1 [(validate.rules).message.required = true];
+}
+
+message InChainDeploymentPlannableResponse {
+    bool plannable = 1;
+}

pkg/datastore/deploymentchainstore.go

@@ -56,6 +56,7 @@ var (
 type DeploymentChainStore interface {
 	AddDeploymentChain(ctx context.Context, d *model.DeploymentChain) error
 	UpdateDeploymentChain(ctx context.Context, id string, updater func(*model.DeploymentChain) error) error
+	GetDeploymentChain(ctx context.Context, id string) (*model.DeploymentChain, error)
 }
 
 type deploymentChainStore struct {
@@ -97,3 +98,11 @@ func (s *deploymentChainStore) UpdateDeploymentChain(ctx context.Context, id str
 		return dc.Validate()
 	})
 }
+
+func (s *deploymentChainStore) GetDeploymentChain(ctx context.Context, id string) (*model.DeploymentChain, error) {
+	var entity model.DeploymentChain
+	if err := s.ds.Get(ctx, DeploymentChainModelKind, id, &entity); err != nil {
+		return nil, err
+	}
+	return &entity, nil
+}

pkg/model/deployment.go

@@ -48,6 +48,11 @@ func IsCompletedDeployment(status DeploymentStatus) bool {
 	return false
 }
 
+// IsCompletedSuccessfullyDeployment checks whether the deployment is at a successfully addressed.
+func IsCompletedSuccessfullyDeployment(status DeploymentStatus) bool {
+	return status == DeploymentStatus_DEPLOYMENT_SUCCESS
+}
+
 // IsCompletedStage checks whether the stage is at a completion state.
 func IsCompletedStage(status StageStatus) bool {
 	if status.String() == "" {