1030

pending completion

Build # 1030

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/uber/cherami-server/commit/<a class=hub.com/uber/cherami-server/commit/0bce9e8ca643809f09486e757ebffe7fe8ca888f">0bce9e8ca<a href="https://github.com/uber/cherami-server/commit/0bce9e8ca643809f09486e757ebffe7fe8ca888f">&quot;&gt;Re-enable replicate-extent test (#106)

* fix test to wait some more for ringpop refresh

* setup test storehosts on init

* remove unnecessary stalls

* test only relicate-extent

* Revert &amp;quot;test only relicate-extent&quot;

This reverts commit </a><a class="double-link" href="https://github.com/uber/cherami-server/commit/<a class="double-link" href="https://github.com/uber/cherami-server/commit/495c716ca13341891fcc4c7040e62926555b4bbe">495c716ca</a>">495c716ca</a><a href="https://github.com/uber/cherami-server/commit/0bce9e8ca643809f09486e757ebffe7fe8ca888f">.

* ~

* ~~

* remove newline

* revert removed newline

* revert .travis.yml to orig

Run Details

19542 of 28520 relevant lines covered (68.52%)

2428.56 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

75.48

/services/outputhost/ackmanager.go

// 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 outputhost

import (
        "errors"
        "sync"
        "time"

        "github.com/uber-common/bark"
        "github.com/uber/tchannel-go/thrift"

        "github.com/uber/cherami-server/common"
        "github.com/uber/cherami-server/common/metrics"
        "github.com/uber/cherami-thrift/.generated/go/admin"
        "github.com/uber/cherami-thrift/.generated/go/metadata"
)

const ackLevelInterval = 5 * time.Second
const metaContextTimeout = 10 * time.Second

type storeHostAddress int64

type (
        // internalMsg is the message which is stored locally on the ackMgr
        internalMsg struct {
                addr  storeHostAddress
                acked bool
        }

        levels struct {
                asOf          common.UnixNanoTime   // Timestamp when this backLogLevel was calculated
                readLevel     common.SequenceNumber // -1 = nothing received, 0 = 1 message (#0) received, etc.
                ackLevel      common.SequenceNumber // -1 = nothing acked
                ackLevelRate  float64               // Rate of change in ackedTo, seconds⁽⁻¹⁾
                readLevelRate float64               // Rate of change in receivedTo, seconds⁽⁻¹⁾
                ackLevelAddr  storeHostAddress      // Storehost address corresponding to ackLevel, -1 = nothing acked
                readLevelAddr storeHostAddress      // Storehost address corresponding to recievedLevel, -1 = nothing read
                lastAckedSeq  common.SequenceNumber // the latest sequence which is acked
        }

        // ackManager is held per CG extent and it holds the addresses that we get from the store.
        ackManager struct {
                addrs              map[common.SequenceNumber]*internalMsg // ‡
                sealed             bool                                   // ‡
                outputHostUUID     string
                cgUUID             string
                extUUID            string
                connectedStoreUUID *string
                *levels                    // ‡ the current levels
                prev               *levels // ‡ the previous levels
                ackLevelTicker     *time.Ticker
                closeChannel       chan struct{}
                waitConsumed       chan<- bool // waitConsumed is the channel which will signal if the extent is completely consumed given by extentCache
                metaclient         metadata.TChanMetadataService
                doneWG             sync.WaitGroup
                logger             bark.Logger
                sessionID          uint16
                ackMgrID           uint16                // ID of this ackManager; unique on this host
                cgCache            *consumerGroupCache   // back pointer to the consumer group cache
                levelOffset        common.SequenceNumber // ‡
                lk                 sync.RWMutex          // ‡ = guarded by this mutex
        }
)

func newAckManager(cgCache *consumerGroupCache, ackMgrID uint32, outputHostUUID string, cgUUID string, extUUID string, connectedStoreUUID *string, waitConsumedCh chan<- bool, cge *metadata.ConsumerGroupExtent, metaclient metadata.TChanMetadataService, logger bark.Logger) *ackManager {
        ackMgr := &ackManager{
                addrs:              make(map[common.SequenceNumber]*internalMsg),
                cgCache:            cgCache,
                outputHostUUID:     outputHostUUID,
                cgUUID:             cgUUID,
                extUUID:            extUUID,
                connectedStoreUUID: connectedStoreUUID,
                sessionID:          cgCache.sessionID, //sessionID,
                ackMgrID:           uint16(ackMgrID),  //ackMgrID,
                metaclient:         metaclient,
                ackLevelTicker:     time.NewTicker(ackLevelInterval),
                waitConsumed:       waitConsumedCh,
                logger:             logger.WithField(common.TagModule, `ackMgr`),
        }

        // Set the previous levels now, so that on our first update, we will calculate rates correctly
        ackMgr.prev = &levels{
                asOf:          common.Now(),
                readLevel:     common.SequenceNumber(cge.GetAckLevelSeqNo()),
                ackLevel:      common.SequenceNumber(cge.GetAckLevelSeqNo()),
                ackLevelAddr:  storeHostAddress(cge.GetAckLevelOffset()),
                readLevelAddr: storeHostAddress(cge.GetReadLevelOffset()),
        }

        ackMgr.levels = &levels{
                readLevel:     common.SequenceNumber(cge.GetAckLevelSeqNo()),
                ackLevel:      common.SequenceNumber(cge.GetAckLevelSeqNo()),
                ackLevelAddr:  storeHostAddress(cge.GetAckLevelOffset()),
                readLevelAddr: storeHostAddress(cge.GetReadLevelOffset()),
        }

        return ackMgr
}

