func (mcp *Mcp) CreateConsumerGroup(ctx thrift.Context, createRequest *shared.CreateConsumerGroupRequest) (*shared.ConsumerGroupDescription, error) {
	if !mcp.isStarted() {
		return nil, &shared.InternalServiceError{Message: "Controller not started"}
	}
	context := mcp.context
	context.m3Client.IncCounter(metrics.ControllerCreateConsumerGroupScope, metrics.ControllerRequests)
	sw := context.m3Client.StartTimer(metrics.ControllerCreateConsumerGroupScope, metrics.ControllerLatencyTimer)
	defer sw.Stop()
	lclLg := context.log.WithFields(bark.Fields{
		common.TagDstPth: common.FmtDstPth(createRequest.GetDestinationPath()),
		common.TagCnsPth: common.FmtCnsPth(createRequest.GetConsumerGroupName()),
	})
	cgDesc, err := mcp.mClient.CreateConsumerGroup(ctx, createRequest)
	if err != nil {
		lclLg.WithField(common.TagErr, err).Error("CreateConsumerGroup: local CreateConsumerGroup failed")
		context.m3Client.IncCounter(metrics.ControllerCreateConsumerGroupScope, metrics.ControllerFailures)
		return nil, err
	}
	if createRequest.GetIsMultiZone() {
		cgUUID := cgDesc.GetConsumerGroupUUID()
		createCGUUIDRequest := shared.NewCreateConsumerGroupUUIDRequest()
		createCGUUIDRequest.Request = createRequest
		createCGUUIDRequest.ConsumerGroupUUID = common.StringPtr(cgUUID)
		localReplicator, replicatorErr := mcp.GetClientFactory().GetReplicatorClient()
		lclLg = lclLg.WithField(common.TagCnsm, common.FmtCnsm(cgUUID))
		if replicatorErr != nil {
			lclLg.WithField(common.TagErr, replicatorErr).Error("CreateConsumerGroup: GetReplicatorClient failed")
			context.m3Client.IncCounter(metrics.ControllerCreateConsumerGroupScope, metrics.ControllerErrCallReplicatorCounter)
			return cgDesc, nil
		}
		replicatorErr = localReplicator.CreateRemoteConsumerGroupUUID(ctx, createCGUUIDRequest)
		if replicatorErr != nil {
			lclLg.WithField(common.TagErr, replicatorErr).Error("CreateConsumerGroup: CreateRemoteConsumerGroupUUID error")
			context.m3Client.IncCounter(metrics.ControllerCreateConsumerGroupScope, metrics.ControllerErrCallReplicatorCounter)
			return cgDesc, nil
		}
	}
	return cgDesc, nil
}
CreateConsumerGroup creates local and remote consumer group
package controllerhost
import (
	"encoding/json"
	"fmt"
	"math/rand"
	"strings"
	"sync/atomic"
	"time"
	"github.com/pborman/uuid"
	"github.com/uber-common/bark"
	"github.com/uber/cherami-server/common"
	"github.com/uber/cherami-server/common/configure"
	"github.com/uber/cherami-server/common/dconfig"
	"github.com/uber/cherami-server/common/metadata"
	"github.com/uber/cherami-server/common/metrics"
	"github.com/uber/cherami-server/services/controllerhost/load"
	a "github.com/uber/cherami-thrift/.generated/go/admin"
	c "github.com/uber/cherami-thrift/.generated/go/controller"
	m "github.com/uber/cherami-thrift/.generated/go/metadata"
	"github.com/uber/cherami-thrift/.generated/go/shared"
	tchannel "github.com/uber/tchannel-go"
	"github.com/uber/tchannel-go/thrift"
)
var (
	ErrMalformedUUID = &shared.BadRequestError{Message: "Malformed UUID in request"}
	ErrNoHealthyExtent = &shared.InternalServiceError{Message: "No healthy extent found for destination"}
	ErrEventQueueFull = &shared.InternalServiceError{Message: "EventQueue full"}
	ErrTryLock = &shared.InternalServiceError{Message: "Failed to acquire lock, backoff and retry"}
	ErrUnavailable = &shared.InternalServiceError{Message: "Service unavailable, backoff and retry"}
	ErrDestinationNotExists = &shared.EntityNotExistsError{Message: "Destination does not exist"}
	ErrConsumerGroupNotExists = &shared.EntityNotExistsError{Message: "ConsumerGroup does not exist"}
	ErrDestinationDisabled = &shared.EntityDisabledError{Message: "Destination is not enabled"}
	ErrConsumerGroupDisabled = &shared.EntityDisabledError{Message: "Consumer group is not enabled"}
)
const (
	nEventPipelineWorkers      = 2048  
	hashLockTableSize          = 1024
	maxFailedExtentSealSetSize = 8192  
	maxExtentSealsPerSecond    = 200
)
type (
	Mcp struct {
		*common.Service
		hostIDHeartbeater common.HostIDHeartbeater
		mClient           m.TChanMetadataService
		context           *Context
		started           int32
	}
	Context struct {
		hostID              string
		localZone           string
		mm                  MetadataMgr
		rpm                 common.RingpopMonitor
		zoneFailoverManager common.ZoneFailoverManager
		failureDetector     Dfdd
		log                 bark.Logger
		dstLock             LockMgr
		eventPipeline       EventPipeline
		resultCache         *resultCache
		extentMonitor       *extentStateMonitor
		timeSource          common.TimeSource
		channel             *tchannel.Channel
		clientFactory       common.ClientFactory
		retMgr              *retMgrRunner
		appConfig           configure.CommonAppConfig
		m3Client            metrics.Client
		cfgMgr              dconfig.ConfigManager
		loadMetrics         load.MetricsAggregator
		placement           Placement
		extentSeals         struct {
			inProgress common.ConcurrentMap
			failed common.ConcurrentMap
			tokenBucket common.TokenBucket
		}
	}
	Boolean bool
)
var _ c.TChanController = (*Mcp)(nil)
func NewController(cfg configure.CommonAppConfig, sVice *common.Service, metadataClient m.TChanMetadataService, zoneFailoverManager common.ZoneFailoverManager) (*Mcp, []thrift.TChanServer) {
	hostID := uuid.New()
	instance := new(Mcp)
	instance.Service = sVice
	deploymentName := sVice.GetConfig().GetDeploymentName()
	logger := (sVice.GetConfig().GetLogger()).WithFields(bark.Fields{common.TagCtrl: common.FmtCtrl(hostID), common.TagDplName: common.FmtDplName(deploymentName)})
	lockMgr, err := NewLockMgr(hashLockTableSize, common.UUIDHashCode, logger)
	if err != nil {
		logger.WithField(common.TagErr, err).Fatal(`Failed to create hash lock`)
	}
	context := &Context{
		appConfig:  cfg,
		timeSource: common.NewRealTimeSource(),
		log:        logger,
	}
	context.localZone, _ = common.GetLocalClusterInfo(strings.ToLower(deploymentName))
	context.zoneFailoverManager = zoneFailoverManager
	context.dstLock = lockMgr
	context.m3Client = metrics.NewClient(instance.Service.GetMetricsReporter(), metrics.Controller)
	instance.mClient = metadata.NewMetadataMetricsMgr(metadataClient, context.m3Client, context.log)
	context.mm = NewMetadataMgr(instance.mClient, context.m3Client, context.log)
	context.extentSeals.inProgress = common.NewShardedConcurrentMap(1024, common.UUIDHashCode)
	context.extentSeals.failed = common.NewShardedConcurrentMap(1024, common.UUIDHashCode)
	context.extentSeals.tokenBucket = common.NewTokenBucket(maxExtentSealsPerSecond, common.NewRealTimeSource())
	context.resultCache = newResultCache(context)
	context.cfgMgr = newConfigManager(metadataClient, context.log)
	context.loadMetrics = load.NewTimeSlotAggregator(common.NewRealTimeSource(), context.log)
	if context.placement, err = NewDistancePlacement(context); err != nil {
		context.log.WithField(common.TagErr, err).Error("Cannot initialize topology for placement")
	}
	instance.context = context
	return instance, []thrift.TChanServer{c.NewTChanControllerServer(instance),
		a.NewTChanControllerHostAdminServer(instance)}
}
func (mcp *Mcp) Start(thriftService []thrift.TChanServer) {
	mcp.Service.Start(thriftService)
	context := mcp.context
	context.hostID = mcp.GetHostUUID()
	context.clientFactory = mcp.GetClientFactory()
	context.channel = mcp.GetTChannel()
	context.rpm = mcp.GetRingpopMonitor()
	context.zoneFailoverManager.Start()
	context.eventPipeline = NewEventPipeline(context, nEventPipelineWorkers)
	context.eventPipeline.Start()
	context.loadMetrics.Start()
	context.cfgMgr.Start()
	context.failureDetector = NewDfdd(context, common.NewRealTimeSource())
	context.failureDetector.Start()
	context.retMgr = newRetMgrRunner(&retMgrRunnerContext{
		hostID:         mcp.GetHostUUID(),
		ringpop:        context.rpm,
		metadataClient: mcp.mClient,
		clientFactory:  context.clientFactory,
		log:            context.log,
		m3Client:       context.m3Client,
		localZone:      context.localZone,
	})
	context.retMgr.Start()
	mcp.hostIDHeartbeater = common.NewHostIDHeartbeater(mcp.mClient, mcp.GetHostUUID(), mcp.GetHostPort(), mcp.GetHostName(), mcp.context.log)
	mcp.hostIDHeartbeater.Start()
	context.extentMonitor = newExtentStateMonitor(context)
	context.extentMonitor.Start()
	atomic.StoreInt32(&mcp.started, 1)
}
func (mcp *Mcp) Stop() {
	mcp.hostIDHeartbeater.Stop()
	mcp.context.zoneFailoverManager.Stop()
	mcp.context.extentMonitor.Stop()
	mcp.context.retMgr.Stop()
	mcp.context.failureDetector.Stop()
	mcp.context.eventPipeline.Stop()
	mcp.context.loadMetrics.Stop()
	mcp.context.cfgMgr.Stop()
	mcp.Service.Stop()
}
func (mcp *Mcp) isStarted() bool {
	return (atomic.LoadInt32(&mcp.started) == 1)
}
func (mcp *Mcp) GetInputHosts(ctx thrift.Context, inReq *c.GetInputHostsRequest) (*c.GetInputHostsResult_, error) {
	if !mcp.isStarted() {
		return nil, &shared.InternalServiceError{Message: "Controller not started"}
	}
	context := mcp.context
	context.m3Client.IncCounter(metrics.GetInputHostsScope, metrics.ControllerRequests)
	sw := context.m3Client.StartTimer(metrics.GetInputHostsScope, metrics.ControllerLatencyTimer)
	defer sw.Stop()
	var result *resultCacheReadResult
	var dstUUID = inReq.GetDestinationUUID()
	if !isUUIDLengthValid(dstUUID) {
		context.m3Client.IncCounter(metrics.GetInputHostsScope, metrics.ControllerErrBadRequestCounter)
		return nil, ErrMalformedUUID
	}
	response := func(err error) (*c.GetInputHostsResult_, error) {
		if len(result.cachedResult) < 1 {
			context.m3Client.IncCounter(metrics.GetInputHostsScope, metrics.ControllerFailures)
			return nil, err
		}
		return &c.GetInputHostsResult_{InputHostIds: result.cachedResult}, nil
	}
	var now = context.timeSource.Now().UnixNano()
	result = context.resultCache.readInputHosts(dstUUID, now)
	if result.cacheHit && !result.refreshCache {
		return response(ErrUnavailable)
	}
	if !context.dstLock.TryLock(dstUUID, getLockTimeout(result)) {
		context.m3Client.IncCounter(metrics.GetInputHostsScope, metrics.ControllerErrTryLockCounter)
		return response(ErrTryLock)
	}
	result = context.resultCache.readInputHosts(dstUUID, now)
	if result.cacheHit && !result.refreshCache {
		context.dstLock.Unlock(dstUUID)
		return response(ErrUnavailable)
	}
	hostIDs, err := refreshInputHostsForDst(context, dstUUID, now)
	context.dstLock.Unlock(dstUUID)
	if err != nil {
		switch {
		case isEntityError(err):
			context.m3Client.IncCounter(metrics.GetInputHostsScope, metrics.ControllerErrBadEntityCounter)
			return nil, err
		case isBadRequestError(err):
			context.m3Client.IncCounter(metrics.GetInputHostsScope, metrics.ControllerErrBadRequestCounter)
			return nil, err
		default:
			return response(&shared.InternalServiceError{Message: err.Error()})
		}
	}
	return &c.GetInputHostsResult_{InputHostIds: hostIDs}, nil
}
func (mcp *Mcp) GetOutputHosts(ctx thrift.Context, inReq *c.GetOutputHostsRequest) (*c.GetOutputHostsResult_, error) {
	if !mcp.isStarted() {
		return nil, &shared.InternalServiceError{Message: "Controller not started"}
	}
	context := mcp.context
	context.m3Client.IncCounter(metrics.GetOutputHostsScope, metrics.ControllerRequests)
	sw := context.m3Client.StartTimer(metrics.GetOutputHostsScope, metrics.ControllerLatencyTimer)
	defer sw.Stop()
	var result *resultCacheReadResult
	var cgUUID = inReq.GetConsumerGroupUUID()
	var dstUUID = inReq.GetDestinationUUID()
	if !isUUIDLengthValid(dstUUID) || !isUUIDLengthValid(cgUUID) {
		context.m3Client.IncCounter(metrics.GetOutputHostsScope, metrics.ControllerErrBadRequestCounter)
		return nil, ErrMalformedUUID
	}
	response := func() (*c.GetOutputHostsResult_, error) {
		if len(result.cachedResult) < 1 && !result.consumeDisabled {
			context.m3Client.IncCounter(metrics.GetOutputHostsScope, metrics.ControllerFailures)
			return nil, ErrUnavailable
		}
		return &c.GetOutputHostsResult_{OutputHostIds: result.cachedResult}, nil
	}
	var now = context.timeSource.Now().UnixNano()
	result = context.resultCache.readOutputHosts(cgUUID, now)
	if result.cacheHit && !result.refreshCache {
		return response()
	}
	if !context.dstLock.TryLock(dstUUID, getLockTimeout(result)) {
		context.m3Client.IncCounter(metrics.GetOutputHostsScope, metrics.ControllerErrTryLockCounter)
		return response()
	}
	result = context.resultCache.readOutputHosts(cgUUID, now)
	if result.cacheHit && !result.refreshCache {
		context.dstLock.Unlock(dstUUID)
		return response()
	}
	hostIDs, err := refreshOutputHostsForConsGroup(context, dstUUID, cgUUID, *result, now)
	context.dstLock.Unlock(dstUUID)
	if err != nil {
		if isEntityError(err) {
			context.m3Client.IncCounter(metrics.GetOutputHostsScope, metrics.ControllerErrBadEntityCounter)
			return nil, err
		}
		return response()
	}
	return &c.GetOutputHostsResult_{OutputHostIds: hostIDs}, nil
}
func (mcp *Mcp) GetQueueDepthInfo(ctx thrift.Context, inReq *c.GetQueueDepthInfoRequest) (*c.GetQueueDepthInfoResult_, error) {
	if inReq == nil {
		return nil, &shared.BadRequestError{Message: "request is nil"}
	}
	cgUUID := inReq.GetKey()
	if !isUUIDLengthValid(cgUUID) {
		mcp.context.log.WithFields(bark.Fields{common.TagCnsm: common.FmtCnsm(cgUUID)}).Error(`GetQueueDepthInfo: Malformed UUID ( cgUUID: )`)
		return nil, ErrMalformedUUID
	}
	queueInfo, err := mcp.context.extentMonitor.queueDepth.GetQueueDepthResult(cgUUID)
	if err == nil {
		output := &QueueDepthCacheJSONFields{
			CacheTime:        queueInfo.Time,
			BacklogAvailable: queueInfo.BacklogAvailable,
			BacklogInflight:  queueInfo.BacklogInflight,
			BacklogDLQ:       queueInfo.BacklogDLQ,
		}
		queueInfo, _ := json.Marshal(output)
		queueInfoStr := string(queueInfo)
		return &c.GetQueueDepthInfoResult_{Value: &queueInfoStr}, nil
	}
	return nil, err
}
func (mcp *Mcp) ExtentsUnreachable(ctx thrift.Context, extentsUnreachableRequest *a.ExtentsUnreachableRequest) error {
	if !mcp.isStarted() {
		return &shared.InternalServiceError{Message: "Controller not started"}
	}
	context := mcp.context
	context.m3Client.IncCounter(metrics.ExtentsUnreachableScope, metrics.ControllerRequests)
	sw := context.m3Client.StartTimer(metrics.ExtentsUnreachableScope, metrics.ControllerLatencyTimer)
	defer sw.Stop()
	dstIDs := make(map[string]struct{})
	for _, update := range extentsUnreachableRequest.GetUpdates() {
		dstID := update.GetDestinationUUID()
		extID := update.GetExtentUUID()
		if !isExtentBeingSealed(context, extID) {
			dstIDs[dstID] = struct{}{}
			addExtentDownEvent(context, update.GetSealSequenceNumber(), dstID, extID)
		}
		context.log.WithFields(bark.Fields{
			common.TagDst: common.FmtDst(dstID),
			common.TagExt: common.FmtExt(string(extID)),
		}).Info("ExtentsUnreachable notificiation from inputhost")
	}
	for id := range dstIDs {
		if !context.dstLock.TryLock(id, time.Second) {
			break
		}
		refreshInputHostsForDst(context, id, context.timeSource.Now().UnixNano())
		context.dstLock.Unlock(id)
	}
	return nil
}
func (mcp *Mcp) ReportNodeMetric(ctx thrift.Context, request *c.ReportNodeMetricRequest) error {
	if !mcp.isStarted() {
		return &shared.InternalServiceError{Message: "Controller not started"}
	}
	context := mcp.context
	context.m3Client.IncCounter(metrics.ReportNodeMetricScope, metrics.ControllerRequests)
	sw := context.m3Client.StartTimer(metrics.ReportNodeMetricScope, metrics.ControllerLatencyTimer)
	defer sw.Stop()
	hostID := request.GetHostId()
	if len(hostID) < 1 {
		return &shared.BadRequestError{Message: "HostID cannot be empty"}
	}
	metrics := request.GetMetrics()
	timestamp := request.GetTimestamp()
	loadMetrics := mcp.context.loadMetrics
	if metrics.IsSetRemainingDiskSpace() {
		loadMetrics.Put(hostID, load.EmptyTag, load.RemDiskSpaceBytes, metrics.GetRemainingDiskSpace(), timestamp)
	}
	if metrics.IsSetNumberOfActiveExtents() {
		loadMetrics.Put(hostID, load.EmptyTag, load.NumExtentsActive, metrics.GetNumberOfActiveExtents(), timestamp)
	}
	if metrics.IsSetNumberOfConnections() {
		loadMetrics.Put(hostID, load.EmptyTag, load.NumConns, metrics.GetNumberOfConnections(), timestamp)
	}
	if metrics.IsSetIncomingMessagesCounter() {
		loadMetrics.Put(hostID, load.EmptyTag, load.MsgsInPerSec, metrics.GetIncomingMessagesCounter(), timestamp)
	}
	if metrics.IsSetIncomingBytesCounter() {
		loadMetrics.Put(hostID, load.EmptyTag, load.BytesInPerSec, metrics.GetIncomingBytesCounter(), timestamp)
	}
	if metrics.IsSetOutgoingMessagesCounter() {
		loadMetrics.Put(hostID, load.EmptyTag, load.MsgsOutPerSec, metrics.GetOutgoingMessagesCounter(), timestamp)
	}
	if metrics.IsSetOutgoingBytesCounter() {
		loadMetrics.Put(hostID, load.EmptyTag, load.BytesOutPerSec, metrics.GetOutgoingBytesCounter(), timestamp)
	}
	if metrics.IsSetNodeStatus() && request.IsSetRole() && metrics.GetNodeStatus() == c.NodeStatus_GOING_DOWN {
		switch request.GetRole() {
		case c.Role_IN:
			context.failureDetector.ReportHostGoingDown(common.InputServiceName, hostID)
		case c.Role_STORE:
			context.failureDetector.ReportHostGoingDown(common.StoreServiceName, hostID)
		}
	}
	return nil
}
func (mcp *Mcp) ReportDestinationMetric(ctx thrift.Context, request *c.ReportDestinationMetricRequest) error {
	if !mcp.isStarted() {
		return &shared.InternalServiceError{Message: "Controller not started"}
	}
	context := mcp.context
	context.m3Client.IncCounter(metrics.ReportDestinationMetricScope, metrics.ControllerRequests)
	sw := context.m3Client.StartTimer(metrics.ReportDestinationMetricScope, metrics.ControllerLatencyTimer)
	defer sw.Stop()
	hostID := request.GetHostId()
	dstID := request.GetDestinationUUID()
	if len(hostID) < 1 || len(dstID) < 1 {
		return &shared.BadRequestError{Message: "HostID or DestinationUUID cannot be empty"}
	}
	metrics := request.GetMetrics()
	timestamp := request.GetTimestamp()
	loadMetrics := mcp.context.loadMetrics
	if metrics.IsSetNumberOfConnections() {
		loadMetrics.Put(hostID, dstID, load.NumConns, metrics.GetNumberOfConnections(), timestamp)
	}
	if metrics.IsSetNumberOfActiveExtents() {
		loadMetrics.Put(hostID, dstID, load.NumExtentsActive, metrics.GetNumberOfActiveExtents(), timestamp)
	}
	if metrics.IsSetIncomingMessagesCounter() {
		loadMetrics.Put(hostID, dstID, load.MsgsInPerSec, metrics.GetIncomingMessagesCounter(), timestamp)
	}
	if metrics.IsSetIncomingBytesCounter() {
		loadMetrics.Put(hostID, dstID, load.BytesInPerSec, metrics.GetIncomingBytesCounter(), timestamp)
	}
	return nil
}
func (mcp *Mcp) ReportDestinationExtentMetric(ctx thrift.Context, request *c.ReportDestinationExtentMetricRequest) error {
	if !mcp.isStarted() {
		return &shared.InternalServiceError{Message: "Controller not started"}
	}
	context := mcp.context
	context.m3Client.IncCounter(metrics.ReportDestinationExtentMetricScope, metrics.ControllerRequests)
	sw := context.m3Client.StartTimer(metrics.ReportDestinationExtentMetricScope, metrics.ControllerLatencyTimer)
	defer sw.Stop()
	hostID := request.GetHostId()
	extID := request.GetExtentUUID()
	if len(hostID) < 1 || len(extID) < 1 {
		return &shared.BadRequestError{Message: "HostID or ExtentUUID cannot be empty"}
	}
	metrics := request.GetMetrics()
	timestamp := request.GetTimestamp()
	loadMetrics := mcp.context.loadMetrics
	if metrics.IsSetIncomingMessagesCounter() {
		loadMetrics.Put(hostID, extID, load.MsgsInPerSec, metrics.GetIncomingMessagesCounter(), timestamp)
	}
	if metrics.IsSetIncomingBytesCounter() {
		loadMetrics.Put(hostID, extID, load.BytesInPerSec, metrics.GetIncomingBytesCounter(), timestamp)
	}
	return nil
}
func (mcp *Mcp) ReportConsumerGroupMetric(ctx thrift.Context, request *c.ReportConsumerGroupMetricRequest) error {
	if !mcp.isStarted() {
		return &shared.InternalServiceError{Message: "Controller not started"}
	}
	context := mcp.context
	context.m3Client.IncCounter(metrics.ReportConsumerGroupMetricScope, metrics.ControllerRequests)
	sw := context.m3Client.StartTimer(metrics.ReportConsumerGroupMetricScope, metrics.ControllerLatencyTimer)
	defer sw.Stop()
	hostID := request.GetHostId()
	cgID := request.GetConsumerGroupUUID()
	if len(hostID) < 1 || len(cgID) < 1 {
		return &shared.BadRequestError{Message: "HostID or ConsumerGroupUUID cannot be empty"}
	}
	metrics := request.GetMetrics()
	timestamp := request.GetTimestamp()
	loadMetrics := mcp.context.loadMetrics
	if metrics.IsSetNumberOfConnections() {
		loadMetrics.Put(hostID, cgID, load.NumConns, metrics.GetNumberOfConnections(), timestamp)
	}
	if metrics.IsSetNumberOfActiveExtents() {
		loadMetrics.Put(hostID, cgID, load.NumExtentsActive, metrics.GetNumberOfActiveExtents(), timestamp)
	}
	if metrics.IsSetOutgoingMessagesCounter() {
		loadMetrics.Put(hostID, cgID, load.MsgsOutPerSec, metrics.GetOutgoingMessagesCounter(), timestamp)
	}
	if metrics.IsSetOutgoingBytesCounter() {
		loadMetrics.Put(hostID, cgID, load.BytesOutPerSec, metrics.GetOutgoingBytesCounter(), timestamp)
	}
	if metrics.IsSetSmartRetryOnCounter() {
		loadMetrics.Put(hostID, cgID, load.SmartRetryOn, metrics.GetSmartRetryOnCounter(), timestamp)
	}
	return nil
}
func (mcp *Mcp) ReportConsumerGroupExtentMetric(ctx thrift.Context, request *c.ReportConsumerGroupExtentMetricRequest) error {
	if !mcp.isStarted() {
		return &shared.InternalServiceError{Message: "Controller not started"}
	}
	context := mcp.context
	context.m3Client.IncCounter(metrics.ReportConsumerGroupExtentMetricScope, metrics.ControllerRequests)
	sw := context.m3Client.StartTimer(metrics.ReportConsumerGroupExtentMetricScope, metrics.ControllerLatencyTimer)
	defer sw.Stop()
	hostID := request.GetHostId()
	extID := request.GetExtentUUID()
	if len(hostID) < 1 || len(extID) < 1 {
		return &shared.BadRequestError{Message: "HostID or ExtentUUID cannot be empty"}
	}
	metrics := request.GetMetrics()
	timestamp := request.GetTimestamp()
	loadMetrics := mcp.context.loadMetrics
	if metrics.IsSetOutgoingMessagesCounter() {
		loadMetrics.Put(hostID, extID, load.MsgsOutPerSec, metrics.GetOutgoingMessagesCounter(), timestamp)
	}
	if metrics.IsSetOutgoingBytesCounter() {
		loadMetrics.Put(hostID, extID, load.BytesOutPerSec, metrics.GetOutgoingBytesCounter(), timestamp)
	}
	return nil
}
func (mcp *Mcp) ReportStoreExtentMetric(ctx thrift.Context, request *c.ReportStoreExtentMetricRequest) error {
	if !mcp.isStarted() {
		return &shared.InternalServiceError{Message: "Controller not started"}
	}
	context := mcp.context
	context.m3Client.IncCounter(metrics.ReportStoreExtentMetricScope, metrics.ControllerRequests)
	sw := context.m3Client.StartTimer(metrics.ReportStoreExtentMetricScope, metrics.ControllerLatencyTimer)
	defer sw.Stop()
	hostID := request.GetStoreId()
	extID := request.GetExtentUUID()
	if len(hostID) < 1 || len(extID) < 1 {
		return &shared.BadRequestError{Message: "HostID or ExtentUUID cannot be empty"}
	}
	metrics := request.GetMetrics()
	timestamp := request.GetTimestamp()
	loadMetrics := mcp.context.loadMetrics
	if metrics.IsSetNumberOfConnections() {
		loadMetrics.Put(hostID, extID, load.NumConns, metrics.GetNumberOfConnections(), timestamp)
	}
	if metrics.IsSetIncomingMessagesCounter() {
		loadMetrics.Put(hostID, extID, load.MsgsInPerSec, metrics.GetIncomingMessagesCounter(), timestamp)
	}
	if metrics.IsSetIncomingBytesCounter() {
		loadMetrics.Put(hostID, extID, load.BytesInPerSec, metrics.GetIncomingBytesCounter(), timestamp)
	}
	if metrics.IsSetOutgoingMessagesCounter() {
		loadMetrics.Put(hostID, extID, load.MsgsOutPerSec, metrics.GetOutgoingMessagesCounter(), timestamp)
	}
	if metrics.IsSetOutgoingBytesCounter() {
		loadMetrics.Put(hostID, extID, load.BytesOutPerSec, metrics.GetOutgoingBytesCounter(), timestamp)
	}
	if metrics.IsSetExtentStatus() {
		status := shared.ExtentStatus(metrics.GetExtentStatus())
		context.extentMonitor.RecvStoreExtentHeartbeat(hostID, extID, status)
	}
	return nil
}
func (mcp *Mcp) GetCapacities(ctx thrift.Context, getCapacitiesRequest *c.GetCapacitiesRequest) (*c.GetCapacitiesResult_, error) {
	return nil, fmt.Errorf("Not implemented")
}
func (mcp *Mcp) UpsertInputHostCapacities(ctx thrift.Context, upsertCapacitiesRequest *c.UpsertInputHostCapacitiesRequest) error {
	return fmt.Errorf("Not implemented")
}
func (mcp *Mcp) UpsertOutputHostCapacities(ctx thrift.Context, upsertCapacitiesRequest *c.UpsertOutputHostCapacitiesRequest) error {
	return fmt.Errorf("Not implemented")
}
func (mcp *Mcp) UpsertStoreCapacities(ctx thrift.Context, upsertCapacitiesRequest *c.UpsertStoreCapacitiesRequest) error {
	return fmt.Errorf("Not implemented")
}
func (mcp *Mcp) RemoveCapacities(ctx thrift.Context, removeCapacitiesRequest *c.RemoveCapacitiesRequest) error {
	return fmt.Errorf("Not implemented")
}
func (mcp *Mcp) CreateDestination(ctx thrift.Context, createRequest *shared.CreateDestinationRequest) (*shared.DestinationDescription, error) {
	if !mcp.isStarted() {
		return nil, &shared.InternalServiceError{Message: "Controller not started"}
	}
	context := mcp.context
	context.m3Client.IncCounter(metrics.ControllerCreateDestinationScope, metrics.ControllerRequests)
	sw := context.m3Client.StartTimer(metrics.ControllerCreateDestinationScope, metrics.ControllerLatencyTimer)
	defer sw.Stop()
	lclLg := context.log.WithField(common.TagDstPth, common.FmtDstPth(createRequest.GetPath()))
	destDesc, err := mcp.mClient.CreateDestination(ctx, createRequest)
	if err != nil {
		lclLg.WithField(common.TagErr, err).Error("CreateDestination: local CreateDestination failed")
		context.m3Client.IncCounter(metrics.ControllerCreateDestinationScope, metrics.ControllerFailures)
		return nil, err
	}
	if createRequest.IsSetIsMultiZone() && createRequest.GetIsMultiZone() {
		destUUID := destDesc.GetDestinationUUID()
		createDestUUIDRequest := shared.NewCreateDestinationUUIDRequest()
		createDestUUIDRequest.Request = createRequest
		createDestUUIDRequest.DestinationUUID = common.StringPtr(destUUID)
		localReplicator, replicatorErr := mcp.GetClientFactory().GetReplicatorClient()
		lclLg = lclLg.WithField(common.TagDst, common.FmtDst(destUUID))
		if replicatorErr != nil {
			lclLg.WithField(common.TagErr, replicatorErr).Error("CreateDestination: GetReplicatorClient failed")
			context.m3Client.IncCounter(metrics.ControllerCreateDestinationScope, metrics.ControllerErrCallReplicatorCounter)
			return destDesc, nil
		}
		replicatorErr = localReplicator.CreateRemoteDestinationUUID(ctx, createDestUUIDRequest)
		if replicatorErr != nil {
			lclLg.WithField(common.TagErr, replicatorErr).Error("CreateDestination: CreateRemoteDestinationUUID failed")
			context.m3Client.IncCounter(metrics.ControllerCreateDestinationScope, metrics.ControllerErrCallReplicatorCounter)
			return destDesc, nil
		}
	}
	return destDesc, nil
}
func (mcp *Mcp) UpdateDestination(ctx thrift.Context, updateRequest *shared.UpdateDestinationRequest) (*shared.DestinationDescription, error) {
	if !mcp.isStarted() {
		return nil, &shared.InternalServiceError{Message: "Controller not started"}
	}
	context := mcp.context
	context.m3Client.IncCounter(metrics.ControllerUpdateDestinationScope, metrics.ControllerRequests)
	sw := context.m3Client.StartTimer(metrics.ControllerUpdateDestinationScope, metrics.ControllerLatencyTimer)
	defer sw.Stop()
	lclLg := context.log.WithField(common.TagDst, common.FmtDst(updateRequest.GetDestinationUUID()))
	if updateRequest.IsSetZoneConfigs() {
		valid, err := mcp.ValidateDestZoneConfig(ctx, lclLg, updateRequest)
		if err != nil {
			context.m3Client.IncCounter(metrics.ControllerUpdateDestinationScope, metrics.ControllerFailures)
			lclLg.WithField(common.TagErr, err).Error("UpdateDestination: ValidateDestZoneConfig returned error")
			return nil, err
		}
		if !valid {
			context.m3Client.IncCounter(metrics.ControllerUpdateDestinationScope, metrics.ControllerFailures)
			lclLg.Error("UpdateDestination: zone config validation failed")
			return nil, &shared.BadRequestError{Message: "zone config validation failed"}
		}
	}
	destDesc, err := mcp.mClient.UpdateDestination(ctx, updateRequest)
	if err != nil {
		lclLg.WithField(common.TagErr, err).Error("UpdateDestination: local UpdateDestination failed")
		context.m3Client.IncCounter(metrics.ControllerUpdateDestinationScope, metrics.ControllerFailures)
		return nil, err
	}
	if destDesc.GetIsMultiZone() {
		localReplicator, replicatorErr := mcp.GetClientFactory().GetReplicatorClient()
		lclLg = lclLg.WithField(common.TagDst, common.FmtDst(destDesc.GetDestinationUUID()))
		if replicatorErr != nil {
			lclLg.WithField(common.TagErr, replicatorErr).Error("UpdateDestination: GetReplicatorClient failed")
			context.m3Client.IncCounter(metrics.ControllerUpdateDestinationScope, metrics.ControllerErrCallReplicatorCounter)
			return destDesc, nil
		}
		replicatorErr = localReplicator.UpdateRemoteDestination(ctx, updateRequest)
		if replicatorErr != nil {
			lclLg.WithField(common.TagErr, replicatorErr).Error("UpdateDestination: UpdateRemoteDestination failed")
			context.m3Client.IncCounter(metrics.ControllerUpdateDestinationScope, metrics.ControllerErrCallReplicatorCounter)
			return destDesc, nil
		}
	}
	return destDesc, nil
}
func (mcp *Mcp) DeleteDestination(ctx thrift.Context, deleteRequest *shared.DeleteDestinationRequest) error {
	if !mcp.isStarted() {
		return &shared.InternalServiceError{Message: "Controller not started"}
	}
	context := mcp.context
	context.m3Client.IncCounter(metrics.ControllerDeleteDestinationScope, metrics.ControllerRequests)
	sw := context.m3Client.StartTimer(metrics.ControllerDeleteDestinationScope, metrics.ControllerLatencyTimer)
	defer sw.Stop()
	lclLg := context.log.WithField(common.TagDstPth, common.FmtDstPth(deleteRequest.GetPath()))
	readDestinationRequest := &shared.ReadDestinationRequest{
		Path: common.StringPtr(deleteRequest.GetPath()),
	}
	destDesc, err := mcp.mClient.ReadDestination(ctx, readDestinationRequest)
	if err != nil {
		lclLg.WithField(common.TagErr, err).Error("DeleteDestination: ReadDestination failed")
		context.m3Client.IncCounter(metrics.ControllerDeleteDestinationScope, metrics.ControllerFailures)
		return err
	}
	lclLg = lclLg.WithField(common.TagDst, common.FmtDst(destDesc.GetDestinationUUID()))
	err = mcp.mClient.DeleteDestination(ctx, deleteRequest)
	if err != nil {
		lclLg.WithField(common.TagErr, err).Error("DeleteDestination: local DeleteDestination failed")
		context.m3Client.IncCounter(metrics.ControllerDeleteDestinationScope, metrics.ControllerFailures)
		return err
	}
	if destDesc.GetIsMultiZone() {
		localReplicator, replicatorErr := mcp.GetClientFactory().GetReplicatorClient()
		if replicatorErr != nil {
			lclLg.WithField(common.TagErr, replicatorErr).Error("DeleteDestination: GetReplicatorClient failed")
			context.m3Client.IncCounter(metrics.ControllerDeleteDestinationScope, metrics.ControllerErrCallReplicatorCounter)
			return nil
		}
		replicatorErr = localReplicator.DeleteRemoteDestination(ctx, deleteRequest)
		if replicatorErr != nil {
			lclLg.WithField(common.TagErr, replicatorErr).Error("DeleteDestination: DeleteRemoteDestination failed")
			context.m3Client.IncCounter(metrics.ControllerDeleteDestinationScope, metrics.ControllerErrCallReplicatorCounter)
			return nil
		}
	}
	return nil
}