mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
Rodrigo Zhouandrodrigozhou
46497a04da
<!-- Describe what has changed in this PR --> Dynamic config to suppress error when setting system search attributes <!-- Tell your future self why have you made these changes --> Create a way to prevent errors when new system search attributes are added, but there is already a custom search attribute registered with the same name. <!-- How have you verified this change? Tested locally? Added a unit test? Checked in staging env? --> Added unit test. <!-- Assuming the worst case, what can be broken when deploying this change to production? --> <!-- Have you made sure this change doesn't falsify anything currently stated in `docs/`? If significant new behavior is added, have you described that in `docs/`? --> <!-- Is this PR a hotfix candidate or does it require a notification to be sent to the broader community? (Yes/No) -->
949 lines
36 KiB
Go
949 lines
36 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 history
|
|
|
|
import (
|
|
"math/rand"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/golang/mock/gomock"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
"github.com/stretchr/testify/suite"
|
|
commandpb "go.temporal.io/api/command/v1"
|
|
commonpb "go.temporal.io/api/common/v1"
|
|
enumspb "go.temporal.io/api/enums/v1"
|
|
"go.temporal.io/api/serviceerror"
|
|
taskqueuepb "go.temporal.io/api/taskqueue/v1"
|
|
"google.golang.org/protobuf/types/known/durationpb"
|
|
|
|
persistencespb "go.temporal.io/server/api/persistence/v1"
|
|
"go.temporal.io/server/common"
|
|
"go.temporal.io/server/common/cluster"
|
|
"go.temporal.io/server/common/definition"
|
|
"go.temporal.io/server/common/dynamicconfig"
|
|
"go.temporal.io/server/common/log"
|
|
"go.temporal.io/server/common/log/tag"
|
|
"go.temporal.io/server/common/metrics"
|
|
"go.temporal.io/server/common/namespace"
|
|
"go.temporal.io/server/common/payloads"
|
|
"go.temporal.io/server/common/persistence/visibility/manager"
|
|
"go.temporal.io/server/common/searchattribute"
|
|
"go.temporal.io/server/service/history/configs"
|
|
"go.temporal.io/server/service/history/tests"
|
|
"go.temporal.io/server/service/history/workflow"
|
|
)
|
|
|
|
var (
|
|
nonTerminalCommands = []*commandpb.Command{
|
|
{CommandType: enumspb.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK},
|
|
{CommandType: enumspb.COMMAND_TYPE_REQUEST_CANCEL_ACTIVITY_TASK},
|
|
{CommandType: enumspb.COMMAND_TYPE_CANCEL_TIMER},
|
|
{CommandType: enumspb.COMMAND_TYPE_CANCEL_TIMER},
|
|
{CommandType: enumspb.COMMAND_TYPE_RECORD_MARKER},
|
|
{CommandType: enumspb.COMMAND_TYPE_REQUEST_CANCEL_EXTERNAL_WORKFLOW_EXECUTION},
|
|
{CommandType: enumspb.COMMAND_TYPE_SIGNAL_EXTERNAL_WORKFLOW_EXECUTION},
|
|
{CommandType: enumspb.COMMAND_TYPE_START_CHILD_WORKFLOW_EXECUTION},
|
|
{CommandType: enumspb.COMMAND_TYPE_UPSERT_WORKFLOW_SEARCH_ATTRIBUTES},
|
|
{CommandType: enumspb.COMMAND_TYPE_MODIFY_WORKFLOW_PROPERTIES},
|
|
}
|
|
|
|
terminalCommands = []*commandpb.Command{
|
|
{CommandType: enumspb.COMMAND_TYPE_CONTINUE_AS_NEW_WORKFLOW_EXECUTION},
|
|
{CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION},
|
|
{CommandType: enumspb.COMMAND_TYPE_FAIL_WORKFLOW_EXECUTION},
|
|
{CommandType: enumspb.COMMAND_TYPE_CANCEL_WORKFLOW_EXECUTION},
|
|
}
|
|
)
|
|
|
|
type (
|
|
commandAttrValidatorSuite struct {
|
|
suite.Suite
|
|
*require.Assertions
|
|
|
|
controller *gomock.Controller
|
|
mockNamespaceCache *namespace.MockRegistry
|
|
mockVisibilityManager *manager.MockVisibilityManager
|
|
|
|
validator *commandAttrValidator
|
|
|
|
testNamespaceID namespace.ID
|
|
testTargetNamespaceID namespace.ID
|
|
}
|
|
)
|
|
|
|
func TestCommandAttrValidatorSuite(t *testing.T) {
|
|
s := new(commandAttrValidatorSuite)
|
|
suite.Run(t, s)
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) SetupSuite() {
|
|
s.testNamespaceID = "test namespace ID"
|
|
s.testTargetNamespaceID = "test target namespace ID"
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TearDownSuite() {
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) SetupTest() {
|
|
s.Assertions = require.New(s.T())
|
|
|
|
s.controller = gomock.NewController(s.T())
|
|
s.mockNamespaceCache = namespace.NewMockRegistry(s.controller)
|
|
|
|
s.mockVisibilityManager = manager.NewMockVisibilityManager(s.controller)
|
|
s.mockVisibilityManager.EXPECT().GetIndexName().Return("index-name").AnyTimes()
|
|
s.mockVisibilityManager.EXPECT().
|
|
ValidateCustomSearchAttributes(gomock.Any()).
|
|
DoAndReturn(
|
|
func(searchAttributes map[string]any) (map[string]any, error) {
|
|
return searchAttributes, nil
|
|
},
|
|
).
|
|
AnyTimes()
|
|
|
|
config := &configs.Config{
|
|
MaxIDLengthLimit: dynamicconfig.GetIntPropertyFn(1000),
|
|
SearchAttributesNumberOfKeysLimit: dynamicconfig.GetIntPropertyFilteredByNamespace(100),
|
|
SearchAttributesSizeOfValueLimit: dynamicconfig.GetIntPropertyFilteredByNamespace(2 * 1024),
|
|
SearchAttributesTotalSizeLimit: dynamicconfig.GetIntPropertyFilteredByNamespace(40 * 1024),
|
|
DefaultActivityRetryPolicy: dynamicconfig.GetMapPropertyFnWithNamespaceFilter(common.GetDefaultRetryPolicyConfigOptions()),
|
|
DefaultWorkflowRetryPolicy: dynamicconfig.GetMapPropertyFnWithNamespaceFilter(common.GetDefaultRetryPolicyConfigOptions()),
|
|
EnableCrossNamespaceCommands: dynamicconfig.GetBoolPropertyFn(true),
|
|
DefaultWorkflowTaskTimeout: dynamicconfig.GetDurationPropertyFnFilteredByNamespace(common.DefaultWorkflowTaskTimeout),
|
|
}
|
|
s.validator = newCommandAttrValidator(
|
|
s.mockNamespaceCache,
|
|
config,
|
|
searchattribute.NewValidator(
|
|
searchattribute.NewTestProvider(),
|
|
searchattribute.NewTestMapperProvider(nil),
|
|
config.SearchAttributesNumberOfKeysLimit,
|
|
config.SearchAttributesSizeOfValueLimit,
|
|
config.SearchAttributesTotalSizeLimit,
|
|
s.mockVisibilityManager,
|
|
dynamicconfig.GetBoolPropertyFnFilteredByNamespace(false),
|
|
dynamicconfig.GetBoolPropertyFnFilteredByNamespace(false),
|
|
))
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TearDownTest() {
|
|
s.controller.Finish()
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateSignalExternalWorkflowExecutionAttributes() {
|
|
namespaceEntry := namespace.NewLocalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testNamespaceID.String()},
|
|
nil,
|
|
cluster.TestCurrentClusterName,
|
|
)
|
|
targetNamespaceEntry := namespace.NewLocalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testTargetNamespaceID.String()},
|
|
nil,
|
|
cluster.TestCurrentClusterName,
|
|
)
|
|
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testNamespaceID).Return(namespaceEntry, nil).AnyTimes()
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testTargetNamespaceID).Return(targetNamespaceEntry, nil).AnyTimes()
|
|
|
|
var attributes *commandpb.SignalExternalWorkflowExecutionCommandAttributes
|
|
|
|
fc, err := s.validator.validateSignalExternalWorkflowExecutionAttributes(s.testNamespaceID, s.testTargetNamespaceID, attributes)
|
|
s.EqualError(err, "SignalExternalWorkflowExecutionCommandAttributes is not set on SignalExternalWorkflowExecutionCommand.")
|
|
s.Equal(enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SIGNAL_WORKFLOW_EXECUTION_ATTRIBUTES, fc)
|
|
|
|
attributes = &commandpb.SignalExternalWorkflowExecutionCommandAttributes{}
|
|
fc, err = s.validator.validateSignalExternalWorkflowExecutionAttributes(s.testNamespaceID, s.testTargetNamespaceID, attributes)
|
|
s.EqualError(err, "Execution is not set on SignalExternalWorkflowExecutionCommand.")
|
|
s.Equal(enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SIGNAL_WORKFLOW_EXECUTION_ATTRIBUTES, fc)
|
|
|
|
attributes.Execution = &commonpb.WorkflowExecution{}
|
|
attributes.Execution.WorkflowId = "workflow-id"
|
|
fc, err = s.validator.validateSignalExternalWorkflowExecutionAttributes(s.testNamespaceID, s.testTargetNamespaceID, attributes)
|
|
s.EqualError(err, "SignalName is not set on SignalExternalWorkflowExecutionCommand. WorkflowId=workflow-id Namespace= RunId=")
|
|
s.Equal(enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SIGNAL_WORKFLOW_EXECUTION_ATTRIBUTES, fc)
|
|
|
|
attributes.Execution.RunId = "run-id"
|
|
fc, err = s.validator.validateSignalExternalWorkflowExecutionAttributes(s.testNamespaceID, s.testTargetNamespaceID, attributes)
|
|
s.EqualError(err, "Invalid RunId set on SignalExternalWorkflowExecutionCommand. WorkflowId=workflow-id Namespace= RunId=run-id SignalName=")
|
|
attributes.Execution.RunId = tests.RunID
|
|
s.Equal(enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SIGNAL_WORKFLOW_EXECUTION_ATTRIBUTES, fc)
|
|
|
|
attributes.SignalName = "my signal name"
|
|
fc, err = s.validator.validateSignalExternalWorkflowExecutionAttributes(s.testNamespaceID, s.testTargetNamespaceID, attributes)
|
|
s.NoError(err)
|
|
s.Equal(enumspb.WORKFLOW_TASK_FAILED_CAUSE_UNSPECIFIED, fc)
|
|
|
|
attributes.Input = payloads.EncodeString("test input")
|
|
fc, err = s.validator.validateSignalExternalWorkflowExecutionAttributes(s.testNamespaceID, s.testTargetNamespaceID, attributes)
|
|
s.NoError(err)
|
|
s.Equal(enumspb.WORKFLOW_TASK_FAILED_CAUSE_UNSPECIFIED, fc)
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateUpsertWorkflowSearchAttributes() {
|
|
namespace := namespace.Name("tests.Namespace")
|
|
var attributes *commandpb.UpsertWorkflowSearchAttributesCommandAttributes
|
|
|
|
fc, err := s.validator.validateUpsertWorkflowSearchAttributes(namespace, attributes)
|
|
s.EqualError(err, "UpsertWorkflowSearchAttributesCommandAttributes is not set on UpsertWorkflowSearchAttributesCommand.")
|
|
s.Equal(enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SEARCH_ATTRIBUTES, fc)
|
|
|
|
attributes = &commandpb.UpsertWorkflowSearchAttributesCommandAttributes{}
|
|
fc, err = s.validator.validateUpsertWorkflowSearchAttributes(namespace, attributes)
|
|
s.EqualError(err, "SearchAttributes is not set on UpsertWorkflowSearchAttributesCommand.")
|
|
s.Equal(enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SEARCH_ATTRIBUTES, fc)
|
|
|
|
attributes.SearchAttributes = &commonpb.SearchAttributes{}
|
|
fc, err = s.validator.validateUpsertWorkflowSearchAttributes(namespace, attributes)
|
|
s.EqualError(err, "IndexedFields is not set on UpsertWorkflowSearchAttributesCommand.")
|
|
s.Equal(enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SEARCH_ATTRIBUTES, fc)
|
|
|
|
saPayload, err := searchattribute.EncodeValue("bytes", enumspb.INDEXED_VALUE_TYPE_KEYWORD)
|
|
s.NoError(err)
|
|
attributes.SearchAttributes.IndexedFields = map[string]*commonpb.Payload{
|
|
"CustomKeywordField": saPayload,
|
|
}
|
|
fc, err = s.validator.validateUpsertWorkflowSearchAttributes(namespace, attributes)
|
|
s.NoError(err)
|
|
s.Equal(enumspb.WORKFLOW_TASK_FAILED_CAUSE_UNSPECIFIED, fc)
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateContinueAsNewWorkflowExecutionAttributes() {
|
|
executionTimeout := time.Hour
|
|
workflowTypeName := "workflowType"
|
|
taskQueue := "taskQueue"
|
|
|
|
attributes := &commandpb.ContinueAsNewWorkflowExecutionCommandAttributes{
|
|
// workflow type name and task queue name should be retrieved from existing workflow info
|
|
|
|
// WorkflowRunTimeout should be shorten to execution timeout
|
|
WorkflowRunTimeout: durationpb.New(executionTimeout * 2),
|
|
// WorkflowTaskTimeout should be shorten to max workflow task timeout
|
|
WorkflowTaskTimeout: durationpb.New(common.MaxWorkflowTaskStartToCloseTimeout * 2),
|
|
}
|
|
|
|
executionInfo := &persistencespb.WorkflowExecutionInfo{
|
|
WorkflowTypeName: workflowTypeName,
|
|
TaskQueue: taskQueue,
|
|
WorkflowExecutionTimeout: durationpb.New(executionTimeout),
|
|
}
|
|
|
|
fc, err := s.validator.validateContinueAsNewWorkflowExecutionAttributes(
|
|
tests.Namespace,
|
|
attributes,
|
|
executionInfo,
|
|
)
|
|
s.NoError(err)
|
|
s.Equal(enumspb.WORKFLOW_TASK_FAILED_CAUSE_UNSPECIFIED, fc)
|
|
|
|
s.Equal(workflowTypeName, attributes.GetWorkflowType().GetName())
|
|
s.Equal(taskQueue, attributes.GetTaskQueue().GetName())
|
|
s.Equal(executionTimeout, attributes.GetWorkflowRunTimeout().AsDuration())
|
|
s.Equal(common.MaxWorkflowTaskStartToCloseTimeout, attributes.GetWorkflowTaskTimeout().AsDuration())
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateModifyWorkflowProperties() {
|
|
namespace := namespace.Name("tests.Namespace")
|
|
var attributes *commandpb.ModifyWorkflowPropertiesCommandAttributes
|
|
|
|
fc, err := s.validator.validateModifyWorkflowProperties(namespace, attributes)
|
|
s.EqualError(err, "ModifyWorkflowPropertiesCommandAttributes is not set on ModifyWorkflowPropertiesCommand.")
|
|
s.Equal(enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_MODIFY_WORKFLOW_PROPERTIES_ATTRIBUTES, fc)
|
|
|
|
// test attributes has at least one non-nil attribute
|
|
attributes = &commandpb.ModifyWorkflowPropertiesCommandAttributes{}
|
|
fc, err = s.validator.validateModifyWorkflowProperties(namespace, attributes)
|
|
s.EqualError(err, "UpsertedMemo is not set on ModifyWorkflowPropertiesCommand.")
|
|
s.Equal(enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_MODIFY_WORKFLOW_PROPERTIES_ATTRIBUTES, fc)
|
|
|
|
// test UpsertedMemo cannot be an empty map
|
|
attributes = &commandpb.ModifyWorkflowPropertiesCommandAttributes{
|
|
UpsertedMemo: &commonpb.Memo{},
|
|
}
|
|
fc, err = s.validator.validateModifyWorkflowProperties(namespace, attributes)
|
|
s.EqualError(err, "UpsertedMemo.Fields is not set on ModifyWorkflowPropertiesCommand.")
|
|
s.Equal(enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_MODIFY_WORKFLOW_PROPERTIES_ATTRIBUTES, fc)
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateCrossNamespaceCall_LocalToLocal() {
|
|
namespaceEntry := namespace.NewLocalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testNamespaceID.String()},
|
|
nil,
|
|
cluster.TestCurrentClusterName,
|
|
)
|
|
targetNamespaceEntry := namespace.NewLocalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testTargetNamespaceID.String()},
|
|
nil,
|
|
cluster.TestCurrentClusterName,
|
|
)
|
|
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testNamespaceID).Return(namespaceEntry, nil)
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testTargetNamespaceID).Return(targetNamespaceEntry, nil)
|
|
|
|
err := s.validator.validateCrossNamespaceCall(s.testNamespaceID, s.testTargetNamespaceID)
|
|
s.Nil(err)
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateCrossNamespaceCall_LocalToEffectiveLocal_SameCluster() {
|
|
namespaceEntry := namespace.NewLocalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testNamespaceID.String()},
|
|
nil,
|
|
cluster.TestCurrentClusterName,
|
|
)
|
|
targetNamespaceEntry := namespace.NewGlobalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testTargetNamespaceID.String()},
|
|
nil,
|
|
&persistencespb.NamespaceReplicationConfig{
|
|
ActiveClusterName: cluster.TestCurrentClusterName,
|
|
Clusters: []string{cluster.TestCurrentClusterName},
|
|
},
|
|
1234,
|
|
)
|
|
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testNamespaceID).Return(namespaceEntry, nil)
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testTargetNamespaceID).Return(targetNamespaceEntry, nil)
|
|
|
|
err := s.validator.validateCrossNamespaceCall(s.testNamespaceID, s.testTargetNamespaceID)
|
|
s.Nil(err)
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateCrossNamespaceCall_LocalToEffectiveLocal_DiffCluster() {
|
|
namespaceEntry := namespace.NewLocalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testNamespaceID.String()},
|
|
nil,
|
|
cluster.TestCurrentClusterName,
|
|
)
|
|
targetNamespaceEntry := namespace.NewGlobalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testTargetNamespaceID.String()},
|
|
nil,
|
|
&persistencespb.NamespaceReplicationConfig{
|
|
ActiveClusterName: cluster.TestAlternativeClusterName,
|
|
Clusters: []string{cluster.TestAlternativeClusterName},
|
|
},
|
|
1234,
|
|
)
|
|
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testNamespaceID).Return(namespaceEntry, nil)
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testTargetNamespaceID).Return(targetNamespaceEntry, nil)
|
|
|
|
err := s.validator.validateCrossNamespaceCall(s.testNamespaceID, s.testTargetNamespaceID)
|
|
s.IsType(&serviceerror.InvalidArgument{}, err)
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateCrossNamespaceCall_LocalToGlobal() {
|
|
namespaceEntry := namespace.NewLocalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testNamespaceID.String()},
|
|
nil,
|
|
cluster.TestCurrentClusterName,
|
|
)
|
|
targetNamespaceEntry := namespace.NewGlobalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testTargetNamespaceID.String()},
|
|
nil,
|
|
&persistencespb.NamespaceReplicationConfig{
|
|
ActiveClusterName: cluster.TestCurrentClusterName,
|
|
Clusters: []string{
|
|
cluster.TestCurrentClusterName,
|
|
cluster.TestAlternativeClusterName,
|
|
},
|
|
},
|
|
1234,
|
|
)
|
|
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testNamespaceID).Return(namespaceEntry, nil)
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testTargetNamespaceID).Return(targetNamespaceEntry, nil)
|
|
|
|
err := s.validator.validateCrossNamespaceCall(s.testNamespaceID, s.testTargetNamespaceID)
|
|
s.IsType(&serviceerror.InvalidArgument{}, err)
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateCrossNamespaceCall_EffectiveLocalToLocal_SameCluster() {
|
|
namespaceEntry := namespace.NewGlobalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testNamespaceID.String()},
|
|
nil,
|
|
&persistencespb.NamespaceReplicationConfig{
|
|
ActiveClusterName: cluster.TestCurrentClusterName,
|
|
Clusters: []string{cluster.TestCurrentClusterName},
|
|
},
|
|
1234,
|
|
)
|
|
targetNamespaceEntry := namespace.NewLocalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testTargetNamespaceID.String()},
|
|
nil,
|
|
cluster.TestCurrentClusterName,
|
|
)
|
|
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testNamespaceID).Return(namespaceEntry, nil)
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testTargetNamespaceID).Return(targetNamespaceEntry, nil)
|
|
|
|
err := s.validator.validateCrossNamespaceCall(s.testNamespaceID, s.testTargetNamespaceID)
|
|
s.Nil(err)
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateCrossNamespaceCall_EffectiveLocalToLocal_DiffCluster() {
|
|
namespaceEntry := namespace.NewGlobalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testNamespaceID.String()},
|
|
nil,
|
|
&persistencespb.NamespaceReplicationConfig{
|
|
ActiveClusterName: cluster.TestAlternativeClusterName,
|
|
Clusters: []string{cluster.TestAlternativeClusterName},
|
|
},
|
|
1234,
|
|
)
|
|
targetNamespaceEntry := namespace.NewLocalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testTargetNamespaceID.String()},
|
|
nil,
|
|
cluster.TestCurrentClusterName,
|
|
)
|
|
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testNamespaceID).Return(namespaceEntry, nil)
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testTargetNamespaceID).Return(targetNamespaceEntry, nil)
|
|
|
|
err := s.validator.validateCrossNamespaceCall(s.testNamespaceID, s.testTargetNamespaceID)
|
|
s.IsType(&serviceerror.InvalidArgument{}, err)
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateCrossNamespaceCall_EffectiveLocalToEffectiveLocal_SameCluster() {
|
|
namespaceEntry := namespace.NewGlobalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testNamespaceID.String()},
|
|
nil,
|
|
&persistencespb.NamespaceReplicationConfig{
|
|
ActiveClusterName: cluster.TestCurrentClusterName,
|
|
Clusters: []string{cluster.TestCurrentClusterName},
|
|
},
|
|
1234,
|
|
)
|
|
targetNamespaceEntry := namespace.NewGlobalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testTargetNamespaceID.String()},
|
|
nil,
|
|
&persistencespb.NamespaceReplicationConfig{
|
|
ActiveClusterName: cluster.TestCurrentClusterName,
|
|
Clusters: []string{cluster.TestCurrentClusterName},
|
|
},
|
|
5678,
|
|
)
|
|
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testNamespaceID).Return(namespaceEntry, nil)
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testTargetNamespaceID).Return(targetNamespaceEntry, nil)
|
|
|
|
err := s.validator.validateCrossNamespaceCall(s.testNamespaceID, s.testTargetNamespaceID)
|
|
s.Nil(err)
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateCrossNamespaceCall_EffectiveLocalToEffectiveLocal_DiffCluster() {
|
|
namespaceEntry := namespace.NewGlobalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testNamespaceID.String()},
|
|
nil,
|
|
&persistencespb.NamespaceReplicationConfig{
|
|
ActiveClusterName: cluster.TestCurrentClusterName,
|
|
Clusters: []string{cluster.TestCurrentClusterName},
|
|
},
|
|
1234,
|
|
)
|
|
targetNamespaceEntry := namespace.NewGlobalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testTargetNamespaceID.String()},
|
|
nil,
|
|
&persistencespb.NamespaceReplicationConfig{
|
|
ActiveClusterName: cluster.TestAlternativeClusterName,
|
|
Clusters: []string{cluster.TestAlternativeClusterName},
|
|
},
|
|
5678,
|
|
)
|
|
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testNamespaceID).Return(namespaceEntry, nil)
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testTargetNamespaceID).Return(targetNamespaceEntry, nil)
|
|
|
|
err := s.validator.validateCrossNamespaceCall(s.testNamespaceID, s.testTargetNamespaceID)
|
|
s.IsType(&serviceerror.InvalidArgument{}, err)
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateCrossNamespaceCall_EffectiveLocalToGlobal() {
|
|
namespaceEntry := namespace.NewGlobalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testNamespaceID.String()},
|
|
nil,
|
|
&persistencespb.NamespaceReplicationConfig{
|
|
ActiveClusterName: cluster.TestCurrentClusterName,
|
|
Clusters: []string{
|
|
cluster.TestCurrentClusterName,
|
|
},
|
|
},
|
|
5678,
|
|
)
|
|
targetNamespaceEntry := namespace.NewGlobalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testTargetNamespaceID.String()},
|
|
nil,
|
|
&persistencespb.NamespaceReplicationConfig{
|
|
ActiveClusterName: cluster.TestCurrentClusterName,
|
|
Clusters: []string{
|
|
cluster.TestCurrentClusterName,
|
|
cluster.TestAlternativeClusterName,
|
|
},
|
|
},
|
|
1234,
|
|
)
|
|
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testNamespaceID).Return(namespaceEntry, nil)
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testTargetNamespaceID).Return(targetNamespaceEntry, nil)
|
|
|
|
err := s.validator.validateCrossNamespaceCall(s.testNamespaceID, s.testTargetNamespaceID)
|
|
s.IsType(&serviceerror.InvalidArgument{}, err)
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateCrossNamespaceCall_GlobalToLocal() {
|
|
namespaceEntry := namespace.NewGlobalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testNamespaceID.String()},
|
|
nil,
|
|
&persistencespb.NamespaceReplicationConfig{
|
|
ActiveClusterName: cluster.TestCurrentClusterName,
|
|
Clusters: []string{
|
|
cluster.TestCurrentClusterName,
|
|
cluster.TestAlternativeClusterName,
|
|
},
|
|
},
|
|
1234,
|
|
)
|
|
targetNamespaceEntry := namespace.NewLocalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testTargetNamespaceID.String()},
|
|
nil,
|
|
cluster.TestCurrentClusterName,
|
|
)
|
|
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testNamespaceID).Return(namespaceEntry, nil)
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testTargetNamespaceID).Return(targetNamespaceEntry, nil)
|
|
|
|
err := s.validator.validateCrossNamespaceCall(s.testNamespaceID, s.testTargetNamespaceID)
|
|
s.IsType(&serviceerror.InvalidArgument{}, err)
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateCrossNamespaceCall_GlobalToEffectiveLocal() {
|
|
namespaceEntry := namespace.NewGlobalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testNamespaceID.String()},
|
|
nil,
|
|
&persistencespb.NamespaceReplicationConfig{
|
|
ActiveClusterName: cluster.TestCurrentClusterName,
|
|
Clusters: []string{
|
|
cluster.TestCurrentClusterName,
|
|
cluster.TestAlternativeClusterName,
|
|
},
|
|
},
|
|
5678,
|
|
)
|
|
targetNamespaceEntry := namespace.NewGlobalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testTargetNamespaceID.String()},
|
|
nil,
|
|
&persistencespb.NamespaceReplicationConfig{
|
|
ActiveClusterName: cluster.TestCurrentClusterName,
|
|
Clusters: []string{
|
|
cluster.TestCurrentClusterName,
|
|
},
|
|
},
|
|
1234,
|
|
)
|
|
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testNamespaceID).Return(namespaceEntry, nil)
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testTargetNamespaceID).Return(targetNamespaceEntry, nil)
|
|
|
|
err := s.validator.validateCrossNamespaceCall(s.testNamespaceID, s.testTargetNamespaceID)
|
|
s.IsType(&serviceerror.InvalidArgument{}, err)
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateCrossNamespaceCall_GlobalToGlobal_DiffNamespace() {
|
|
namespaceEntry := namespace.NewGlobalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testNamespaceID.String()},
|
|
nil,
|
|
&persistencespb.NamespaceReplicationConfig{
|
|
ActiveClusterName: cluster.TestCurrentClusterName,
|
|
Clusters: []string{
|
|
cluster.TestAlternativeClusterName,
|
|
cluster.TestCurrentClusterName,
|
|
},
|
|
},
|
|
1234,
|
|
)
|
|
targetNamespaceEntry := namespace.NewGlobalNamespaceForTest(
|
|
&persistencespb.NamespaceInfo{Name: s.testTargetNamespaceID.String()},
|
|
nil,
|
|
&persistencespb.NamespaceReplicationConfig{
|
|
ActiveClusterName: cluster.TestCurrentClusterName,
|
|
Clusters: []string{
|
|
cluster.TestCurrentClusterName,
|
|
cluster.TestAlternativeClusterName,
|
|
},
|
|
},
|
|
1234,
|
|
)
|
|
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testNamespaceID).Return(namespaceEntry, nil)
|
|
s.mockNamespaceCache.EXPECT().GetNamespaceByID(s.testTargetNamespaceID).Return(targetNamespaceEntry, nil)
|
|
|
|
err := s.validator.validateCrossNamespaceCall(s.testNamespaceID, s.testTargetNamespaceID)
|
|
s.IsType(&serviceerror.InvalidArgument{}, err)
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateCrossNamespaceCall_GlobalToGlobal_SameNamespace() {
|
|
targetNamespaceID := s.testNamespaceID
|
|
|
|
err := s.validator.validateCrossNamespaceCall(s.testNamespaceID, targetNamespaceID)
|
|
s.Nil(err)
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateTaskQueueName() {
|
|
newTaskQueue := func(name string) *taskqueuepb.TaskQueue {
|
|
return &taskqueuepb.TaskQueue{
|
|
Name: name,
|
|
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
|
|
}
|
|
}
|
|
|
|
testCases := []struct {
|
|
defaultVal string
|
|
input *taskqueuepb.TaskQueue
|
|
output *taskqueuepb.TaskQueue
|
|
isOutputErr bool
|
|
}{
|
|
{"tq-1", nil, newTaskQueue("tq-1"), false},
|
|
{"", newTaskQueue("tq-1"), newTaskQueue("tq-1"), false},
|
|
{"tq-1", newTaskQueue("tq-1"), newTaskQueue("tq-1"), false},
|
|
{"", newTaskQueue("/tl-1"), newTaskQueue("/tl-1"), false},
|
|
{"", newTaskQueue("/__temporal_sys"), newTaskQueue("/__temporal_sys"), false},
|
|
{"", nil, newTaskQueue(""), true},
|
|
{"", newTaskQueue(""), newTaskQueue(""), true},
|
|
{"", newTaskQueue(reservedTaskQueuePrefix), newTaskQueue(reservedTaskQueuePrefix), true},
|
|
{"tq-1", newTaskQueue(reservedTaskQueuePrefix), newTaskQueue(reservedTaskQueuePrefix), true},
|
|
{"", newTaskQueue(reservedTaskQueuePrefix + "tq-1"), newTaskQueue(reservedTaskQueuePrefix + "tq-1"), true},
|
|
{"tq-1", newTaskQueue(reservedTaskQueuePrefix + "tq-1"), newTaskQueue(reservedTaskQueuePrefix + "tq-1"), true},
|
|
}
|
|
|
|
for _, tc := range testCases {
|
|
key := tc.defaultVal + "#"
|
|
if tc.input != nil {
|
|
key += tc.input.GetName()
|
|
} else {
|
|
key += "nil"
|
|
}
|
|
s.Run(key, func() {
|
|
output, err := s.validator.validateTaskQueue(tc.input, tc.defaultVal)
|
|
if tc.isOutputErr {
|
|
s.Error(err)
|
|
} else {
|
|
s.NoError(err)
|
|
}
|
|
s.EqualValues(tc.output, output)
|
|
})
|
|
}
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateActivityRetryPolicy() {
|
|
testCases := []struct {
|
|
name string
|
|
input *commonpb.RetryPolicy
|
|
want *commonpb.RetryPolicy
|
|
}{
|
|
{
|
|
name: "override non-set policy",
|
|
input: nil,
|
|
want: &commonpb.RetryPolicy{
|
|
InitialInterval: durationpb.New(1 * time.Second),
|
|
BackoffCoefficient: 2,
|
|
MaximumInterval: durationpb.New(100 * time.Second),
|
|
MaximumAttempts: 0,
|
|
},
|
|
},
|
|
{
|
|
name: "do not override fully set policy",
|
|
input: &commonpb.RetryPolicy{
|
|
InitialInterval: durationpb.New(5 * time.Second),
|
|
BackoffCoefficient: 10,
|
|
MaximumInterval: durationpb.New(20 * time.Second),
|
|
MaximumAttempts: 8,
|
|
},
|
|
want: &commonpb.RetryPolicy{
|
|
InitialInterval: durationpb.New(5 * time.Second),
|
|
BackoffCoefficient: 10,
|
|
MaximumInterval: durationpb.New(20 * time.Second),
|
|
MaximumAttempts: 8,
|
|
},
|
|
},
|
|
{
|
|
name: "partial override of fields",
|
|
input: &commonpb.RetryPolicy{
|
|
InitialInterval: durationpb.New(0 * time.Second),
|
|
BackoffCoefficient: 1.2,
|
|
MaximumInterval: durationpb.New(0 * time.Second),
|
|
MaximumAttempts: 7,
|
|
},
|
|
want: &commonpb.RetryPolicy{
|
|
InitialInterval: durationpb.New(1 * time.Second),
|
|
BackoffCoefficient: 1.2,
|
|
MaximumInterval: durationpb.New(100 * time.Second),
|
|
MaximumAttempts: 7,
|
|
},
|
|
},
|
|
{
|
|
name: "set expected max interval if only init interval set",
|
|
input: &commonpb.RetryPolicy{
|
|
InitialInterval: durationpb.New(3 * time.Second),
|
|
MaximumInterval: durationpb.New(0 * time.Second),
|
|
},
|
|
want: &commonpb.RetryPolicy{
|
|
InitialInterval: durationpb.New(3 * time.Second),
|
|
BackoffCoefficient: 2,
|
|
MaximumInterval: durationpb.New(300 * time.Second),
|
|
MaximumAttempts: 0,
|
|
},
|
|
},
|
|
{
|
|
name: "override all defaults",
|
|
input: &commonpb.RetryPolicy{
|
|
InitialInterval: durationpb.New(0 * time.Second),
|
|
BackoffCoefficient: 0,
|
|
MaximumInterval: durationpb.New(0 * time.Second),
|
|
MaximumAttempts: 0,
|
|
},
|
|
want: &commonpb.RetryPolicy{
|
|
InitialInterval: durationpb.New(1 * time.Second),
|
|
BackoffCoefficient: 2,
|
|
MaximumInterval: durationpb.New(100 * time.Second),
|
|
MaximumAttempts: 0,
|
|
},
|
|
},
|
|
}
|
|
|
|
for _, tt := range testCases {
|
|
s.Run(tt.name, func() {
|
|
attr := &commandpb.ScheduleActivityTaskCommandAttributes{
|
|
RetryPolicy: tt.input,
|
|
}
|
|
|
|
err := s.validator.validateActivityRetryPolicy(s.testNamespaceID, attr)
|
|
assert.Nil(s.T(), err, "expected no error")
|
|
assert.Equal(s.T(), tt.want, attr.RetryPolicy, "unexpected retry policy")
|
|
})
|
|
}
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateCommandSequence_NoTerminalCommand() {
|
|
err := s.validator.validateCommandSequence(nonTerminalCommands)
|
|
s.NoError(err)
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateCommandSequence_ValidTerminalCommand() {
|
|
for _, terminalCommand := range terminalCommands {
|
|
err := s.validator.validateCommandSequence(append(nonTerminalCommands, terminalCommand))
|
|
s.NoError(err)
|
|
}
|
|
}
|
|
|
|
func (s *commandAttrValidatorSuite) TestValidateCommandSequence_InvalidTerminalCommand() {
|
|
for _, terminalCommand := range terminalCommands {
|
|
err := s.validator.validateCommandSequence(append(
|
|
[]*commandpb.Command{terminalCommand},
|
|
nonTerminalCommands[int(rand.Int31n(int32(len(nonTerminalCommands))))],
|
|
))
|
|
s.Error(err)
|
|
s.IsType(&serviceerror.InvalidArgument{}, err)
|
|
}
|
|
}
|
|
|
|
func TestWorkflowSizeChecker_NumChildWorkflows(t *testing.T) {
|
|
|
|
for _, c := range []struct {
|
|
Name string
|
|
NumPendingChildExecutions int
|
|
NumPendingActivities int
|
|
NumPendingCancelRequests int
|
|
NumPendingSignals int
|
|
|
|
PendingChildExecutionsLimit int
|
|
PendingActivitiesLimit int
|
|
PendingCancelRequestsLimit int
|
|
PendingSignalsLimit int
|
|
|
|
ExpectedMetric string
|
|
ExpectedChildExecutionsErrorMsg string
|
|
ExpectedActivitiesErrorMsg string
|
|
ExpectedCancelRequestsErrorMsg string
|
|
ExpectedSignalsErrorMsg string
|
|
}{
|
|
{
|
|
Name: "No limits and no data",
|
|
},
|
|
{
|
|
Name: "Limits but no workflow data",
|
|
PendingChildExecutionsLimit: 1,
|
|
PendingActivitiesLimit: 1,
|
|
PendingCancelRequestsLimit: 1,
|
|
PendingSignalsLimit: 1,
|
|
},
|
|
{
|
|
Name: "Limits not exceeded",
|
|
NumPendingChildExecutions: 1,
|
|
NumPendingActivities: 1,
|
|
NumPendingCancelRequests: 1,
|
|
NumPendingSignals: 1,
|
|
PendingChildExecutionsLimit: 2,
|
|
PendingActivitiesLimit: 2,
|
|
PendingCancelRequestsLimit: 2,
|
|
PendingSignalsLimit: 2,
|
|
},
|
|
{
|
|
Name: "Pending child executions limit exceeded",
|
|
NumPendingChildExecutions: 1,
|
|
PendingChildExecutionsLimit: 1,
|
|
ExpectedMetric: "wf_too_many_pending_child_workflows",
|
|
ExpectedChildExecutionsErrorMsg: "the number of pending child workflow executions, 1, has reached the " +
|
|
"per-workflow limit of 1",
|
|
},
|
|
{
|
|
Name: "Pending activities limit exceeded",
|
|
NumPendingActivities: 1,
|
|
PendingActivitiesLimit: 1,
|
|
ExpectedMetric: "wf_too_many_pending_activities",
|
|
ExpectedActivitiesErrorMsg: "the number of pending activities, 1, has reached the per-workflow limit of 1",
|
|
},
|
|
{
|
|
Name: "Pending cancel requests limit exceeded",
|
|
NumPendingCancelRequests: 1,
|
|
PendingCancelRequestsLimit: 1,
|
|
ExpectedMetric: "wf_too_many_pending_cancel_requests",
|
|
ExpectedCancelRequestsErrorMsg: "the number of pending requests to cancel external workflows, 1, has " +
|
|
"reached the per-workflow limit of 1",
|
|
},
|
|
{
|
|
Name: "Pending signals limit exceeded",
|
|
NumPendingSignals: 1,
|
|
PendingSignalsLimit: 1,
|
|
ExpectedMetric: "wf_too_many_pending_external_workflow_signals",
|
|
ExpectedSignalsErrorMsg: "the number of pending signals to external workflows, 1, has reached the " +
|
|
"per-workflow limit of 1",
|
|
},
|
|
} {
|
|
t.Run(c.Name, func(t *testing.T) {
|
|
ctrl := gomock.NewController(t)
|
|
mutableState := workflow.NewMockMutableState(ctrl)
|
|
logger := log.NewMockLogger(ctrl)
|
|
metricsHandler := metrics.NewMockHandler(ctrl)
|
|
|
|
workflowKey := definition.NewWorkflowKey(
|
|
"test-namespace-id",
|
|
"test-workflow-id",
|
|
"test-run-id",
|
|
)
|
|
mutableState.EXPECT().GetWorkflowKey().Return(workflowKey).AnyTimes()
|
|
|
|
executionInfos := make(map[int64]*persistencespb.ChildExecutionInfo)
|
|
activityInfos := make(map[int64]*persistencespb.ActivityInfo)
|
|
requestCancelInfos := make(map[int64]*persistencespb.RequestCancelInfo)
|
|
signalInfos := make(map[int64]*persistencespb.SignalInfo)
|
|
for i := 0; i < c.NumPendingChildExecutions; i++ {
|
|
executionInfos[int64(i)] = new(persistencespb.ChildExecutionInfo)
|
|
}
|
|
for i := 0; i < c.NumPendingActivities; i++ {
|
|
activityInfos[int64(i)] = new(persistencespb.ActivityInfo)
|
|
}
|
|
for i := 0; i < c.NumPendingCancelRequests; i++ {
|
|
requestCancelInfos[int64(i)] = new(persistencespb.RequestCancelInfo)
|
|
}
|
|
for i := 0; i < c.NumPendingSignals; i++ {
|
|
signalInfos[int64(i)] = new(persistencespb.SignalInfo)
|
|
}
|
|
mutableState.EXPECT().GetPendingChildExecutionInfos().Return(executionInfos)
|
|
mutableState.EXPECT().GetPendingActivityInfos().Return(activityInfos)
|
|
mutableState.EXPECT().GetPendingRequestCancelExternalInfos().Return(requestCancelInfos)
|
|
mutableState.EXPECT().GetPendingSignalExternalInfos().Return(signalInfos)
|
|
|
|
if len(c.ExpectedMetric) > 0 {
|
|
counterMetric := metrics.NewMockCounterIface(ctrl)
|
|
metricsHandler.EXPECT().Counter(c.ExpectedMetric).Return(counterMetric)
|
|
counterMetric.EXPECT().Record(int64(1))
|
|
}
|
|
|
|
for _, msg := range []string{
|
|
c.ExpectedChildExecutionsErrorMsg,
|
|
c.ExpectedActivitiesErrorMsg,
|
|
c.ExpectedCancelRequestsErrorMsg,
|
|
c.ExpectedSignalsErrorMsg,
|
|
} {
|
|
if len(msg) > 0 {
|
|
logger.EXPECT().Error(msg, gomock.Any()).Do(func(msg string, tags ...tag.Tag) {
|
|
var namespaceID, workflowID, runID interface{}
|
|
for _, t := range tags {
|
|
if t.Key() == "wf-namespace-id" {
|
|
namespaceID = t.Value()
|
|
} else if t.Key() == "wf-id" {
|
|
workflowID = t.Value()
|
|
} else if t.Key() == "wf-run-id" {
|
|
runID = t.Value()
|
|
}
|
|
}
|
|
assert.Equal(t, "test-namespace-id", namespaceID)
|
|
assert.Equal(t, "test-workflow-id", workflowID)
|
|
assert.Equal(t, "test-run-id", runID)
|
|
})
|
|
}
|
|
}
|
|
|
|
checker := newWorkflowSizeChecker(workflowSizeLimits{
|
|
numPendingChildExecutionsLimit: c.PendingChildExecutionsLimit,
|
|
numPendingActivitiesLimit: c.PendingActivitiesLimit,
|
|
numPendingCancelsRequestLimit: c.PendingCancelRequestsLimit,
|
|
numPendingSignalsLimit: c.PendingSignalsLimit,
|
|
}, mutableState, nil, metricsHandler, logger)
|
|
|
|
err := checker.checkIfNumChildWorkflowsExceedsLimit()
|
|
if len(c.ExpectedChildExecutionsErrorMsg) > 0 {
|
|
require.Error(t, err)
|
|
assert.Equal(t, c.ExpectedChildExecutionsErrorMsg, err.Error())
|
|
} else {
|
|
assert.NoError(t, err)
|
|
}
|
|
|
|
err = checker.checkIfNumPendingActivitiesExceedsLimit()
|
|
if len(c.ExpectedActivitiesErrorMsg) > 0 {
|
|
require.Error(t, err)
|
|
assert.Equal(t, c.ExpectedActivitiesErrorMsg, err.Error())
|
|
} else {
|
|
assert.NoError(t, err)
|
|
}
|
|
|
|
err = checker.checkIfNumPendingCancelRequestsExceedsLimit()
|
|
if len(c.ExpectedCancelRequestsErrorMsg) > 0 {
|
|
require.Error(t, err)
|
|
assert.Equal(t, c.ExpectedCancelRequestsErrorMsg, err.Error())
|
|
} else {
|
|
assert.NoError(t, err)
|
|
}
|
|
|
|
err = checker.checkIfNumPendingSignalsExceedsLimit()
|
|
if len(c.ExpectedSignalsErrorMsg) > 0 {
|
|
require.Error(t, err)
|
|
assert.Equal(t, c.ExpectedSignalsErrorMsg, err.Error())
|
|
} else {
|
|
assert.NoError(t, err)
|
|
}
|
|
})
|
|
}
|
|
}
|