// ackID is a string which is a base64 encoded string
// First we get the ackID and store the address locally in our data structure
// for maintaining the ack level
func (ackMgr *ackManager) getNextAckID(address int64, sequence common.SequenceNumber) (ackID string) {
        ackMgr.lk.Lock()
        ackMgr.readLevel++ // This means that the first ID is '1'
        ackMgr.readLevelAddr = storeHostAddress(address)

        expectedReadLevel := ackMgr.levelOffset + ackMgr.readLevel

        // NOTE: sequence should be zero for timer destinations, since they usually have discontinuous sequence numbers
        if sequence != 0 && expectedReadLevel != sequence {
                skippedMessages := sequence - expectedReadLevel

                if skippedMessages < 0 {
                        ackMgr.logger.Error(`negative discontinuity detected (rollback)`)
                        // Don't update gauge, since negative numbers aren't supported for M3 gauges
                } else {
                        // update gauge here to say we skipped messages (potentially due to retention?)
                        ackMgr.cgCache.consumerM3Client.UpdateGauge(metrics.ConsConnectionScope, metrics.OutputhostCGSkippedMessages, int64(skippedMessages))
                }
                // TODO: Support message auditing by creating discontinuity events, rather than just updating the offset here
                // TODO: Consumption rates will be spuriously high
                // NOTE: Increasing the offset here also affects the AckLevelSeqNo, which will tend to push it into the discontinuity,
                //       which makes it appear that a retained message was received. This is OK, and it will cause queue depth to react
                //       to retention events a bit faster.
                // NOTE: The same skipped messages can be reported twice if the outputhost is restarted before the ack level
                //       passes the discontinuity
                ackMgr.addLevelOffsetImpl(skippedMessages)
        }

        ackID = common.ConstructAckID(ackMgr.sessionID, ackMgr.ackMgrID, uint32(ackMgr.readLevel), address)

        // now store the message in the data structure internally
        ackMgr.addrs[ackMgr.readLevel] = &internalMsg{
                addr: storeHostAddress(address),
        }

        ackMgr.lk.Unlock()

        return
}

// addLevelOffset adds an offset to the sequence number stored in cassandra. This ensures that our internal records always have
// continguous sequence numbers, but we can adjust the offset when/if the replica connection gets loaded (asynchronously)
func (ackMgr *ackManager) addLevelOffset(offset common.SequenceNumber) {
        ackMgr.lk.Lock()
        ackMgr.addLevelOffsetImpl(offset)
        ackMgr.lk.Unlock()
}

func (ackMgr *ackManager) addLevelOffsetImpl(offset common.SequenceNumber) {
        ackMgr.logger.WithFields(bark.Fields{
                `oldOffset`: ackMgr.levelOffset,
                `newOffset`: ackMgr.levelOffset + offset,
                `change`:    offset,
        }).Info(`adjusting sequence number offset`)
        ackMgr.levelOffset += offset
}

func (ackMgr *ackManager) stop() {
        close(ackMgr.closeChannel)
        ackMgr.doneWG.Wait()
}

func (ackMgr *ackManager) start() {
        ackMgr.closeChannel = make(chan struct{})
        ackMgr.doneWG.Add(1)
        go ackMgr.manageAckLevel()
}

func (ackMgr *ackManager) getCurrentAckLevelOffset() (addr int64) {
        ackMgr.lk.RLock()
        if addrs, ok := ackMgr.addrs[ackMgr.ackLevel]; ok {
                addr = int64(addrs.addr)
        }
        ackMgr.lk.RUnlock()

        return
}

func (ackMgr *ackManager) getCurrentAckLevelSeqNo() (seqNo common.SequenceNumber) {
        ackMgr.lk.RLock()
        seqNo = ackMgr.levelOffset + ackMgr.ackLevel
        ackMgr.lk.RUnlock()
        return
}

// resetMsg is a utility routine which is used to reset the readLevel because
// we couldn't even write this to the client msgsCh
func (ackMgr *ackManager) resetMsg(offset int64) {
        ackMgr.lk.Lock()
        // make sure the address of the current readLevel matches with this offset
        if addrs, ok := ackMgr.addrs[ackMgr.readLevel]; ok {
                expectedOffset := int64(addrs.addr)
                if expectedOffset != offset {
                        // this should *never* happen and we should panic here because the only way to
                        // get here is only because of memory corruption.
                        ackMgr.cgCache.consumerM3Client.UpdateGauge(metrics.ConsConnectionScope, metrics.OutputhostCGAckMgrResetMsgError, 1)
                        ackMgr.logger.WithFields(bark.Fields{
                                `expectedOffset`: expectedOffset,
                                `offset`:         offset,
                        }).Panic(`resetMsg: offsets don't match!`)
                } else {
                        // report that we removed a read level
                        ackMgr.cgCache.consumerM3Client.UpdateGauge(metrics.ConsConnectionScope, metrics.OutputhostCGAckMgrResetMsg, 1)
                        delete(ackMgr.addrs, ackMgr.readLevel)
                        // move the readLevel one below, since this message is not here anymore
                        ackMgr.readLevel--
                }
        }
        ackMgr.lk.Unlock()
}

// notifySealed just makes a note that this extent is sealed
func (ackMgr *ackManager) notifySealed() {
        ackMgr.lk.Lock()
        ackMgr.sealed = true
        ackMgr.lk.Unlock()
        return
}

