func (monitor *extentStateMonitor) fixOutOfSyncStoreExtents(dstID string, extent *metadata.DestinationExtent) {
	if common.IsRemoteZoneExtent(extent.GetOriginZone(), monitor.context.localZone) {
		return
	}
	for _, storeh := range extent.GetStoreUUIDs() {
		key := buildExtentCacheKey(storeh, extent.GetExtentUUID())
		entry, ok := monitor.storeExtents.Get(key).(*extentCacheEntry)
		if !ok {
			continue
		}
		if entry.status == shared.ExtentStatus_OPEN {
			addStoreExtentStatusOutOfSyncEvent(monitor.context, dstID, extent.GetExtentUUID(), storeh)
			monitor.ll.WithFields(bark.Fields{
				common.TagDst:  common.FmtDst(dstID),
				common.TagExt:  common.FmtExt(extent.GetExtentUUID()),
				common.TagStor: common.FmtStor(storeh),
			}).Warn("Extent status out of sync on store; expected=SEALED; found=OPEN")
		}
	}
}
fixOutOfSyncStoreExtents checks if all stores have the same view of
the extent status and if not, issues a SEAL to the out of sync
store to bring it up to speed
For a remote zone extent, store will mark a store extent as sealed
only after it gets the sealed marker from replication
So we don't need to sync the status for remote zone extent
package controllerhost
import (
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/pborman/uuid"
	"github.com/uber-common/bark"
	"github.com/uber/cherami-client-go/common/backoff"
	"github.com/uber/cherami-server/common"
	"github.com/uber/cherami-server/common/cache"
	"github.com/uber/cherami-server/common/metrics"
	"github.com/uber/cherami-thrift/.generated/go/admin"
	"github.com/uber/cherami-thrift/.generated/go/metadata"
	"github.com/uber/cherami-thrift/.generated/go/shared"
	"github.com/uber/tchannel-go/thrift"
)
type (
	extentStateMonitor struct {
		started int32
		context *Context
		ll           bark.Logger
		storeExtents cache.Cache
		rateLimiter  common.TokenBucket
		shutdownC    chan struct{}
		shutdownWG   sync.WaitGroup
		queueDepth   *queueDepthCalculator
		mi           *mIterator
		loopStats struct {
			nExtentsByStatus    [maxExtentStates]int64
			nDLQExtentsByStatus [maxExtentStates]int64
			nCGExtentsByStatus  [maxCGExtentStates]int64
		}
	}
	extentCacheEntry struct {
		status shared.ExtentStatus
	}
	consumerGroupInfo struct {
		desc        *shared.ConsumerGroupDescription
		outputHosts map[string]struct{}
	}
)
var (
	maxExtentDownEventsPerSec = 5
	IntervalBtwnScans   = time.Minute
	intervalBtwnRetries = time.Millisecond * 500
	extentCacheTTL        = time.Minute
	extentCacheInitialCap = 1024
	extentCacheMaxSize = 10000
)
const maxExtentStates = int(shared.ExtentStatus_DELETED) + 1
const maxCGExtentStates = int(shared.ConsumerGroupExtentStatus_DELETED) + 1
func newExtentStateMonitor(context *Context) *extentStateMonitor {
	monitor := new(extentStateMonitor)
	monitor.context = context
	monitor.ll = context.log.WithField(`module`, `extentMon`)
	monitor.shutdownC = make(chan struct{})
	opts := &cache.Options{}
	opts.InitialCapacity = extentCacheInitialCap
	opts.TTL = extentCacheTTL
	monitor.storeExtents = cache.New(extentCacheMaxSize, opts)
	monitor.rateLimiter = common.NewTokenBucket(maxExtentDownEventsPerSec, common.NewRealTimeSource())
	monitor.queueDepth = newQueueDepthCalculator(monitor.context)
	monitor.mi = newMIterator(context, monitor.queueDepth, newDlqMonitor(context))
	return monitor
}
func (monitor *extentStateMonitor) Start() {
	if !atomic.CompareAndSwapInt32(&monitor.started, 0, 1) {
		return
	}
	monitor.shutdownWG.Add(1)
	go monitor.run()
	monitor.ll.Info("ExtentStateMonitor started")
}
func (monitor *extentStateMonitor) Stop() {
	close(monitor.shutdownC)
	if !common.AwaitWaitGroup(&monitor.shutdownWG, time.Second) {
		monitor.ll.Error("Timed out waiting for ExtentStateMonitor to stop")
		return
	}
	monitor.ll.Error("ExtentStateMonitor stopped")
}
func (monitor *extentStateMonitor) RecvStoreExtentHeartbeat(hostID string, extID string, status shared.ExtentStatus) {
	key := buildExtentCacheKey(hostID, extID)
	monitor.storeExtents.Put(key, &extentCacheEntry{status: status})
}
func (monitor *extentStateMonitor) isPrimary() bool {
	context := monitor.context
	hi, err := context.rpm.FindHostForKey(common.ControllerServiceName, common.ControllerServiceName)
	if err != nil {
		return false
	}
	return (hi.UUID == context.hostID)
}
func (monitor *extentStateMonitor) waitUntilBootstrap() {
	var done = false
	var context = monitor.context
	var services = []string{common.InputServiceName, common.OutputServiceName, common.StoreServiceName}
	for !monitor.isShutdown() && !done {
		done = true
		monitor.sleep(intervalBtwnRetries)
		for _, svc := range services {
			hosts, err := context.rpm.GetHosts(svc)
			if err != nil || len(hosts) < 1 {
				done = false
				if err != nil {
					monitor.ll.WithField(common.TagErr, err).Error(`RingPop GetHosts failed`)
				} else if len(hosts) < 1 {
					monitor.ll.Warn(`No peers found; waiting for bootstrap...`)
				}
				break
			}
		}
	}
	monitor.ll.Info("ExtentStateMonitor bootstrap complete")
}
func (monitor *extentStateMonitor) run() {
	var sleepTime time.Duration
	monitor.waitUntilBootstrap()
shutdown:
	for !monitor.isShutdown() {
		sleepTime = IntervalBtwnScans
		if !monitor.isPrimary() {
			monitor.ll.Info("ExtentStateMonitor won't run, controller is not primary")
			monitor.sleep(30 * time.Second)
			continue shutdown
		}
		monitor.ll.Info("ExtentStateMonitor beginning to scan all extents")
		monitor.resetLoopStats()
		monitor.runIterator()
		monitor.emitLoopStats()
		monitor.ll.Info("ExtentStateMonitor done with scanning all extents")
		monitor.sleep(sleepTime)
	}
	monitor.shutdownWG.Done()
}
func (monitor *extentStateMonitor) runIterator() {
	monitor.mi.publishEvent(eIterStart, nil)
	dests := monitor.listDestinations()
	for _, dstDesc := range dests {
		if monitor.isShutdown() {
			return
		}
		if dstDesc.GetStatus() == shared.DestinationStatus_DELETED {
			continue
		}
		if common.IsDLQDestinationPath(dstDesc.GetPath()) {
			monitor.handleDestination(dstDesc)  
			continue
		}
		monitor.mi.publishEvent(eDestStart, dstDesc)
		monitor.mi.publishEvent(eExtentIterStart, nil)
		monitor.iterateDestinationExtents(dstDesc)
		monitor.mi.publishEvent(eExtentIterEnd, nil)
		consgroups, err := monitor.listConsumerGroups(dstDesc.GetDestinationUUID())
		if err != nil {
			monitor.mi.publishEvent(eDestEnd, dstDesc)
			continue
		}
		monitor.mi.publishEvent(eCnsmIterStart, nil)
		for _, cgDesc := range consgroups {
			if monitor.isShutdown() {
				break
			}
			if cgDesc.GetStatus() == shared.ConsumerGroupStatus_DELETED {
				continue
			}
			monitor.mi.publishEvent(eCnsmStart, cgDesc)
			monitor.mi.publishEvent(eCnsmExtentIterStart, nil)
			monitor.iterateConsumerGroupExtents(dstDesc, cgDesc)
			monitor.mi.publishEvent(eCnsmExtentIterEnd, nil)
			monitor.mi.publishEvent(eCnsmEnd, cgDesc)
		}
		monitor.mi.publishEvent(eCnsmIterEnd, nil)
		monitor.mi.publishEvent(eDestEnd, dstDesc)
		for _, cgDesc := range consgroups {
			if monitor.isShutdown() {
				break
			}
			if cgDesc.GetStatus() == shared.ConsumerGroupStatus_DELETED {
				continue
			}
			monitor.iterateDLQDestination(cgDesc)
		}
		monitor.handleDestination(dstDesc)  
	}
	monitor.mi.publishEvent(eIterEnd, nil)
}
func (monitor *extentStateMonitor) iterateDestinationExtents(dstDesc *shared.DestinationDescription) {
	var err error
	var context = monitor.context
	var extents []*metadata.DestinationExtent
	isDLQ := common.IsDLQDestinationPath(dstDesc.GetPath())
extentIter:
	for _, status := range []shared.ExtentStatus{
		shared.ExtentStatus_OPEN,
		shared.ExtentStatus_SEALED,
		shared.ExtentStatus_CONSUMED,
	} {
		filterBy := []shared.ExtentStatus{status}
		extents, err = context.mm.ListDestinationExtentsByStatus(dstDesc.GetDestinationUUID(), filterBy)
		if err != nil {
			monitor.ll.WithFields(bark.Fields{
				common.TagDst:  dstDesc.GetDestinationUUID(),
				common.TagErr:  err,
				`filterStatus`: status}).
				Error(`ListDestinationExtentsByStatus failed`)
			if monitor.sleep(intervalBtwnRetries) {
				break extentIter
			}
			continue extentIter
		}
		for _, extent := range extents {
			monitor.handleDestinationExtent(dstDesc, isDLQ, extent)  
			monitor.mi.publishEvent(eExtent, extent)
		}
	}
}
func (monitor *extentStateMonitor) iterateConsumerGroupExtents(dstDesc *shared.DestinationDescription, cgDesc *shared.ConsumerGroupDescription) {
	dstID := dstDesc.GetDestinationUUID()
	cgID := cgDesc.GetConsumerGroupUUID()
	for _, status := range []shared.ConsumerGroupExtentStatus{
		shared.ConsumerGroupExtentStatus_OPEN,
		shared.ConsumerGroupExtentStatus_CONSUMED,
	} {
		filterBy := []shared.ConsumerGroupExtentStatus{status}
		extents, err := monitor.context.mm.ListExtentsByConsumerGroup(dstID, cgID, filterBy)
		if err != nil {
			monitor.ll.WithFields(bark.Fields{
				common.TagErr:  err,
				common.TagDst:  dstID,
				common.TagCnsm: cgID,
				`statusFilter`: status,
			}).Error(`ListExtentsByConsumerGroup failed`)
			continue
		}
	nextExtent:
		for _, ext := range extents {
			if ext.GetStatus() == shared.ConsumerGroupExtentStatus_DELETED {
				continue nextExtent
			}
			monitor.mi.publishEvent(eCnsmExtent, ext)
			monitor.loopStats.nCGExtentsByStatus[int(ext.GetStatus())]++
		}
	}
}
func (monitor *extentStateMonitor) iterateDLQDestination(cgDesc *shared.ConsumerGroupDescription) {
	dstID := cgDesc.GetDeadLetterQueueDestinationUUID()
	if len(dstID) == 0 {
		return
	}
	dstDesc, err := monitor.context.mm.ReadDestination(dstID, "")
	if err != nil || dstDesc.GetStatus() == shared.DestinationStatus_DELETED {
		return
	}
	monitor.mi.publishEvent(eDestStart, dstDesc)
	monitor.mi.publishEvent(eExtentIterStart, nil)
	monitor.iterateDestinationExtents(dstDesc)
	monitor.mi.publishEvent(eExtentIterEnd, nil)
	monitor.mi.publishEvent(eCnsmIterStart, nil)
	monitor.mi.publishEvent(eCnsmStart, cgDesc)
	monitor.mi.publishEvent(eCnsmEnd, cgDesc)
	monitor.mi.publishEvent(eCnsmIterEnd, nil)
	monitor.mi.publishEvent(eDestEnd, dstDesc)
}
func (monitor *extentStateMonitor) handleDestination(dstDesc *shared.DestinationDescription) {
	if dstDesc.GetStatus() == shared.DestinationStatus_DELETING {
		monitor.context.resultCache.Delete(dstDesc.GetDestinationUUID())  
		consgroups, err := monitor.listConsumerGroups(dstDesc.GetDestinationUUID())
		if err == nil {
			for _, cgDesc := range consgroups {
				monitor.deleteConsumerGroup(dstDesc, cgDesc)
			}
		}
	}
}
func (monitor *extentStateMonitor) deleteConsumerGroup(dstDesc *shared.DestinationDescription, cgDesc *shared.ConsumerGroupDescription) {
	if cgDesc.GetStatus() == shared.ConsumerGroupStatus_DELETING ||
		cgDesc.GetStatus() == shared.ConsumerGroupStatus_DELETED {
		return
	}
	context := monitor.context
	dstID := dstDesc.GetDestinationUUID()
	cgID := cgDesc.GetConsumerGroupUUID()
	filterBy := []shared.ConsumerGroupExtentStatus{shared.ConsumerGroupExtentStatus_OPEN}
	extents, e := context.mm.ListExtentsByConsumerGroupLite(dstID, cgID, filterBy)
	if e != nil {
		monitor.ll.WithFields(bark.Fields{
			common.TagErr:  e,
			common.TagDst:  dstID,
			common.TagCnsm: cgID,
			`statusFilter`: filterBy[0],
		}).Error(`ListExtentsByConsumerGroupLite failed`)
		return
	}
	e = context.mm.DeleteConsumerGroup(dstDesc.GetDestinationUUID(), cgDesc.GetConsumerGroupName())
	if e != nil {
		context.m3Client.IncCounter(metrics.ExtentMonitorScope, metrics.ControllerErrMetadataUpdateCounter)
		monitor.ll.WithFields(bark.Fields{
			common.TagDst:  common.FmtDst(dstID),
			common.TagCnsm: common.FmtCnsm(cgID),
			common.TagErr:  e,
		}).Warn("Failed to update consumer group status to DELETED")
		return
	}
	monitor.ll.WithFields(bark.Fields{
		common.TagDstPth: common.FmtDstPth(dstDesc.GetPath()),
		common.TagDst:    common.FmtDst(dstDesc.GetDestinationUUID()),
		common.TagCnsPth: common.FmtCnsPth(cgDesc.GetConsumerGroupName()),
		common.TagCnsm:   common.FmtCnsm(cgDesc.GetConsumerGroupUUID()),
	}).Info("ExtentMon: ConsumerGroup DELETED")
	context.resultCache.Delete(cgID)  
	outputHosts := make(map[string]struct{}, 4)
	for _, ext := range extents {
		if ext.GetStatus() != shared.ConsumerGroupExtentStatus_OPEN {
			continue
		}
		outputHosts[ext.GetOutputHostUUID()] = struct{}{}
	}
	monitor.notifyOutputHosts(dstID, cgID, outputHosts, notifyCGDeleted, cgID)
}
func (monitor *extentStateMonitor) handleDestinationExtent(dstDesc *shared.DestinationDescription, isDLQ bool, extent *metadata.DestinationExtent) {
	monitor.loopStats.nExtentsByStatus[int(extent.GetStatus())]++
	if isDLQ {
		monitor.loopStats.nDLQExtentsByStatus[int(extent.GetStatus())]++
	}
	context := monitor.context
	switch extent.GetStatus() {
	case shared.ExtentStatus_OPEN:
		if !common.IsRemoteZoneExtent(extent.GetOriginZone(), context.localZone) &&
			(dstDesc.GetStatus() == shared.DestinationStatus_DELETING ||
				!monitor.isExtentHealthy(dstDesc, extent)) {
			if !monitor.rateLimiter.Consume(1, 2*time.Second) {
				return
			}
			addExtentDownEvent(context, 0, dstDesc.GetDestinationUUID(), extent.GetExtentUUID())
		}
	default:
		lastUpdateTime := time.Unix(0, extent.GetStatusUpdatedTimeMillis()*int64(time.Millisecond))
		if time.Since(lastUpdateTime) > 2*extentCacheTTL {
			monitor.fixOutOfSyncStoreExtents(dstDesc.GetDestinationUUID(), extent)
		}
	}
	if common.IsRemoteZoneExtent(extent.GetOriginZone(), context.localZone) {
		monitor.handleRemoteZoneDestinationExtent(dstDesc, extent)
	}
}
func (monitor *extentStateMonitor) handleRemoteZoneDestinationExtent(dstDesc *shared.DestinationDescription, extent *metadata.DestinationExtent) {
	context := monitor.context
	var failedStores []string
	for _, storeID := range extent.GetStoreUUIDs() {
		state, duration := context.failureDetector.GetHostState(common.StoreServiceName, storeID)
		switch state {
		case dfddHostStateUnknown:
			failedStores = append(failedStores, storeID)
		case dfddHostStateDown:
			if duration >= maxHostRestartDuration {
				failedStores = append(failedStores, storeID)
			}
		}
	}
	if len(failedStores) > 0 {
		stats, err := context.mm.ReadExtentStats(dstDesc.GetDestinationUUID(), extent.GetExtentUUID())
		if err != nil {
			return
		}
		for _, s := range failedStores {
			if s == stats.GetExtent().GetRemoteExtentPrimaryStore() {
				event := NewRemoteExtentPrimaryStoreDownEvent(s, extent.GetExtentUUID())
				context.eventPipeline.Add(event)
			}
		}
	}
}