mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
1e31c76ed1
## What changed? Cherry-pick versioning PRs - #9168 - Cache for system protection - #9262 - Cache for system protection - #9239 - Critical PR to enable sending `TargetVersionChanged` flag for Upgrade-on-CaN feature - #9147 - Tracks version drainage properly when version receives workflows via `VersioningOverride`. Needed for automated worker controllers to correctly scale versioned workers that received workflows via `VersioningOverride`. - #9300 - Needed for `approximate_backlog_count` metric to track Current and Ramping version tasks correctly - #9316 - Needed for `approximate_backlog_count` metric to track Current and Ramping version tasks correctly - #8957 - Contains minor metric improvement. Included because it adds a test harness that is used in the two metrics PRs above - #9250 - Bug fix of task rescheduling edge case during AutoUpgrade Transition ## Why? For OSS v1.30.2 ## How did you test it? - [x] built - [ ] run locally and tested manually - [x] covered by existing tests - [ ] added new unit test(s) - [ ] added new functional test(s) --- > [!NOTE] > **Medium Risk** > Touches history/matching worker-versioning paths, adding new caches and changing workflow task/start handling and backlog metric emission; incorrect caching or signaling could affect dispatch/upgrade behavior and observability. > > **Overview** > Adds new worker-versioning protections and upgrade signaling: workflow task started events now persist a `workflow_task_target_worker_deployment_version_changed` flag (and emit a new `workflow_target_version_changed_count` metric) under a new `EnableSendTargetVersionChanged` dynamic config. > > Introduces two new caches with metrics and dynamic config knobs: a `RoutingInfoCache` to avoid repeated `GetTaskQueueUserData` lookups during activity start/transition logic, and a `ReactivationSignalCache` plus `EnableVersionReactivationSignals` to dedupe and asynchronously send “reactivation” signals when workflows are pinned (via start/signal-with-start/reset/update-options) to potentially drained/inactive worker versions. > > Extends matching backlog metrics to support version-attributed reporting by adding `BacklogMetricsEmitInterval` and switching queue DB emission to *physical* backlog gauges (`physical_approximate_backlog_*`) when attribution is enabled, while keeping legacy gauges when disabled. > > Adds frontend scaffolding for a new visibility RPC `CountSchedules` (client plumbing, interception/metadata/quota wiring) but leaves the frontend handler unimplemented, and bumps `go.temporal.io/api` to `v1.62.2`. > > <sup>Written by [Cursor Bugbot]() for commit cb8ae14c86666654c51dd66ae91cb225722e1ba7. This will update automatically on new commits. Configure [here]().</sup> --------- Co-authored-by: Shahab Tajik <[email protected]> Co-authored-by: Shivam <[email protected]>
339 lines
10 KiB
Go
339 lines
10 KiB
Go
//nolint:staticcheck
|
|
package worker_versioning
|
|
|
|
import (
|
|
"sync"
|
|
"testing"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
enumspb "go.temporal.io/api/enums/v1"
|
|
deploymentspb "go.temporal.io/server/api/deployment/v1"
|
|
"go.temporal.io/server/common/cache"
|
|
"go.temporal.io/server/common/metrics"
|
|
"go.temporal.io/server/common/metrics/metricstest"
|
|
)
|
|
|
|
func TestRoutingInfoCache_GetPut(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
metricsHandler := metricstest.NewCaptureHandler()
|
|
c := cache.NewLRU(10, metrics.NoopMetricsHandler)
|
|
routingCache := NewRoutingInfoCache(c, metricsHandler)
|
|
|
|
namespaceID := "test-namespace"
|
|
taskQueue := "test-task-queue"
|
|
taskQueueType := enumspb.TASK_QUEUE_TYPE_WORKFLOW
|
|
|
|
currentVersion := &deploymentspb.WorkerDeploymentVersion{
|
|
DeploymentName: "deployment-1",
|
|
BuildId: "build-1",
|
|
}
|
|
currentRevNum := int64(100)
|
|
|
|
rampingVersion := &deploymentspb.WorkerDeploymentVersion{
|
|
DeploymentName: "deployment-2",
|
|
BuildId: "build-2",
|
|
}
|
|
rampPercentage := float32(0.3)
|
|
rampingRevNum := int64(200)
|
|
|
|
// Test cache miss
|
|
_, ok := routingCache.Get(namespaceID, taskQueue, taskQueueType)
|
|
assert.False(t, ok, "Cache should be empty initially")
|
|
|
|
// Test Put
|
|
routingCache.Put(
|
|
namespaceID,
|
|
taskQueue,
|
|
taskQueueType,
|
|
currentVersion,
|
|
currentRevNum,
|
|
rampingVersion,
|
|
rampPercentage,
|
|
rampingRevNum,
|
|
)
|
|
|
|
// Test cache hit
|
|
result, ok := routingCache.Get(
|
|
namespaceID,
|
|
taskQueue,
|
|
taskQueueType,
|
|
)
|
|
require.True(t, ok, "Cache should contain the entry")
|
|
assert.Equal(t, currentVersion, result.Current)
|
|
assert.Equal(t, currentRevNum, result.CurrentRevisionNumber)
|
|
assert.Equal(t, rampingVersion, result.Ramping)
|
|
assert.InDelta(t, rampPercentage, result.RampPercentage, 0.0001)
|
|
assert.Equal(t, rampingRevNum, result.RampingRevisionNumber)
|
|
}
|
|
|
|
func TestRoutingInfoCache_DifferentKeys(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
metricsHandler := metricstest.NewCaptureHandler()
|
|
c := cache.NewLRU(10, metrics.NoopMetricsHandler)
|
|
routingCache := NewRoutingInfoCache(c, metricsHandler)
|
|
|
|
// Setup two different cache entries
|
|
namespace1 := "namespace-1"
|
|
namespace2 := "namespace-2"
|
|
taskQueue := "test-queue"
|
|
|
|
version1 := &deploymentspb.WorkerDeploymentVersion{
|
|
DeploymentName: "deployment-1",
|
|
BuildId: "build-1",
|
|
}
|
|
version2 := &deploymentspb.WorkerDeploymentVersion{
|
|
DeploymentName: "deployment-2",
|
|
BuildId: "build-2",
|
|
}
|
|
|
|
routingCache.Put(namespace1, taskQueue, enumspb.TASK_QUEUE_TYPE_WORKFLOW, version1, 100, nil, 0, 0)
|
|
routingCache.Put(namespace2, taskQueue, enumspb.TASK_QUEUE_TYPE_WORKFLOW, version2, 200, nil, 0, 0)
|
|
|
|
// Verify both entries exist and are independent
|
|
result1, ok1 := routingCache.Get(namespace1, taskQueue, enumspb.TASK_QUEUE_TYPE_WORKFLOW)
|
|
assert.True(t, ok1)
|
|
assert.Equal(t, version1, result1.Current)
|
|
assert.Equal(t, int64(100), result1.CurrentRevisionNumber)
|
|
|
|
result2, ok2 := routingCache.Get(namespace2, taskQueue, enumspb.TASK_QUEUE_TYPE_WORKFLOW)
|
|
assert.True(t, ok2)
|
|
assert.Equal(t, version2, result2.Current)
|
|
assert.Equal(t, int64(200), result2.CurrentRevisionNumber)
|
|
}
|
|
|
|
func TestRoutingInfoCache_TaskQueueTypes(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
metricsHandler := metricstest.NewCaptureHandler()
|
|
c := cache.NewLRU(10, metrics.NoopMetricsHandler)
|
|
routingCache := NewRoutingInfoCache(c, metricsHandler)
|
|
|
|
namespace := "namespace-1"
|
|
taskQueue := "test-queue"
|
|
|
|
workflowVersion := &deploymentspb.WorkerDeploymentVersion{
|
|
DeploymentName: "workflow-deployment",
|
|
BuildId: "workflow-build",
|
|
}
|
|
activityVersion := &deploymentspb.WorkerDeploymentVersion{
|
|
DeploymentName: "activity-deployment",
|
|
BuildId: "activity-build",
|
|
}
|
|
|
|
// Put different versions for different task queue types
|
|
routingCache.Put(namespace, taskQueue, enumspb.TASK_QUEUE_TYPE_WORKFLOW, workflowVersion, 100, nil, 0, 0)
|
|
routingCache.Put(namespace, taskQueue, enumspb.TASK_QUEUE_TYPE_ACTIVITY, activityVersion, 200, nil, 0, 0)
|
|
|
|
// Verify workflow task queue type
|
|
result, ok := routingCache.Get(namespace, taskQueue, enumspb.TASK_QUEUE_TYPE_WORKFLOW)
|
|
assert.True(t, ok)
|
|
assert.Equal(t, workflowVersion, result.Current)
|
|
assert.Equal(t, int64(100), result.CurrentRevisionNumber)
|
|
|
|
// Verify activity task queue type
|
|
result, ok = routingCache.Get(namespace, taskQueue, enumspb.TASK_QUEUE_TYPE_ACTIVITY)
|
|
assert.True(t, ok)
|
|
assert.Equal(t, activityVersion, result.Current)
|
|
assert.Equal(t, int64(200), result.CurrentRevisionNumber)
|
|
}
|
|
|
|
func TestRoutingInfoCache_NilValues(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
metricsHandler := metricstest.NewCaptureHandler()
|
|
c := cache.NewLRU(10, metrics.NoopMetricsHandler)
|
|
routingCache := NewRoutingInfoCache(c, metricsHandler)
|
|
|
|
namespace := "namespace-1"
|
|
taskQueue := "test-queue"
|
|
taskQueueType := enumspb.TASK_QUEUE_TYPE_WORKFLOW
|
|
|
|
// Put with nil current and ramping (unversioned task queue)
|
|
routingCache.Put(namespace, taskQueue, taskQueueType, nil, 0, nil, 0, 0)
|
|
|
|
// Verify we can retrieve nil values
|
|
result, ok := routingCache.Get(namespace, taskQueue, taskQueueType)
|
|
assert.True(t, ok)
|
|
assert.Nil(t, result.Current)
|
|
assert.Equal(t, int64(0), result.CurrentRevisionNumber)
|
|
assert.Nil(t, result.Ramping)
|
|
assert.InDelta(t, float32(0), result.RampPercentage, 0.0001)
|
|
assert.Equal(t, int64(0), result.RampingRevisionNumber)
|
|
}
|
|
|
|
func TestRoutingInfoCache_UpdateExisting(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
metricsHandler := metricstest.NewCaptureHandler()
|
|
c := cache.NewLRU(10, metrics.NoopMetricsHandler)
|
|
routingCache := NewRoutingInfoCache(c, metricsHandler)
|
|
|
|
namespace := "namespace-1"
|
|
taskQueue := "test-queue"
|
|
taskQueueType := enumspb.TASK_QUEUE_TYPE_WORKFLOW
|
|
|
|
version1 := &deploymentspb.WorkerDeploymentVersion{
|
|
DeploymentName: "deployment-1",
|
|
BuildId: "build-1",
|
|
}
|
|
version2 := &deploymentspb.WorkerDeploymentVersion{
|
|
DeploymentName: "deployment-2",
|
|
BuildId: "build-2",
|
|
}
|
|
|
|
// Put initial value
|
|
routingCache.Put(namespace, taskQueue, taskQueueType, version1, 100, nil, 0, 0)
|
|
|
|
// Update with new value
|
|
routingCache.Put(namespace, taskQueue, taskQueueType, version2, 200, nil, 0, 0)
|
|
|
|
// Verify updated value
|
|
result, ok := routingCache.Get(namespace, taskQueue, taskQueueType)
|
|
assert.True(t, ok)
|
|
assert.Equal(t, version2, result.Current)
|
|
assert.Equal(t, int64(200), result.CurrentRevisionNumber)
|
|
}
|
|
|
|
func TestRoutingInfoCache_WithRamping(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
metricsHandler := metricstest.NewCaptureHandler()
|
|
c := cache.NewLRU(10, metrics.NoopMetricsHandler)
|
|
routingCache := NewRoutingInfoCache(c, metricsHandler)
|
|
|
|
namespace := "namespace-1"
|
|
taskQueue := "test-queue"
|
|
taskQueueType := enumspb.TASK_QUEUE_TYPE_WORKFLOW
|
|
|
|
currentVersion := &deploymentspb.WorkerDeploymentVersion{
|
|
DeploymentName: "current-deployment",
|
|
BuildId: "current-build",
|
|
}
|
|
rampingVersion := &deploymentspb.WorkerDeploymentVersion{
|
|
DeploymentName: "ramping-deployment",
|
|
BuildId: "ramping-build",
|
|
}
|
|
|
|
// Put with ramping info
|
|
routingCache.Put(
|
|
namespace,
|
|
taskQueue,
|
|
taskQueueType,
|
|
currentVersion,
|
|
150,
|
|
rampingVersion,
|
|
0.25,
|
|
250,
|
|
)
|
|
|
|
// Verify all values including ramping
|
|
result, ok := routingCache.Get(namespace, taskQueue, taskQueueType)
|
|
assert.True(t, ok)
|
|
assert.Equal(t, currentVersion, result.Current)
|
|
assert.Equal(t, int64(150), result.CurrentRevisionNumber)
|
|
assert.Equal(t, rampingVersion, result.Ramping)
|
|
assert.InDelta(t, float32(0.25), result.RampPercentage, 0.0001)
|
|
assert.Equal(t, int64(250), result.RampingRevisionNumber)
|
|
}
|
|
|
|
func TestRoutingInfoCache_Concurrent(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
metricsHandler := metricstest.NewCaptureHandler()
|
|
c := cache.NewLRU(100, metrics.NoopMetricsHandler)
|
|
routingCache := NewRoutingInfoCache(c, metricsHandler)
|
|
|
|
namespace := "namespace-1"
|
|
taskQueue := "test-queue"
|
|
taskQueueType := enumspb.TASK_QUEUE_TYPE_WORKFLOW
|
|
|
|
var wg sync.WaitGroup
|
|
numGoroutines := 10
|
|
numOperations := 100
|
|
|
|
// Concurrent writes
|
|
for i := 0; i < numGoroutines; i++ {
|
|
idx := i
|
|
wg.Add(1) //nolint:revive // use-waitgroup-go: standard sync.WaitGroup doesn't have Go() method
|
|
go func() {
|
|
defer wg.Done()
|
|
for j := 0; j < numOperations; j++ {
|
|
version := &deploymentspb.WorkerDeploymentVersion{
|
|
DeploymentName: "deployment",
|
|
BuildId: "build",
|
|
}
|
|
routingCache.Put(
|
|
namespace,
|
|
taskQueue,
|
|
taskQueueType,
|
|
version,
|
|
int64(idx*numOperations+j),
|
|
nil,
|
|
0,
|
|
0,
|
|
)
|
|
}
|
|
}()
|
|
}
|
|
|
|
// Concurrent reads
|
|
for i := 0; i < numGoroutines; i++ {
|
|
wg.Add(1) //nolint:revive // use-waitgroup-go: standard sync.WaitGroup doesn't have Go() method
|
|
go func() {
|
|
defer wg.Done()
|
|
for j := 0; j < numOperations; j++ {
|
|
routingCache.Get(namespace, taskQueue, taskQueueType)
|
|
}
|
|
}()
|
|
}
|
|
|
|
wg.Wait()
|
|
|
|
// Verify cache is in a consistent state
|
|
_, ok := routingCache.Get(namespace, taskQueue, taskQueueType)
|
|
assert.True(t, ok, "Cache should contain an entry after concurrent operations")
|
|
}
|
|
|
|
func TestRoutingInfoCache_Metrics(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
metricsHandler := metricstest.NewCaptureHandler()
|
|
capture := metricsHandler.StartCapture()
|
|
|
|
c := cache.NewLRU(10, metrics.NoopMetricsHandler)
|
|
routingCache := NewRoutingInfoCache(c, metricsHandler)
|
|
|
|
namespace := "namespace-1"
|
|
taskQueue := "test-queue"
|
|
taskQueueType := enumspb.TASK_QUEUE_TYPE_WORKFLOW
|
|
|
|
version := &deploymentspb.WorkerDeploymentVersion{
|
|
DeploymentName: "deployment",
|
|
BuildId: "build",
|
|
}
|
|
|
|
// Test cache miss - should record cache request and cache miss
|
|
routingCache.Get(namespace, taskQueue, taskQueueType)
|
|
|
|
// Test Put - should record cache request
|
|
routingCache.Put(namespace, taskQueue, taskQueueType, version, 100, nil, 0, 0)
|
|
|
|
// Test cache hit - should record cache request only (no miss)
|
|
routingCache.Get(namespace, taskQueue, taskQueueType)
|
|
|
|
snapshot := capture.Snapshot()
|
|
|
|
// Verify cache requests were recorded
|
|
cacheRequests, ok := snapshot[metrics.CacheRequests.Name()]
|
|
assert.True(t, ok, "CacheRequests metric should be present")
|
|
assert.NotEmpty(t, cacheRequests, "CacheRequests should have entries")
|
|
|
|
// Verify cache misses were recorded
|
|
cacheMisses, ok := snapshot[metrics.CacheMissCounter.Name()]
|
|
assert.True(t, ok, "CacheMissCounter metric should be present")
|
|
assert.NotEmpty(t, cacheMisses, "CacheMissCounter should have entries")
|
|
}
|