func (h helper) computeInstanceMetadatas(
	placementInstances []placement.Instance,
	deploymentInstances []Instance,
) (instanceMetadatas, error) {
	if len(placementInstances) != len(deploymentInstances) {
		errMsg := "number of instances is %d in the placement and %d in the deployment"
		return nil, fmt.Errorf(errMsg, len(placementInstances), len(deploymentInstances))
	}
	metadatas := make(instanceMetadatas, len(placementInstances))
	unique := make(map[string]int)
	for i, pi := range placementInstances {
		id := pi.ID()
		_, exists := unique[id]
		if exists {
			return nil, fmt.Errorf("instance %s not unique in the placement", id)
		}
		endpoint := pi.Endpoint()
		apiEndpoint, err := h.toAPIEndpointFn(endpoint)
		if err != nil {
			return nil, fmt.Errorf("unable to convert placement endpoint %s to api endpoint: %v", endpoint, err)
		}
		unique[id] = i
		metadatas[i].PlacementInstanceID = id
		metadatas[i].ShardSetID = pi.ShardSetID()
		metadatas[i].APIEndpoint = apiEndpoint
	}
	for _, di := range deploymentInstances {
		id := di.ID()
		placementInstanceID, err := h.toPlacementInstanceIDFn(id)
		if err != nil {
			return nil, fmt.Errorf("unable to convert deployment instance id %s to placement instance id", id)
		}
		idx, exists := unique[placementInstanceID]
		if !exists {
			return nil, fmt.Errorf("instance %s is in deployment but not in placement", id)
		}
		if metadatas[idx].DeploymentInstanceID != "" {
			return nil, fmt.Errorf("instance %s not unique in the deployment", id)
		}
		metadatas[idx].DeploymentInstanceID = id
		metadatas[idx].Revision = di.Revision()
	}
	return metadatas, nil
}
validateInstances validates instances derived from placement against
instances derived from deployment, ensuring there are no duplicate instances
and the instances derived from two sources match against each other.
package deploy
import (
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
	"github.com/m3db/m3cluster/placement"
	xerrors "github.com/m3db/m3x/errors"
	"github.com/m3db/m3x/log"
	"github.com/m3db/m3x/retry"
	xsync "github.com/m3db/m3x/sync"
)
var (
	errNoDeploymentProgress = errors.New("no deployment progress")
	errInvalidRevision      = errors.New("invalid revision")
)
type Mode int
const (
	DryRunMode Mode = iota
	ForceMode
)
type Helper interface {
	Deploy(revision string, placement placement.Placement, mode Mode) error
}
type helper struct {
	logger                  log.Logger
	planner                 planner
	client                  aggregatorClient
	mgr                     Manager
	retrier                 retry.Retrier
	foreverRetrier          retry.Retrier
	workers                 xsync.WorkerPool
	toPlacementInstanceIDFn ToPlacementInstanceIDFn
	toAPIEndpointFn         ToAPIEndpointFn
	settleBetweenSteps      time.Duration
}
func NewHelper(opts HelperOptions) (Helper, error) {
	client := newAggregatorClient(opts.HTTPClient())
	planner := newPlanner(client, opts.PlannerOptions())
	retryOpts := opts.RetryOptions()
	retrier := retry.NewRetrier(retryOpts)
	foreverRetrier := retry.NewRetrier(retryOpts.SetForever(true))
	return helper{
		logger:                  opts.InstrumentOptions().Logger(),
		planner:                 planner,
		client:                  client,
		mgr:                     opts.Manager(),
		retrier:                 retrier,
		foreverRetrier:          foreverRetrier,
		workers:                 opts.WorkerPool(),
		toPlacementInstanceIDFn: opts.ToPlacementInstanceIDFn(),
		toAPIEndpointFn:         opts.ToAPIEndpointFn(),
		settleBetweenSteps:      opts.SettleDurationBetweenSteps(),
	}, nil
}
func (h helper) Deploy(revision string, placement placement.Placement, mode Mode) error {
	if revision == "" {
		return errInvalidRevision
	}
	all, err := h.allInstanceMetadatas(placement)
	if err != nil {
		return fmt.Errorf("unable to get all instance metadatas: %v", err)
	}
	filtered := all.WithoutRevision(revision)
	plan, err := h.planner.GeneratePlan(filtered, all)
	if err != nil {
		return fmt.Errorf("unable to generate deployment plan: %v", err)
	}
	h.logger.Infof("generated deployment plan: %+v", plan)
	if mode == DryRunMode {
		return nil
	}
	if err = h.execute(plan, revision, all); err != nil {
		return fmt.Errorf("unable to execute deployment plan: %v", err)
	}
	return nil
}
func (h helper) execute(
	plan deploymentPlan,
	revision string,
	all instanceMetadatas,
) error {
	numSteps := len(plan.Steps)
	for i, step := range plan.Steps {
		h.logger.Infof("deploying step %d of %d", i+1, numSteps)
		if err := h.executeStep(step, revision, all); err != nil {
			return err
		}
		h.logger.Infof("deploying step %d succeeded", i+1)
		if h.settleBetweenSteps > 0 {
			h.logger.Infof("waiting settle duration after step: %s", h.settleBetweenSteps.String())
			time.Sleep(h.settleBetweenSteps)
		}
	}
	return nil
}
func (h helper) executeStep(
	step deploymentStep,
	revision string,
	all instanceMetadatas,
) error {
	h.logger.Infof("waiting until safe to deploy for step %v", step)
	if err := h.waitUntilSafe(all); err != nil {
		return err
	}
	h.logger.Infof("waiting until all targets are validated for step %v", step)
	if err := h.validate(step.Targets); err != nil {
		return err
	}
	h.logger.Infof("waiting until all targets have resigned for step %v", step)
	if err := h.resign(step.Targets); err != nil {
		return err
	}
	h.logger.Infof("beginning to deploy instances for step %v", step)
	targetIDs := step.Targets.DeploymentInstanceIDs()
	if err := h.deploy(targetIDs, revision); err != nil {
		return err
	}
	h.logger.Infof("deployment started, waiting for progress: %v", step)
	if err := h.waitUntilProgressing(targetIDs, revision); err != nil {
		return err
	}
	h.logger.Infof("deployment progressed, waiting for completion: %v", step)
	return h.waitUntilSafe(all)
}
func (h helper) waitUntilSafe(instances instanceMetadatas) error {
	deploymentInstanceIDs := instances.DeploymentInstanceIDs()
	return h.foreverRetrier.Attempt(func() error {
		deploymentInstances, err := h.mgr.Query(deploymentInstanceIDs)
		if err != nil {
			return fmt.Errorf("error querying instances: %v", err)
		}
		var (
			wg   sync.WaitGroup
			safe int64
		)
		for i := range deploymentInstances {
			i := i
			wg.Add(1)
			h.workers.Go(func() {
				defer wg.Done()
				if !deploymentInstances[i].IsHealthy() || deploymentInstances[i].IsDeploying() {
					return
				}
				if err := h.client.IsHealthy(instances[i].APIEndpoint); err != nil {
					return
				}
				atomic.AddInt64(&safe, 1)
			})
		}
		wg.Wait()
		if safe != int64(len(instances)) {
			return fmt.Errorf("only %d out of %d instances are safe to deploy", safe, len(instances))
		}
		return nil
	})
}
func (h helper) validate(targets deploymentTargets) error {
	return h.forEachTarget(targets, func(target deploymentTarget) error {
		return h.foreverRetrier.Attempt(func() error {
			validator := target.Validator
			if validator == nil {
				return nil
			}
			if err := validator(); err != nil {
				err = fmt.Errorf("validation error for instance %s: %v", target.Instance.PlacementInstanceID, err)
				return err
			}
			return nil
		})
	})
}
func (h helper) resign(targets deploymentTargets) error {
	return h.forEachTarget(targets, func(target deploymentTarget) error {
		return h.retrier.Attempt(func() error {
			instance := target.Instance
			if err := h.client.Resign(instance.APIEndpoint); err != nil {
				err = fmt.Errorf("resign error for instance %s: %v", instance.PlacementInstanceID, err)
				return err
			}
			return nil
		})
	})
}
func (h helper) deploy(targetIDs []string, revision string) error {
	return h.retrier.Attempt(func() error {
		return h.mgr.Deploy(targetIDs, revision)
	})
}
func (h helper) waitUntilProgressing(targetIDs []string, revision string) error {
	return h.foreverRetrier.Attempt(func() error {
		targetInstances, err := h.mgr.Query(targetIDs)
		if err != nil {
			return fmt.Errorf("error querying instances: %v", err)
		}
		for _, di := range targetInstances {
			if di.IsDeploying() || di.Revision() == revision {
				return nil
			}
		}
		return errNoDeploymentProgress
	})
}
func (h helper) forEachTarget(targets deploymentTargets, workFn targetWorkFn) error {
	var (
		wg    sync.WaitGroup
		errCh = make(chan error, len(targets))
	)
	for i := range targets {
		i := i
		wg.Add(1)
		h.workers.Go(func() {
			defer wg.Done()
			if err := workFn(targets[i]); err != nil {
				errCh <- err
			}
		})
	}
	wg.Wait()
	close(errCh)
	multiErr := xerrors.NewMultiError()
	for err := range errCh {
		multiErr = multiErr.Add(err)
	}
	return multiErr.FinalError()
}
func (h helper) allInstanceMetadatas(placement placement.Placement) (instanceMetadatas, error) {
	placementInstances := placement.Instances()
	deploymentInstances, err := h.mgr.QueryAll()
	if err != nil {
		return nil, fmt.Errorf("unable to query all instances from deployment: %v", err)
	}
	metadatas, err := h.computeInstanceMetadatas(placementInstances, deploymentInstances)
	if err != nil {
		return nil, fmt.Errorf("unable to compute instance metadatas: %v", err)
	}
	return metadatas, nil
}