Support multiple manual approvals

pkg/app/piped/executor/waitapproval/BUILD.bazel

@@ -1,4 +1,4 @@
-load("@io_bazel_rules_go//go:def.bzl", "go_library")
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
@@ -12,3 +12,18 @@ go_library(
         "@org_uber_go_zap//:go_default_library",
     ],
 )
+
+go_test(
+    name = "go_default_test",
+    size = "small",
+    srcs = ["waitapproval_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//pkg/app/api/service/pipedservice:go_default_library",
+        "//pkg/app/piped/executor:go_default_library",
+        "//pkg/app/piped/metadatastore:go_default_library",
+        "//pkg/model:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@org_golang_google_grpc//:go_default_library",
+    ],
+)

pkg/app/piped/executor/waitapproval/waitapproval.go

@@ -18,6 +18,8 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"strconv"
+	"strings"
 	"time"
 
 	"go.uber.org/zap"
@@ -28,7 +30,9 @@ import (
 )
 
 const (
-	approvedByKey = "ApprovedBy"
+	approvedByKey  = "ApprovedBy"
+	minApproverNum = "MinApproverNum"
+	approversKey   = "CurrentApprovers"
 )
 
 type Executor struct {
@@ -61,13 +65,33 @@ func (e *Executor) Execute(sig executor.StopSignal) model.StageStatus {
 	timer := time.NewTimer(timeout)
 
 	e.reportRequiringApproval()
-	e.LogPersister.Info("Waiting for an approval...")
+	n, ok := e.MetadataStore.Stage(e.Stage.Id).Get(minApproverNum)
+	if !ok {
+		e.LogPersister.Errorf("Unabled to retrive %s from metadata", minApproverNum)
+		return model.StageStatus_STAGE_FAILURE
+	}
+
+	num, err := strconv.Atoi(n)
+	if err != nil {
+		e.LogPersister.Errorf("%s could not be converted to integer: %v", num, err)
+		return model.StageStatus_STAGE_FAILURE
+	}
+	if num > 1 {
+		e.LogPersister.Infof("Waiting for an approval from at least %d users...", num)
+	} else {
+		e.LogPersister.Infof("Waiting for an approval from at least %d user...", num)
+	}
 	for {
 		select {
 		case <-ticker.C:
-			if commander, ok := e.checkApproval(ctx); ok {
-				e.reportApproved(commander)
-				e.LogPersister.Infof("Got an approval from %s", commander)
+			if as, ok := e.checkApproval(ctx, num); ok {
+				e.reportApproved(as)
+				approvers := strings.Split(as, ", ")
+				if n := len(approvers); n > 1 {
+					e.LogPersister.Infof("This stage has been approved by %d users (%s)", n, as)
+				} else {
+					e.LogPersister.Infof("This stage has been approved by %d user (%s)", n, as)
+				}
 				return model.StageStatus_STAGE_SUCCESS
 			}
 
@@ -87,7 +111,7 @@ func (e *Executor) Execute(sig executor.StopSignal) model.StageStatus {
 	}
 }
 
-func (e *Executor) checkApproval(ctx context.Context) (string, bool) {
+func (e *Executor) checkApproval(ctx context.Context, num int) (string, bool) {
 	var approveCmd *model.ReportableCommand
 	commands := e.CommandLister.ListCommands()
 
@@ -101,8 +125,20 @@ func (e *Executor) checkApproval(ctx context.Context) (string, bool) {
 		return "", false
 	}
 
+	as, ok := e.validateApproverNum(approveCmd.Commander, num)
+	if !ok {
+		if len(as) > 0 {
+			if err := e.MetadataStore.Stage(e.Stage.Id).Put(ctx, approversKey, as); err != nil {
+				e.LogPersister.Errorf("Unabled to save approver information to deployment, %v", err)
+			}
+		}
+		return "", false
+	}
+	e.LogPersister.Info("Received all needed approvals")
+	e.LogPersister.Info("")
+
 	metadata := map[string]string{
-		approvedByKey: approveCmd.Commander,
+		approvedByKey: as,
 	}
 	if err := e.MetadataStore.Stage(e.Stage.Id).PutMulti(ctx, metadata); err != nil {
 		e.LogPersister.Errorf("Unabled to save approver information to deployment, %v", err)
@@ -112,7 +148,7 @@ func (e *Executor) checkApproval(ctx context.Context) (string, bool) {
 	if err := approveCmd.Report(ctx, model.CommandStatus_COMMAND_SUCCEEDED, nil, nil); err != nil {
 		e.Logger.Error("failed to report handled command", zap.Error(err))
 	}
-	return approveCmd.Commander, true
+	return as, true
 }
 
 func (e *Executor) reportApproved(approver string) {
@@ -161,3 +197,28 @@ func (e *Executor) getMentionedAccounts(event model.NotificationEventType) ([]st
 
 	return notification.FindSlackAccounts(event), nil
 }
+
+func (e *Executor) validateApproverNum(approver string, num int) (string, bool) {
+	if num <= 1 {
+		e.LogPersister.Infof("Got approval from \"%s\"", approver)
+		return approver, true
+	}
+	as, ok := e.MetadataStore.Stage(e.Stage.Id).Get(approversKey)
+	if !ok {
+		e.LogPersister.Infof("Got approval from \"%s\"", approver)
+		e.LogPersister.Infof("Waiting for other approvers...")
+		return approver, false
+	}
+	approvers := strings.Split(as, " ,")
+	for _, a := range approvers {
+		if a == approver {
+			return "", false
+		}
+	}
+	e.LogPersister.Infof("Got approval from \"%s\"", approver)
+	if d := num - len(approvers) - 1; d > 0 {
+		e.LogPersister.Infof("Waiting for other approvers...")
+		return as + ", " + approver, false
+	}
+	return as + ", " + approver, true
+}

pkg/app/piped/executor/waitapproval/waitapproval_test.go

@@ -0,0 +1,214 @@
+// Copyright 2021 The PipeCD 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 waitapproval
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/grpc"
+
+	"github.com/pipe-cd/pipe/pkg/app/api/service/pipedservice"
+	"github.com/pipe-cd/pipe/pkg/app/piped/executor"
+	"github.com/pipe-cd/pipe/pkg/app/piped/metadatastore"
+	"github.com/pipe-cd/pipe/pkg/model"
+)
+
+type fakeLogPersister struct{}
+
+func (l *fakeLogPersister) Write(_ []byte) (int, error)         { return 0, nil }
+func (l *fakeLogPersister) Info(_ string)                       {}
+func (l *fakeLogPersister) Infof(_ string, _ ...interface{})    {}
+func (l *fakeLogPersister) Success(_ string)                    {}
+func (l *fakeLogPersister) Successf(_ string, _ ...interface{}) {}
+func (l *fakeLogPersister) Error(_ string)                      {}
+func (l *fakeLogPersister) Errorf(_ string, _ ...interface{})   {}
+
+type metadata map[string]string
+
+type fakeAPIClient struct {
+	shared metadata
+	stages map[string]metadata
+}
+
+func (c *fakeAPIClient) SaveDeploymentMetadata(_ context.Context, req *pipedservice.SaveDeploymentMetadataRequest, _ ...grpc.CallOption) (*pipedservice.SaveDeploymentMetadataResponse, error) {
+	md := make(map[string]string, len(c.shared)+len(req.Metadata))
+	for k, v := range c.shared {
+		md[k] = v
+	}
+	for k, v := range req.Metadata {
+		md[k] = v
+	}
+	c.shared = md
+	return &pipedservice.SaveDeploymentMetadataResponse{}, nil
+}
+
+func (c *fakeAPIClient) SaveStageMetadata(_ context.Context, req *pipedservice.SaveStageMetadataRequest, _ ...grpc.CallOption) (*pipedservice.SaveStageMetadataResponse, error) {
+	ori := c.stages[req.StageId]
+	md := make(map[string]string, len(ori)+len(req.Metadata))
+	for k, v := range ori {
+		md[k] = v
+	}
+	for k, v := range req.Metadata {
+		md[k] = v
+	}
+	c.stages[req.StageId] = md
+	return &pipedservice.SaveStageMetadataResponse{}, nil
+}
+
+func TestValidateApproverNum(t *testing.T) {
+	ac := &fakeAPIClient{
+		shared: make(map[string]string, 0),
+		stages: make(map[string]metadata, 0),
+	}
+	testcases := []struct {
+		name           string
+		approver       string
+		minApproverNum int
+		executor       *Executor
+		wantBool       bool
+		wantApprovers  string
+	}{
+		{
+			name:           "return the person who just approved it and true",
+			approver:       "user-1",
+			minApproverNum: 0,
+			executor: &Executor{
+				Input: executor.Input{
+					Stage: &model.PipelineStage{
+						Id: "stage-1",
+					},
+					LogPersister: &fakeLogPersister{},
+					MetadataStore: metadatastore.NewMetadataStore(ac, &model.Deployment{
+						Stages: []*model.PipelineStage{
+							{
+								Id:       "stage-1",
+								Metadata: map[string]string{},
+							},
+						},
+					}),
+				},
+			},
+			wantBool:      true,
+			wantApprovers: "user-1",
+		},
+		{
+			name:           "return the person who just approved it and false",
+			approver:       "user-1",
+			minApproverNum: 2,
+			executor: &Executor{
+				Input: executor.Input{
+					Stage: &model.PipelineStage{
+						Id: "stage-1",
+					},
+					LogPersister: &fakeLogPersister{},
+					MetadataStore: metadatastore.NewMetadataStore(ac, &model.Deployment{
+						Stages: []*model.PipelineStage{
+							{
+								Id:       "stage-1",
+								Metadata: map[string]string{},
+							},
+						},
+					}),
+				},
+			},
+			wantBool:      false,
+			wantApprovers: "user-1",
+		},
+		{
+			name:           "return nothing and false",
+			approver:       "user-1",
+			minApproverNum: 2,
+			executor: &Executor{
+				Input: executor.Input{
+					Stage: &model.PipelineStage{
+						Id: "stage-1",
+					},
+					LogPersister: &fakeLogPersister{},
+					MetadataStore: metadatastore.NewMetadataStore(ac, &model.Deployment{
+						Stages: []*model.PipelineStage{
+							{
+								Id: "stage-1",
+								Metadata: map[string]string{
+									approversKey: "user-1",
+								},
+							},
+						},
+					}),
+				},
+			},
+			wantBool:      false,
+			wantApprovers: "",
+		},
+		{
+			name:           "return all approvers and false",
+			approver:       "user-2",
+			minApproverNum: 3,
+			executor: &Executor{
+				Input: executor.Input{
+					Stage: &model.PipelineStage{
+						Id: "stage-1",
+					},
+					LogPersister: &fakeLogPersister{},
+					MetadataStore: metadatastore.NewMetadataStore(ac, &model.Deployment{
+						Stages: []*model.PipelineStage{
+							{
+								Id: "stage-1",
+								Metadata: map[string]string{
+									approversKey: "user-1",
+								},
+							},
+						},
+					}),
+				},
+			},
+			wantBool:      false,
+			wantApprovers: "user-1, user-2",
+		},
+		{
+			name:           "return all approvers and true",
+			approver:       "user-2",
+			minApproverNum: 2,
+			executor: &Executor{
+				Input: executor.Input{
+					Stage: &model.PipelineStage{
+						Id: "stage-1",
+					},
+					LogPersister: &fakeLogPersister{},
+					MetadataStore: metadatastore.NewMetadataStore(ac, &model.Deployment{
+						Stages: []*model.PipelineStage{
+							{
+								Id: "stage-1",
+								Metadata: map[string]string{
+									approversKey: "user-1",
+								},
+							},
+						},
+					}),
+				},
+			},
+			wantBool:      true,
+			wantApprovers: "user-1, user-2",
+		},
+	}
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			as, ok := tc.executor.validateApproverNum(tc.approver, tc.minApproverNum)
+			assert.Equal(t, tc.wantBool, ok)
+			assert.Equal(t, tc.wantApprovers, as)
+		})
+	}
+}

pkg/app/piped/planner/planner.go

@@ -21,6 +21,7 @@ package planner
 
 import (
 	"context"
+	"strconv"
 	"strings"
 
 	"go.uber.org/zap"
@@ -62,7 +63,8 @@ func MakeInitialStageMetadata(cfg config.PipelineStage) map[string]string {
 	switch cfg.Name {
 	case model.StageWaitApproval:
 		return map[string]string{
-			"Approvers": strings.Join(cfg.WaitApprovalStageOptions.Approvers, ","),
+			"Approvers":      strings.Join(cfg.WaitApprovalStageOptions.Approvers, ","),
+			"MinApproverNum": strconv.Itoa(cfg.WaitApprovalStageOptions.MinApproverNum),
 		}
 	default:
 		return nil

pkg/config/deployment.go

@@ -365,8 +365,9 @@ type WaitStageOptions struct {
 type WaitApprovalStageOptions struct {
 	// The maximum length of time to wait before giving up.
 	// Defaults to 6h.
-	Timeout   Duration `json:"timeout"`
-	Approvers []string `json:"approvers"`
+	Timeout        Duration `json:"timeout"`
+	Approvers      []string `json:"approvers"`
+	MinApproverNum int      `json:"minApproverNum"`
 }
 
 // AnalysisStageOptions contains all configurable values for a K8S_ANALYSIS stage.