func (agg *aggregator) closeShardsAsync(shards []*aggregatorShard) {
	pendingClose := atomic.AddInt32(&agg.shardsPendingClose, int32(len(shards)))
	agg.metrics.shards.pendingClose.Update(float64(pendingClose))
	for _, shard := range shards {
		shard := shard
		go func() {
			shard.Close()
			pendingClose := atomic.AddInt32(&agg.shardsPendingClose, -1)
			agg.metrics.shards.pendingClose.Update(float64(pendingClose))
			agg.metrics.shards.close.Inc(1)
		}()
	}
}
closeShardsAsync asynchronously closes the shards to avoid blocking writes.
Because each shard write happens while holding the shard read lock, the shard
may only close itself after all its pending writes are finished.
package aggregator
import (
	"context"
	"errors"
	"math"
	"strconv"
	"sync"
	"sync/atomic"
	"time"
	"github.com/m3db/m3aggregator/aggregator/handler"
	"github.com/m3db/m3aggregator/client"
	"github.com/m3db/m3aggregator/sharding"
	"github.com/m3db/m3cluster/placement"
	"github.com/m3db/m3cluster/shard"
	"github.com/m3db/m3metrics/metadata"
	"github.com/m3db/m3metrics/metric"
	"github.com/m3db/m3metrics/metric/aggregated"
	"github.com/m3db/m3metrics/metric/id"
	"github.com/m3db/m3metrics/metric/unaggregated"
	"github.com/m3db/m3x/clock"
	"github.com/m3db/m3x/instrument"
	"github.com/uber-go/tally"
)
const (
	uninitializedCutoverNanos = math.MinInt64
	uninitializedShardSetID   = 0
)
var (
	errAggregatorNotOpenOrClosed     = errors.New("aggregator is not open or closed")
	errAggregatorAlreadyOpenOrClosed = errors.New("aggregator is already open or closed")
	errInvalidMetricType             = errors.New("invalid metric type")
	errActivePlacementChanged        = errors.New("active placement has changed")
	errShardNotOwned                 = errors.New("aggregator shard is not owned")
)
type Aggregator interface {
	Open() error
	AddUntimed(metric unaggregated.MetricUnion, metas metadata.StagedMetadatas) error
	AddTimed(metric aggregated.Metric, metadata metadata.TimedMetadata) error
	AddForwarded(metric aggregated.ForwardedMetric, metadata metadata.ForwardMetadata) error
	Resign() error
	Status() RuntimeStatus
	Close() error
}
type aggregator struct {
	sync.RWMutex
	opts              Options
	nowFn             clock.NowFn
	shardFn           sharding.ShardFn
	checkInterval     time.Duration
	placementManager  PlacementManager
	flushTimesManager FlushTimesManager
	flushTimesChecker flushTimesChecker
	electionManager   ElectionManager
	flushManager      FlushManager
	flushHandler      handler.Handler
	adminClient       client.AdminClient
	resignTimeout     time.Duration
	shardSetID          uint32
	shardSetOpen        bool
	shardIDs            []uint32
	shards              []*aggregatorShard
	currStagedPlacement placement.ActiveStagedPlacement
	currPlacement       placement.Placement
	state               aggregatorState
	doneCh              chan struct{}
	wg                  sync.WaitGroup
	sleepFn             sleepFn
	shardsPendingClose  int32
	metrics             aggregatorMetrics
}
func NewAggregator(opts Options) Aggregator {
	iOpts := opts.InstrumentOptions()
	scope := iOpts.MetricsScope()
	samplingRate := iOpts.MetricsSamplingRate()
	return &aggregator{
		opts:              opts,
		nowFn:             opts.ClockOptions().NowFn(),
		shardFn:           opts.ShardFn(),
		checkInterval:     opts.EntryCheckInterval(),
		placementManager:  opts.PlacementManager(),
		flushTimesManager: opts.FlushTimesManager(),
		flushTimesChecker: newFlushTimesChecker(scope.SubScope("tick.shard-check")),
		electionManager:   opts.ElectionManager(),
		flushManager:      opts.FlushManager(),
		flushHandler:      opts.FlushHandler(),
		adminClient:       opts.AdminClient(),
		resignTimeout:     opts.ResignTimeout(),
		metrics:           newAggregatorMetrics(scope, samplingRate, opts.MaxAllowedForwardingDelayFn()),
		doneCh:            make(chan struct{}),
		sleepFn:           time.Sleep,
	}
}
func (agg *aggregator) Open() error {
	agg.Lock()
	defer agg.Unlock()
	if agg.state != aggregatorNotOpen {
		return errAggregatorAlreadyOpenOrClosed
	}
	if err := agg.placementManager.Open(); err != nil {
		return err
	}
	stagedPlacement, placement, err := agg.placementManager.Placement()
	if err != nil {
		return err
	}
	if err := agg.processPlacementWithLock(stagedPlacement, placement); err != nil {
		return err
	}
	if agg.checkInterval > 0 {
		agg.wg.Add(1)
		go agg.tick()
	}
	agg.state = aggregatorOpen
	return nil
}
func (agg *aggregator) AddUntimed(
	metric unaggregated.MetricUnion,
	metadatas metadata.StagedMetadatas,
) error {
	callStart := agg.nowFn()
	if err := agg.checkMetricType(metric); err != nil {
		agg.metrics.addUntimed.ReportError(err)
		return err
	}
	shard, err := agg.shardFor(metric.ID)
	if err != nil {
		agg.metrics.addUntimed.ReportError(err)
		return err
	}
	if err = shard.AddUntimed(metric, metadatas); err != nil {
		agg.metrics.addUntimed.ReportError(err)
		return err
	}
	agg.metrics.addUntimed.ReportSuccess(agg.nowFn().Sub(callStart))
	return nil
}
func (agg *aggregator) AddTimed(
	metric aggregated.Metric,
	metadata metadata.TimedMetadata,
) error {
	callStart := agg.nowFn()
	agg.metrics.timed.Inc(1)
	shard, err := agg.shardFor(metric.ID)
	if err != nil {
		agg.metrics.addTimed.ReportError(err)
		return err
	}
	if err = shard.AddTimed(metric, metadata); err != nil {
		agg.metrics.addTimed.ReportError(err)
		return err
	}
	agg.metrics.addTimed.ReportSuccess(agg.nowFn().Sub(callStart))
	return nil
}
func (agg *aggregator) AddForwarded(
	metric aggregated.ForwardedMetric,
	metadata metadata.ForwardMetadata,
) error {
	callStart := agg.nowFn()
	agg.metrics.forwarded.Inc(1)
	shard, err := agg.shardFor(metric.ID)
	if err != nil {
		agg.metrics.addForwarded.ReportError(err)
		return err
	}
	if err = shard.AddForwarded(metric, metadata); err != nil {
		agg.metrics.addForwarded.ReportError(err)
		return err
	}
	callEnd := agg.nowFn()
	agg.metrics.addForwarded.ReportSuccess(callEnd.Sub(callStart))
	forwardingDelay := time.Duration(callEnd.UnixNano() - metric.TimeNanos)
	agg.metrics.addForwarded.ReportForwardingLatency(
		metadata.StoragePolicy.Resolution().Window,
		metadata.NumForwardedTimes,
		forwardingDelay,
	)
	return nil
}
func (agg *aggregator) Resign() error {
	ctx, cancel := context.WithTimeout(context.Background(), agg.resignTimeout)
	defer cancel()
	return agg.electionManager.Resign(ctx)
}
func (agg *aggregator) Status() RuntimeStatus {
	return RuntimeStatus{
		FlushStatus: agg.flushManager.Status(),
	}
}
func (agg *aggregator) Close() error {
	agg.Lock()
	defer agg.Unlock()
	if agg.state != aggregatorOpen {
		return errAggregatorNotOpenOrClosed
	}
	close(agg.doneCh)
	for _, shardID := range agg.shardIDs {
		agg.shards[shardID].Close()
	}
	if agg.shardSetOpen {
		agg.closeShardSetWithLock()
	}
	agg.flushHandler.Close()
	if agg.adminClient != nil {
		agg.adminClient.Close()
	}
	agg.state = aggregatorClosed
	return nil
}
func (agg *aggregator) shardFor(id id.RawID) (*aggregatorShard, error) {
	agg.RLock()
	shard, err := agg.shardForWithLock(id, noUpdateShards)
	if err == nil || err != errActivePlacementChanged {
		agg.RUnlock()
		return shard, err
	}
	agg.RUnlock()
	agg.Lock()
	shard, err = agg.shardForWithLock(id, updateShards)
	agg.Unlock()
	return shard, err
}
func (agg *aggregator) shardForWithLock(id id.RawID, updateShardsType updateShardsType) (*aggregatorShard, error) {
	if agg.state != aggregatorOpen {
		return nil, errAggregatorNotOpenOrClosed
	}
	stagedPlacement, placement, err := agg.placementManager.Placement()
	if err != nil {
		return nil, err
	}
	if agg.shouldProcessPlacementWithLock(stagedPlacement, placement) {
		if updateShardsType == noUpdateShards {
			return nil, errActivePlacementChanged
		}
		if err := agg.processPlacementWithLock(stagedPlacement, placement); err != nil {
			return nil, err
		}
	}
	numShards := placement.NumShards()
	shardID := agg.shardFn([]byte(id), numShards)
	if int(shardID) >= len(agg.shards) || agg.shards[shardID] == nil {
		return nil, errShardNotOwned
	}
	return agg.shards[shardID], nil
}
func (agg *aggregator) processPlacementWithLock(
	newStagedPlacement placement.ActiveStagedPlacement,
	newPlacement placement.Placement,
) error {
	if !agg.shouldProcessPlacementWithLock(newStagedPlacement, newPlacement) {
		return nil
	}
	var newShardSet shard.Shards
	instance, err := agg.placementManager.InstanceFrom(newPlacement)
	if err == nil {
		newShardSet = instance.Shards()
	} else if err == ErrInstanceNotFoundInPlacement {
		newShardSet = shard.NewShards(nil)
	} else {
		return err
	}
	agg.updateShardsWithLock(newStagedPlacement, newPlacement, newShardSet)
	if err := agg.updateShardSetIDWithLock(instance); err != nil {
		return err
	}
	agg.metrics.placement.updated.Inc(1)
	return nil
}
func (agg *aggregator) shouldProcessPlacementWithLock(
	newStagedPlacement placement.ActiveStagedPlacement,
	newPlacement placement.Placement,
) bool {
	if agg.currStagedPlacement == nil || agg.currStagedPlacement != newStagedPlacement {
		agg.metrics.placement.stagedPlacementChanged.Inc(1)
		return true
	}
	if agg.currPlacement == nil || agg.currPlacement.CutoverNanos() < newPlacement.CutoverNanos() {
		agg.metrics.placement.cutoverChanged.Inc(1)
		return true
	}
	return false
}
func (agg *aggregator) updateShardSetIDWithLock(instance placement.Instance) error {
	if instance == nil {
		return agg.clearShardSetIDWithLock()
	}
	return agg.resetShardSetIDWithLock(instance)
}
func (agg *aggregator) clearShardSetIDWithLock() error {
	agg.metrics.shardSetID.clear.Inc(1)
	if !agg.shardSetOpen {
		return nil
	}
	if err := agg.closeShardSetWithLock(); err != nil {
		return err
	}
	agg.shardSetID = uninitializedShardSetID
	agg.shardSetOpen = false
	return nil
}
func (agg *aggregator) resetShardSetIDWithLock(instance placement.Instance) error {
	agg.metrics.shardSetID.reset.Inc(1)
	if !agg.shardSetOpen {
		shardSetID := instance.ShardSetID()
		if err := agg.openShardSetWithLock(shardSetID); err != nil {
			return err
		}
		agg.shardSetID = shardSetID
		agg.shardSetOpen = true
		return nil
	}
	if instance.ShardSetID() == agg.shardSetID {
		agg.metrics.shardSetID.same.Inc(1)
		return nil
	}
	agg.metrics.shardSetID.changed.Inc(1)
	if err := agg.closeShardSetWithLock(); err != nil {
		return err
	}
	newShardSetID := instance.ShardSetID()
	if err := agg.openShardSetWithLock(newShardSetID); err != nil {
		return err
	}
	agg.shardSetID = newShardSetID
	agg.shardSetOpen = true
	return nil
}
func (agg *aggregator) openShardSetWithLock(shardSetID uint32) error {
	agg.metrics.shardSetID.open.Inc(1)
	if err := agg.flushTimesManager.Open(shardSetID); err != nil {
		return err
	}
	if err := agg.electionManager.Open(shardSetID); err != nil {
		return err
	}
	return agg.flushManager.Open()
}
func (agg *aggregator) closeShardSetWithLock() error {
	agg.metrics.shardSetID.close.Inc(1)
	if err := agg.flushManager.Close(); err != nil {
		return err
	}
	if err := agg.flushManager.Reset(); err != nil {
		return err
	}
	if err := agg.electionManager.Close(); err != nil {
		return err
	}
	if err := agg.electionManager.Reset(); err != nil {
		return err
	}
	if err := agg.flushTimesManager.Close(); err != nil {
		return err
	}
	return agg.flushTimesManager.Reset()
}
func (agg *aggregator) updateShardsWithLock(
	newStagedPlacement placement.ActiveStagedPlacement,
	newPlacement placement.Placement,
	newShardSet shard.Shards,
) {
	var (
		incoming []*aggregatorShard
		closing  = make([]*aggregatorShard, 0, len(agg.shardIDs))
	)
	for _, shard := range agg.shards {
		if shard == nil {
			continue
		}
		if !newShardSet.Contains(shard.ID()) {
			closing = append(closing, shard)
		}
	}
	var (
		newShards   = newShardSet.All()
		newShardIDs []uint32
	)
	if numShards := len(newShards); numShards > 0 {
		newShardIDs = make([]uint32, 0, numShards)
		maxShardID := newShards[numShards-1].ID()
		incoming = make([]*aggregatorShard, maxShardID+1)
	}
	for _, shard := range newShards {
		shardID := shard.ID()
		newShardIDs = append(newShardIDs, shardID)
		if int(shardID) < len(agg.shards) && agg.shards[shardID] != nil {
			incoming[shardID] = agg.shards[shardID]
		} else {
			incoming[shardID] = newAggregatorShard(shardID, agg.opts)
			agg.metrics.shards.add.Inc(1)
		}
		shardTimeRange := timeRange{
			cutoverNanos: shard.CutoverNanos(),
			cutoffNanos:  shard.CutoffNanos(),
		}
		incoming[shardID].SetWriteableRange(shardTimeRange)
	}
	agg.shardIDs = newShardIDs
	agg.shards = incoming
	agg.currStagedPlacement = newStagedPlacement
	agg.currPlacement = newPlacement
	agg.closeShardsAsync(closing)
}
func (agg *aggregator) checkMetricType(mu unaggregated.MetricUnion) error {
	switch mu.Type {
	case metric.CounterType:
		agg.metrics.counters.Inc(1)
		return nil
	case metric.TimerType:
		agg.metrics.timerBatches.Inc(1)
		agg.metrics.timers.Inc(int64(len(mu.BatchTimerVal)))
		return nil
	case metric.GaugeType:
		agg.metrics.gauges.Inc(1)
		return nil
	default:
		return errInvalidMetricType
	}
}
func (agg *aggregator) ownedShards() (owned, toClose []*aggregatorShard) {
	agg.Lock()
	defer agg.Unlock()
	if len(agg.shardIDs) == 0 {
		return nil, nil
	}
	flushTimes, err := agg.flushTimesManager.Get()
	if err != nil {
		agg.metrics.tick.flushTimesErrors.Inc(1)
	}
	owned = make([]*aggregatorShard, 0, len(agg.shardIDs))
	for i := 0; i < len(agg.shardIDs); i++ {
		shardID := agg.shardIDs[i]
		shard := agg.shards[shardID]
		canCloseShard := !shard.IsWritable() &&
			shard.IsCutoff() &&
			agg.flushTimesChecker.HasFlushed(
				shard.ID(),
				shard.CutoffNanos(),
				flushTimes,
			)
		if !canCloseShard {
			owned = append(owned, shard)
		} else {
			lastIdx := len(agg.shardIDs) - 1
			agg.shardIDs[i], agg.shardIDs[lastIdx] = agg.shardIDs[lastIdx], agg.shardIDs[i]
			agg.shardIDs = agg.shardIDs[:lastIdx]
			i--
			agg.shards[shardID] = nil
			toClose = append(toClose, shard)
		}
	}
	return owned, toClose
}