func (h *InputHost) GetTestShortExtentsByPath() (override string) {
	return h.testShortExtentsByPath
}
GetTestShortExtentsByPath gets path override that enables testing short extents
package inputhost
import (
	"fmt"
	"net/http"
	"os"
	"sync"
	"sync/atomic"
	"time"
	"github.com/uber-common/bark"
	"github.com/uber/tchannel-go/thrift"
	ccommon "github.com/uber/cherami-client-go/common"
	"github.com/uber/cherami-server/common"
	dconfig "github.com/uber/cherami-server/common/dconfigclient"
	mm "github.com/uber/cherami-server/common/metadata"
	"github.com/uber/cherami-server/common/metrics"
	"github.com/uber/cherami-server/services/inputhost/load"
	"github.com/uber/cherami-server/stream"
	"github.com/uber/cherami-thrift/.generated/go/admin"
	"github.com/uber/cherami-thrift/.generated/go/cherami"
	"github.com/uber/cherami-thrift/.generated/go/controller"
	"github.com/uber/cherami-thrift/.generated/go/metadata"
	"github.com/uber/cherami-thrift/.generated/go/shared"
)
const (
	defaultIdleTimeout = 10 * time.Minute
	defaultBufferSize = 1000
	hostLoadReportingInterval = 2 * time.Second
	minimumAllowedMessageDelaySeconds = 60
	minimumAllowedMessageDelaySecondsTest = 3
	extentRolloverSeqnumMin, extentRolloverSeqnumMax = 10000000, 20000000
	defaultDrainTimeout = 1 * time.Minute
	connWGTimeout = 5 * time.Second
	drainAllUUID = "D3A9C5DC-AE62-4465-9898-7FE71BD1FCA"
)
var (
	batchMsgAckTimeout = 1 * time.Minute  
)
type (
	InputHost struct {
		logger                 bark.Logger
		loadShutdownRef        int32
		pathMutex              sync.RWMutex
		pathCache              map[string]*inPathCache
		pathCacheByDestPath    map[string]string
		shutdownWG             sync.WaitGroup
		shutdown               chan struct{}
		cacheTimeout           time.Duration
		mClient                metadata.TChanMetadataService
		hostIDHeartbeater      common.HostIDHeartbeater
		loadReporter           common.LoadReporterDaemon
		m3Client               metrics.Client
		dConfigClient          dconfig.Client
		tokenBucketValue       atomic.Value  
		hostConnLimit          int32
		hostConnLimitPerSecond int32
		maxConnLimit           int32
		extMsgsLimitPerSecond  int32
		connMsgsLimitPerSecond int32
		hostMetrics            *load.HostMetrics
		lastLoadReportedTime   int64         
		nodeStatus             atomic.Value  
		testShortExtentsByPath string        
		common.SCommon
	}
	InOptions struct {
		CacheIdleTimeout time.Duration
	}
	extentInfo struct {
		uuid     string
		replicas []string
	}
	inReconfigInfo struct {
		req        *admin.DestinationUpdatedNotification
		updateUUID string
	}
)
var _ cherami.TChanBIn = &InputHost{}
var _ admin.TChanInputHostAdmin = &InputHost{}
var _ common.WSService = &InputHost{}
var ErrHostShutdown = &cherami.InternalServiceError{Message: "InputHost already shutdown"}
var ErrThrottled = &cherami.InternalServiceError{Message: "InputHost throttling publisher cconnection"}
var ErrDstNotLoaded = &cherami.InternalServiceError{Message: "Destination no longer served by this input host"}
var ErrDrainTimedout = &cherami.InternalServiceError{Message: "Draining of Extents timedout"}
func (h *InputHost) isDestinationWritable(destDesc *shared.DestinationDescription) bool {
	status := destDesc.GetStatus()
	if status != shared.DestinationStatus_ENABLED && status != shared.DestinationStatus_SENDONLY {
		return false
	}
	return true
}
func (h *InputHost) listExtentsByStatus(ctx thrift.Context, destUUID string, status *shared.ExtentStatus) (*metadata.ListInputHostExtentsStatsResult_, error) {
	listReq := metadata.ListInputHostExtentsStatsRequest{
		InputHostUUID:   common.StringPtr(h.GetHostUUID()),
		DestinationUUID: common.StringPtr(destUUID),
		Status:          status,
	}
	listResp, err := h.mClient.ListInputHostExtentsStats(ctx, &listReq)
	if err != nil {
		return nil, fmt.Errorf("ListInputHostExtentStats() returned error dst=%v, err=%v",
			destUUID, err)
	}
	if listResp == nil || len(listResp.GetExtentStatsList()) == 0 {
		return nil, fmt.Errorf("getExtentsInfoForDestination - Found zero extents for destination, dst=%v",
			destUUID)
	}
	return listResp, nil
}
func (h *InputHost) createExtentInfo(extentUUID string, storeIDs []string) *extentInfo {
	if extentUUID == "" || storeIDs == nil || len(storeIDs) == 0 {
		return nil
	}
	var rpm = h.GetRingpopMonitor()
	var info = &extentInfo{
		uuid:     extentUUID,
		replicas: make([]string, 0, 3),
	}
	for _, storeID := range storeIDs {
		addr, err := rpm.ResolveUUID(common.StoreServiceName, storeID)
		if err != nil {
			h.logger.WithField(`storeID`, storeID).
				WithField(common.TagErr, err).
				Error("Cannot resolve UUID for store")
			return nil
		}
		info.replicas = append(info.replicas, addr)
	}
	return info
}
func (h *InputHost) getExtentsInfoForDestination(ctx thrift.Context, destUUID string) ([]*extentInfo, error) {
	listResp, err := h.listExtentsByStatus(ctx, destUUID, common.MetadataExtentStatusPtr(shared.ExtentStatus_OPEN))
	if err != nil {
		return nil, err
	}
	var extents = make([]*extentInfo, 0, 2)
	for _, item := range listResp.ExtentStatsList {
		extent := item.Extent
		info := h.createExtentInfo(extent.GetExtentUUID(), extent.StoreUUIDs)
		if info == nil {
			continue
		}
		extents = append(extents, info)
	}
	if len(extents) < 1 {
		h.logger.WithField(common.TagDst, common.FmtDst(destUUID)).
			WithField("nExtents", len(listResp.GetExtentStatsList())).
			Error("Can't find extents for destination, no healthy extent found")
		return nil, fmt.Errorf("No healthy extent found")
	}
	return extents, nil
}
func (h *InputHost) getExtentsAndLoadPathCache(ctx thrift.Context, destPath, destUUID string, destType shared.DestinationType) (*inPathCache, error) {
	extents, err := h.getExtentsInfoForDestination(ctx, destUUID)
	if err != nil {
		return nil, err
	}
	pathCache := h.loadPath(extents, destPath, destUUID, destType, h.m3Client)
	return pathCache, nil
}
func (h *InputHost) checkDestination(ctx thrift.Context, path string) (string, shared.DestinationType, metrics.ErrorClass, error) {
	mGetRequest := shared.ReadDestinationRequest{Path: common.StringPtr(path)}
	destDesc, err := h.mClient.ReadDestination(ctx, &mGetRequest)
	if err != nil {
		errC, newErr := common.ConvertDownstreamErrors(h.logger, err)
		return "", shared.DestinationType_UNKNOWN, errC, newErr
	}
	if destDesc == nil {
		errMsg := fmt.Sprintf("unable to get destination description from metadata for dst=%v", path)
		errC, newErr := common.ConvertDownstreamErrors(h.logger, &cherami.BadRequestError{Message: errMsg})
		return "", shared.DestinationType_UNKNOWN, errC, newErr
	}
	if !h.isDestinationWritable(destDesc) {
		errMsg := fmt.Sprintf("Destination is not writable, dst=%v, status=%v", path, destDesc.GetStatus())
		errC, newErr := common.ConvertDownstreamErrors(h.logger, &cherami.BadRequestError{Message: errMsg})
		return "", shared.DestinationType_UNKNOWN, errC, newErr
	}
	return destDesc.GetDestinationUUID(), destDesc.GetType(), metrics.NoError, nil
}
func (h *InputHost) OpenPublisherStreamHandler(w http.ResponseWriter, r *http.Request) {
	path := r.Header.Get("path")
	if len(path) == 0 {
		err := &cherami.BadRequestError{}
		h.logger.Warn("please set the path as part of the header. BadRequestError")
		h.m3Client.IncCounter(metrics.OpenPublisherStreamScope, metrics.InputhostUserFailures)
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	wsStream, err := h.GetWSConnector().AcceptPublisherStream(w, r)
	if err != nil {
		h.logger.WithField(common.TagDstPth, common.FmtDstPth(path)).
			WithField(common.TagErr, err).Error("unable to upgrade websocket connection")
		h.m3Client.IncCounter(metrics.OpenPublisherStreamScope, metrics.InputhostInternalFailures)
		return
	}
	ctx, cancel := thrift.NewContext(common.MaxDuration)
	defer cancel()
	ctx = thrift.WithHeaders(ctx, map[string]string{
		"path": path,
	})
	if err = h.OpenPublisherStream(ctx, wsStream); err != nil {
		h.logger.WithField(common.TagDstPth, common.FmtDstPth(path)).
			WithField(common.TagErr, err).Error("unable to open publish stream")
		return
	}
}
func (h *InputHost) OpenPublisherStream(ctx thrift.Context, call stream.BInOpenPublisherStreamInCall) error {
	h.m3Client.IncCounter(metrics.OpenPublisherStreamScope, metrics.InputhostRequests)
	path, ok := (ctx.Headers()["path"])
	if !ok {
		err := &cherami.BadRequestError{}
		h.logger.Warn("please set the path as part of the header. BadRequestError")
		call.Done()
		h.m3Client.IncCounter(metrics.OpenPublisherStreamScope, metrics.InputhostUserFailures)
		return err
	}
	h.logger.WithField(common.TagDstPth, common.FmtDstPth(path)).Debug("inputhost: OpenPublisherStream called with path")
	if h.IsLimitsEnabled() {
		if ok, _ = h.GetTokenBucketValue().TryConsume(1); !ok {
			h.logger.WithField(common.TagHostConnLimit,
				common.FmtHostConnLimit(h.GetHostConnLimitPerSecond())).
				Warn("Too many open publisher streams on this inputhost within a second")
			call.Done()
			h.m3Client.IncCounter(metrics.OpenPublisherStreamScope, metrics.InputhostUserFailures)
			return ErrThrottled
		}
	}
	if h.IsLimitsEnabled() && (h.GetNumConnections() > h.GetHostConnLimitOverall()) {
		h.logger.WithField(common.TagHostConnLimit,
			common.FmtHostConnLimit(h.GetNumConnections())).
			Warn("Too many open connections on this inputhost. Rejecting this open")
		call.Done()
		h.m3Client.IncCounter(metrics.OpenPublisherStreamScope, metrics.InputhostUserFailures)
		return ErrThrottled
	}
	if atomic.AddInt32(&h.loadShutdownRef, 1) <= 0 {
		atomic.AddInt32(&h.loadShutdownRef, -1)
		_, newErr := common.ConvertDownstreamErrors(h.logger, ErrHostShutdown)
		call.Done()
		return newErr
	}
	pathCache, ok := h.getPathCacheByDestPath(path)
	if !ok || pathCache == nil {
		destinationUUID, destType, errC, err := h.checkDestination(ctx, path)
		if err != nil {
			h.logger.WithField(common.TagDstPth, common.FmtDstPth(path)).
				WithField(common.TagErr, err).Error("failed on destination check")
			atomic.AddInt32(&h.loadShutdownRef, -1)
			call.Done()
			h.incFailureCounter(metrics.OpenPublisherStreamScope, errC)
			return err
		}
		pathCache, err = h.getExtentsAndLoadPathCache(ctx, path, destinationUUID, destType)
		if err != nil || pathCache == nil {
			h.logger.WithField(common.TagDstPth, common.FmtDstPth(path)).
				WithField(common.TagErr, err).Error("unable to load path cache for destination")
			atomic.AddInt32(&h.loadShutdownRef, -1)
			call.Done()
			h.m3Client.IncCounter(metrics.OpenPublisherStreamScope, metrics.InputhostInternalFailures)
			return &ReplicaNotExistsError{}
		}
	}
	doneCh := make(chan bool, 5)
	pathCache.Lock()
	errCleanup := func() {
		pathCache.Unlock()
		atomic.AddInt32(&h.loadShutdownRef, -1)
		call.Done()
	}
	if !pathCache.isActive() {
		errCleanup()
		h.m3Client.IncCounter(metrics.OpenPublisherStreamScope, metrics.InputhostInternalFailures)
		return ErrDstNotLoaded
	}
	hostMaxConnPerDestination := h.GetMaxConnPerDest()
	if h.IsLimitsEnabled() && pathCache.dstMetrics.Get(load.DstMetricNumOpenConns) > int64(hostMaxConnPerDestination) {
		pathCache.logger.WithField(common.TagHostConnLimit,
			common.FmtHostConnLimit(hostMaxConnPerDestination)).
			Warn("Too many open connections on this path. Rejecting this open")
		errCleanup()
		h.m3Client.IncCounter(metrics.OpenPublisherStreamScope, metrics.InputhostUserFailures)
		return ErrThrottled
	}
	conn := newPubConnection(path, call, pathCache, h.m3Client, h.IsLimitsEnabled(), h.cacheTimeout, doneCh)
	pathCache.connections[pathCache.currID] = conn
	conn.open()
	pathCache.currID++
	pathCache.dstMetrics.Increment(load.DstMetricNumOpenConns)
	pathCache.Unlock()
	h.hostMetrics.Increment(load.HostMetricNumOpenConns)
	atomic.AddInt32(&h.loadShutdownRef, -1)
	<-conn.doneCh
	pathCache.dstMetrics.Decrement(load.DstMetricNumOpenConns)
	h.hostMetrics.Decrement(load.HostMetricNumOpenConns)
	return nil
}
func (h *InputHost) PutMessageBatch(ctx thrift.Context, request *cherami.PutMessageBatchRequest) (*cherami.PutMessageBatchResult_, error) {
	sw := h.m3Client.StartTimer(metrics.PutMessageBatchInputHostScope, metrics.InputhostLatencyTimer)
	defer sw.Stop()
	h.m3Client.IncCounter(metrics.PutMessageBatchInputHostScope, metrics.InputhostRequests)
	path := request.GetDestinationPath()
	messages := request.GetMessages()
	lclLg := h.logger.WithField(common.TagDstPth, common.FmtDstPth(path))
	var msgLength int64
	if atomic.AddInt32(&h.loadShutdownRef, 1) <= 0 {
		atomic.AddInt32(&h.loadShutdownRef, -1)
		_, newErr := common.ConvertDownstreamErrors(h.logger, ErrHostShutdown)
		return nil, newErr
	}
	pathCache, ok := h.getPathCacheByDestPath(path)
	if !ok || pathCache == nil {
		destinationUUID, destType, errC, err := h.checkDestination(ctx, path)
		if err != nil {
			lclLg.WithField(common.TagErr, err).Error("failed on destination check")
			h.incFailureCounter(metrics.PutMessageBatchInputHostScope, errC)
			return nil, err
		}
		pathCache, err = h.getExtentsAndLoadPathCache(ctx, path, destinationUUID, destType)
		if err != nil || pathCache == nil {
			lclLg.WithField(common.TagErr, err).Error("unable to load path cache for destination")
			atomic.AddInt32(&h.loadShutdownRef, -1)
			h.m3Client.IncCounter(metrics.PutMessageBatchInputHostScope, metrics.InputhostInternalFailures)
			return nil, &ReplicaNotExistsError{}
		}
	}
	result := cherami.NewPutMessageBatchResult_()
	ackChannel := make(chan *cherami.PutMessageAck, defaultBufferSize)
	inflightRequestCnt := 0
	inflightMsgMap := make(map[string]struct{})
	for _, msg := range messages {
		msgLength = int64(len(msg.Data))
		pathCache.dstMetrics.Add(load.DstMetricBytesIn, msgLength)
		pathCache.hostMetrics.Add(load.HostMetricBytesIn, msgLength)
		inMsg := &inPutMessage{
			putMsg:         msg,
			putMsgAckCh:    ackChannel,
			putMsgRecvTime: time.Now(),
		}
		select {
		case pathCache.putMsgCh <- inMsg:
			inflightRequestCnt++
			inflightMsgMap[msg.GetID()] = struct{}{}
		default:
			result.FailedMessages = append(result.FailedMessages, &cherami.PutMessageAck{
				ID:      common.StringPtr(msg.GetID()),
				Status:  common.CheramiStatusPtr(cherami.Status_THROTTLED),
				Message: common.StringPtr("throttling; inputhost is busy"),
			})
		}
	}
	internalErrs, userErrs := int64(0), int64(0)
	var respStatus cherami.Status
	var respMsg string
	ackReceived := func(ack *cherami.PutMessageAck) {
		if ack.GetStatus() != cherami.Status_OK {
			if ack.GetStatus() != cherami.Status_THROTTLED {
				internalErrs++
			} else {
				userErrs++
			}
			result.FailedMessages = append(result.FailedMessages, ack)
		} else {
			result.SuccessMessages = append(result.SuccessMessages, ack)
		}
		delete(inflightMsgMap, ack.GetID())
	}
	msgTimer := common.NewTimer(batchMsgAckTimeout)
	defer msgTimer.Stop()
ACKDRAIN:
	for i := 0; i < inflightRequestCnt; i++ {
		select {
		case ack := <-ackChannel:
			ackReceived(ack)
		default:
			select {
			case ack := <-ackChannel:
				ackReceived(ack)
			case <-pathCache.closeCh:
				respStatus = cherami.Status_FAILED
				respMsg = "pathCache unloaded"
				break ACKDRAIN
			case <-msgTimer.C:
				respStatus = cherami.Status_TIMEDOUT
				respMsg = "message timedout"
				break ACKDRAIN
			}
		}
	}
	if len(inflightMsgMap) > 0 {
		pathCache.logger.WithFields(bark.Fields{
			`numFailedMessages`: len(inflightMsgMap),
			`respMsg`:           respMsg,
		}).Info("failing putMessageBatch")
		for id := range inflightMsgMap {
			result.FailedMessages = append(result.FailedMessages, &cherami.PutMessageAck{
				ID:      common.StringPtr(id),
				Status:  common.CheramiStatusPtr(respStatus),
				Message: common.StringPtr(respMsg),
			})
			internalErrs++
		}
	}
	pathCache.updateLastDisconnectTime()
	atomic.AddInt32(&h.loadShutdownRef, -1)
	h.m3Client.AddCounter(metrics.PutMessageBatchInputHostScope, metrics.InputhostMessageReceived, int64(len(result.SuccessMessages)))
	pathCache.destM3Client.AddCounter(metrics.PutMessageBatchInputHostDestScope, metrics.InputhostDestMessageReceived, int64(len(result.SuccessMessages)))
	h.m3Client.AddCounter(metrics.PutMessageBatchInputHostScope, metrics.InputhostMessageUserFailures, userErrs)
	pathCache.destM3Client.AddCounter(metrics.PutMessageBatchInputHostDestScope, metrics.InputhostDestMessageUserFailures, userErrs)
	h.m3Client.AddCounter(metrics.PutMessageBatchInputHostScope, metrics.InputhostMessageInternalFailures, internalErrs)
	pathCache.destM3Client.AddCounter(metrics.PutMessageBatchInputHostDestScope, metrics.InputhostDestMessageInternalFailures, internalErrs)
	pathCache.dstMetrics.Add(load.DstMetricOverallNumMsgs, int64(len(messages)))
	pathCache.dstMetrics.Add(load.DstMetricNumFailed, (internalErrs + userErrs))
	pathCache.dstMetrics.Add(load.DstMetricNumAcks, int64(len(result.SuccessMessages)))
	return result, nil
}
func (h *InputHost) DestinationsUpdated(ctx thrift.Context, request *admin.DestinationsUpdatedRequest) (err error) {
	defer atomic.AddInt32(&h.loadShutdownRef, -1)
	sw := h.m3Client.StartTimer(metrics.DestinationsUpdatedScope, metrics.InputhostLatencyTimer)
	defer sw.Stop()
	h.m3Client.IncCounter(metrics.DestinationsUpdatedScope, metrics.InputhostRequests)
	if atomic.AddInt32(&h.loadShutdownRef, 1) <= 0 {
		h.logger.WithField(common.TagReconfigureID, common.FmtReconfigureID(request.GetUpdateUUID())).Error("inputhost: DestinationsUpdated: dropping reconfiguration due to shutdown")
		h.m3Client.IncCounter(metrics.DestinationsUpdatedScope, metrics.InputhostFailures)
		return ErrHostShutdown
	}
	var intErr error
	updateUUID := request.GetUpdateUUID()
	h.logger.WithField(common.TagReconfigureID, common.FmtReconfigureID(updateUUID)).
		Debug("inputhost: DestinationsUpdated: processing reconfiguration")
	for _, req := range request.Updates {
		destUUID := req.GetDestinationUUID()
		pathCache, ok := h.getPathCacheByDestUUID(destUUID)
		if ok {
			if pathCache.isActiveNoLock() {
				pathCache.reconfigureCh <- inReconfigInfo{req: req, updateUUID: updateUUID}
			} else {
				intErr = errPathCacheUnloading
			}
		} else {
			intErr = &cherami.EntityNotExistsError{}
		}
		if intErr != nil {
			err = intErr
			h.m3Client.IncCounter(metrics.DestinationsUpdatedScope, metrics.InputhostFailures)
			h.logger.WithFields(bark.Fields{
				common.TagDst:           common.FmtDst(destUUID),
				common.TagReconfigureID: common.FmtReconfigureID(updateUUID),
				common.TagErr:           intErr,
			}).Error("inputhost: DestinationsUpdated: dropping reconfiguration")
		}
	}
	h.logger.WithField(common.TagReconfigureID, common.FmtReconfigureID(updateUUID)).
		Debug("inputhost: DestinationsUpdated: finished reconfiguration")
	return
}
func (h *InputHost) UnloadDestinations(ctx thrift.Context, request *admin.UnloadDestinationsRequest) (err error) {
	defer atomic.AddInt32(&h.loadShutdownRef, -1)
	sw := h.m3Client.StartTimer(metrics.UnloadDestinationsScope, metrics.InputhostLatencyTimer)
	defer sw.Stop()
	h.m3Client.IncCounter(metrics.UnloadDestinationsScope, metrics.InputhostRequests)
	if atomic.AddInt32(&h.loadShutdownRef, 1) <= 0 {
		h.logger.Error("not unloading the path cache; inputHost already shutdown")
		h.m3Client.IncCounter(metrics.UnloadDestinationsScope, metrics.InputhostFailures)
		return ErrHostShutdown
	}
	for _, destUUID := range request.DestUUIDs {
		h.pathMutex.RLock()
		pathCache, ok := h.pathCache[destUUID]
		if ok {
			pathCache.Lock()
			pathCache.prepareForUnload()
			go pathCache.unload()
			pathCache.Unlock()
		} else {
			h.logger.WithField(common.TagDst, common.FmtDst(destUUID)).
				Error("destination is not cached at all")
			err = ErrDstNotLoaded
			h.m3Client.IncCounter(metrics.UnloadDestinationsScope, metrics.InputhostFailures)
		}
		h.pathMutex.RUnlock()
	}
	return err
}
func (h *InputHost) ListLoadedDestinations(ctx thrift.Context) (result *admin.ListLoadedDestinationsResult_, err error) {
	defer atomic.AddInt32(&h.loadShutdownRef, -1)
	sw := h.m3Client.StartTimer(metrics.ListLoadedDestinationsScope, metrics.InputhostLatencyTimer)
	defer sw.Stop()
	h.m3Client.IncCounter(metrics.ListLoadedDestinationsScope, metrics.InputhostRequests)
	if atomic.AddInt32(&h.loadShutdownRef, 1) <= 0 {
		h.logger.Error("inputHost already shutdown")
		h.m3Client.IncCounter(metrics.ListLoadedDestinationsScope, metrics.InputhostFailures)
		return nil, ErrHostShutdown
	}
	result = admin.NewListLoadedDestinationsResult_()
	h.pathMutex.RLock()
	result.Dests = make([]*admin.Destinations, len(h.pathCache))
	count := 0
	for destUUID, pathCache := range h.pathCache {
		destRes := admin.NewDestinations()
		destRes.DestUUID = common.StringPtr(destUUID)
		destRes.DestPath = common.StringPtr(pathCache.destinationPath)
		result.Dests[count] = destRes
		count++
	}
	h.pathMutex.RUnlock()
	return result, err
}
func (h *InputHost) ReadDestState(ctx thrift.Context, request *admin.ReadDestinationStateRequest) (result *admin.ReadDestinationStateResult_, err error) {
	defer atomic.AddInt32(&h.loadShutdownRef, -1)
	sw := h.m3Client.StartTimer(metrics.ReadDestStateScope, metrics.InputhostLatencyTimer)
	defer sw.Stop()
	h.m3Client.IncCounter(metrics.ReadDestStateScope, metrics.InputhostRequests)
	if atomic.AddInt32(&h.loadShutdownRef, 1) <= 0 {
		h.logger.Error("inputHost already shutdown")
		h.m3Client.IncCounter(metrics.ReadDestStateScope, metrics.InputhostFailures)
		return nil, ErrHostShutdown
	}
	result = admin.NewReadDestinationStateResult_()
	result.InputHostUUID = common.StringPtr(h.GetHostUUID())
	result.DestState = make([]*admin.DestinationState, 0)
	h.pathMutex.RLock()
	for _, destUUID := range request.DestUUIDs {
		pathCache, ok := h.pathCache[destUUID]
		if ok {
			destState := pathCache.getState()
			result.DestState = append(result.DestState, destState)
		} else {
			h.logger.WithField(common.TagDst, common.FmtDst(destUUID)).
				Warn("destination is not cached at all")
			err = ErrDstNotLoaded
		}
	}
	h.pathMutex.RUnlock()
	if err != nil {
		h.m3Client.IncCounter(metrics.ReadDestStateScope, metrics.InputhostFailures)
	}
	return result, err
}
func getDrainTimeout(ctx thrift.Context) time.Duration {
	if ctx != nil {
		if deadline, ok := ctx.Deadline(); ok {
			return deadline.Sub(time.Now())
		}
	}
	return defaultDrainTimeout
}
func (h *InputHost) DrainExtent(ctx thrift.Context, request *admin.DrainExtentsRequest) (err error) {
	defer atomic.AddInt32(&h.loadShutdownRef, -1)
	sw := h.m3Client.StartTimer(metrics.DrainExtentsScope, metrics.InputhostLatencyTimer)
	defer sw.Stop()
	h.m3Client.IncCounter(metrics.DrainExtentsScope, metrics.InputhostRequests)
	if atomic.AddInt32(&h.loadShutdownRef, 1) <= 0 {
		h.logger.WithField(common.TagReconfigureID, common.FmtReconfigureID(request.GetUpdateUUID())).Error("inputhost: DrainExtent: dropping due to shutdown")
		h.m3Client.IncCounter(metrics.DrainExtentsScope, metrics.InputhostFailures)
		return ErrHostShutdown
	}
	var intErr error
	updateUUID := request.GetUpdateUUID()
	h.logger.WithField(common.TagReconfigureID, common.FmtReconfigureID(updateUUID)).
		Debug("inputhost: DrainExtent: processing drain")
	timeoutTime := getDrainTimeout(ctx)
	var drainWG sync.WaitGroup
	var extentUUID string
	for _, req := range request.Extents {
		destUUID := req.GetDestinationUUID()
		pathCache, ok := h.getPathCacheByDestUUID(destUUID)
		if ok {
			extentUUID = req.GetExtentUUID()
			if pathCache.isActiveNoLock() {
				drainWG.Add(1)
				go pathCache.drainExtent(extentUUID, updateUUID, &drainWG, timeoutTime)
			} else {
				intErr = errPathCacheUnloading
			}
		} else {
			intErr = &cherami.EntityNotExistsError{}
		}
		if intErr != nil {
			err = intErr
			h.m3Client.IncCounter(metrics.DrainExtentsScope, metrics.InputhostFailures)
			h.logger.WithFields(bark.Fields{
				common.TagDst:           common.FmtDst(destUUID),
				common.TagExt:           common.FmtDst(extentUUID),
				common.TagReconfigureID: common.FmtReconfigureID(updateUUID),
				common.TagErr:           intErr,
			}).Error("inputhost: DrainExtent: dropping reconfiguration")
		}
	}
	h.logger.WithField(common.TagReconfigureID, common.FmtReconfigureID(updateUUID)).
		Debug("inputhost: DrainExtent: finished reconfiguration")
	if ok := common.AwaitWaitGroup(&drainWG, timeoutTime); !ok {
		err = ErrDrainTimedout
		h.m3Client.IncCounter(metrics.DrainExtentsScope, metrics.InputhostFailures)
		h.logger.WithFields(bark.Fields{
			common.TagReconfigureID: common.FmtReconfigureID(updateUUID),
			common.TagErr:           err,
		}).Error("inputhost: DrainExtent: timed out")
	}
	return
}
func (h *InputHost) reportHostMetric(reporter common.LoadReporter, intervalSecs int64) int64 {
	msgsInPerSec := h.hostMetrics.GetAndReset(load.HostMetricMsgsIn) / intervalSecs
	bytesInSinceLastReport := h.hostMetrics.GetAndReset(load.HostMetricBytesIn)
	hostMetrics := controller.NodeMetrics{
		NumberOfActiveExtents:   common.Int64Ptr(h.hostMetrics.Get(load.HostMetricNumOpenExtents)),
		NumberOfConnections:     common.Int64Ptr(h.hostMetrics.Get(load.HostMetricNumOpenConns)),
		IncomingMessagesCounter: common.Int64Ptr(msgsInPerSec),
		IncomingBytesCounter:    common.Int64Ptr(bytesInSinceLastReport),
		NodeStatus:              common.NodeStatusPtr(h.GetNodeStatus()),
	}
	reporter.ReportHostMetric(hostMetrics)
	return *(hostMetrics.NumberOfConnections)
}
func (h *InputHost) Report(reporter common.LoadReporter) {
	now := time.Now().UnixNano()
	intervalSecs := (now - h.lastLoadReportedTime) / int64(time.Second)
	if intervalSecs < 1 {
		return
	}
	numConns := h.reportHostMetric(reporter, intervalSecs)
	h.lastLoadReportedTime = now
	h.m3Client.UpdateGauge(metrics.PubConnectionStreamScope, metrics.InputhostPubConnection, numConns)
}
func (h *InputHost) GetHostConnLimitPerSecond() int {
	return int(atomic.LoadInt32(&h.hostConnLimitPerSecond))
}
func (h *InputHost) SetHostConnLimitPerSecond(connLimit int32) {
	h.logger.WithField(`val`, connLimit).Info(`SetHostConnLimitPerSecond`)
	atomic.StoreInt32(&h.hostConnLimitPerSecond, connLimit)
	h.SetTokenBucketValue(int32(connLimit))
}
func (h *InputHost) GetExtMsgsLimitPerSecond() int {
	return int(atomic.LoadInt32(&h.extMsgsLimitPerSecond))
}
func (h *InputHost) SetExtMsgsLimitPerSecond(connLimit int32) {
	h.logger.WithField(`val`, connLimit).Info(`SetExtMsgsLimitPerSecond`)
	atomic.StoreInt32(&h.extMsgsLimitPerSecond, connLimit)
	h.updateExtTokenBucket(int32(connLimit))
}
func (h *InputHost) GetConnMsgsLimitPerSecond() int {
	return int(atomic.LoadInt32(&h.connMsgsLimitPerSecond))
}
func (h *InputHost) SetConnMsgsLimitPerSecond(connLimit int32) {
	h.logger.WithField(`val`, connLimit).Info(`SetConnMsgsLimitPerSecond`)
	atomic.StoreInt32(&h.connMsgsLimitPerSecond, connLimit)
	h.updateConnTokenBucket(int32(connLimit))
}
func (h *InputHost) GetTokenBucketValue() common.TokenBucket {
	return h.tokenBucketValue.Load().(common.TokenBucket)
}
func (h *InputHost) SetTokenBucketValue(connLimit int32) {
	h.logger.WithField(`val`, connLimit).Info(`SetTokenBucketValue`)
	tokenBucket := common.NewTokenBucket(int(connLimit), common.NewRealTimeSource())
	h.tokenBucketValue.Store(tokenBucket)
}
func (h *InputHost) SetHostConnLimit(connLimit int32) {
	h.logger.WithField(`val`, connLimit).Info(`SetHostConnLimit`)
	atomic.StoreInt32(&h.hostConnLimit, connLimit)
}
func (h *InputHost) GetHostConnLimitOverall() int {
	return int(atomic.LoadInt32(&h.hostConnLimit))
}
func (h *InputHost) SetMaxConnPerDest(connLimit int32) {
	h.logger.WithField(`val`, connLimit).Info(`SetMaxConnPerDest`)
	atomic.StoreInt32(&h.maxConnLimit, connLimit)
}
func (h *InputHost) GetMaxConnPerDest() int {
	return int(atomic.LoadInt32(&h.maxConnLimit))
}
func (h *InputHost) GetNumConnections() int {
	return int(h.hostMetrics.Get(load.HostMetricNumOpenConns))
}
func (h *InputHost) GetNodeStatus() controller.NodeStatus {
	return h.nodeStatus.Load().(controller.NodeStatus)
}
func (h *InputHost) SetNodeStatus(status controller.NodeStatus) {
	h.logger.WithField(`val`, status).Info(`SetNodeStatus`)
	h.nodeStatus.Store(status)
}
func (h *InputHost) SetTestShortExtentsByPath(override string) {
	h.logger.WithField(`val`, override).Info(`SetTestShortExtentsByPath`)
	h.testShortExtentsByPath = override
}