0186dd68-6020-4844-b02a-bb5e79dca3e8

Committed 14 Mar 2023 12:16AM UTC coverage: 57.084% (+0.001%) from 57.083%

Build # 0186dd68-6020-4844-b02a-bb5e79dca3e8

Build Type

push

buildkite

Committed by David Porter

Commit Message

Merge branch 'master' into feature/zonal-partitioning

Run Details

335 of 335 new or added lines in 21 files covered. (100.0%)

85415 of 149631 relevant lines covered (57.08%)

2276.13 hits per line

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

56.06

/service/history/queue/transfer_queue_processor.go

// Copyright (c) 2017-2020 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 queue

import (
        "context"
        "errors"
        "fmt"
        "math"
        "sync"
        "sync/atomic"
        "time"

        "github.com/pborman/uuid"

        "github.com/uber/cadence/common"
        "github.com/uber/cadence/common/log"
        "github.com/uber/cadence/common/log/tag"
        "github.com/uber/cadence/common/metrics"
        "github.com/uber/cadence/common/ndc"
        "github.com/uber/cadence/common/persistence"
        "github.com/uber/cadence/common/reconciliation/invariant"
        "github.com/uber/cadence/common/types"
        hcommon "github.com/uber/cadence/service/history/common"
        "github.com/uber/cadence/service/history/config"
        "github.com/uber/cadence/service/history/engine"
        "github.com/uber/cadence/service/history/execution"
        "github.com/uber/cadence/service/history/reset"
        "github.com/uber/cadence/service/history/shard"
        "github.com/uber/cadence/service/history/task"
        "github.com/uber/cadence/service/worker/archiver"
)

const (
        defaultProcessingQueueLevel = 0
)

var (
        errUnexpectedQueueTask = errors.New("unexpected queue task")
        errProcessorShutdown   = errors.New("queue processor has been shutdown")

        maximumTransferTaskKey = newTransferTaskKey(math.MaxInt64)
)

type (
        transferQueueProcessor struct {
                shard         shard.Context
                historyEngine engine.Engine
                taskProcessor task.Processor

                config             *config.Config
                currentClusterName string

                metricsClient metrics.Client
                logger        log.Logger

                status       int32
                shutdownChan chan struct{}
                shutdownWG   sync.WaitGroup

                ackLevel               int64
                taskAllocator          TaskAllocator
                activeTaskExecutor     task.Executor
                activeQueueProcessor   *transferQueueProcessorBase
                standbyQueueProcessors map[string]*transferQueueProcessorBase
        }
)

// NewTransferQueueProcessor creates a new transfer QueueProcessor
func NewTransferQueueProcessor(
        shard shard.Context,
        historyEngine engine.Engine,
        taskProcessor task.Processor,
        executionCache *execution.Cache,
        workflowResetter reset.WorkflowResetter,
        archivalClient archiver.Client,
        executionCheck invariant.Invariant,
) Processor {
        logger := shard.GetLogger().WithTags(tag.ComponentTransferQueue)
        currentClusterName := shard.GetClusterMetadata().GetCurrentClusterName()
        config := shard.GetConfig()
        taskAllocator := NewTaskAllocator(shard)

        activeTaskExecutor := task.NewTransferActiveTaskExecutor(
                shard,
                archivalClient,
                executionCache,
                workflowResetter,
                logger,
                config,
        )

        activeQueueProcessor := newTransferQueueActiveProcessor(
                shard,
                historyEngine,
                taskProcessor,
                taskAllocator,
                activeTaskExecutor,
                logger,
        )

        standbyQueueProcessors := make(map[string]*transferQueueProcessorBase)
        for clusterName := range shard.GetClusterMetadata().GetRemoteClusterInfo() {
                historyResender := ndc.NewHistoryResender(
                        shard.GetDomainCache(),
                        shard.GetService().GetClientBean().GetRemoteAdminClient(clusterName),
                        func(ctx context.Context, request *types.ReplicateEventsV2Request) error {
                                return historyEngine.ReplicateEventsV2(ctx, request)
                        },
                        config.StandbyTaskReReplicationContextTimeout,
                        executionCheck,
                        shard.GetLogger(),
                )
                standbyTaskExecutor := task.NewTransferStandbyTaskExecutor(
                        shard,
                        archivalClient,
                        executionCache,
                        historyResender,
                        logger,
                        clusterName,
                        config,
                )
                standbyQueueProcessors[clusterName] = newTransferQueueStandbyProcessor(
                        clusterName,
                        shard,
                        historyEngine,
                        taskProcessor,
                        taskAllocator,
                        standbyTaskExecutor,
                        logger,
                )
        }

        return &transferQueueProcessor{
                shard:         shard,
                historyEngine: historyEngine,
                taskProcessor: taskProcessor,

                config:             config,
                currentClusterName: currentClusterName,

                metricsClient: shard.GetMetricsClient(),
                logger:        logger,

                status:       common.DaemonStatusInitialized,
                shutdownChan: make(chan struct{}),

                ackLevel:               shard.GetTransferAckLevel(),
                taskAllocator:          taskAllocator,
                activeTaskExecutor:     activeTaskExecutor,
                activeQueueProcessor:   activeQueueProcessor,
                standbyQueueProcessors: standbyQueueProcessors,
        }
}