func (ackMgr *ackManager) updateAckLevel() {
        update := false
        consumed := false
        var oReq *metadata.SetAckOffsetRequest

        ackMgr.lk.Lock()

        count := 0
        stop := ackMgr.ackLevel + common.SequenceNumber(int64(len(ackMgr.addrs)))

        // We go through the map here and see if the messages are acked,
        // moving the acklevel as we go forward.
        for curr := ackMgr.ackLevel + 1; curr <= stop; curr++ {
                if addrs, ok := ackMgr.addrs[curr]; ok {
                        if addrs.acked {
                                update = true
                                ackMgr.ackLevel = curr

                                // getCurrentAckLevelOffset needs addr[ackMgr.ackLevel], so delete the previous one if it exists
                                count++
                                delete(ackMgr.addrs, curr-1)
                        } else {
                                break
                        }
                }
        }

        // check if the extent can be marked as consumed
        // We can mark an extent as consumed, if we have both these conditions:
        // 1. The extent is sealed (which means we have it marked after receiving the last message)
        // 2. The ackLevel has reached the end (which means that the ackLevel equals the readLevel)
        if ackMgr.sealed && ackMgr.ackLevel == ackMgr.readLevel {
                ackMgr.logger.Debug("extent sealed and consumed")
                consumed = true
                update = true
        }

        // check if ackLevel is valid here and get the addr here
        if _, ok := ackMgr.addrs[ackMgr.ackLevel]; ok {
                ackMgr.ackLevelAddr = ackMgr.addrs[ackMgr.ackLevel].addr
        }

        if update {
                ackMgr.asOf = common.Now()
                oReq = &metadata.SetAckOffsetRequest{
                        OutputHostUUID:     common.StringPtr(ackMgr.outputHostUUID),
                        ConsumerGroupUUID:  common.StringPtr(ackMgr.cgUUID),
                        ExtentUUID:         common.StringPtr(ackMgr.extUUID),
                        ConnectedStoreUUID: common.StringPtr(*ackMgr.connectedStoreUUID),
                        AckLevelAddress:    common.Int64Ptr(int64(ackMgr.ackLevelAddr)),
                        AckLevelSeqNo:      common.Int64Ptr(int64(ackMgr.levelOffset + ackMgr.ackLevel)), // levelOffset adjusts the level according to what was returned from getAddressFromTimestamp
                        ReadLevelAddress:   common.Int64Ptr(int64(ackMgr.readLevelAddr)),
                        ReadLevelSeqNo:     common.Int64Ptr(int64(ackMgr.levelOffset + ackMgr.readLevel)),
                }

                // check if we can set the status as consumed
                if consumed {
                        oReq.Status = common.CheramiConsumerGroupExtentStatusPtr(metadata.ConsumerGroupExtentStatus_CONSUMED)
                        // Rates are forced to zero in the consumed case
                        oReq.AckLevelSeqNoRate = common.Float64Ptr(0e0)
                        oReq.ReadLevelSeqNoRate = common.Float64Ptr(0e0)
                } else {
                        oReq.Status = common.CheramiConsumerGroupExtentStatusPtr(metadata.ConsumerGroupExtentStatus_OPEN)
                        oReq.AckLevelSeqNoRate = common.Float64Ptr(common.CalculateRate(ackMgr.prev.ackLevel, ackMgr.ackLevel, ackMgr.prev.asOf, ackMgr.asOf))
                        oReq.ReadLevelSeqNoRate = common.Float64Ptr(common.CalculateRate(ackMgr.prev.readLevel, ackMgr.readLevel, ackMgr.prev.asOf, ackMgr.asOf))
                }

                // move the current levels to prev and replace the current levels with a deep copy
                ackMgr.prev = ackMgr.levels
                ackMgr.levels = &levels{}
                *ackMgr.levels = *ackMgr.prev
        }

        updatedSize := len(ackMgr.addrs)
        ackMgr.lk.Unlock()

        if update {
                ctx, cancel := thrift.NewContext(metaContextTimeout)
                defer cancel()
                // check if we can set the status as consumed
                // ackMgr.logger.
                //        WithField(common.TagCnsm, common.FmtCnsm(ackMgr.cgUUID)).
                //        WithField(common.TagExt, common.FmtExt(ackMgr.extUUID)).
                //        WithField(`ackLevelAddress`, oReq.GetAckLevelAddress()).
                //        Debug(`outputhost: Setting ackLevel`)

                if err := ackMgr.metaclient.SetAckOffset(ctx, oReq); err != nil {
                        ackMgr.logger.WithFields(bark.Fields{
                                common.TagErr:     err,
                                `ackLevelAddress`: oReq.GetAckLevelAddress(),
                        }).Error(`error updating ackLevel`)
                } else {
                        // Updating metadata succeeded; report some metrics and mark the extent as consumed if necessary
                        // report the count of updates we did this round
                        ackMgr.cgCache.consumerM3Client.UpdateGauge(metrics.ConsConnectionScope, metrics.OutputhostCGAckMgrLevelUpdate, int64(count))
                        if consumed {
                                // report that the extent is consumed
                                ackMgr.cgCache.consumerM3Client.UpdateGauge(metrics.ConsConnectionScope, metrics.OutputhostCGAckMgrConsumed, 1)
                                // notify extentCache that this extent is consumed
                                // We can do this in a non-blocking way because, if we fail to
                                // notify this time, we will anyway notify it the next round.
                                // We do it in a non-blocking way because we could potentially
                                // deadlock during unload
                                select {
                                case ackMgr.waitConsumed <- true:
                                        ackMgr.logger.WithField(`ackLevelAddress`, oReq.GetAckLevelAddress()).Info(`extent consumed`)
                                default:
                                }
                        }
                }
        }

        // Report the size of the ackMgr map, if greater than 0
        if updatedSize > 0 {
                ackMgr.cgCache.consumerM3Client.UpdateGauge(metrics.ConsConnectionScope, metrics.OutputhostCGAckMgrSize, int64(updatedSize))
        }
}

