<commit_before>// Copyright (c) 2016 Uber Technologies, Inc.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.

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 (
	// T471438, ADDR_SEAL is being written into replica_stats. We ignore small negative 26-bit 2's-compliment numbers,
	// which are larger than this constant in signed 64-bit space
	storageMetaValuesLimit = int64(1<<26) - (1 << 4)

	// QueueDepthTabulationString can be added to a destination or CG owner email to request queue depth tabulation
	// Note that Google allows something like this: <EMAIL>
	// The above is still a valid email and will be delivered to <EMAIL>
	QueueDepthTabulationString = `queueDepthTabulation`

	gaftTimeout    = 2 * time.Second // getAddressFromTimestamp timeout
	gaftCacheLimit = 1 << 20         // We expect cache entries to take less than 2^7 bytes. 2^27 = 128MB. 27-7 = 20. This means that our cache should be less than 128MB
	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 represents the state during a
		// single iteration of a destination and/or
		// consumer group
		iter struct {
			isDLQ                    bool // is the current destination a DLQ
			dstDesc                  *shared.DestinationDescription
			dstExtents               map[extentID]*metadata.DestinationExtent // set of extents for the current dst
			storeExtentMetadataCache *storeExtentMetadataCache

			cg struct {
				now                   common.UnixNanoTime
				backlogAvailable      int64
				backlogInflight       int64
				nOpenExtents          int64 // stat for stallness check
				nStalledExtents       int64 // stat for stallness check
				desc                  *shared.ConsumerGroupDescription
				coverMap              map[extentID]struct{} // list of extents that account towards backlog
				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`),
	}
}

// GetQueueDepthResult gives a queue depth result for the given cgUUID
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)}
}

// handleEvent handles an event from the metadata iterator
func (qdc *queueDepthCalculator) handleEvent(e *mIteratorEvent) {
	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)
}<commit_msg>disable queue-depth for Kafka destinations (#144)<commit_after>// Copyright (c) 2016 Uber Technologies, Inc.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.

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 (
	// T471438, ADDR_SEAL is being written into replica_stats. We ignore small negative 26-bit 2's-compliment numbers,
	// which are larger than this constant in signed 64-bit space
	storageMetaValuesLimit = int64(1<<26) - (1 << 4)

	// QueueDepthTabulationString can be added to a destination or CG owner email to request queue depth tabulation
	// Note that Google allows something like this: <EMAIL>
	// The above is still a valid email and will be delivered to <EMAIL>
	QueueDepthTabulationString = `queueDepthTabulation`

	gaftTimeout    = 2 * time.Second // getAddressFromTimestamp timeout
	gaftCacheLimit = 1 << 20         // We expect cache entries to take less than 2^7 bytes. 2^27 = 128MB. 27-7 = 20. This means that our cache should be less than 128MB
	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 represents the state during a
		// single iteration of a destination and/or
		// consumer group
		iter struct {
			isDLQ                    bool // is the current destination a DLQ
			dstDesc                  *shared.DestinationDescription
			dstExtents               map[extentID]*metadata.DestinationExtent // set of extents for the current dst
			storeExtentMetadataCache *storeExtentMetadataCache

			cg struct {
				now                   common.UnixNanoTime
				backlogAvailable      int64
				backlogInflight       int64
				nOpenExtents          int64 // stat for stallness check
				nStalledExtents       int64 // stat for stallness check
				desc                  *shared.ConsumerGroupDescription
				coverMap              map[extentID]struct{} // list of extents that account towards backlog
				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`),
	}
}

// GetQueueDepthResult gives a queue depth result for the given cgUUID
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)}
}

// handleEvent handles an event from the metadata iterator
func (qdc *queueDepthCalculator) handleEvent(e *mIteratorEvent) {

	// ignore Kafka destinations; we do not support queue-depth for Kafka destinations
	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)
}
{'fraction_non_alphanumeric': 0.08517350157728706, 'fraction_numerical': 0.01790684728149935, 'mean_word_length': 3.6906005221932117, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 4, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7233219', 'n_tokens_mistral': 3546, 'n_tokens_neox': 3234, 'n_words': 1006}