func (t *transferQueueProcessor) Start() {
        if !atomic.CompareAndSwapInt32(&t.status, common.DaemonStatusInitialized, common.DaemonStatusStarted) {
                return
        }

        t.activeQueueProcessor.Start()
        for _, standbyQueueProcessor := range t.standbyQueueProcessors {
                standbyQueueProcessor.Start()
        }

        t.shutdownWG.Add(1)
        go t.completeTransferLoop()
}

func (t *transferQueueProcessor) Stop() {
        if !atomic.CompareAndSwapInt32(&t.status, common.DaemonStatusStarted, common.DaemonStatusStopped) {
                return
        }

        t.activeQueueProcessor.Stop()
        for _, standbyQueueProcessor := range t.standbyQueueProcessors {
                standbyQueueProcessor.Stop()
        }

        close(t.shutdownChan)
        common.AwaitWaitGroup(&t.shutdownWG, time.Minute)
}

func (t *transferQueueProcessor) NotifyNewTask(
        clusterName string,
        info *hcommon.NotifyTaskInfo,
) {
        if len(info.Tasks) == 0 {
                return
        }

        if clusterName == t.currentClusterName {
                t.activeQueueProcessor.notifyNewTask(info)
                return
        }

        standbyQueueProcessor, ok := t.standbyQueueProcessors[clusterName]
        if !ok {
                panic(fmt.Sprintf("Cannot find transfer processor for %s.", clusterName))
        }
        standbyQueueProcessor.notifyNewTask(info)
}

