mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
ast2023andGitHub
fdea23d608
<!-- Describe what has changed in this PR --> **What changed?** To find number of per namespace workers we now use ringpop LookupN method instead of sequentially calling Lookup. The lookup key also changed: worker id was dropped from the hashed string. Added util.SliceMap function and renamed util.SliceReduce to util.SliceFold <!-- Tell your future self why have you made these changes --> **Why?** more evenly distribute work across ring <!-- How have you verified this change? Tested locally? Added a unit test? Checked in staging env? --> **How did you test it?** local testing only <!-- Assuming the worst case, what can be broken when deploying this change to production? --> **Potential risks** not sure <!-- Is this PR a hotfix candidate or require that a notification be sent to the broader community? (Yes/No) --> **Is hotfix candidate?** no
546 lines
20 KiB
Go
546 lines
20 KiB
Go
// The MIT License
|
|
//
|
|
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
|
|
//
|
|
// Copyright (c) 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 worker
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/golang/mock/gomock"
|
|
"github.com/stretchr/testify/suite"
|
|
enumspb "go.temporal.io/api/enums/v1"
|
|
sdkclient "go.temporal.io/sdk/client"
|
|
sdkworker "go.temporal.io/sdk/worker"
|
|
|
|
persistencespb "go.temporal.io/server/api/persistence/v1"
|
|
"go.temporal.io/server/common/cluster"
|
|
"go.temporal.io/server/common/log"
|
|
"go.temporal.io/server/common/membership"
|
|
"go.temporal.io/server/common/namespace"
|
|
"go.temporal.io/server/common/primitives"
|
|
"go.temporal.io/server/common/sdk"
|
|
"go.temporal.io/server/common/testing/mocksdk"
|
|
workercommon "go.temporal.io/server/service/worker/common"
|
|
)
|
|
|
|
type perNsWorkerManagerSuite struct {
|
|
suite.Suite
|
|
|
|
controller *gomock.Controller
|
|
logger log.Logger
|
|
cfactory *sdk.MockClientFactory
|
|
registry *namespace.MockRegistry
|
|
hostInfo membership.HostInfo
|
|
serviceResolver *membership.MockServiceResolver
|
|
|
|
cmp1 *workercommon.MockPerNSWorkerComponent
|
|
cmp2 *workercommon.MockPerNSWorkerComponent
|
|
|
|
manager *perNamespaceWorkerManager
|
|
}
|
|
|
|
func TestPerNsWorkerManager(t *testing.T) {
|
|
suite.Run(t, new(perNsWorkerManagerSuite))
|
|
}
|
|
|
|
func (s *perNsWorkerManagerSuite) SetupTest() {
|
|
s.controller = gomock.NewController(s.T())
|
|
|
|
s.logger = log.NewTestLogger()
|
|
s.cfactory = sdk.NewMockClientFactory(s.controller)
|
|
s.registry = namespace.NewMockRegistry(s.controller)
|
|
s.hostInfo = membership.NewHostInfoFromAddress("self")
|
|
s.serviceResolver = membership.NewMockServiceResolver(s.controller)
|
|
s.cmp1 = workercommon.NewMockPerNSWorkerComponent(s.controller)
|
|
s.cmp2 = workercommon.NewMockPerNSWorkerComponent(s.controller)
|
|
|
|
s.manager = NewPerNamespaceWorkerManager(perNamespaceWorkerManagerInitParams{
|
|
Logger: s.logger,
|
|
SdkClientFactory: s.cfactory,
|
|
NamespaceRegistry: s.registry,
|
|
HostName: "self",
|
|
Config: &Config{
|
|
PerNamespaceWorkerCount: func(ns string) int {
|
|
return max(1, map[string]int{"ns1": 1, "ns2": 2, "ns3": 6}[ns])
|
|
},
|
|
PerNamespaceWorkerOptions: func(ns string) map[string]any {
|
|
switch ns {
|
|
case "ns1":
|
|
return map[string]any{
|
|
"MaxConcurrentWorkflowTaskPollers": 100,
|
|
}
|
|
case "ns2":
|
|
return map[string]any{
|
|
"WorkerLocalActivitiesPerSecond": 200.0,
|
|
"StickyScheduleToStartTimeout": "7.5s",
|
|
}
|
|
default:
|
|
return map[string]any{}
|
|
}
|
|
},
|
|
},
|
|
Components: []workercommon.PerNSWorkerComponent{s.cmp1, s.cmp2},
|
|
ClusterMetadata: cluster.NewMetadataForTest(cluster.NewTestClusterMetadataConfig(false, true)),
|
|
})
|
|
s.manager.initialRetry = 1 * time.Millisecond
|
|
|
|
s.registry.EXPECT().RegisterStateChangeCallback(gomock.Any(), gomock.Any())
|
|
s.serviceResolver.EXPECT().AddListener(gomock.Any(), gomock.Any())
|
|
|
|
s.manager.Start(s.hostInfo, s.serviceResolver)
|
|
}
|
|
|
|
func (s *perNsWorkerManagerSuite) TearDownTest() {
|
|
s.registry.EXPECT().UnregisterStateChangeCallback(gomock.Any())
|
|
s.serviceResolver.EXPECT().RemoveListener(gomock.Any())
|
|
s.manager.Stop()
|
|
s.controller.Finish()
|
|
}
|
|
|
|
func (s *perNsWorkerManagerSuite) TestDisabled() {
|
|
ns := testns("ns1", enumspb.NAMESPACE_STATE_REGISTERED)
|
|
|
|
s.cmp1.EXPECT().DedicatedWorkerOptions(gomock.Any()).Return(&workercommon.PerNSDedicatedWorkerOptions{
|
|
Enabled: false,
|
|
}).AnyTimes()
|
|
s.cmp2.EXPECT().DedicatedWorkerOptions(gomock.Any()).Return(&workercommon.PerNSDedicatedWorkerOptions{
|
|
Enabled: false,
|
|
}).AnyTimes()
|
|
|
|
s.manager.namespaceCallback(ns, false)
|
|
time.Sleep(50 * time.Millisecond)
|
|
}
|
|
|
|
func (s *perNsWorkerManagerSuite) TestInactive() {
|
|
ns := testInactiveNS("ns1", enumspb.NAMESPACE_STATE_REGISTERED)
|
|
|
|
s.cmp1.EXPECT().DedicatedWorkerOptions(gomock.Any()).Return(&workercommon.PerNSDedicatedWorkerOptions{
|
|
Enabled: true,
|
|
}).AnyTimes()
|
|
s.cmp2.EXPECT().DedicatedWorkerOptions(gomock.Any()).Return(&workercommon.PerNSDedicatedWorkerOptions{
|
|
Enabled: true,
|
|
}).AnyTimes()
|
|
|
|
s.manager.namespaceCallback(ns, false)
|
|
time.Sleep(50 * time.Millisecond)
|
|
}
|
|
|
|
func (s *perNsWorkerManagerSuite) TestEnabledButResolvedToOther() {
|
|
ns := testns("ns1", enumspb.NAMESPACE_STATE_REGISTERED)
|
|
|
|
s.cmp1.EXPECT().DedicatedWorkerOptions(gomock.Any()).Return(&workercommon.PerNSDedicatedWorkerOptions{
|
|
Enabled: true,
|
|
}).AnyTimes()
|
|
s.cmp2.EXPECT().DedicatedWorkerOptions(gomock.Any()).Return(&workercommon.PerNSDedicatedWorkerOptions{
|
|
Enabled: false,
|
|
}).AnyTimes()
|
|
|
|
s.serviceResolver.EXPECT().LookupN("ns1", 1).Return([]membership.HostInfo{membership.NewHostInfoFromAddress("other1")})
|
|
|
|
s.manager.namespaceCallback(ns, false)
|
|
// main work happens in a goroutine
|
|
time.Sleep(50 * time.Millisecond)
|
|
}
|
|
|
|
func (s *perNsWorkerManagerSuite) TestEnabled() {
|
|
ns := testns("ns1", enumspb.NAMESPACE_STATE_REGISTERED)
|
|
|
|
s.cmp1.EXPECT().DedicatedWorkerOptions(gomock.Any()).Return(&workercommon.PerNSDedicatedWorkerOptions{
|
|
Enabled: true,
|
|
}).AnyTimes()
|
|
s.cmp2.EXPECT().DedicatedWorkerOptions(gomock.Any()).Return(&workercommon.PerNSDedicatedWorkerOptions{
|
|
Enabled: false,
|
|
}).AnyTimes()
|
|
|
|
s.serviceResolver.EXPECT().LookupN("ns1", 1).Return([]membership.HostInfo{membership.NewHostInfoFromAddress("self")})
|
|
cli1 := mocksdk.NewMockClient(s.controller)
|
|
s.cfactory.EXPECT().NewClient(matchOptions("ns1")).Return(cli1)
|
|
wkr1 := mocksdk.NewMockWorker(s.controller)
|
|
s.cfactory.EXPECT().NewWorker(matchStrict{cli1}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Return(wkr1)
|
|
s.cmp1.EXPECT().Register(wkr1, ns, workercommon.RegistrationDetails{TotalWorkers: 1, Multiplicity: 1})
|
|
wkr1.EXPECT().Start()
|
|
|
|
s.manager.namespaceCallback(ns, false)
|
|
time.Sleep(50 * time.Millisecond)
|
|
|
|
// shutdown
|
|
wkr1.EXPECT().Stop()
|
|
cli1.EXPECT().Close()
|
|
}
|
|
|
|
/*
|
|
Given machine has ownership of the namespace
|
|
When name space change reported
|
|
Then worker should be started with configuration proportional to administratively set configuration ensuring
|
|
fair distribution across all the machines owning the namespace
|
|
*/
|
|
func (s *perNsWorkerManagerSuite) TestMultiplicity() {
|
|
machinesOwningNamespace := []membership.HostInfo{
|
|
membership.NewHostInfoFromAddress("other-1"),
|
|
membership.NewHostInfoFromAddress("other-2"),
|
|
membership.NewHostInfoFromAddress("self"),
|
|
}
|
|
desiredWorkersNumber := 6 // should match mock configuration in line 88
|
|
expectedMultiplicity := 2
|
|
|
|
ns := testns("ns3", enumspb.NAMESPACE_STATE_REGISTERED)
|
|
|
|
s.cmp1.EXPECT().DedicatedWorkerOptions(gomock.Any()).Return(&workercommon.PerNSDedicatedWorkerOptions{
|
|
Enabled: true,
|
|
}).AnyTimes()
|
|
s.cmp2.EXPECT().DedicatedWorkerOptions(gomock.Any()).Return(&workercommon.PerNSDedicatedWorkerOptions{
|
|
Enabled: false,
|
|
}).AnyTimes()
|
|
|
|
s.serviceResolver.EXPECT().LookupN("ns3", desiredWorkersNumber).Return(machinesOwningNamespace)
|
|
cli1 := mocksdk.NewMockClient(s.controller)
|
|
s.cfactory.EXPECT().NewClient(matchOptions("ns3")).Return(cli1)
|
|
wkr1 := mocksdk.NewMockWorker(s.controller)
|
|
s.cfactory.EXPECT().NewWorker(matchStrict{cli1}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Do(func(_, _ any, options sdkworker.Options) {
|
|
s.Equal(4, options.MaxConcurrentWorkflowTaskPollers)
|
|
s.Equal(4, options.MaxConcurrentActivityTaskPollers)
|
|
s.Equal(2000, options.MaxConcurrentWorkflowTaskExecutionSize)
|
|
s.Equal(2000, options.MaxConcurrentLocalActivityExecutionSize)
|
|
s.Equal(2000, options.MaxConcurrentActivityExecutionSize)
|
|
}).Return(wkr1)
|
|
s.cmp1.EXPECT().Register(wkr1, ns, workercommon.RegistrationDetails{TotalWorkers: desiredWorkersNumber, Multiplicity: expectedMultiplicity})
|
|
wkr1.EXPECT().Start()
|
|
|
|
s.manager.namespaceCallback(ns, false)
|
|
|
|
time.Sleep(50 * time.Millisecond)
|
|
|
|
wkr1.EXPECT().Stop()
|
|
cli1.EXPECT().Close()
|
|
}
|
|
|
|
func (s *perNsWorkerManagerSuite) TestOptions() {
|
|
ns1 := testns("ns1", enumspb.NAMESPACE_STATE_REGISTERED)
|
|
ns2 := testns("ns2", enumspb.NAMESPACE_STATE_REGISTERED)
|
|
ns3 := testns("ns3", enumspb.NAMESPACE_STATE_REGISTERED)
|
|
|
|
s.cmp1.EXPECT().DedicatedWorkerOptions(gomock.Any()).Return(&workercommon.PerNSDedicatedWorkerOptions{
|
|
Enabled: true,
|
|
}).AnyTimes()
|
|
|
|
s.cmp2.EXPECT().DedicatedWorkerOptions(gomock.Any()).Return(&workercommon.PerNSDedicatedWorkerOptions{
|
|
Enabled: false,
|
|
}).AnyTimes()
|
|
|
|
s.serviceResolver.EXPECT().LookupN("ns1", 1).Return([]membership.HostInfo{
|
|
membership.NewHostInfoFromAddress("self"),
|
|
}).AnyTimes()
|
|
s.serviceResolver.EXPECT().LookupN("ns2", 2).Return([]membership.HostInfo{
|
|
membership.NewHostInfoFromAddress("self"),
|
|
}).AnyTimes()
|
|
s.serviceResolver.EXPECT().LookupN("ns3", 6).Return([]membership.HostInfo{
|
|
membership.NewHostInfoFromAddress("self"),
|
|
}).AnyTimes()
|
|
|
|
cli1 := mocksdk.NewMockClient(s.controller)
|
|
s.cfactory.EXPECT().NewClient(matchOptions("ns1")).Return(cli1)
|
|
wkr := mocksdk.NewMockWorker(s.controller)
|
|
s.cfactory.EXPECT().NewWorker(matchStrict{cli1}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Do(func(_, _ any, options sdkworker.Options) {
|
|
s.Equal(100, options.MaxConcurrentWorkflowTaskPollers)
|
|
s.Equal(2, options.MaxConcurrentActivityTaskPollers)
|
|
s.Equal(0.0, options.WorkerLocalActivitiesPerSecond)
|
|
}).Return(wkr)
|
|
|
|
cli2 := mocksdk.NewMockClient(s.controller)
|
|
s.cfactory.EXPECT().NewClient(matchOptions("ns2")).Return(cli2)
|
|
s.cfactory.EXPECT().NewWorker(matchStrict{cli2}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Do(func(_, _ any, options sdkworker.Options) {
|
|
s.Equal(4, options.MaxConcurrentWorkflowTaskPollers)
|
|
s.Equal(200.0, options.WorkerLocalActivitiesPerSecond)
|
|
s.Equal(7500*time.Millisecond, options.StickyScheduleToStartTimeout)
|
|
}).Return(wkr)
|
|
|
|
cli3 := mocksdk.NewMockClient(s.controller)
|
|
s.cfactory.EXPECT().NewClient(matchOptions("ns3")).Return(cli3)
|
|
s.cfactory.EXPECT().NewWorker(matchStrict{cli3}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Do(func(_, _ any, options sdkworker.Options) {
|
|
s.Equal(12, options.MaxConcurrentWorkflowTaskPollers)
|
|
s.Equal(0.0, options.WorkerLocalActivitiesPerSecond)
|
|
s.Equal(0*time.Millisecond, options.StickyScheduleToStartTimeout)
|
|
}).Return(wkr)
|
|
s.cmp1.EXPECT().Register(wkr, gomock.Any(), gomock.Any()).AnyTimes()
|
|
wkr.EXPECT().Start().AnyTimes()
|
|
|
|
s.manager.namespaceCallback(ns1, false)
|
|
s.manager.namespaceCallback(ns2, false)
|
|
s.manager.namespaceCallback(ns3, false)
|
|
|
|
time.Sleep(50 * time.Millisecond)
|
|
|
|
wkr.EXPECT().Stop().AnyTimes()
|
|
cli1.EXPECT().Close()
|
|
cli2.EXPECT().Close()
|
|
cli3.EXPECT().Close()
|
|
}
|
|
|
|
func (s *perNsWorkerManagerSuite) TestTwoNamespacesTwoComponents() {
|
|
ns1 := testns("ns1", enumspb.NAMESPACE_STATE_REGISTERED)
|
|
ns2 := testns("ns2", enumspb.NAMESPACE_STATE_REGISTERED)
|
|
|
|
s.cmp1.EXPECT().DedicatedWorkerOptions(gomock.Any()).DoAndReturn(
|
|
func(ns *namespace.Namespace) *workercommon.PerNSDedicatedWorkerOptions {
|
|
return &workercommon.PerNSDedicatedWorkerOptions{Enabled: true}
|
|
}).AnyTimes()
|
|
s.cmp2.EXPECT().DedicatedWorkerOptions(gomock.Any()).DoAndReturn(
|
|
func(ns *namespace.Namespace) *workercommon.PerNSDedicatedWorkerOptions {
|
|
// only enabled on ns1
|
|
return &workercommon.PerNSDedicatedWorkerOptions{Enabled: ns.Name().String() == "ns1"}
|
|
}).AnyTimes()
|
|
|
|
s.serviceResolver.EXPECT().LookupN("ns1", 1).Return([]membership.HostInfo{membership.NewHostInfoFromAddress("self")})
|
|
s.serviceResolver.EXPECT().LookupN("ns2", 2).Return([]membership.HostInfo{membership.NewHostInfoFromAddress("self")})
|
|
|
|
cli1 := mocksdk.NewMockClient(s.controller)
|
|
cli2 := mocksdk.NewMockClient(s.controller)
|
|
|
|
wkr1 := mocksdk.NewMockWorker(s.controller)
|
|
wkr2 := mocksdk.NewMockWorker(s.controller)
|
|
|
|
s.cfactory.EXPECT().NewClient(matchOptions("ns1")).Return(cli1)
|
|
s.cfactory.EXPECT().NewClient(matchOptions("ns2")).Return(cli2)
|
|
|
|
s.cfactory.EXPECT().NewWorker(matchStrict{cli1}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Return(wkr1)
|
|
s.cfactory.EXPECT().NewWorker(matchStrict{cli2}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Return(wkr2)
|
|
|
|
s.cmp1.EXPECT().Register(wkr1, ns1, workercommon.RegistrationDetails{TotalWorkers: 1, Multiplicity: 1})
|
|
s.cmp1.EXPECT().Register(wkr2, ns2, workercommon.RegistrationDetails{TotalWorkers: 2, Multiplicity: 2})
|
|
s.cmp2.EXPECT().Register(wkr1, ns1, workercommon.RegistrationDetails{TotalWorkers: 1, Multiplicity: 1})
|
|
|
|
wkr1.EXPECT().Start()
|
|
wkr2.EXPECT().Start()
|
|
|
|
s.manager.namespaceCallback(ns1, false)
|
|
s.manager.namespaceCallback(ns2, false)
|
|
time.Sleep(50 * time.Millisecond)
|
|
|
|
wkr1.EXPECT().Stop()
|
|
wkr2.EXPECT().Stop()
|
|
cli1.EXPECT().Close()
|
|
cli2.EXPECT().Close()
|
|
}
|
|
|
|
func (s *perNsWorkerManagerSuite) TestDeleteNs() {
|
|
ns := testns("ns1", enumspb.NAMESPACE_STATE_REGISTERED)
|
|
|
|
s.cmp1.EXPECT().DedicatedWorkerOptions(gomock.Any()).Return(&workercommon.PerNSDedicatedWorkerOptions{
|
|
Enabled: true,
|
|
}).AnyTimes()
|
|
s.cmp2.EXPECT().DedicatedWorkerOptions(gomock.Any()).Return(&workercommon.PerNSDedicatedWorkerOptions{
|
|
Enabled: false,
|
|
}).AnyTimes()
|
|
|
|
s.serviceResolver.EXPECT().LookupN("ns1", 1).Return([]membership.HostInfo{membership.NewHostInfoFromAddress("self")})
|
|
cli1 := mocksdk.NewMockClient(s.controller)
|
|
s.cfactory.EXPECT().NewClient(matchOptions("ns1")).Return(cli1)
|
|
wkr1 := mocksdk.NewMockWorker(s.controller)
|
|
s.cfactory.EXPECT().NewWorker(matchStrict{cli1}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Return(wkr1)
|
|
s.cmp1.EXPECT().Register(wkr1, ns, workercommon.RegistrationDetails{TotalWorkers: 1, Multiplicity: 1})
|
|
wkr1.EXPECT().Start()
|
|
|
|
s.manager.namespaceCallback(ns, false)
|
|
time.Sleep(50 * time.Millisecond)
|
|
|
|
// now delete it
|
|
nsDeleted := testns("ns1", enumspb.NAMESPACE_STATE_DELETED)
|
|
wkr1.EXPECT().Stop()
|
|
cli1.EXPECT().Close()
|
|
s.manager.namespaceCallback(nsDeleted, false)
|
|
time.Sleep(50 * time.Millisecond)
|
|
|
|
// restore it
|
|
nsRestored := testns("ns1", enumspb.NAMESPACE_STATE_REGISTERED)
|
|
s.serviceResolver.EXPECT().LookupN("ns1", 1).Return([]membership.HostInfo{membership.NewHostInfoFromAddress("self")})
|
|
cli2 := mocksdk.NewMockClient(s.controller)
|
|
s.cfactory.EXPECT().NewClient(matchOptions("ns1")).Return(cli2)
|
|
wkr2 := mocksdk.NewMockWorker(s.controller)
|
|
s.cfactory.EXPECT().NewWorker(matchStrict{cli2}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Return(wkr2)
|
|
s.cmp1.EXPECT().Register(wkr2, ns, workercommon.RegistrationDetails{TotalWorkers: 1, Multiplicity: 1})
|
|
wkr2.EXPECT().Start()
|
|
|
|
s.manager.namespaceCallback(nsRestored, false)
|
|
time.Sleep(50 * time.Millisecond)
|
|
|
|
// now delete from db without changing state
|
|
wkr2.EXPECT().Stop()
|
|
cli2.EXPECT().Close()
|
|
s.manager.namespaceCallback(nsRestored, true)
|
|
time.Sleep(50 * time.Millisecond)
|
|
}
|
|
|
|
func (s *perNsWorkerManagerSuite) TestMembershipChanged() {
|
|
ns := testns("ns1", enumspb.NAMESPACE_STATE_REGISTERED)
|
|
|
|
s.cmp1.EXPECT().DedicatedWorkerOptions(gomock.Any()).Return(&workercommon.PerNSDedicatedWorkerOptions{
|
|
Enabled: true,
|
|
}).AnyTimes()
|
|
s.cmp2.EXPECT().DedicatedWorkerOptions(gomock.Any()).Return(&workercommon.PerNSDedicatedWorkerOptions{
|
|
Enabled: false,
|
|
}).AnyTimes()
|
|
|
|
// we don't own it at first
|
|
s.serviceResolver.EXPECT().LookupN("ns1", 1).Return([]membership.HostInfo{membership.NewHostInfoFromAddress("other")})
|
|
|
|
s.manager.namespaceCallback(ns, false)
|
|
time.Sleep(50 * time.Millisecond)
|
|
|
|
// now we own it
|
|
s.serviceResolver.EXPECT().LookupN("ns1", 1).Return([]membership.HostInfo{membership.NewHostInfoFromAddress("self")})
|
|
cli1 := mocksdk.NewMockClient(s.controller)
|
|
s.cfactory.EXPECT().NewClient(matchOptions("ns1")).Return(cli1)
|
|
wkr1 := mocksdk.NewMockWorker(s.controller)
|
|
s.cfactory.EXPECT().NewWorker(matchStrict{cli1}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Return(wkr1)
|
|
s.cmp1.EXPECT().Register(wkr1, ns, workercommon.RegistrationDetails{TotalWorkers: 1, Multiplicity: 1})
|
|
wkr1.EXPECT().Start()
|
|
|
|
s.manager.membershipChangedCh <- nil
|
|
time.Sleep(50 * time.Millisecond)
|
|
|
|
// now we don't own it anymore
|
|
s.serviceResolver.EXPECT().LookupN("ns1", 1).Return([]membership.HostInfo{membership.NewHostInfoFromAddress("other")})
|
|
wkr1.EXPECT().Stop()
|
|
cli1.EXPECT().Close()
|
|
|
|
s.manager.membershipChangedCh <- nil
|
|
time.Sleep(50 * time.Millisecond)
|
|
}
|
|
|
|
func (s *perNsWorkerManagerSuite) TestServiceResolverError() {
|
|
ns := testns("ns1", enumspb.NAMESPACE_STATE_REGISTERED)
|
|
|
|
s.cmp1.EXPECT().DedicatedWorkerOptions(gomock.Any()).Return(&workercommon.PerNSDedicatedWorkerOptions{
|
|
Enabled: true,
|
|
}).AnyTimes()
|
|
s.cmp2.EXPECT().DedicatedWorkerOptions(gomock.Any()).Return(&workercommon.PerNSDedicatedWorkerOptions{
|
|
Enabled: false,
|
|
}).AnyTimes()
|
|
|
|
s.serviceResolver.EXPECT().LookupN("ns1", 1).Return([]membership.HostInfo{})
|
|
s.serviceResolver.EXPECT().LookupN("ns1", 1).Return([]membership.HostInfo{})
|
|
s.serviceResolver.EXPECT().LookupN("ns1", 1).Return([]membership.HostInfo{membership.NewHostInfoFromAddress("self")})
|
|
|
|
cli1 := mocksdk.NewMockClient(s.controller)
|
|
s.cfactory.EXPECT().NewClient(matchOptions("ns1")).Return(cli1)
|
|
wkr1 := mocksdk.NewMockWorker(s.controller)
|
|
s.cfactory.EXPECT().NewWorker(matchStrict{cli1}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Return(wkr1)
|
|
s.cmp1.EXPECT().Register(wkr1, ns, workercommon.RegistrationDetails{TotalWorkers: 1, Multiplicity: 1})
|
|
wkr1.EXPECT().Start()
|
|
|
|
s.manager.namespaceCallback(ns, false)
|
|
time.Sleep(50 * time.Millisecond)
|
|
|
|
// shutdown
|
|
wkr1.EXPECT().Stop()
|
|
cli1.EXPECT().Close()
|
|
}
|
|
|
|
func (s *perNsWorkerManagerSuite) TestStartWorkerError() {
|
|
ns := testns("ns1", enumspb.NAMESPACE_STATE_REGISTERED)
|
|
|
|
s.cmp1.EXPECT().DedicatedWorkerOptions(gomock.Any()).Return(&workercommon.PerNSDedicatedWorkerOptions{
|
|
Enabled: true,
|
|
}).AnyTimes()
|
|
s.cmp2.EXPECT().DedicatedWorkerOptions(gomock.Any()).Return(&workercommon.PerNSDedicatedWorkerOptions{
|
|
Enabled: false,
|
|
}).AnyTimes()
|
|
|
|
s.serviceResolver.EXPECT().LookupN("ns1", 1).Return([]membership.HostInfo{membership.NewHostInfoFromAddress("self")}).AnyTimes()
|
|
|
|
cli1 := mocksdk.NewMockClient(s.controller)
|
|
s.cfactory.EXPECT().NewClient(matchOptions("ns1")).Return(cli1)
|
|
wkr1 := mocksdk.NewMockWorker(s.controller)
|
|
s.cfactory.EXPECT().NewWorker(matchStrict{cli1}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Return(wkr1)
|
|
s.cmp1.EXPECT().Register(wkr1, ns, workercommon.RegistrationDetails{TotalWorkers: 1, Multiplicity: 1})
|
|
|
|
// first try fails to start
|
|
wkr1.EXPECT().Start().Return(errors.New("start worker error"))
|
|
cli1.EXPECT().Close()
|
|
|
|
// second try
|
|
cli2 := mocksdk.NewMockClient(s.controller)
|
|
s.cfactory.EXPECT().NewClient(matchOptions("ns1")).Return(cli2)
|
|
wkr2 := mocksdk.NewMockWorker(s.controller)
|
|
s.cfactory.EXPECT().NewWorker(matchStrict{cli2}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Return(wkr2)
|
|
s.cmp1.EXPECT().Register(wkr2, ns, workercommon.RegistrationDetails{TotalWorkers: 1, Multiplicity: 1})
|
|
wkr2.EXPECT().Start()
|
|
|
|
s.manager.namespaceCallback(ns, false)
|
|
time.Sleep(50 * time.Millisecond)
|
|
|
|
// shutdown
|
|
wkr2.EXPECT().Stop()
|
|
cli2.EXPECT().Close()
|
|
}
|
|
|
|
func testns(name string, state enumspb.NamespaceState) *namespace.Namespace {
|
|
return namespace.NewLocalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{
|
|
Id: name,
|
|
State: state,
|
|
Name: name,
|
|
},
|
|
nil,
|
|
cluster.TestCurrentClusterName,
|
|
)
|
|
}
|
|
|
|
func testInactiveNS(name string, state enumspb.NamespaceState) *namespace.Namespace {
|
|
return namespace.NewNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{
|
|
Id: name,
|
|
State: state,
|
|
Name: name,
|
|
},
|
|
nil,
|
|
true,
|
|
&persistencespb.NamespaceReplicationConfig{
|
|
ActiveClusterName: cluster.TestAlternativeClusterName, // not active
|
|
Clusters: cluster.TestAllClusterNames,
|
|
},
|
|
0,
|
|
)
|
|
}
|
|
|
|
type matchOptions string
|
|
|
|
func (ns matchOptions) Matches(v any) bool {
|
|
options, ok := v.(sdkclient.Options)
|
|
return ok && options.Namespace == string(ns)
|
|
}
|
|
|
|
func (ns matchOptions) String() string {
|
|
return fmt.Sprintf("namespace=%q", string(ns))
|
|
}
|
|
|
|
// matchStrict implements shallow matching for gomock (default uses reflect.DeepEqual)
|
|
type matchStrict struct{ v any }
|
|
|
|
func (m matchStrict) Matches(v any) bool { return v == m.v }
|
|
func (m matchStrict) String() string { return fmt.Sprintf("%#v", m) }
|