func (qdc *queueDepthCalculator) getExtentCoverMap() map[extentID]struct{} {
	result := make(map[extentID]struct{}, len(qdc.iter.dstExtents))
	for extID := range qdc.iter.dstExtents {
		result[extID] = struct{}{}
	}
	return result
}
getExtentCoverMap returns a copy of all the extent
ids for the currently iterated destination
package controllerhost
import (
	"fmt"
	"math"
	"math/rand"
	"strings"
	"time"
	"github.com/uber/cherami-server/common"
	"github.com/uber/cherami-server/common/metrics"
	"github.com/uber/cherami-server/services/controllerhost/load"
	c "github.com/uber/cherami-thrift/.generated/go/controller"
	"github.com/uber/cherami-thrift/.generated/go/metadata"
	"github.com/uber/cherami-thrift/.generated/go/shared"
	storeGen "github.com/uber/cherami-thrift/.generated/go/store"
	"github.com/uber-common/bark"
	"github.com/uber/tchannel-go/thrift"
)
const (
	storageMetaValuesLimit = int64(1<<26) - (1 << 4)
	QueueDepthTabulationString = `queueDepthTabulation`
	gaftTimeout    = 2 * time.Second  
	gaftCacheLimit = 1 << 20          
	gaftSealed     = common.SequenceNumber(-1)
	backlogProgessInfinity    = 6666
	minBacklogForCGStallCheck = 1000
	zeroUUID = "00000000-0000-0000-0000-000000000000"
)
var (
	gaftCache = make(map[gaftKey]common.SequenceNumber)
)
type (
	extentID string
	storeID  string
	queueDepthCalculator struct {
		context         *Context
		ll              bark.Logger
		queueDepthCache *QueueDepthCache
		iter struct {
			isDLQ                    bool  
			dstDesc                  *shared.DestinationDescription
			dstExtents               map[extentID]*metadata.DestinationExtent  
			storeExtentMetadataCache *storeExtentMetadataCache
			cg struct {
				now                   common.UnixNanoTime
				backlogAvailable      int64
				backlogInflight       int64
				nOpenExtents          int64  
				nStalledExtents       int64  
				desc                  *shared.ConsumerGroupDescription
				coverMap              map[extentID]struct{}  
				isTabulationRequested bool
			}
		}
	}
	gaftKey struct {
		extentUUID string
		timestamp  int64
	}
)
func newQueueDepthCalculator(context *Context) *queueDepthCalculator {
	return &queueDepthCalculator{
		context:         context,
		queueDepthCache: newQueueDepthCache(),
		ll:              context.log.WithField(`module`, `queueDepth`),
	}
}
func (qdc *queueDepthCalculator) GetQueueDepthResult(cgUUID string) (QueueDepthCacheEntry, error) {
	if entry, ok := qdc.queueDepthCache.get(cgUUID); ok {
		return entry, nil
	}
	return QueueDepthCacheEntry{}, &c.QueueCacheMissError{Message: fmt.Sprintf("queue depth entity cache miss for %s", cgUUID)}
}
func (qdc *queueDepthCalculator) handleEvent(e *mIteratorEvent) {
	if e.dest != nil && e.dest.GetType() == shared.DestinationType_KAFKA {
		return
	}
	switch e.t {
	case eDestStart:
		qdc.handleDestinationStart(e.dest)
	case eDestEnd:
		qdc.handleDestinationEnd(e.dest)
	case eExtent:
		qdc.handleDestinationExtent(e.dest, e.extent)
	case eCnsmStart:
		qdc.iter.cg.desc = e.cnsm
		qdc.handleConsumerGroupStart(e.dest, e.cnsm)
	case eCnsmEnd:
		qdc.handleConsumerGroupEnd(e.dest, e.cnsm)
	case eCnsmExtent:
		qdc.handleConsumerGroupExtent(e.dest, e.cnsm, e.cnsmExtent)
	}
}
func (qdc *queueDepthCalculator) handleDestinationStart(dstDesc *shared.DestinationDescription) {
	qdc.iter.dstDesc = dstDesc
	qdc.iter.isDLQ = common.IsDLQDestination(dstDesc)
	qdc.iter.dstExtents = make(map[extentID]*metadata.DestinationExtent, 16)
	qdc.iter.storeExtentMetadataCache = newStoreExtentMetadataCache(qdc.context.mm, qdc.ll)
}
func (qdc *queueDepthCalculator) handleDestinationEnd(dstDesc *shared.DestinationDescription) {
	qdc.iter.dstDesc = nil
	qdc.iter.isDLQ = false
	qdc.iter.storeExtentMetadataCache = nil
	qdc.iter.dstExtents = nil
}
func (qdc *queueDepthCalculator) handleConsumerGroupStart(dstDesc *shared.DestinationDescription, cgDesc *shared.ConsumerGroupDescription) {
	iter := &qdc.iter
	iter.cg.desc = cgDesc
	iter.cg.isTabulationRequested = qdc.isTabulationRequested(cgDesc, dstDesc)
	if iter.isDLQ && cgDesc.GetDeadLetterQueueDestinationUUID() != dstDesc.GetDestinationUUID() {
		return  
	}
	iter.cg.now = common.Now()
	iter.cg.coverMap = qdc.getExtentCoverMap()
	if iter.cg.isTabulationRequested {
		qdc.ll.WithFields(bark.Fields{
			common.TagDst:    common.FmtDst(dstDesc.GetDestinationUUID()),
			common.TagDstPth: common.FmtDstPth(dstDesc.GetPath()),
			common.TagCnsm:   common.FmtCnsm(cgDesc.GetConsumerGroupUUID()),
			common.TagCnsPth: common.FmtCnsPth(cgDesc.GetConsumerGroupName()),
			`extentCount`:    len(iter.cg.coverMap),
		}).Info(`Queue Depth Tabulation, start`)
	}
}
func (qdc *queueDepthCalculator) handleConsumerGroupEnd(dstDesc *shared.DestinationDescription, cgDesc *shared.ConsumerGroupDescription) {
	iter := &qdc.iter
	if cgDesc.GetStatus() == shared.ConsumerGroupStatus_DELETING ||
		cgDesc.GetStatus() == shared.ConsumerGroupStatus_DELETED ||
		dstDesc.GetStatus() == shared.DestinationStatus_DELETING ||
		dstDesc.GetStatus() == shared.DestinationStatus_DELETED {
		qdc.reportBacklog(cgDesc, dstDesc, backlogProgessInfinity, 0)
	} else {
		for extent := range iter.cg.coverMap {
			dstExtent, _ := iter.dstExtents[extent]  
			storeUUIDs := dstExtent.GetStoreUUIDs()
			if len(storeUUIDs) == 0 {
				continue  
			}
			cge := &shared.ConsumerGroupExtent{
				ConsumerGroupUUID:  common.StringPtr(cgDesc.GetConsumerGroupUUID()),
				ExtentUUID:         common.StringPtr(string(extent)),
				OutputHostUUID:     common.StringPtr(zeroUUID),
				ConnectedStoreUUID: common.StringPtr(storeUUIDs[rand.Intn(len(storeUUIDs))]),  
			}
			qdc.addExtentBacklog(dstDesc, cgDesc, dstExtent, cge, qdc.makeCGExtentLogger(dstDesc, cgDesc, cge))
		}
		progressScore := qdc.measureBacklogProgress(dstDesc, cgDesc, iter.cg.backlogAvailable)
		qdc.reportBacklog(cgDesc, dstDesc, progressScore, iter.cg.now)
	}
	if iter.cg.isTabulationRequested {
		qdc.ll.WithFields(bark.Fields{
			common.TagDst:    common.FmtDst(dstDesc.GetDestinationUUID()),
			common.TagDstPth: common.FmtDstPth(dstDesc.GetPath()),
			common.TagCnsm:   common.FmtCnsm(cgDesc.GetConsumerGroupUUID()),
			common.TagCnsPth: common.FmtCnsPth(cgDesc.GetConsumerGroupName()),
		}).Info(`Queue Depth Tabulation, end`)
	}
	iter.cg.desc = nil
	iter.cg.coverMap = nil
	iter.cg.nOpenExtents = 0
	iter.cg.nStalledExtents = 0
	iter.cg.backlogAvailable = 0
	iter.cg.backlogInflight = 0
}
func (qdc *queueDepthCalculator) handleDestinationExtent(dstDesc *shared.DestinationDescription, extent *metadata.DestinationExtent) {
	if dstDesc.GetStatus() == shared.DestinationStatus_DELETING ||
		dstDesc.GetStatus() == shared.DestinationStatus_DELETED ||
		extent.GetStatus() > shared.ExtentStatus_SEALED {
		return
	}
	qdc.iter.dstExtents[extentID(extent.GetExtentUUID())] = extent
}
func (qdc *queueDepthCalculator) handleConsumerGroupExtent(dstDesc *shared.DestinationDescription, cgDesc *shared.ConsumerGroupDescription, cgExtent *shared.ConsumerGroupExtent) {
	if cgDesc.GetStatus() == shared.ConsumerGroupStatus_DELETING ||
		cgDesc.GetStatus() == shared.ConsumerGroupStatus_DELETED ||
		dstDesc.GetStatus() == shared.DestinationStatus_DELETING ||
		dstDesc.GetStatus() == shared.DestinationStatus_DELETED {
		return
	}
	iter := &qdc.iter
	logger := qdc.makeCGExtentLogger(dstDesc, cgDesc, cgExtent)
	if cgExtent.GetStatus() == shared.ConsumerGroupExtentStatus_OPEN {
		qdc.iter.cg.nOpenExtents++
		if qdc.isCGExtentStalled(cgExtent) {
			qdc.iter.cg.nStalledExtents++
		}
	}
	if cgExtent.GetAckLevelSeqNo() == 0 && cgExtent.GetConnectedStoreUUID() == `` {
		if iter.cg.isTabulationRequested {
			logger.Info(`Queue Depth Tabulation, forcing assigned but unopened extent to dangling`)
		}
		return
	}
	dstExtent, ok := iter.dstExtents[extentID(cgExtent.GetExtentUUID())]
	if !ok {
		return  
	}
	qdc.addExtentBacklog(dstDesc, cgDesc, dstExtent, cgExtent, logger)
	delete(iter.cg.coverMap, extentID(cgExtent.GetExtentUUID()))
}
func (qdc *queueDepthCalculator) addExtentBacklog(
	dstDesc *shared.DestinationDescription,
	cgDesc *shared.ConsumerGroupDescription,
	dstExtent *metadata.DestinationExtent,
	cgExtent *shared.ConsumerGroupExtent,
	logger bark.Logger) {
	extID := extentID(cgExtent.GetExtentUUID())
	connectedStoreID := storeID(cgExtent.GetConnectedStoreUUID())
	iter := &qdc.iter
	if cgExtent.GetStatus() != shared.ConsumerGroupExtentStatus_OPEN {
		if iter.cg.isTabulationRequested {
			logger.WithFields(bark.Fields{
				common.TagStor:     string(connectedStoreID),
				`cgeStatus`:        cgExtent.GetStatus(),
				`cgeAckLvlSeq`:     cgExtent.GetAckLevelSeqNo(),
				`cgeAckLvlSeqRate`: cgExtent.GetAckLevelSeqNoRate(),
				`cgeWriteTime`:     common.UnixNanoTime(cgExtent.GetWriteTime()).ToSecondsFmt(),
			}).Info(`Queue Depth Tabulation (skipping consumed/deleted extent)`)
		}
		return
	}
	if !qdc.isExtentBelongsToCG(dstExtent, cgDesc) || qdc.isPurgedDLQExtent(dstExtent) {
		if iter.cg.isTabulationRequested {
			logger.Info(`Queue Depth Tabulation, skipping other other CG's merged and/or purged dlq extent`)
		}
		return
	}
	ΔInFlight := common.MaxInt64(0, cgExtent.GetReadLevelSeqNo()-cgExtent.GetAckLevelSeqNo())
	iter.cg.backlogInflight += ΔInFlight
	// Calculations requiring store extent
	storeExtentMetadata := iter.storeExtentMetadataCache.get(connectedStoreID, extID)
	if storeExtentMetadata == nil {
		// Can only happen due to cassandra's eventual consistency
		if iter.cg.isTabulationRequested {
			logger.WithFields(bark.Fields{
				common.TagStor:     string(connectedStoreID),
				`cgeAckLvlSeq`:     cgExtent.GetAckLevelSeqNo(),
				`cgeAckLvlSeqRate`: cgExtent.GetAckLevelSeqNoRate(),
				`cgeWriteTime`:     common.UnixNanoTime(cgExtent.GetWriteTime()).ToSecondsFmt(),
			}).Info(`Queue Depth Tabulation (missing store extent)`)
		}
		return
	}
	if !qdc.handleStartFrom(dstDesc, cgDesc, dstExtent, cgExtent, storeExtentMetadata, logger) {
		return
	}
	iter.cg.backlogAvailable += qdc.computeBacklog(cgDesc, cgExtent, storeExtentMetadata, string(connectedStoreID), logger)
}
func (qdc *queueDepthCalculator) computeBacklog(cgDesc *shared.ConsumerGroupDescription, cgExtent *shared.ConsumerGroupExtent, storeMetadata *storeExtentMetadata, storeID string, logger bark.Logger) int64 {
	var backlog int64  
	var iter = &qdc.iter
	if qdc.iter.isDLQ {
		if storeMetadata.lastSequence != 0 && storeMetadata.lastSequence != math.MaxInt64 &&
			storeMetadata.beginSequence != 0 && storeMetadata.beginSequence != math.MaxInt64 {
			backlog = storeMetadata.lastSequence - storeMetadata.beginSequence + 1
		}
	} else {
		if storeMetadata.availableSequence != 0 && storeMetadata.availableSequence != math.MaxInt64 {
			backlog = storeMetadata.availableSequence - cgExtent.GetAckLevelSeqNo()
		}
	}
	if iter.cg.isTabulationRequested {
		logger.WithFields(bark.Fields{
			common.TagStor:      storeID,
			`rsStore`:           storeMetadata.storeID,
			`rsAvailSeq`:        storeMetadata.availableSequence,
			`rsBeginSeq`:        storeMetadata.beginSequence,
			`rsLastSeq`:         storeMetadata.lastSequence,
			`rsLastSeqRate`:     storeMetadata.lastSequenceRate,
			`cgeAckLvlSeq`:      cgExtent.GetAckLevelSeqNo(),
			`rsAvailSeqRate`:    storeMetadata.availableSequenceRate,
			`cgeAckLvlSeqRate`:  cgExtent.GetAckLevelSeqNoRate(),
			`rsWriteTime`:       storeMetadata.writeTime.ToSecondsFmt(),
			`cgeWriteTime`:      common.UnixNanoTime(cgExtent.GetWriteTime()).ToSecondsFmt(),
			`subTotalAvail`:     backlog,
			`runningTotalAvail`: backlog + iter.cg.backlogAvailable,
		}).Info(`Queue Depth Tabulation`)
	}
	return backlog
}
func getAddressFromTimestampOnStores(context *Context, cgDesc *shared.ConsumerGroupDescription, storeUUIDs []string, extentID string, timestamp int64) common.SequenceNumber {
	var client storeGen.TChanBStore
	var err error
	var ok bool
	var res *storeGen.GetAddressFromTimestampResult_
	var seqNo common.SequenceNumber
	var storeUUID, storeAddr string
	var trace int
	if seqNo, ok = gaftCache[gaftKey{extentUUID: extentID, timestamp: timestamp}]; ok {
		return seqNo
	}
	req := storeGen.NewGetAddressFromTimestampRequest()
	ctx, cancel := thrift.NewContext(gaftTimeout)
	defer cancel()
	for _, storeUUID = range storeUUIDs {  
		storeAddr, err = context.rpm.ResolveUUID(common.StoreServiceName, storeUUID)
		if err == nil {
			break
		}
	}
	if err != nil {
		trace = 1
		goto fail
	}
	client, err = context.clientFactory.GetThriftStoreClient(storeAddr, storeUUID)
	if err != nil {
		trace = 2
		goto fail
	}
	defer context.clientFactory.ReleaseThriftStoreClient(storeUUID)
	req.ExtentUUID = common.StringPtr(extentID)
	req.Timestamp = common.Int64Ptr(timestamp)
	res, err = client.GetAddressFromTimestamp(ctx, req)
	if err != nil {
		trace = 3
		goto fail
	}
	seqNo = common.SequenceNumber(res.GetSequenceNumber())
	if res.GetSealed() {
		seqNo = gaftSealed
	}
	if len(gaftCache) > gaftCacheLimit {  
		for k := range gaftCache {
			delete(gaftCache, k)
			break
		}
	}
	gaftCache[gaftKey{extentUUID: extentID, timestamp: timestamp}] = seqNo
	return seqNo
fail:
	context.log.WithFields(bark.Fields{
		common.TagErr:  err,
		common.TagStor: storeUUID,  
		common.TagExt:  common.FmtExt(extentID),
		common.TagCnsm: common.FmtCnsm(cgDesc.GetConsumerGroupUUID()),
		`timestamp`:    timestamp,
		`storeAddr`:    storeAddr,
		`trace`:        trace,
	}).Error(`Queue Depth: Error getting address for timestamp`)
	return 0  
}
func (qdc *queueDepthCalculator) handleStartFrom(
	dstDesc *shared.DestinationDescription,
	cgDesc *shared.ConsumerGroupDescription,
	dstExtent *metadata.DestinationExtent,
	consumerGroupExtent *shared.ConsumerGroupExtent,
	storeMetadata *storeExtentMetadata,
	logger bark.Logger,
) (qualify bool) {
	if qdc.iter.isDLQ {
		return true  
	}
	var startFromSeq common.SequenceNumber
	var trace int  
	qualify = true
	doGaft := true
	createTime := storeMetadata.createdTime
	now := qdc.iter.cg.now
	isTabulationRequested := qdc.iter.cg.isTabulationRequested
	if cgDesc.GetStartFrom() > int64(now) {
		trace = 1
		qualify = false
		goto done
	}
	if consumerGroupExtent.AckLevelSeqNo == nil {  
		switch {
		case cgDesc.GetStartFrom() == 0:
			trace = 2
			doGaft = false
		case storeMetadata.endTime != 0 && cgDesc.GetStartFrom() > storeMetadata.endTime:
			trace = 3
			qualify = false
			goto done
		case storeMetadata.endTime == 0 && storeMetadata.lastEnqueueTimeUtc != 0 && cgDesc.GetStartFrom() > storeMetadata.lastEnqueueTimeUtc:
			trace = 4
			qualify = false
			goto done
		case storeMetadata.beginTime != 0 && cgDesc.GetStartFrom() >= storeMetadata.beginTime:
			trace = 5
		case storeMetadata.beginTime == 0 && storeMetadata.beginEnqueueTimeUtc != 0 && cgDesc.GetStartFrom() >= storeMetadata.beginEnqueueTimeUtc:
			trace = 6
		case storeMetadata.beginTime == 0 && storeMetadata.beginEnqueueTimeUtc == 0 && cgDesc.GetStartFrom() > int64(createTime):
			trace = 7
		case createTime != 0 && cgDesc.GetStartFrom() <= int64(createTime):  
			trace = 8
			doGaft = false
		default:
			trace = 9
			if isTabulationRequested {
				logger.WithFields(bark.Fields{
					`rsBeginSeq`:     storeMetadata.beginSequence,
					`rsEndtime`:      storeMetadata.endTime,
					`rsLastEnqueue`:  storeMetadata.lastEnqueueTimeUtc,
					`rsBeginTime`:    storeMetadata.beginTime,
					`rsBeginEnqueue`: storeMetadata.beginEnqueueTimeUtc,
					`createTime`:     createTime,
					`startFrom`:      cgDesc.GetStartFrom(),
				}).Warn(`Queue Depth Tabulation (StartFrom) Unhandled case`)
			}
		}
		if doGaft {
			startFromSeq = getAddressFromTimestampOnStores(qdc.context, cgDesc, dstExtent.GetStoreUUIDs(), dstExtent.GetExtentUUID(), cgDesc.GetStartFrom())
			if startFromSeq == gaftSealed {
				trace += 10
				qualify = false  
			} else {  
				trace += 20
			}
		}
		if qualify {
			consumerGroupExtent.WriteTime = common.Int64Ptr(int64(now))
			if storeMetadata.beginSequence != math.MaxInt64 {
				trace += 100
				consumerGroupExtent.AckLevelSeqNo = common.Int64Ptr(common.MaxInt64(
					storeMetadata.beginSequence-1,  
					int64(startFromSeq),            
				))
			} else {
				trace += 200
				consumerGroupExtent.AckLevelSeqNo = common.Int64Ptr(0)
			}
		}
	}
done:
	if isTabulationRequested {
		logger.WithFields(bark.Fields{
			`rsBeginSeq`:   storeMetadata.beginSequence,
			`startFromSeq`: startFromSeq,
			`rsAvailSeq`:   storeMetadata.availableSequence,
			`qualify`:      qualify,
			`trace`:        trace,
		}).Info(`Queue Depth Tabulation (StartFrom)`)
	}
	return
}
func (qdc *queueDepthCalculator) measureBacklogProgress(dstDesc *shared.DestinationDescription, cgDesc *shared.ConsumerGroupDescription, backlog int64) int {
	nOpen := qdc.iter.cg.nOpenExtents
	nStalled := qdc.iter.cg.nStalledExtents
	if nOpen == 0 || backlog < minBacklogForCGStallCheck {
		return backlogProgessInfinity
	}
	if nStalled > 0 && nStalled >= nOpen/2 {
		qdc.ll.WithFields(bark.Fields{
			common.TagDst:    common.FmtDst(dstDesc.GetDestinationUUID()),
			common.TagDstPth: dstDesc.GetPath(),
			common.TagCnsm:   cgDesc.GetConsumerGroupUUID(),
			common.TagCnsPth: cgDesc.GetConsumerGroupName(),
			`openExtents`:    nOpen,
			`stalledExtents`: nStalled,
		}).Warn(`ConsumerGroup stalled`)
		return 0
	}
	return int(nOpen)
}
func (qdc *queueDepthCalculator) isCGExtentStalled(cge *shared.ConsumerGroupExtent) bool {
	context := qdc.context
	cgID := cge.GetConsumerGroupUUID()
	extID := cge.GetExtentUUID()
	hostID := cge.GetOutputHostUUID()
	smartRetryOn, err := context.loadMetrics.Get(hostID, cgID, load.SmartRetryOn, load.OneMinSum)
	if err != nil || smartRetryOn > 0 {
		return false
	}
	nConns, err := context.loadMetrics.Get(hostID, cgID, load.NumConns, load.FiveMinAvg)
	if err != nil {
		return false
	}
	msgsOut, err := context.loadMetrics.Get(hostID, extID, load.MsgsOutPerSec, load.FiveMinSum)
	if err != nil {
		return false
	}
	return (nConns > 0 && msgsOut == 0)
}
func (qdc *queueDepthCalculator) reportStalledMetric(dstDesc *shared.DestinationDescription) bool {
	return !strings.HasPrefix(dstDesc.GetPath(), "/test")
}
func (qdc *queueDepthCalculator) reportNormalBacklog(metricsClient metrics.Client, cgDesc *shared.ConsumerGroupDescription, dstDesc *shared.DestinationDescription, progressScore int, now common.UnixNanoTime) {
	metricsClient.UpdateGauge(metrics.QueueDepthBacklogCGScope, metrics.ControllerCGBacklogAvailable, qdc.iter.cg.backlogAvailable)
	metricsClient.UpdateGauge(metrics.QueueDepthBacklogCGScope, metrics.ControllerCGBacklogInflight, qdc.iter.cg.backlogInflight)
	if qdc.reportStalledMetric(dstDesc) {
		metricsClient.UpdateGauge(metrics.QueueDepthBacklogCGScope, metrics.ControllerCGBacklogProgress, int64(progressScore))
	}
	entry, _ := qdc.queueDepthCache.get(cgDesc.GetConsumerGroupUUID())
	entry.BacklogAvailable = qdc.iter.cg.backlogAvailable
	entry.BacklogInflight = qdc.iter.cg.backlogInflight
	qdc.queueDepthCache.put(cgDesc.GetConsumerGroupUUID(), &entry)
}
func (qdc *queueDepthCalculator) reportDLQBacklog(metricsClient metrics.Client, cgDesc *shared.ConsumerGroupDescription, dstDesc *shared.DestinationDescription, now common.UnixNanoTime) {
	metricsClient.UpdateGauge(metrics.QueueDepthBacklogCGScope, metrics.ControllerCGBacklogDLQ, qdc.iter.cg.backlogAvailable)
	entry, _ := qdc.queueDepthCache.get(cgDesc.GetConsumerGroupUUID())
	entry.BacklogDLQ = qdc.iter.cg.backlogAvailable
	qdc.queueDepthCache.put(cgDesc.GetConsumerGroupUUID(), &entry)
}
func (qdc *queueDepthCalculator) reportBacklog(cgDesc *shared.ConsumerGroupDescription, dstDesc *shared.DestinationDescription, progressScore int, now common.UnixNanoTime) {
	metricsClient := qdc.makeMetricsClient(cgDesc)
	if qdc.iter.isDLQ {
		qdc.reportDLQBacklog(metricsClient, cgDesc, dstDesc, now)
	} else {
		qdc.reportNormalBacklog(metricsClient, cgDesc, dstDesc, progressScore, now)
	}
	if qdc.iter.cg.backlogAvailable != 0 || qdc.iter.cg.backlogInflight != 0 {
		qdc.ll.WithFields(bark.Fields{
			`time`:             now,
			common.TagDst:      common.FmtDst(dstDesc.GetDestinationUUID()),
			common.TagDstPth:   common.FmtDstPth(dstDesc.GetPath()),
			common.TagCnsm:     common.FmtCnsm(cgDesc.GetConsumerGroupUUID()),
			common.TagCnsPth:   common.FmtCnsPth(cgDesc.GetConsumerGroupName()),
			`BacklogAvailable`: qdc.iter.cg.backlogAvailable,
			`BacklogInflight`:  qdc.iter.cg.backlogInflight,
			`progressScore`:    progressScore,
		}).Info(`Queue Depth Metrics`)
	}
}