func (t *transferQueueProcessor) FailoverDomain(
        domainIDs map[string]struct{},
) {
        // Failover queue is used to scan all inflight tasks, if queue processor is not
        // started, there's no inflight task and we don't need to create a failover processor.
        // Also the HandleAction will be blocked if queue processor processing loop is not running.
        if atomic.LoadInt32(&t.status) != common.DaemonStatusStarted {
                return
        }

        minLevel := t.shard.GetTransferClusterAckLevel(t.currentClusterName)
        standbyClusterName := t.currentClusterName
        for clusterName := range t.shard.GetClusterMetadata().GetEnabledClusterInfo() {
                ackLevel := t.shard.GetTransferClusterAckLevel(clusterName)
                if ackLevel < minLevel {
                        minLevel = ackLevel
                        standbyClusterName = clusterName
                }
        }

        maxReadLevel := int64(0)
        actionResult, err := t.HandleAction(context.Background(), t.currentClusterName, NewGetStateAction())
        if err != nil {
                t.logger.Error("Transfer Failover Failed", tag.WorkflowDomainIDs(domainIDs), tag.Error(err))
                if err == errProcessorShutdown {
                        // processor/shard already shutdown, we don't need to create failover queue processor
                        return
                }
                // other errors should never be returned for GetStateAction
                panic(fmt.Sprintf("unknown error for GetStateAction: %v", err))
        }
        for _, queueState := range actionResult.GetStateActionResult.States {
                queueReadLevel := queueState.ReadLevel().(transferTaskKey).taskID
                if maxReadLevel < queueReadLevel {
                        maxReadLevel = queueReadLevel
                }
        }
        // maxReadLevel is exclusive, so add 1
        maxReadLevel++

        t.logger.Info("Transfer Failover Triggered",
                tag.WorkflowDomainIDs(domainIDs),
                tag.MinLevel(minLevel),
                tag.MaxLevel(maxReadLevel))

        updateShardAckLevel, failoverQueueProcessor := newTransferQueueFailoverProcessor(
                t.shard,
                t.historyEngine,
                t.taskProcessor,
                t.taskAllocator,
                t.activeTaskExecutor,
                t.logger,
                minLevel,
                maxReadLevel,
                domainIDs,
                standbyClusterName,
        )

        // NOTE: READ REF BEFORE MODIFICATION
        // ref: historyEngine.go registerDomainFailoverCallback function
        err = updateShardAckLevel(newTransferTaskKey(minLevel))
        if err != nil {
                t.logger.Error("Error update shard ack level", tag.Error(err))
        }
        failoverQueueProcessor.Start()
}

func (t *transferQueueProcessor) HandleAction(
        ctx context.Context,
        clusterName string,
        action *Action,
) (*ActionResult, error) {
        var resultNotificationCh chan actionResultNotification
        var added bool
        if clusterName == t.currentClusterName {
                resultNotificationCh, added = t.activeQueueProcessor.addAction(ctx, action)
        } else {
                found := false
                for standbyClusterName, standbyProcessor := range t.standbyQueueProcessors {
                        if clusterName == standbyClusterName {
                                resultNotificationCh, added = standbyProcessor.addAction(ctx, action)
                                found = true
                                break
                        }
                }

                if !found {
                        return nil, fmt.Errorf("unknown cluster name: %v", clusterName)
                }
        }

        if !added {
                if ctxErr := ctx.Err(); ctxErr != nil {
                        return nil, ctxErr
                }
                return nil, errProcessorShutdown
        }

        select {
        case resultNotification := <-resultNotificationCh:
                return resultNotification.result, resultNotification.err
        case <-t.shutdownChan:
                return nil, errProcessorShutdown
        case <-ctx.Done():
                return nil, ctx.Err()
        }
}

func (t *transferQueueProcessor) LockTaskProcessing() {
        t.taskAllocator.Lock()
}

func (t *transferQueueProcessor) UnlockTaskProcessing() {
        t.taskAllocator.Unlock()
}

func (t *transferQueueProcessor) completeTransferLoop() {
        defer t.shutdownWG.Done()

        completeTimer := time.NewTimer(t.config.TransferProcessorCompleteTransferInterval())
        defer completeTimer.Stop()

        for {
                select {
                case <-t.shutdownChan:
                        // before shutdown, make sure the ack level is up to date
                        if err := t.completeTransfer(); err != nil {
                                t.logger.Error("Error complete transfer task", tag.Error(err))
                        }
                        return
                case <-completeTimer.C:
                        for attempt := 0; attempt < t.config.TransferProcessorCompleteTransferFailureRetryCount(); attempt++ {
                                err := t.completeTransfer()
                                if err == nil {
                                        break
                                }

                                t.logger.Error("Failed to complete transfer task", tag.Error(err))
                                if err == shard.ErrShardClosed {
                                        // shard closed, trigger shutdown and bail out
                                        go t.Stop()
                                        return
                                }
                                backoff := time.Duration(attempt * 100)
                                time.Sleep(backoff * time.Millisecond)

                                select {
                                case <-t.shutdownChan:
                                        // break the retry loop if shutdown chan is closed
                                        break
                                default:
                                }
                        }

                        completeTimer.Reset(t.config.TransferProcessorCompleteTransferInterval())
                }
        }
}

