8000 Add limits on other workflow fields by MichaelSnowden · Pull Request #3599 · temporalio/temporal · GitHub
[go: up one dir, main page]
More Web Proxy on the site http://driver.im/
Skip to content

Add limits on other workflow fields #3599

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Nov 18, 2022
Merged

Add limits on other workflow fields #3599

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 12 additions & 2 deletions common/dynamicconfig/constants.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,8 +113,18 @@ const (
MemoSizeLimitError = "limit.memoSize.error"
// MemoSizeLimitWarn is the per event memo size limit for warning
MemoSizeLimitWarn = "limit.memoSize.warn"
// NumPendingChildExecutionLimitError is the per workflow pending child workflow limit
NumPendingChildExecutionLimitError = "limit.numPendingChildExecution.error"
// NumPendingChildExecutionsLimitError is the maximum number of pending child workflows a workflow can have before
// StartChildWorkflowExecution commands will fail.
NumPendingChildExecutionsLimitError = "limit.numPendingChildExecutions.error"
// NumPendingActivitiesLimitError is the maximum number of pending activities a workflow can have before
// ScheduleActivityTask will fail.
NumPendingActivitiesLimitError = "limit.numPendingActivities.error"
// NumPendingSignalsLimitError is the maximum number of pending signals a workflow can have before
// SignalExternalWorkflowExecution commands from this workflow will fail.
NumPendingSignalsLimitError = "limit.numPendingSignals.error"
// NumPendingCancelRequestsLimitError is the maximum number of pending cancel requests a workflow can have before
// RequestCancelExternalWorkflowExecution commands will fail.
NumPendingCancelRequestsLimitError = "limit.numPendingCancelRequests.error"
// HistorySizeLimitError is the per workflow execution history size limit
HistorySizeLimitError = "limit.historySize.error"
// HistorySizeLimitWarn is the per workflow execution history size limit for warning
Expand Down
4 changes: 2 additions & 2 deletions host/client_integration_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,11 +81,11 @@ func TestClientIntegrationSuite(t *testing.T) {
}

func (s *clientIntegrationSuite) SetupSuite() {
// maxPendingChildWorkflows should be NumPendingChildExecutionLimitError, but that value is 1000, and it takes
// maxPendingChildWorkflows should be NumPendingChildExecutionsLimit, but that value is 1000, and it takes
// too long to test (which is the point of this limit)
s.maxPendingChildExecutions = 10
s.dynamicConfigOverrides = map[dynamicconfig.Key]interface{}{
dynamicconfig.NumPendingChildExecutionLimitError: s.maxPendingChildExecutions,
dynamicconfig.NumPendingChildExecutionsLimitError: s.maxPendingChildExecutions,
}
s.setupSuite("testdata/clientintegrationtestcluster.yaml")

Expand Down
17 changes: 10 additions & 7 deletions service/history/commandChecker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,11 +62,14 @@ type (
}

workflowSizeLimits struct {
blobSizeLimitWarn int
blobSizeLimitError int
memoSizeLimitWarn int
memoSizeLimitError int
numPendingChildExecutionLimitError int
blobSizeLimitWarn int
blobSizeLimitError int
memoSizeLimitWarn int
memoSizeLimitError int
numPendingChildExecutionsLimit int
numPendingActivitiesLimit int
numPendingSignalsLimit int
numPendingCancelsRequestLimit int
}

workflowSizeChecker struct {
Expand Down Expand Up @@ -189,7 +192,7 @@ func (c *workflowSizeChecker) checkIfNumChildWorkflowsExceedsLimit() error {
)

numPending := len(c.mutableState.GetPendingChildExecutionInfos())
errLimit := c.numPendingChildExecutionLimitError
errLimit := c.numPendingChildExecutionsLimit
if withinLimit 8000 (numPending, errLimit) {
return nil
}
Expand All @@ -199,7 +202,7 @@ func (c *workflowSizeChecker) checkIfNumChildWorkflowsExceedsLimit() error {
"has reached the error limit of %d established with %q",
numPending,
errLimit,
dynamicconfig.NumPendingChildExecutionLimitError,
dynamicconfig.NumPendingChildExecutionsLimitError,
)
logger.Error(err.Error(), tag.Error(err))
return err
Expand Down
4 changes: 2 additions & 2 deletions service/history/commandChecker_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -718,7 +718,7 @@ func (s *commandAttrValidatorSuite) TestValidateCommandSequence_InvalidTerminalC
func TestWorkflowSizeChecker_NumChildWorkflows(t *testing.T) {
const (
errMsg = "the number of pending child workflow executions, 1, " +
"has reached the error limit of 1 established with \"limit.numPendingChildExecution.error\""
"has reached the error limit of 1 established with \"limit.numPendingChildExecutions.error\""
)
for _, c := range []struct {
Name string
Expand Down Expand Up @@ -798,7 +798,7 @@ func TestWorkflowSizeChecker_NumChildWorkflows(t *testing.T) {
}

checker := newWorkflowSizeChecker(workflowSizeLimits{
numPendingChildExecutionLimitError: c.ErrorLimit,
numPendingChildExecutionsLimit: c.ErrorLimit,
}, mutableState, nil, nil, metricsHandler, logger)
err := checker.checkIfNumChildWorkflowsExceedsLimit()

Expand Down
42 changes: 24 additions & 18 deletions service/history/configs/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,15 +188,18 @@ type Config struct {
DurableArchivalEnabled dynamicconfig.BoolPropertyFn

// Size limit related settings
BlobSizeLimitError dynamicconfig.IntPropertyFnWithNamespaceFilter
BlobSizeLimitWarn dynamicconfig.IntPropertyFnWithNamespaceFilter
MemoSizeLimitError dynamicconfig.IntPropertyFnWithNamespaceFilter
MemoSizeLimitWarn dynamicconfig.IntPropertyFnWithNamespaceFilter
HistorySizeLimitError dynamicconfig.IntPropertyFnWithNamespaceFilter
HistorySizeLimitWarn dynamicconfig.IntPropertyFnWithNamespaceFilter
HistoryCountLimitError dynamicconfig.IntPropertyFnWithNamespaceFilter
HistoryCountLimitWarn dynamicconfig.IntPropertyFnWithNamespaceFilter
NumPendingChildExecutionLimitError dynamicconfig.IntPropertyFnWithNamespaceFilter
BlobSizeLimitError dynamicconfig.IntPropertyFnWithNamespaceFilter
BlobSizeLimitWarn dynamicconfig.IntPropertyFnWithNamespaceFilter
MemoSizeLimitError dynamicconfig.IntPropertyFnWithNamespaceFilter
MemoSizeLimitWarn dynamicconfig.IntPropertyFnWithNamespaceFilter
HistorySizeLimitError dynamicconfig.IntPropertyFnWithNamespaceFilter
HistorySizeLimitWarn dynamicconfig.IntPropertyFnWithNamespaceFilter
HistoryCountLimitError dynamicconfig.IntPropertyFnWithNamespaceFilter
HistoryCountLimitWarn dynamicconfig.IntPropertyFnWithNamespaceFilter
NumPendingChildExecutionsLimit dynamicconfig.IntPropertyFnWithNamespaceFilter
NumPendingActivitiesLimit dynamicconfig.IntPropertyFnWithNamespaceFilter
NumPendingSignalsLimit dynamicconfig.IntPropertyFnWithNamespaceFilter
NumPendingCancelsRequestLimit dynamicconfig.IntPropertyFnWithNamespaceFilter

// DefaultActivityRetryOptions specifies the out-of-box retry policy if
// none is configured on the Activity by the user.
Expand Down Expand Up @@ -427,15 +430,18 @@ func NewConfig(dc *dynamicconfig.Collection, numberOfShards int32, isAdvancedVis
return false
},

BlobSizeLimitError: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.BlobSizeLimitError, 2*1024*1024),
BlobSizeLimitWarn: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.BlobSizeLimitWarn, 512*1024),
MemoSizeLimitError: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.MemoSizeLimitError, 2*1024*1024),
MemoSizeLimitWarn: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.MemoSizeLimitWarn, 2*1024),
NumPendingChildExecutionLimitError: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.NumPendingChildExecutionLimitError, 1000),
HistorySizeLimitError: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.HistorySizeLimitError, 50*1024*1024),
HistorySizeLimitWarn: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.HistorySizeLimitWarn, 10*1024*1024),
HistoryCountLimitError: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.HistoryCountLimitError, 50*1024),
HistoryCountLimitWarn: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.HistoryCountLimitWarn, 10*1024),
BlobSizeLimitError: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.BlobSizeLimitError, 2*1024*1024),
BlobSizeLimitWarn: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.BlobSizeLimitWarn, 512*1024),
MemoSizeLimitError: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.MemoSizeLimitError, 2*1024*1024),
MemoSizeLimitWarn: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.MemoSizeLimitWarn, 2*1024),
NumPendingChildExecutionsLimit: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.NumPendingChildExecutionsLimitError, 50000),
NumPendingActivitiesLimit: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.NumPendingActivitiesLimitError, 50000),
NumPendingSignalsLimit: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.NumPendingSignalsLimitError, 50000),
NumPendingCancelsRequestLimit: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.NumPendingCancelRequestsLimitError, 50000),
HistorySizeLimitError: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.HistorySizeLimitError, 50*1024*1024),
HistorySizeLimitWarn: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.HistorySizeLimitWarn, 10*1024*1024),
HistoryCountLimitError: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.HistoryCountLimitError, 50*1024),
HistoryCountLimitWarn: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.HistoryCountLimitWarn, 10*1024),

ThrottledLogRPS: dc.GetIntProperty(dynamicconfig.HistoryThrottledLogRPS, 4),
EnableStickyQuery: dc.GetBoolPropertyFnWithNamespaceFilter(dynamicconfig.EnableStickyQuery, true),
Expand Down
4 changes: 2 additions & 2 deletions service/history/historyEngine2_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -991,7 +991,7 @@ func (s *engine2Suite) TestRespondWorkflowTaskCompleted_StartChildWorkflow_Excee
response := &persistence.GetWorkflowExecutionResponse{State: workflow.TestCloneToProto(ms)}
s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).Return(response, nil).AnyTimes()

s.historyEngine.shard.GetConfig().NumPendingChildExecutionLimitError = func(namespace string) int {
s.historyEngine.shard.GetConfig().NumPendingChildExecutionsLimit = func(namespace string) int {
return 5
}
_, err := s.historyEngine.RespondWorkflowTaskCompleted(metrics.AddMetricsContext(context.Background()), &historyservice.RespondWorkflowTaskCompletedRequest{
Expand All @@ -1005,7 +1005,7 @@ func (s *engine2Suite) TestRespondWorkflowTaskCompleted_StartChildWorkflow_Excee

s.Error(err)
s.Assert().Equal([]string{"the number of pending child workflow executions, 5, " +
"has reached the error limit of 5 established with \"limit.numPendingChildExecution.error\""}, s.errorMessages)
"has reached the error limit of 5 established with \"limit.numPendingChildExecutions.error\""}, s.errorMessages)
}

func (s *engine2Suite) TestStartWorkflowExecution_BrandNew() {
Expand Down
14 changes: 8 additions & 6 deletions service/history/workflowTaskHandlerCallbacks.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -450,12 +450,14 @@ func (handler *workflowTaskHandlerCallbacksImpl) handleWorkflowTaskCompleted(
namespace := namespaceEntry.Name()
workflowSizeChecker := newWorkflowSizeChecker(
workflowSizeLimits{
blobSizeLimitWarn: handler.config.BlobSizeLimitWarn(namespace.String()),
blobSizeLimitError: handler.config.BlobSizeLimitError(namespace.String()),
memoSizeLimitWarn: handler.config.MemoSizeLimitWarn(namespace.String()),
memoSizeLimitError: handler.config.MemoSizeLimitError(namespace.String()),
numPendingChildExecutionLimitError: handler.config.NumPendingChildExecutionLimitError(namespace.
String()),
blobSizeLimitWarn: handler.config.BlobSizeLimitWarn(namespace.String()),
blobSizeLimitError: handler.config.BlobSizeLimitError(namespace.String()),
memoSizeLimitWarn: handler.config.MemoSizeLimitWarn(namespace.String()),
memoSizeLimitError: handler.config.MemoSizeLimitError(namespace.String()),
numPendingChildExecutionsLimit: handler.config.NumPendingChildExecutionsLimit(namespace.String()),
numPendingActivitiesLimit: handler.config.NumPendingActivitiesLimit(namespace.String()),
numPendingSignalsLimit: handler.config.NumPendingSignalsLimit(namespace.String()),
numPendingCancelsRequestLimit: handler.config.NumPendingCancelsRequestLimit(namespace.String()),
},
ms,
handler.searchAttributesValidator,
Expand Down
0