mirror of
https://github.com/hanzoai/tasks.git
synced 2026-11-02 12:22:18 +00:00
Jacob MoodyandGitHub
7d1a144404
## What changed? Add a new dynamic config for Auto Enabling fairness and priority if we see the relevant tasks coming in. ## Why? Seamlessly start to transition users who start using the fields over to the new code path. ## How did you test it? - [X] built - [x] run locally and tested manually - [x] covered by existing tests - [x] added new unit test(s) - [x] added new functional test(s) ## Potential risks Due to storing this in the userdata we are using that interface a bit more, we also need to change the initialization such that we start it before being able to substantiate the defaultQ, this change in initialization might have unintended side effects that I'm not currently seeing.
91 lines
5.3 KiB
Go
91 lines
5.3 KiB
Go
package configs
|
|
|
|
import (
|
|
"go.temporal.io/server/common/dynamicconfig"
|
|
"go.temporal.io/server/common/headers"
|
|
"go.temporal.io/server/common/quotas"
|
|
)
|
|
|
|
const (
|
|
// OperatorPriority is used to give precedence to calls coming from web UI or tctl
|
|
OperatorPriority = 0
|
|
)
|
|
|
|
var (
|
|
APIToPriority = map[string]int{
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/AddActivityTask": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/AddWorkflowTask": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/CancelOutstandingPoll": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/DescribeTaskQueue": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/ListTaskQueuePartitions": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/PollActivityTaskQueue": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/PollWorkflowTaskQueue": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/QueryWorkflow": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/RespondQueryTaskCompleted": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/GetWorkerBuildIdCompatibility": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/UpdateWorkerBuildIdCompatibility": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/GetTaskQueueUserData": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/ApplyTaskQueueUserDataReplicationEvent": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/GetBuildIdTaskQueueMapping": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/ForceUnloadTaskQueuePartition": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/ForceUnloadTaskQueue": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/ForceLoadTaskQueuePartition": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/UpdateTaskQueueUserData": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/ReplicateTaskQueueUserData": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/CheckTaskQueueUserDataPropagation": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/PollNexusTaskQueue": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/RespondNexusTaskCompleted": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/RespondNexusTaskFailed": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/DispatchNexusTask": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/CreateNexusEndpoint": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/UpdateNexusEndpoint": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/ListNexusEndpoints": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/DeleteNexusEndpoint": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/UpdateWorkerVersioningRules": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/GetWorkerVersioningRules": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/DescribeTaskQueuePartition": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/DescribeVersionedTaskQueues": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/SyncDeploymentUserData": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/RecordWorkerHeartbeat": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/ListWorkers": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/UpdateTaskQueueConfig": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/DescribeWorker": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/UpdateFairnessState": 1,
|
|
"/temporal.server.api.matchingservice.v1.MatchingService/CheckTaskQueueVersionMembership": 1,
|
|
}
|
|
|
|
APIPrioritiesOrdered = []int{0, 1}
|
|
)
|
|
|
|
func NewPriorityRateLimiter(
|
|
rateFn quotas.RateFn,
|
|
operatorRPSRatio dynamicconfig.FloatPropertyFn,
|
|
) quotas.RequestRateLimiter {
|
|
rateLimiters := make(map[int]quotas.RequestRateLimiter)
|
|
for priority := range APIPrioritiesOrdered {
|
|
if priority == OperatorPriority {
|
|
rateLimiters[priority] = quotas.NewRequestRateLimiterAdapter(quotas.NewDefaultIncomingRateLimiter(operatorRateFn(rateFn, operatorRPSRatio)))
|
|
} else {
|
|
rateLimiters[priority] = quotas.NewRequestRateLimiterAdapter(quotas.NewDefaultIncomingRateLimiter(rateFn))
|
|
}
|
|
}
|
|
return quotas.NewPriorityRateLimiter(func(req quotas.Request) int {
|
|
if req.CallerType == headers.CallerTypeOperator {
|
|
return OperatorPriority
|
|
}
|
|
if priority, ok := APIToPriority[req.API]; ok {
|
|
return priority
|
|
}
|
|
return APIPrioritiesOrdered[len(APIPrioritiesOrdered)-1]
|
|
}, rateLimiters)
|
|
}
|
|
|
|
func operatorRateFn(
|
|
rateFn quotas.RateFn,
|
|
operatorRPSRatio dynamicconfig.FloatPropertyFn,
|
|
) quotas.RateFn {
|
|
return func() float64 {
|
|
return operatorRPSRatio() * rateFn()
|
|
}
|
|
}
|