func (t *transferQueueProcessor) completeTransfer() error {
        newAckLevel := maximumTransferTaskKey
        actionResult, err := t.HandleAction(context.Background(), t.currentClusterName, NewGetStateAction())
        if err != nil {
                return err
        }
        for _, queueState := range actionResult.GetStateActionResult.States {
                newAckLevel = minTaskKey(newAckLevel, queueState.AckLevel())
        }

        for standbyClusterName := range t.standbyQueueProcessors {
                actionResult, err := t.HandleAction(context.Background(), standbyClusterName, NewGetStateAction())
                if err != nil {
                        return err
                }
                for _, queueState := range actionResult.GetStateActionResult.States {
                        newAckLevel = minTaskKey(newAckLevel, queueState.AckLevel())
                }
        }

        for _, failoverInfo := range t.shard.GetAllTransferFailoverLevels() {
                failoverLevel := newTransferTaskKey(failoverInfo.MinLevel)
                if newAckLevel == nil {
                        newAckLevel = failoverLevel
                } else {
                        newAckLevel = minTaskKey(newAckLevel, failoverLevel)
                }
        }

        if newAckLevel == nil {
                panic("Unable to get transfer queue processor ack level")
        }

        newAckLevelTaskID := newAckLevel.(transferTaskKey).taskID
        t.logger.Debug(fmt.Sprintf("Start completing transfer task from: %v, to %v.", t.ackLevel, newAckLevelTaskID))
        if t.ackLevel >= newAckLevelTaskID {
                return nil
        }

        t.metricsClient.IncCounter(metrics.TransferQueueProcessorScope, metrics.TaskBatchCompleteCounter)

        for {
                pageSize := t.config.TransferTaskDeleteBatchSize()
                resp, err := t.shard.GetExecutionManager().RangeCompleteTransferTask(context.Background(), &persistence.RangeCompleteTransferTaskRequest{
                        ExclusiveBeginTaskID: t.ackLevel,
                        InclusiveEndTaskID:   newAckLevelTaskID,
                        PageSize:             pageSize, // pageSize may or may not be honored
                })
                if err != nil {
                        return err
                }
                if !persistence.HasMoreRowsToDelete(resp.TasksCompleted, pageSize) {
                        break
                }
        }

        t.ackLevel = newAckLevelTaskID

        return t.shard.UpdateTransferAckLevel(newAckLevelTaskID)
}

func newTransferQueueActiveProcessor(
        shard shard.Context,
        historyEngine engine.Engine,
        taskProcessor task.Processor,
        taskAllocator TaskAllocator,
        taskExecutor task.Executor,
        logger log.Logger,
) *transferQueueProcessorBase {
        config := shard.GetConfig()
        options := newTransferQueueProcessorOptions(config, true, false)

        currentClusterName := shard.GetClusterMetadata().GetCurrentClusterName()
        logger = logger.WithTags(tag.ClusterName(currentClusterName))

        taskFilter := func(taskInfo task.Info) (bool, error) {
                task, ok := taskInfo.(*persistence.TransferTaskInfo)
                if !ok {
                        return false, errUnexpectedQueueTask
                }
                if notRegistered, err := isDomainNotRegistered(shard, task.DomainID); notRegistered && err == nil {
                        logger.Info("Domain is not in registered status, skip task in active transfer queue.", tag.WorkflowDomainID(task.DomainID), tag.Value(taskInfo))
                        return false, nil
                }
                return taskAllocator.VerifyActiveTask(task.DomainID, task)
        }

        updateMaxReadLevel := func() task.Key {
                return newTransferTaskKey(shard.GetTransferMaxReadLevel())
        }

        updateClusterAckLevel := func(ackLevel task.Key) error {
                taskID := ackLevel.(transferTaskKey).taskID
                return shard.UpdateTransferClusterAckLevel(currentClusterName, taskID)
        }

        updateProcessingQueueStates := func(states []ProcessingQueueState) error {
                pStates := convertToPersistenceTransferProcessingQueueStates(states)
                return shard.UpdateTransferProcessingQueueStates(currentClusterName, pStates)
        }

        queueShutdown := func() error {
                return nil
        }

        return newTransferQueueProcessorBase(
                shard,
                loadTransferProcessingQueueStates(currentClusterName, shard, options, logger),
                taskProcessor,
                options,
                updateMaxReadLevel,
                updateClusterAckLevel,
                updateProcessingQueueStates,
                queueShutdown,
                taskFilter,
                taskExecutor,
                logger,
                shard.GetMetricsClient(),
        )
}