func (ackMgr *ackManager) acknowledgeMessage(ackID AckID, seqNum uint32, address int64, isNack bool) error {
        var err error
        notifyCg := true
        ackMgr.lk.Lock() // Read lock would be OK in this case (except for a benign race with two simultaneous acks for the same ackID), see below
        // check if this id is present
        if addrs, ok := ackMgr.addrs[common.SequenceNumber(seqNum)]; ok {
                // validate the address from the ackID
                if addrs.addr != storeHostAddress(address) {
                        ackMgr.logger.WithFields(bark.Fields{
                                `address`:  address,
                                `expected`: addrs.addr,
                        }).Error(`ack address does not match!`)
                        err = errors.New("address of the ackID doesn't match with ackMgr")
                        notifyCg = false
                } else {
                        if ackMgr.cgCache.cachedCGDesc.GetOwnerEmail() == SmartRetryDisableString {
                                ackMgr.logger.WithFields(bark.Fields{
                                        `Address`:     address,
                                        `addr`:        addrs.addr,
                                        common.TagSeq: seqNum,
                                        `isNack`:      isNack,
                                }).Info(`msg ack`)
                        }
                        if !isNack {
                                addrs.acked = true // This is the only place that this field of addrs is changed. It was initially set under a write lock elsewhere, hence we can have a read lock
                                // update the last acked sequence, if this is the most recent ack
                                if ackMgr.lastAckedSeq < common.SequenceNumber(seqNum) {
                                        ackMgr.lastAckedSeq = common.SequenceNumber(seqNum)
                                }
                        }
                }
        } else {
                // Update metric to reflect that the sequence number is not found
                ackMgr.cgCache.consumerM3Client.IncCounter(metrics.ConsConnectionScope, metrics.OutputhostCGAckMgrSeqNotFound)
        }
        ackMgr.lk.Unlock()

        // Now notify the message cache so that it can update it's state
        // Note: We explicitly do this outside the lock above to prevent us from
        // blocking with a lock held
        // send the ack to the ack channel for the msg cache to cleanup
        if notifyCg {
                if isNack {
                        ackMgr.cgCache.nackMsgCh <- timestampedAckID{AckID: ackID, ts: common.Now()}
                } else {
                        ackMgr.cgCache.ackMsgCh <- timestampedAckID{AckID: ackID, ts: common.Now()}
                }
        }
        return err
}

func (ackMgr *ackManager) manageAckLevel() {
        defer ackMgr.doneWG.Done()
        // this needs to look at all the acked messages and update the ackLevel
        // accordingly.
        for {
                select {
                case <-ackMgr.ackLevelTicker.C:
                        ackMgr.updateAckLevel()
                case <-ackMgr.closeChannel:
                        // before returning make sure we try to set the ack offset
                        ackMgr.updateAckLevel()
                        return
                }
        }
}

// get the number of acked and unacked messages from the last ack level
func (ackMgr *ackManager) getNumAckedAndUnackedMessages() (*int64, *int64) {
        stop := ackMgr.ackLevel + common.SequenceNumber(int64(len(ackMgr.addrs)))

        var acked int64
        var unacked int64
        // We go through the map here and see if the messages are acked,
        for curr := ackMgr.ackLevel + 1; curr <= stop; curr++ {
                if addrs, ok := ackMgr.addrs[curr]; ok {
                        if addrs.acked {
                                acked++
                        } else {
                                unacked++
                        }
                } else {
                        ackMgr.logger.WithFields(bark.Fields{
                                common.TagSeq: curr,
                        }).Error(`sequence number not found in the ack mgr`)
                }
        }

        return common.Int64Ptr(acked), common.Int64Ptr(unacked)
}

func (ackMgr *ackManager) getAckMgrState() *admin.AckMgrState {
        ackMgrState := admin.NewAckMgrState()
        ackMgr.lk.RLock()
        defer ackMgr.lk.RUnlock()
        ackMgrState.AckMgrID = common.Int16Ptr(int16(ackMgr.ackMgrID))
        ackMgrState.IsSealed = common.BoolPtr(ackMgr.sealed)
        ackMgrState.ReadLevelSeq = common.Int64Ptr(int64(ackMgr.readLevel))
        ackMgrState.AckLevelSeq = common.Int64Ptr(int64(ackMgr.ackLevel))
        ackMgrState.ReadLevelOffset = common.Int64Ptr(int64(ackMgr.readLevelAddr))
        ackMgrState.AckLevelOffset = common.Int64Ptr(int64(ackMgr.ackLevelAddr))
        ackMgrState.LastAckLevelUpdateTime = common.Int64Ptr(int64(ackMgr.asOf))
        ackMgrState.LastAckedSeq = common.Int64Ptr(int64(ackMgr.lastAckedSeq))
        ackMgrState.NumAckedMsgs, ackMgrState.NumUnackedMsgs = ackMgr.getNumAckedAndUnackedMessages()

        return ackMgrState
}

