mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-30 14:56:19 +00:00
Alex ShtinandGitHub
8e8f1b0f7c
## What changed?
<!-- Describe what has changed in this PR -->
Remove license header from every file. Because it is really hard to
follow in this PR here is the summary:
1. License header is removed from all `.go` and `.proto` files
:fireworks::fireworks:🎆.
2. `LICENSE` file in the root directory has only Temporal and Uber
copyrights.
3. 5 other `LICENSE` files added to the packages which have copyrights
different from Temporal and Uber: Datadog, Xargin, "Mat Ryer, Tyler
Bunnell and contributors".
4. `license_file` flag is removed from all code generation tools.
5. `copyright_file` flag is removed from `go:generate mockgen`
directive.
6. All copyright related targets are removed from `Makefile`.
7. Updated Temporal copyright year to 2025 everywhere.
## Why?
<!-- Tell your future self why have you made these changes -->
I double checked with legal department that it is not needed to have
license header in every file. One file per repo is enough. I put all
copyrights to the root `LICENSE` file and removed header from all other
files. Also updated tools and `Makefile`.
189 lines
5.5 KiB
Go
189 lines
5.5 KiB
Go
package history
|
|
|
|
import (
|
|
"go.temporal.io/server/client"
|
|
"go.temporal.io/server/common/log"
|
|
"go.temporal.io/server/common/log/tag"
|
|
"go.temporal.io/server/common/metrics"
|
|
"go.temporal.io/server/common/persistence/visibility/manager"
|
|
"go.temporal.io/server/common/resource"
|
|
"go.temporal.io/server/common/sdk"
|
|
"go.temporal.io/server/common/telemetry"
|
|
historyi "go.temporal.io/server/service/history/interfaces"
|
|
"go.temporal.io/server/service/history/queues"
|
|
"go.temporal.io/server/service/history/tasks"
|
|
wcache "go.temporal.io/server/service/history/workflow/cache"
|
|
"go.uber.org/fx"
|
|
)
|
|
|
|
const (
|
|
transferQueuePersistenceMaxRPSRatio = 0.3
|
|
)
|
|
|
|
type (
|
|
transferQueueFactoryParams struct {
|
|
fx.In
|
|
|
|
QueueFactoryBaseParams
|
|
|
|
ClientBean client.Bean
|
|
SdkClientFactory sdk.ClientFactory
|
|
HistoryRawClient resource.HistoryRawClient
|
|
MatchingRawClient resource.MatchingRawClient
|
|
VisibilityManager manager.VisibilityManager
|
|
}
|
|
|
|
transferQueueFactory struct {
|
|
transferQueueFactoryParams
|
|
QueueFactoryBase
|
|
}
|
|
)
|
|
|
|
func NewTransferQueueFactory(
|
|
params transferQueueFactoryParams,
|
|
) QueueFactory {
|
|
return &transferQueueFactory{
|
|
transferQueueFactoryParams: params,
|
|
QueueFactoryBase: QueueFactoryBase{
|
|
HostScheduler: queues.NewScheduler(
|
|
params.ClusterMetadata.GetCurrentClusterName(),
|
|
queues.SchedulerOptions{
|
|
WorkerCount: params.Config.TransferProcessorSchedulerWorkerCount,
|
|
ActiveNamespaceWeights: params.Config.TransferProcessorSchedulerActiveRoundRobinWeights,
|
|
StandbyNamespaceWeights: params.Config.TransferProcessorSchedulerStandbyRoundRobinWeights,
|
|
InactiveNamespaceDeletionDelay: params.Config.TaskSchedulerInactiveChannelDeletionDelay,
|
|
},
|
|
params.NamespaceRegistry,
|
|
params.Logger,
|
|
),
|
|
HostPriorityAssigner: queues.NewPriorityAssigner(),
|
|
HostReaderRateLimiter: queues.NewReaderPriorityRateLimiter(
|
|
NewHostRateLimiterRateFn(
|
|
params.Config.TransferProcessorMaxPollHostRPS,
|
|
params.Config.PersistenceMaxQPS,
|
|
transferQueuePersistenceMaxRPSRatio,
|
|
),
|
|
int64(params.Config.TransferQueueMaxReaderCount()),
|
|
),
|
|
Tracer: params.TracerProvider.Tracer(telemetry.ComponentQueueTransfer),
|
|
},
|
|
}
|
|
}
|
|
|
|
func (f *transferQueueFactory) CreateQueue(
|
|
shardContext historyi.ShardContext,
|
|
workflowCache wcache.Cache,
|
|
) queues.Queue {
|
|
logger := log.With(shardContext.GetLogger(), tag.ComponentTransferQueue)
|
|
metricsHandler := f.MetricsHandler.WithTags(metrics.OperationTag(metrics.OperationTransferQueueProcessorScope))
|
|
|
|
currentClusterName := f.ClusterMetadata.GetCurrentClusterName()
|
|
|
|
var shardScheduler = f.HostScheduler
|
|
if f.Config.TaskSchedulerEnableRateLimiter() {
|
|
shardScheduler = queues.NewRateLimitedScheduler(
|
|
f.HostScheduler,
|
|
queues.RateLimitedSchedulerOptions{
|
|
EnableShadowMode: f.Config.TaskSchedulerEnableRateLimiterShadowMode,
|
|
StartupDelay: f.Config.TaskSchedulerRateLimiterStartupDelay,
|
|
},
|
|
currentClusterName,
|
|
f.NamespaceRegistry,
|
|
f.SchedulerRateLimiter,
|
|
f.TimeSource,
|
|
logger,
|
|
metricsHandler,
|
|
)
|
|
}
|
|
|
|
rescheduler := queues.NewRescheduler(
|
|
shardScheduler,
|
|
shardContext.GetTimeSource(),
|
|
logger,
|
|
metricsHandler,
|
|
)
|
|
|
|
activeExecutor := newTransferQueueActiveTaskExecutor(
|
|
shardContext,
|
|
workflowCache,
|
|
f.SdkClientFactory,
|
|
logger,
|
|
f.MetricsHandler,
|
|
f.Config,
|
|
f.HistoryRawClient,
|
|
f.MatchingRawClient,
|
|
f.VisibilityManager,
|
|
)
|
|
|
|
standbyExecutor := newTransferQueueStandbyTaskExecutor(
|
|
shardContext,
|
|
workflowCache,
|
|
logger,
|
|
f.MetricsHandler,
|
|
currentClusterName,
|
|
f.HistoryRawClient,
|
|
f.MatchingRawClient,
|
|
f.VisibilityManager,
|
|
f.ClientBean,
|
|
)
|
|
|
|
executor := queues.NewActiveStandbyExecutor(
|
|
currentClusterName,
|
|
f.NamespaceRegistry,
|
|
activeExecutor,
|
|
standbyExecutor,
|
|
logger,
|
|
)
|
|
if f.ExecutorWrapper != nil {
|
|
executor = f.ExecutorWrapper.Wrap(executor)
|
|
}
|
|
|
|
factory := queues.NewExecutableFactory(
|
|
executor,
|
|
shardScheduler,
|
|
rescheduler,
|
|
f.HostPriorityAssigner,
|
|
shardContext.GetTimeSource(),
|
|
shardContext.GetNamespaceRegistry(),
|
|
shardContext.GetClusterMetadata(),
|
|
logger,
|
|
metricsHandler,
|
|
f.Tracer,
|
|
f.DLQWriter,
|
|
f.Config.TaskDLQEnabled,
|
|
f.Config.TaskDLQUnexpectedErrorAttempts,
|
|
f.Config.TaskDLQInternalErrors,
|
|
f.Config.TaskDLQErrorPattern,
|
|
)
|
|
return queues.NewImmediateQueue(
|
|
shardContext,
|
|
tasks.CategoryTransfer,
|
|
shardScheduler,
|
|
rescheduler,
|
|
&queues.Options{
|
|
ReaderOptions: queues.ReaderOptions{
|
|
BatchSize: f.Config.TransferTaskBatchSize,
|
|
MaxPendingTasksCount: f.Config.QueuePendingTaskMaxCount,
|
|
PollBackoffInterval: f.Config.TransferProcessorPollBackoffInterval,
|
|
MaxPredicateSize: f.Config.QueueMaxPredicateSize,
|
|
},
|
|
MonitorOptions: queues.MonitorOptions{
|
|
PendingTasksCriticalCount: f.Config.QueuePendingTaskCriticalCount,
|
|
ReaderStuckCriticalAttempts: f.Config.QueueReaderStuckCriticalAttempts,
|
|
SliceCountCriticalThreshold: f.Config.QueueCriticalSlicesCount,
|
|
},
|
|
MaxPollRPS: f.Config.TransferProcessorMaxPollRPS,
|
|
MaxPollInterval: f.Config.TransferProcessorMaxPollInterval,
|
|
MaxPollIntervalJitterCoefficient: f.Config.TransferProcessorMaxPollIntervalJitterCoefficient,
|
|
CheckpointInterval: f.Config.TransferProcessorUpdateAckInterval,
|
|
CheckpointIntervalJitterCoefficient: f.Config.TransferProcessorUpdateAckIntervalJitterCoefficient,
|
|
MaxReaderCount: f.Config.TransferQueueMaxReaderCount,
|
|
},
|
|
f.HostReaderRateLimiter,
|
|
queues.GrouperNamespaceID{},
|
|
logger,
|
|
metricsHandler,
|
|
factory,
|
|
)
|
|
}
|