func newTransferQueueStandbyProcessor(
        clusterName string,
        shard shard.Context,
        historyEngine engine.Engine,
        taskProcessor task.Processor,
        taskAllocator TaskAllocator,
        taskExecutor task.Executor,
        logger log.Logger,
) *transferQueueProcessorBase {
        config := shard.GetConfig()
        options := newTransferQueueProcessorOptions(config, false, false)

        logger = logger.WithTags(tag.ClusterName(clusterName))

        taskFilter := func(taskInfo task.Info) (bool, error) {
                task, ok := taskInfo.(*persistence.TransferTaskInfo)
                if !ok {
                        return false, errUnexpectedQueueTask
                }
                if notRegistered, err := isDomainNotRegistered(shard, task.DomainID); notRegistered && err == nil {
                        logger.Info("Domain is not in registered status, skip task in standby transfer queue.", tag.WorkflowDomainID(task.DomainID), tag.Value(taskInfo))
                        return false, nil
                }
                if task.TaskType == persistence.TransferTaskTypeCloseExecution ||
                        task.TaskType == persistence.TransferTaskTypeRecordWorkflowClosed {
                        domainEntry, err := shard.GetDomainCache().GetDomainByID(task.DomainID)
                        if err == nil {
                                if domainEntry.HasReplicationCluster(clusterName) {
                                        // guarantee the processing of workflow execution close
                                        return true, nil
                                }
                        } else {
                                if _, ok := err.(*types.EntityNotExistsError); !ok {
                                        // retry the task if failed to find the domain
                                        logger.Warn("Cannot find domain", tag.WorkflowDomainID(task.DomainID))
                                        return false, err
                                }
                                logger.Warn("Cannot find domain, default to not process task.", tag.WorkflowDomainID(task.DomainID), tag.Value(task))
                                return false, nil
                        }
                }
                return taskAllocator.VerifyStandbyTask(clusterName, task.DomainID, task)
        }

        updateMaxReadLevel := func() task.Key {
                return newTransferTaskKey(shard.GetTransferMaxReadLevel())
        }

        updateClusterAckLevel := func(ackLevel task.Key) error {
                taskID := ackLevel.(transferTaskKey).taskID
                return shard.UpdateTransferClusterAckLevel(clusterName, taskID)
        }

        updateProcessingQueueStates := func(states []ProcessingQueueState) error {
                pStates := convertToPersistenceTransferProcessingQueueStates(states)
                return shard.UpdateTransferProcessingQueueStates(clusterName, pStates)
        }

        queueShutdown := func() error {
                return nil
        }

        return newTransferQueueProcessorBase(
                shard,
                loadTransferProcessingQueueStates(clusterName, shard, options, logger),
                taskProcessor,
                options,
                updateMaxReadLevel,
                updateClusterAckLevel,
                updateProcessingQueueStates,
                queueShutdown,
                taskFilter,
                taskExecutor,
                logger,
                shard.GetMetricsClient(),
        )
}

func newTransferQueueFailoverProcessor(
        shardContext shard.Context,
        historyEngine engine.Engine,
        taskProcessor task.Processor,
        taskAllocator TaskAllocator,
        taskExecutor task.Executor,
        logger log.Logger,
        minLevel, maxLevel int64,
        domainIDs map[string]struct{},
        standbyClusterName string,
) (updateClusterAckLevelFn, *transferQueueProcessorBase) {
        config := shardContext.GetConfig()
        options := newTransferQueueProcessorOptions(config, true, true)

        currentClusterName := shardContext.GetService().GetClusterMetadata().GetCurrentClusterName()
        failoverUUID := uuid.New()
        logger = logger.WithTags(
                tag.ClusterName(currentClusterName),
                tag.WorkflowDomainIDs(domainIDs),
                tag.FailoverMsg("from: "+standbyClusterName),
        )

        taskFilter := func(taskInfo task.Info) (bool, error) {
                task, ok := taskInfo.(*persistence.TransferTaskInfo)
                if !ok {
                        return false, errUnexpectedQueueTask
                }
                if notRegistered, err := isDomainNotRegistered(shardContext, task.DomainID); notRegistered && err == nil {
                        logger.Info("Domain is not in registered status, skip task in failover transfer queue.", tag.WorkflowDomainID(task.DomainID), tag.Value(taskInfo))
                        return false, nil
                }
                return taskAllocator.VerifyFailoverActiveTask(domainIDs, task.DomainID, task)
        }

        maxReadLevelTaskKey := newTransferTaskKey(maxLevel)
        updateMaxReadLevel := func() task.Key {
                return maxReadLevelTaskKey // this is a const
        }

        updateClusterAckLevel := func(ackLevel task.Key) error {
                taskID := ackLevel.(transferTaskKey).taskID
                return shardContext.UpdateTransferFailoverLevel(
                        failoverUUID,
                        shard.TransferFailoverLevel{
                                StartTime:    shardContext.GetTimeSource().Now(),
                                MinLevel:     minLevel,
                                CurrentLevel: taskID,
                                MaxLevel:     maxLevel,
                                DomainIDs:    domainIDs,
                        },
                )
        }

        queueShutdown := func() error {
                return shardContext.DeleteTransferFailoverLevel(failoverUUID)
        }

        processingQueueStates := []ProcessingQueueState{
                NewProcessingQueueState(
                        defaultProcessingQueueLevel,
                        newTransferTaskKey(minLevel),
                        maxReadLevelTaskKey,
                        NewDomainFilter(domainIDs, false),
                ),
        }

        return updateClusterAckLevel, newTransferQueueProcessorBase(
                shardContext,
                processingQueueStates,
                taskProcessor,
                options,
                updateMaxReadLevel,
                updateClusterAckLevel,
                nil,
                queueShutdown,
                taskFilter,
                taskExecutor,
                logger,
                shardContext.GetMetricsClient(),
        )
}

func loadTransferProcessingQueueStates(
        clusterName string,
        shard shard.Context,
        options *queueProcessorOptions,
        logger log.Logger,
) []ProcessingQueueState {
        ackLevel := shard.GetTransferClusterAckLevel(clusterName)
        if options.EnableLoadQueueStates() {
                pStates := shard.GetTransferProcessingQueueStates(clusterName)
                if validateProcessingQueueStates(pStates, ackLevel) {
                        return convertFromPersistenceTransferProcessingQueueStates(pStates)
                }

                logger.Error("Incompatible processing queue states and ackLevel",
                        tag.Value(pStates),
                        tag.ShardTransferAcks(ackLevel),
                )
        }

        // LoadQueueStates is disabled or sanity check failed
        // fallback to use ackLevel
        return []ProcessingQueueState{
                NewProcessingQueueState(
                        defaultProcessingQueueLevel,
                        newTransferTaskKey(ackLevel),
                        maximumTransferTaskKey,
                        NewDomainFilter(nil, true),
                ),
        }
}

uber / cadence / 0186dd68-6020-4844-b02a-bb5e79dca3e8

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