func (c *AggregatorConfiguration) NewAggregatorOptions(
	address string,
	client client.Client,
	runtimeOptsManager aggruntime.OptionsManager,
	instrumentOpts instrument.Options,
) (aggregator.Options, error) {
	opts := aggregator.NewOptions().
		SetInstrumentOptions(instrumentOpts).
		SetRuntimeOptionsManager(runtimeOptsManager)
	aggTypesOpts, err := c.AggregationTypes.NewOptions(instrumentOpts)
	if err != nil {
		return nil, err
	}
	opts = opts.SetAggregationTypesOptions(aggTypesOpts)
	opts = setMetricPrefix(opts, c.MetricPrefix, opts.SetMetricPrefix)
	opts = setMetricPrefix(opts, c.CounterPrefix, opts.SetCounterPrefix)
	opts = setMetricPrefix(opts, c.TimerPrefix, opts.SetTimerPrefix)
	opts = setMetricPrefix(opts, c.GaugePrefix, opts.SetGaugePrefix)
	scope := instrumentOpts.MetricsScope()
	iOpts := instrumentOpts.SetMetricsScope(scope.SubScope("stream"))
	streamOpts, err := c.Stream.NewStreamOptions(iOpts)
	if err != nil {
		return nil, err
	}
	opts = opts.SetStreamOptions(streamOpts)
	iOpts = instrumentOpts.SetMetricsScope(scope.SubScope("client"))
	adminClient, err := c.Client.NewAdminClient(client, clock.NewOptions(), iOpts)
	if err != nil {
		return nil, err
	}
	if err = adminClient.Init(); err != nil {
		return nil, err
	}
	opts = opts.SetAdminClient(adminClient)
	instanceID, err := instanceID(address)
	if err != nil {
		return nil, err
	}
	iOpts = instrumentOpts.SetMetricsScope(scope.SubScope("placement-manager"))
	placementManager, err := c.PlacementManager.NewPlacementManager(client, instanceID, iOpts)
	if err != nil {
		return nil, err
	}
	opts = opts.SetPlacementManager(placementManager)
	hashType := sharding.DefaultHash
	if c.HashType != nil {
		hashType = *c.HashType
	}
	shardFn, err := hashType.ShardFn()
	if err != nil {
		return nil, err
	}
	opts = opts.SetShardFn(shardFn)
	if c.BufferDurationBeforeShardCutover != 0 {
		opts = opts.SetBufferDurationBeforeShardCutover(c.BufferDurationBeforeShardCutover)
	}
	if c.BufferDurationAfterShardCutoff != 0 {
		opts = opts.SetBufferDurationAfterShardCutoff(c.BufferDurationAfterShardCutoff)
	}
	if c.BufferDurationForPastTimedMetric != 0 {
		opts = opts.SetBufferForPastTimedMetricFn(bufferForPastTimedMetricFn(c.BufferDurationForPastTimedMetric))
	}
	if c.BufferDurationForFutureTimedMetric != 0 {
		opts = opts.SetBufferForFutureTimedMetric(c.BufferDurationForFutureTimedMetric)
	}
	if c.ResignTimeout != 0 {
		opts = opts.SetResignTimeout(c.ResignTimeout)
	}
	iOpts = instrumentOpts.SetMetricsScope(scope.SubScope("flush-times-manager"))
	flushTimesManager, err := c.FlushTimesManager.NewFlushTimesManager(client, iOpts)
	if err != nil {
		return nil, err
	}
	opts = opts.SetFlushTimesManager(flushTimesManager)
	iOpts = instrumentOpts.SetMetricsScope(scope.SubScope("election-manager"))
	placementNamespace := c.PlacementManager.KVConfig.Namespace
	electionManager, err := c.ElectionManager.NewElectionManager(
		client,
		instanceID,
		placementNamespace,
		placementManager,
		flushTimesManager,
		iOpts,
	)
	if err != nil {
		return nil, err
	}
	opts = opts.SetElectionManager(electionManager)
	iOpts = instrumentOpts.SetMetricsScope(scope.SubScope("flush-manager"))
	flushManagerOpts, err := c.FlushManager.NewFlushManagerOptions(
		placementManager,
		electionManager,
		flushTimesManager,
		iOpts,
	)
	if err != nil {
		return nil, err
	}
	flushManager := aggregator.NewFlushManager(flushManagerOpts)
	opts = opts.SetFlushManager(flushManager)
	iOpts = instrumentOpts.SetMetricsScope(scope.SubScope("flush-handler"))
	flushHandler, err := c.Flush.NewHandler(client, iOpts)
	if err != nil {
		return nil, err
	}
	opts = opts.SetFlushHandler(flushHandler)
	jitterEnabled := flushManagerOpts.JitterEnabled()
	maxJitterFn := flushManagerOpts.MaxJitterFn()
	maxAllowedForwardingDelayFn := c.Forwarding.MaxAllowedForwardingDelayFn(jitterEnabled, maxJitterFn)
	opts = opts.SetMaxAllowedForwardingDelayFn(maxAllowedForwardingDelayFn)
	if c.EntryTTL != 0 {
		opts = opts.SetEntryTTL(c.EntryTTL)
	}
	if c.EntryCheckInterval != 0 {
		opts = opts.SetEntryCheckInterval(c.EntryCheckInterval)
	}
	if c.EntryCheckBatchPercent != 0.0 {
		opts = opts.SetEntryCheckBatchPercent(c.EntryCheckBatchPercent)
	}
	if c.MaxTimerBatchSizePerWrite != 0 {
		opts = opts.SetMaxTimerBatchSizePerWrite(c.MaxTimerBatchSizePerWrite)
	}
	storagePolicies := make([]policy.StoragePolicy, len(c.DefaultStoragePolicies))
	copy(storagePolicies, c.DefaultStoragePolicies)
	opts = opts.SetDefaultStoragePolicies(storagePolicies)
	if c.MaxNumCachedSourceSets != nil {
		opts = opts.SetMaxNumCachedSourceSets(*c.MaxNumCachedSourceSets)
	}
	if c.DiscardNaNAggregatedValues != nil {
		opts = opts.SetDiscardNaNAggregatedValues(*c.DiscardNaNAggregatedValues)
	}
	iOpts = instrumentOpts.SetMetricsScope(scope.SubScope("counter-elem-pool"))
	counterElemPoolOpts := c.CounterElemPool.NewObjectPoolOptions(iOpts)
	counterElemPool := aggregator.NewCounterElemPool(counterElemPoolOpts)
	opts = opts.SetCounterElemPool(counterElemPool)
	counterElemPool.Init(func() *aggregator.CounterElem {
		return aggregator.MustNewCounterElem(nil, policy.EmptyStoragePolicy, aggregation.DefaultTypes, applied.DefaultPipeline, 0, aggregator.NoPrefixNoSuffix, opts)
	})
	iOpts = instrumentOpts.SetMetricsScope(scope.SubScope("timer-elem-pool"))
	timerElemPoolOpts := c.TimerElemPool.NewObjectPoolOptions(iOpts)
	timerElemPool := aggregator.NewTimerElemPool(timerElemPoolOpts)
	opts = opts.SetTimerElemPool(timerElemPool)
	timerElemPool.Init(func() *aggregator.TimerElem {
		return aggregator.MustNewTimerElem(nil, policy.EmptyStoragePolicy, aggregation.DefaultTypes, applied.DefaultPipeline, 0, aggregator.NoPrefixNoSuffix, opts)
	})
	iOpts = instrumentOpts.SetMetricsScope(scope.SubScope("gauge-elem-pool"))
	gaugeElemPoolOpts := c.GaugeElemPool.NewObjectPoolOptions(iOpts)
	gaugeElemPool := aggregator.NewGaugeElemPool(gaugeElemPoolOpts)
	opts = opts.SetGaugeElemPool(gaugeElemPool)
	gaugeElemPool.Init(func() *aggregator.GaugeElem {
		return aggregator.MustNewGaugeElem(nil, policy.EmptyStoragePolicy, aggregation.DefaultTypes, applied.DefaultPipeline, 0, aggregator.NoPrefixNoSuffix, opts)
	})
	iOpts = instrumentOpts.SetMetricsScope(scope.SubScope("entry-pool"))
	entryPoolOpts := c.EntryPool.NewObjectPoolOptions(iOpts)
	entryPool := aggregator.NewEntryPool(entryPoolOpts)
	runtimeOpts := runtimeOptsManager.RuntimeOptions()
	opts = opts.SetEntryPool(entryPool)
	entryPool.Init(func() *aggregator.Entry { return aggregator.NewEntry(nil, runtimeOpts, opts) })
	return opts, nil
}
NewAggregatorOptions creates a new set of aggregator options.
package config
import (
	"errors"
	"fmt"
	"net"
	"os"
	"runtime"
	"sort"
	"time"
	"github.com/m3db/m3aggregator/aggregation/quantile/cm"
	"github.com/m3db/m3aggregator/aggregator"
	"github.com/m3db/m3aggregator/aggregator/handler"
	aggclient "github.com/m3db/m3aggregator/client"
	aggruntime "github.com/m3db/m3aggregator/runtime"
	"github.com/m3db/m3aggregator/sharding"
	"github.com/m3db/m3cluster/client"
	"github.com/m3db/m3cluster/kv"
	"github.com/m3db/m3cluster/placement"
	"github.com/m3db/m3cluster/services"
	"github.com/m3db/m3metrics/aggregation"
	"github.com/m3db/m3metrics/pipeline/applied"
	"github.com/m3db/m3metrics/policy"
	"github.com/m3db/m3x/clock"
	"github.com/m3db/m3x/instrument"
	"github.com/m3db/m3x/pool"
	"github.com/m3db/m3x/retry"
	"github.com/m3db/m3x/sync"
)
var (
	errNoKVClientConfiguration = errors.New("no kv client configuration")
	errEmptyJitterBucketList   = errors.New("empty jitter bucket list")
)
type AggregatorConfiguration struct {
	AggregationTypes aggregation.TypesConfiguration `yaml:"aggregationTypes"`
	MetricPrefix string `yaml:"metricPrefix"`
	CounterPrefix string `yaml:"counterPrefix"`
	TimerPrefix string `yaml:"timerPrefix"`
	GaugePrefix string `yaml:"gaugePrefix"`
	Stream streamConfiguration `yaml:"stream"`
	Client aggclient.Configuration `yaml:"client"`
	PlacementManager placementManagerConfiguration `yaml:"placementManager"`
	HashType *sharding.HashType `yaml:"hashType"`
	BufferDurationBeforeShardCutover time.Duration `yaml:"bufferDurationBeforeShardCutover"`
	BufferDurationAfterShardCutoff time.Duration `yaml:"bufferDurationAfterShardCutoff"`
	BufferDurationForPastTimedMetric time.Duration `yaml:"bufferDurationForPastTimedMetric"`
	BufferDurationForFutureTimedMetric time.Duration `yaml:"bufferDurationForFutureTimedMetric"`
	ResignTimeout time.Duration `yaml:"resignTimeout"`
	FlushTimesManager flushTimesManagerConfiguration `yaml:"flushTimesManager"`
	ElectionManager electionManagerConfiguration `yaml:"electionManager"`
	FlushManager flushManagerConfiguration `yaml:"flushManager"`
	Flush handler.FlushHandlerConfiguration `yaml:"flush"`
	Forwarding forwardingConfiguration `yaml:"forwarding"`
	EntryTTL time.Duration `yaml:"entryTTL"`
	EntryCheckInterval time.Duration `yaml:"entryCheckInterval"`
	EntryCheckBatchPercent float64 `yaml:"entryCheckBatchPercent" validate:"min=0.0,max=1.0"`
	MaxTimerBatchSizePerWrite int `yaml:"maxTimerBatchSizePerWrite" validate:"min=0"`
	DefaultStoragePolicies []policy.StoragePolicy `yaml:"defaultStoragePolicies" validate:"nonzero"`
	MaxNumCachedSourceSets *int `yaml:"maxNumCachedSourceSets"`
	DiscardNaNAggregatedValues *bool `yaml:"discardNaNAggregatedValues"`
	CounterElemPool pool.ObjectPoolConfiguration `yaml:"counterElemPool"`
	TimerElemPool pool.ObjectPoolConfiguration `yaml:"timerElemPool"`
	GaugeElemPool pool.ObjectPoolConfiguration `yaml:"gaugeElemPool"`
	EntryPool pool.ObjectPoolConfiguration `yaml:"entryPool"`
}