1	// Copyright (c) 2016 Uber Technologies, Inc.
2	//
3	// Permission is hereby granted, free of charge, to any person obtaining a copy
4	// of this software and associated documentation files (the "Software"), to deal
5	// in the Software without restriction, including without limitation the rights
6	// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
7	// copies of the Software, and to permit persons to whom the Software is
8	// furnished to do so, subject to the following conditions:
9	//
10	// The above copyright notice and this permission notice shall be included in
11	// all copies or substantial portions of the Software.
12	//
13	// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
14	// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
15	// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
16	// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
17	// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
18	// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
19	// THE SOFTWARE.
20
21	package outputhost
22
23	import (
24	"errors"
25	"sync"
26	"time"
27
28	"github.com/uber-common/bark"
29	"github.com/uber/tchannel-go/thrift"
30
31	"github.com/uber/cherami-server/common"
32	"github.com/uber/cherami-server/common/metrics"
33	"github.com/uber/cherami-thrift/.generated/go/admin"
34	"github.com/uber/cherami-thrift/.generated/go/metadata"
35	)
36
37	const ackLevelInterval = 5 * time.Second
38	const metaContextTimeout = 10 * time.Second
39
40	type storeHostAddress int64
41
42	type (
43	// internalMsg is the message which is stored locally on the ackMgr
44	internalMsg struct {
45	addr storeHostAddress
46	acked bool
47	}
48
49	levels struct {
50	asOf common.UnixNanoTime // Timestamp when this backLogLevel was calculated
51	readLevel common.SequenceNumber // -1 = nothing received, 0 = 1 message (#0) received, etc.
52	ackLevel common.SequenceNumber // -1 = nothing acked
53	ackLevelRate float64 // Rate of change in ackedTo, seconds⁽⁻¹⁾
54	readLevelRate float64 // Rate of change in receivedTo, seconds⁽⁻¹⁾
55	ackLevelAddr storeHostAddress // Storehost address corresponding to ackLevel, -1 = nothing acked
56	readLevelAddr storeHostAddress // Storehost address corresponding to recievedLevel, -1 = nothing read
57	lastAckedSeq common.SequenceNumber // the latest sequence which is acked
58	}
59
60	// ackManager is held per CG extent and it holds the addresses that we get from the store.
61	ackManager struct {
62	addrs map[common.SequenceNumber]*internalMsg // ‡
63	sealed bool // ‡
64	outputHostUUID string
65	cgUUID string
66	extUUID string
67	connectedStoreUUID *string
68	*levels // ‡ the current levels
69	prev *levels // ‡ the previous levels
70	ackLevelTicker *time.Ticker
71	closeChannel chan struct{}
72	waitConsumed chan<- bool // waitConsumed is the channel which will signal if the extent is completely consumed given by extentCache
73	metaclient metadata.TChanMetadataService
74	doneWG sync.WaitGroup
75	logger bark.Logger
76	sessionID uint16
77	ackMgrID uint16 // ID of this ackManager; unique on this host
78	cgCache *consumerGroupCache // back pointer to the consumer group cache
79	levelOffset common.SequenceNumber // ‡
80	lk sync.RWMutex // ‡ = guarded by this mutex
81	}
82	)
83
84	func newAckManager(cgCache consumerGroupCache, ackMgrID uint32, outputHostUUID string, cgUUID string, extUUID string, connectedStoreUUID string, waitConsumedCh chan<- bool, cge metadata.ConsumerGroupExtent, metaclient metadata.TChanMetadataService, logger bark.Logger) ackManager {	39✔
85	ackMgr := &ackManager{	39✔
86	addrs: make(map[common.SequenceNumber]*internalMsg),	39✔
87	cgCache: cgCache,	39✔
88	outputHostUUID: outputHostUUID,	39✔
89	cgUUID: cgUUID,	39✔
90	extUUID: extUUID,	39✔
91	connectedStoreUUID: connectedStoreUUID,	39✔
92	sessionID: cgCache.sessionID, //sessionID,	39✔
93	ackMgrID: uint16(ackMgrID), //ackMgrID,	39✔
94	metaclient: metaclient,	39✔
95	ackLevelTicker: time.NewTicker(ackLevelInterval),	39✔
96	waitConsumed: waitConsumedCh,	39✔
97	logger: logger.WithField(common.TagModule, `ackMgr`),	39✔
98	}	39✔
99		39✔
100	// Set the previous levels now, so that on our first update, we will calculate rates correctly	39✔
101	ackMgr.prev = &levels{	39✔
102	asOf: common.Now(),	39✔
103	readLevel: common.SequenceNumber(cge.GetAckLevelSeqNo()),	39✔
104	ackLevel: common.SequenceNumber(cge.GetAckLevelSeqNo()),	39✔
105	ackLevelAddr: storeHostAddress(cge.GetAckLevelOffset()),	39✔
106	readLevelAddr: storeHostAddress(cge.GetReadLevelOffset()),	39✔
107	}	39✔
108		39✔
109	ackMgr.levels = &levels{	39✔
110	readLevel: common.SequenceNumber(cge.GetAckLevelSeqNo()),	39✔
111	ackLevel: common.SequenceNumber(cge.GetAckLevelSeqNo()),	39✔
112	ackLevelAddr: storeHostAddress(cge.GetAckLevelOffset()),	39✔
113	readLevelAddr: storeHostAddress(cge.GetReadLevelOffset()),	39✔
114	}	39✔
115		39✔
116	return ackMgr	39✔
117	}	39✔
118
119	// ackID is a string which is a base64 encoded string
120	// First we get the ackID and store the address locally in our data structure
121	// for maintaining the ack level
122	func (ackMgr *ackManager) getNextAckID(address int64, sequence common.SequenceNumber) (ackID string) {	47,281✔
123	ackMgr.lk.Lock()	47,281✔
124	ackMgr.readLevel++ // This means that the first ID is '1'	47,281✔
125	ackMgr.readLevelAddr = storeHostAddress(address)	47,281✔
126		47,281✔
127	expectedReadLevel := ackMgr.levelOffset + ackMgr.readLevel	47,281✔
128		47,281✔
129	// NOTE: sequence should be zero for timer destinations, since they usually have discontinuous sequence numbers	47,281✔
130	if sequence != 0 && expectedReadLevel != sequence {	93,594✔
131	skippedMessages := sequence - expectedReadLevel	46,313✔
132		46,313✔
133	if skippedMessages < 0 {	92,626✔
134	ackMgr.logger.Error(`negative discontinuity detected (rollback)`)	46,313✔
135	// Don't update gauge, since negative numbers aren't supported for M3 gauges	46,313✔
136	} else {	46,313✔
137	// update gauge here to say we skipped messages (potentially due to retention?)	×
138	ackMgr.cgCache.consumerM3Client.UpdateGauge(metrics.ConsConnectionScope, metrics.OutputhostCGSkippedMessages, int64(skippedMessages))	×
139	}	×
140	// TODO: Support message auditing by creating discontinuity events, rather than just updating the offset here
141	// TODO: Consumption rates will be spuriously high
142	// NOTE: Increasing the offset here also affects the AckLevelSeqNo, which will tend to push it into the discontinuity,
143	// which makes it appear that a retained message was received. This is OK, and it will cause queue depth to react
144	// to retention events a bit faster.
145	// NOTE: The same skipped messages can be reported twice if the outputhost is restarted before the ack level
146	// passes the discontinuity
147	ackMgr.addLevelOffsetImpl(skippedMessages)	46,313✔
148	}
149
150	ackID = common.ConstructAckID(ackMgr.sessionID, ackMgr.ackMgrID, uint32(ackMgr.readLevel), address)	47,281✔
151		47,281✔
152	// now store the message in the data structure internally	47,281✔
153	ackMgr.addrs[ackMgr.readLevel] = &internalMsg{	47,281✔
154	addr: storeHostAddress(address),	47,281✔
155	}	47,281✔
156		47,281✔
157	ackMgr.lk.Unlock()	47,281✔
158		47,281✔
159	return	47,281✔
160	}
161
162	// addLevelOffset adds an offset to the sequence number stored in cassandra. This ensures that our internal records always have
163	// continguous sequence numbers, but we can adjust the offset when/if the replica connection gets loaded (asynchronously)
164	func (ackMgr *ackManager) addLevelOffset(offset common.SequenceNumber) {	5✔
165	ackMgr.lk.Lock()	5✔
166	ackMgr.addLevelOffsetImpl(offset)	5✔
167	ackMgr.lk.Unlock()	5✔
168	}	5✔
169
170	func (ackMgr *ackManager) addLevelOffsetImpl(offset common.SequenceNumber) {	46,318✔
171	ackMgr.logger.WithFields(bark.Fields{	46,318✔
172	`oldOffset`: ackMgr.levelOffset,	46,318✔
173	`newOffset`: ackMgr.levelOffset + offset,	46,318✔
174	`change`: offset,	46,318✔
175	}).Info(`adjusting sequence number offset`)	46,318✔
176	ackMgr.levelOffset += offset	46,318✔
177	}	46,318✔
178
179	func (ackMgr *ackManager) stop() {	16✔
180	close(ackMgr.closeChannel)	16✔
181	ackMgr.doneWG.Wait()	16✔
182	}	16✔
183
184	func (ackMgr *ackManager) start() {	39✔
185	ackMgr.closeChannel = make(chan struct{})	39✔
186	ackMgr.doneWG.Add(1)	39✔
187	go ackMgr.manageAckLevel()	39✔
188	}	39✔
189
190	func (ackMgr *ackManager) getCurrentAckLevelOffset() (addr int64) {	×
191	ackMgr.lk.RLock()	×
192	if addrs, ok := ackMgr.addrs[ackMgr.ackLevel]; ok {	×
193	addr = int64(addrs.addr)	×
194	}	×
195	ackMgr.lk.RUnlock()	×
196		×
197	return	×
198	}
199
200	func (ackMgr *ackManager) getCurrentAckLevelSeqNo() (seqNo common.SequenceNumber) {	×
201	ackMgr.lk.RLock()	×
202	seqNo = ackMgr.levelOffset + ackMgr.ackLevel	×
203	ackMgr.lk.RUnlock()	×
204	return	×
205	}	×
206
207	// resetMsg is a utility routine which is used to reset the readLevel because
208	// we couldn't even write this to the client msgsCh
209	func (ackMgr *ackManager) resetMsg(offset int64) {	5✔
210	ackMgr.lk.Lock()	5✔
211	// make sure the address of the current readLevel matches with this offset	5✔
212	if addrs, ok := ackMgr.addrs[ackMgr.readLevel]; ok {	10✔
213	expectedOffset := int64(addrs.addr)	5✔
214	if expectedOffset != offset {	5✔
215	// this should never happen and we should panic here because the only way to	×
216	// get here is only because of memory corruption.	×
217	ackMgr.cgCache.consumerM3Client.UpdateGauge(metrics.ConsConnectionScope, metrics.OutputhostCGAckMgrResetMsgError, 1)	×
218	ackMgr.logger.WithFields(bark.Fields{	×
219	`expectedOffset`: expectedOffset,	×
220	`offset`: offset,	×
221	}).Panic(`resetMsg: offsets don't match!`)	×
222	} else {	5✔
223	// report that we removed a read level	5✔
224	ackMgr.cgCache.consumerM3Client.UpdateGauge(metrics.ConsConnectionScope, metrics.OutputhostCGAckMgrResetMsg, 1)	5✔
225	delete(ackMgr.addrs, ackMgr.readLevel)	5✔
226	// move the readLevel one below, since this message is not here anymore	5✔
227	ackMgr.readLevel--	5✔
228	}	5✔
229	}
230	ackMgr.lk.Unlock()	5✔
231	}
232
233	// notifySealed just makes a note that this extent is sealed
234	func (ackMgr *ackManager) notifySealed() {	9✔
235	ackMgr.lk.Lock()	9✔
236	ackMgr.sealed = true	9✔
237	ackMgr.lk.Unlock()	9✔
238	return	9✔
239	}	9✔
240
241	func (ackMgr *ackManager) updateAckLevel() {	518✔
242	update := false	518✔
243	consumed := false	518✔
244	var oReq *metadata.SetAckOffsetRequest	518✔
245		518✔
246	ackMgr.lk.Lock()	518✔
247		518✔
248	count := 0	518✔
249	stop := ackMgr.ackLevel + common.SequenceNumber(int64(len(ackMgr.addrs)))	518✔
250		518✔
251	// We go through the map here and see if the messages are acked,	518✔
252	// moving the acklevel as we go forward.	518✔
253	for curr := ackMgr.ackLevel + 1; curr <= stop; curr++ {	1,867✔
254	if addrs, ok := ackMgr.addrs[curr]; ok {	2,453✔
255	if addrs.acked {	1,948✔
256	update = true	844✔
257	ackMgr.ackLevel = curr	844✔
258		844✔
259	// getCurrentAckLevelOffset needs addr[ackMgr.ackLevel], so delete the previous one if it exists	844✔
260	count++	844✔
261	delete(ackMgr.addrs, curr-1)	844✔
262	} else {	1,104✔
263	break	260✔
264	}
265	}
266	}
267
268	// check if the extent can be marked as consumed
269	// We can mark an extent as consumed, if we have both these conditions:
270	// 1. The extent is sealed (which means we have it marked after receiving the last message)
271	// 2. The ackLevel has reached the end (which means that the ackLevel equals the readLevel)
272	if ackMgr.sealed && ackMgr.ackLevel == ackMgr.readLevel {	522✔
273	ackMgr.logger.Debug("extent sealed and consumed")	4✔
274	consumed = true	4✔
275	update = true	4✔
276	}	4✔
277
278	// check if ackLevel is valid here and get the addr here
279	if _, ok := ackMgr.addrs[ackMgr.ackLevel]; ok {	908✔
280	ackMgr.ackLevelAddr = ackMgr.addrs[ackMgr.ackLevel].addr	390✔
281	}	390✔
282
283	if update {	615✔
284	ackMgr.asOf = common.Now()	97✔
285	oReq = &metadata.SetAckOffsetRequest{	97✔
286	OutputHostUUID: common.StringPtr(ackMgr.outputHostUUID),	97✔
287	ConsumerGroupUUID: common.StringPtr(ackMgr.cgUUID),	97✔
288	ExtentUUID: common.StringPtr(ackMgr.extUUID),	97✔
289	ConnectedStoreUUID: common.StringPtr(*ackMgr.connectedStoreUUID),	97✔
290	AckLevelAddress: common.Int64Ptr(int64(ackMgr.ackLevelAddr)),	97✔
291	AckLevelSeqNo: common.Int64Ptr(int64(ackMgr.levelOffset + ackMgr.ackLevel)), // levelOffset adjusts the level according to what was returned from getAddressFromTimestamp	97✔
292	ReadLevelAddress: common.Int64Ptr(int64(ackMgr.readLevelAddr)),	97✔
293	ReadLevelSeqNo: common.Int64Ptr(int64(ackMgr.levelOffset + ackMgr.readLevel)),	97✔
294	}	97✔
295		97✔
296	// check if we can set the status as consumed	97✔
297	if consumed {	101✔
298	oReq.Status = common.CheramiConsumerGroupExtentStatusPtr(metadata.ConsumerGroupExtentStatus_CONSUMED)	4✔
299	// Rates are forced to zero in the consumed case	4✔
300	oReq.AckLevelSeqNoRate = common.Float64Ptr(0e0)	4✔
301	oReq.ReadLevelSeqNoRate = common.Float64Ptr(0e0)	4✔
302	} else {	97✔
303	oReq.Status = common.CheramiConsumerGroupExtentStatusPtr(metadata.ConsumerGroupExtentStatus_OPEN)	93✔
304	oReq.AckLevelSeqNoRate = common.Float64Ptr(common.CalculateRate(ackMgr.prev.ackLevel, ackMgr.ackLevel, ackMgr.prev.asOf, ackMgr.asOf))	93✔
305	oReq.ReadLevelSeqNoRate = common.Float64Ptr(common.CalculateRate(ackMgr.prev.readLevel, ackMgr.readLevel, ackMgr.prev.asOf, ackMgr.asOf))	93✔
306	}	93✔
307
308	// move the current levels to prev and replace the current levels with a deep copy
309	ackMgr.prev = ackMgr.levels	97✔
310	ackMgr.levels = &levels{}	97✔
311	ackMgr.levels = ackMgr.prev	97✔
312	}
313
314	updatedSize := len(ackMgr.addrs)	518✔
315	ackMgr.lk.Unlock()	518✔
316		518✔
317	if update {	615✔
318	ctx, cancel := thrift.NewContext(metaContextTimeout)	97✔
319	defer cancel()	97✔
320	// check if we can set the status as consumed	97✔
321	// ackMgr.logger.	97✔
322	// WithField(common.TagCnsm, common.FmtCnsm(ackMgr.cgUUID)).	97✔
323	// WithField(common.TagExt, common.FmtExt(ackMgr.extUUID)).	97✔
324	// WithField(`ackLevelAddress`, oReq.GetAckLevelAddress()).	97✔
325	// Debug(`outputhost: Setting ackLevel`)	97✔
326		97✔
327	if err := ackMgr.metaclient.SetAckOffset(ctx, oReq); err != nil {	97✔
328	ackMgr.logger.WithFields(bark.Fields{	×
329	common.TagErr: err,	×
330	`ackLevelAddress`: oReq.GetAckLevelAddress(),	×
331	}).Error(`error updating ackLevel`)	×
332	} else {	97✔
333	// Updating metadata succeeded; report some metrics and mark the extent as consumed if necessary	97✔
334	// report the count of updates we did this round	97✔
335	ackMgr.cgCache.consumerM3Client.UpdateGauge(metrics.ConsConnectionScope, metrics.OutputhostCGAckMgrLevelUpdate, int64(count))	97✔
336	if consumed {	101✔
337	// report that the extent is consumed	4✔
338	ackMgr.cgCache.consumerM3Client.UpdateGauge(metrics.ConsConnectionScope, metrics.OutputhostCGAckMgrConsumed, 1)	4✔
339	// notify extentCache that this extent is consumed	4✔
340	// We can do this in a non-blocking way because, if we fail to	4✔
341	// notify this time, we will anyway notify it the next round.	4✔
342	// We do it in a non-blocking way because we could potentially	4✔
343	// deadlock during unload	4✔
344	select {	4✔
345	case ackMgr.waitConsumed <- true:	4✔
346	ackMgr.logger.WithField(`ackLevelAddress`, oReq.GetAckLevelAddress()).Info(`extent consumed`)	4✔
347	default:	×
348	}
349	}
350	}
351	}
352
353	// Report the size of the ackMgr map, if greater than 0
354	if updatedSize > 0 {	1,034✔
355	ackMgr.cgCache.consumerM3Client.UpdateGauge(metrics.ConsConnectionScope, metrics.OutputhostCGAckMgrSize, int64(updatedSize))	516✔
356	}	516✔
357	}
358
359	func (ackMgr *ackManager) acknowledgeMessage(ackID AckID, seqNum uint32, address int64, isNack bool) error {	1,039✔
360	var err error	1,039✔
361	notifyCg := true	1,039✔
362	ackMgr.lk.Lock() // Read lock would be OK in this case (except for a benign race with two simultaneous acks for the same ackID), see below	1,039✔
363	// check if this id is present	1,039✔
364	if addrs, ok := ackMgr.addrs[common.SequenceNumber(seqNum)]; ok {	2,069✔
365	// validate the address from the ackID	1,030✔
366	if addrs.addr != storeHostAddress(address) {	1,030✔
367	ackMgr.logger.WithFields(bark.Fields{	×
368	`address`: address,	×
369	`expected`: addrs.addr,	×
370	}).Error(`ack address does not match!`)	×
371	err = errors.New("address of the ackID doesn't match with ackMgr")	×
372	notifyCg = false	×
373	} else {	1,030✔
374	if ackMgr.cgCache.cachedCGDesc.GetOwnerEmail() == SmartRetryDisableString {	1,030✔
375	ackMgr.logger.WithFields(bark.Fields{	×
376	`Address`: address,	×
377	`addr`: addrs.addr,	×
378	common.TagSeq: seqNum,	×
379	`isNack`: isNack,	×
380	}).Info(`msg ack`)	×
381	}	×
382	if !isNack {	1,884✔
383	addrs.acked = true // This is the only place that this field of addrs is changed. It was initially set under a write lock elsewhere, hence we can have a read lock	854✔
384	// update the last acked sequence, if this is the most recent ack	854✔
385	if ackMgr.lastAckedSeq < common.SequenceNumber(seqNum) {	1,675✔
386	ackMgr.lastAckedSeq = common.SequenceNumber(seqNum)	821✔
387	}	821✔
388	}
389	}
390	} else {	9✔
391	// Update metric to reflect that the sequence number is not found	9✔
392	ackMgr.cgCache.consumerM3Client.IncCounter(metrics.ConsConnectionScope, metrics.OutputhostCGAckMgrSeqNotFound)	9✔
393	}	9✔
394	ackMgr.lk.Unlock()	1,039✔
395		1,039✔
396	// Now notify the message cache so that it can update it's state	1,039✔
397	// Note: We explicitly do this outside the lock above to prevent us from	1,039✔
398	// blocking with a lock held	1,039✔
399	// send the ack to the ack channel for the msg cache to cleanup	1,039✔
400	if notifyCg {	2,078✔
401	if isNack {	1,223✔
402	ackMgr.cgCache.nackMsgCh <- timestampedAckID{AckID: ackID, ts: common.Now()}	184✔
403	} else {	1,039✔
404	ackMgr.cgCache.ackMsgCh <- timestampedAckID{AckID: ackID, ts: common.Now()}	855✔
405	}	855✔
406	}
407	return err	1,039✔
408	}
409
410	func (ackMgr *ackManager) manageAckLevel() {	39✔
411	defer ackMgr.doneWG.Done()	39✔
412	// this needs to look at all the acked messages and update the ackLevel	39✔
413	// accordingly.	39✔
414	for {	580✔
415	select {	541✔
416	case <-ackMgr.ackLevelTicker.C:	502✔
417	ackMgr.updateAckLevel()	502✔
418	case <-ackMgr.closeChannel:	16✔
419	// before returning make sure we try to set the ack offset	16✔
420	ackMgr.updateAckLevel()	16✔
421	return	16✔
422	}
423	}
424	}
425
426	// get the number of acked and unacked messages from the last ack level
427	func (ackMgr ackManager) getNumAckedAndUnackedMessages() (int64, *int64) {	×
428	stop := ackMgr.ackLevel + common.SequenceNumber(int64(len(ackMgr.addrs)))	×
429		×
430	var acked int64	×
431	var unacked int64	×
432	// We go through the map here and see if the messages are acked,	×
433	for curr := ackMgr.ackLevel + 1; curr <= stop; curr++ {	×
434	if addrs, ok := ackMgr.addrs[curr]; ok {	×
435	if addrs.acked {	×
436	acked++	×
437	} else {	×
438	unacked++	×
439	}	×
440	} else {	×
441	ackMgr.logger.WithFields(bark.Fields{	×
442	common.TagSeq: curr,	×
443	}).Error(`sequence number not found in the ack mgr`)	×
444	}	×
445	}
446
447	return common.Int64Ptr(acked), common.Int64Ptr(unacked)	×
448	}
449
450	func (ackMgr ackManager) getAckMgrState() admin.AckMgrState {	×
451	ackMgrState := admin.NewAckMgrState()	×
452	ackMgr.lk.RLock()	×
453	defer ackMgr.lk.RUnlock()	×
454	ackMgrState.AckMgrID = common.Int16Ptr(int16(ackMgr.ackMgrID))	×
455	ackMgrState.IsSealed = common.BoolPtr(ackMgr.sealed)	×
456	ackMgrState.ReadLevelSeq = common.Int64Ptr(int64(ackMgr.readLevel))	×
457	ackMgrState.AckLevelSeq = common.Int64Ptr(int64(ackMgr.ackLevel))	×
458	ackMgrState.ReadLevelOffset = common.Int64Ptr(int64(ackMgr.readLevelAddr))	×
459	ackMgrState.AckLevelOffset = common.Int64Ptr(int64(ackMgr.ackLevelAddr))	×
460	ackMgrState.LastAckLevelUpdateTime = common.Int64Ptr(int64(ackMgr.asOf))	×
461	ackMgrState.LastAckedSeq = common.Int64Ptr(int64(ackMgr.lastAckedSeq))	×
462	ackMgrState.NumAckedMsgs, ackMgrState.NumUnackedMsgs = ackMgr.getNumAckedAndUnackedMessages()	×
463		×
464	return ackMgrState	×
465	}	×

uber / cherami-server / 1030

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous