mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-30 14:56:19 +00:00
Vladyslav SimonenkoandGitHub
1dabb876de
## What changed? Start to track the size of the header field in the requests ## Why? There are few reasons to start tracking the header size. First, it helps to picture the size of the request used by header field. Second, unlike payload and memo fields, we currently do not enforce the size of the header size. Better visibility into its size will determine at what level we would need to set up the enforcement. ## How did you test it? - [X] built - [X] run locally and tested manually - ran `samples-go/ctxpropagation` and checked that the header size metric is populated in Prometheus - [ ] covered by existing tests - [ ] added new unit test(s) - [ ] added new functional test(s) ## Potential risks No
74 lines
2.3 KiB
Go
74 lines
2.3 KiB
Go
package api
|
|
|
|
import (
|
|
"context"
|
|
|
|
enumspb "go.temporal.io/api/enums/v1"
|
|
"go.temporal.io/server/common"
|
|
"go.temporal.io/server/common/log/tag"
|
|
"go.temporal.io/server/common/metrics"
|
|
"go.temporal.io/server/common/rpc/interceptor"
|
|
"go.temporal.io/server/service/history/consts"
|
|
historyi "go.temporal.io/server/service/history/interfaces"
|
|
)
|
|
|
|
func ValidateSignal(
|
|
ctx context.Context,
|
|
shard historyi.ShardContext,
|
|
mutableState historyi.MutableState,
|
|
signalPayloadSize int,
|
|
signalHeaderSize int,
|
|
operation string,
|
|
) error {
|
|
config := shard.GetConfig()
|
|
namespaceEntry := mutableState.GetNamespaceEntry()
|
|
namespaceID := namespaceEntry.ID().String()
|
|
namespaceName := namespaceEntry.Name().String()
|
|
workflowID := mutableState.GetExecutionInfo().WorkflowId
|
|
runID := mutableState.GetExecutionState().RunId
|
|
|
|
executionInfo := mutableState.GetExecutionInfo()
|
|
maxAllowedSignals := config.MaximumSignalsPerExecution(namespaceName)
|
|
blobSizeLimitWarn := config.BlobSizeLimitWarn(namespaceName)
|
|
blobSizeLimitError := config.BlobSizeLimitError(namespaceName)
|
|
|
|
metricsHandler := interceptor.GetMetricsHandlerFromContext(ctx, shard.GetLogger())
|
|
metrics.HeaderSize.With(metricsHandler.WithTags(metrics.HeaderCallsiteTag(operation))).Record(int64(signalHeaderSize))
|
|
if err := common.CheckEventBlobSizeLimit(
|
|
signalPayloadSize,
|
|
blobSizeLimitWarn,
|
|
blobSizeLimitError,
|
|
namespaceName,
|
|
workflowID,
|
|
runID,
|
|
metricsHandler.WithTags(
|
|
metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String()),
|
|
),
|
|
shard.GetThrottledLogger(),
|
|
tag.BlobSizeViolationOperation(operation),
|
|
); err != nil {
|
|
return err
|
|
}
|
|
|
|
if maxAllowedSignals > 0 && int(executionInfo.SignalCount) >= maxAllowedSignals {
|
|
shard.GetLogger().Info("Execution limit reached for maximum signals",
|
|
tag.WorkflowNamespaceID(namespaceID),
|
|
tag.WorkflowID(workflowID),
|
|
tag.WorkflowRunID(runID),
|
|
tag.WorkflowSignalCount(executionInfo.SignalCount),
|
|
)
|
|
return consts.ErrSignalsLimitExceeded
|
|
}
|
|
|
|
if mutableState.IsWorkflowCloseAttempted() && mutableState.HasStartedWorkflowTask() {
|
|
shard.GetThrottledLogger().Info("Signal rejected because workflow is closing",
|
|
tag.WorkflowNamespaceID(namespaceID),
|
|
tag.WorkflowID(workflowID),
|
|
tag.WorkflowRunID(runID),
|
|
)
|
|
return consts.ErrWorkflowClosing
|
|
}
|
|
|
|
return nil
|
|
}
|