Skip to content

Commit

Permalink
Emit metrics on customer ids being too long (cadence-workflow#3529)
Browse files Browse the repository at this point in the history
  • Loading branch information
@andrewjdawson2016
andrewjdawson2016 authored Sep 21, 2020
1 parent 280dba2 commit 3a15e73
Show file tree
Hide file tree
Showing 5 changed files with 63 additions and 25 deletions.
22 changes: 22 additions & 0 deletions .gen/proto/message.pb.go
View file

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 2 additions & 0 deletions common/metrics/defs.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1596,6 +1596,7 @@ const (
CadenceErrUnauthorizedCounter
CadenceErrAuthorizeFailedCounter
CadenceErrRemoteSyncMatchFailedCounter
CadenceErrIDLengthExceededWarnLimit
PersistenceRequests
PersistenceFailures
PersistenceLatency
Expand Down Expand Up @@ -2042,6 +2043,7 @@ var MetricDefs = map[ServiceIdx]map[int]metricDefinition{
CadenceErrUnauthorizedCounter: {metricName: "cadence_errors_unauthorized", metricType: Counter},
CadenceErrAuthorizeFailedCounter: {metricName: "cadence_errors_authorize_failed", metricType: Counter},
CadenceErrRemoteSyncMatchFailedCounter: {metricName: "cadence_errors_remote_syncmatch_failed", metricType: Counter},
CadenceErrIDLengthExceededWarnLimit: {metricName: "cadence_errors_id_length_exceeded_warn_limit", metricType: Counter},
PersistenceRequests: {metricName: "persistence_requests", metricType: Counter},
PersistenceFailures: {metricName: "persistence_errors", metricType: Counter},
PersistenceLatency: {metricName: "persistence_latency", metricType: Timer},
Expand Down
4 changes: 4 additions & 0 deletions common/service/dynamicconfig/constants.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,7 @@ var keys = map[Key]string{
HistoryCountLimitError: "limit.historyCount.error",
HistoryCountLimitWarn: "limit.historyCount.warn",
MaxIDLengthLimit: "limit.maxIDLength",
MaxIDLengthWarnLimit: "limit.maxIDWarnLength",

// frontend settings
FrontendPersistenceMaxQPS: "frontend.persistenceMaxQPS",
Expand Down Expand Up @@ -403,6 +404,9 @@ const (
// MaxIDLengthLimit is the length limit for various IDs, including: Domain, TaskList, WorkflowID, ActivityID, TimerID,
// WorkflowType, ActivityType, SignalName, MarkerName, ErrorReason/FailureReason/CancelCause, Identity, RequestID
MaxIDLengthLimit
// MaxIDLengthWarnLimit is the warn length limit for various IDs, including: Domain, TaskList, WorkflowID, ActivityID, TimerID,
// WorkflowType, ActivityType, SignalName, MarkerName, ErrorReason/FailureReason/CancelCause, Identity, RequestID
MaxIDLengthWarnLimit

// key for frontend

Expand Down
2 changes: 2 additions & 0 deletions service/frontend/service.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,7 @@ type Config struct {
MaxDomainRPSPerInstance dynamicconfig.IntPropertyFnWithDomainFilter
GlobalDomainRPS dynamicconfig.IntPropertyFnWithDomainFilter
MaxIDLengthLimit dynamicconfig.IntPropertyFn
MaxIDLengthWarnLimit dynamicconfig.IntPropertyFn
EnableClientVersionCheck dynamicconfig.BoolPropertyFn
MinRetentionDays dynamicconfig.IntPropertyFn
DisallowQuery dynamicconfig.BoolPropertyFnWithDomainFilter
Expand Down Expand Up @@ -113,6 +114,7 @@ func NewConfig(dc *dynamicconfig.Collection, numHistoryShards int, enableReadFro
MaxDomainRPSPerInstance: dc.GetIntPropertyFilteredByDomain(dynamicconfig.FrontendMaxDomainRPSPerInstance, 1200),
GlobalDomainRPS: dc.GetIntPropertyFilteredByDomain(dynamicconfig.FrontendGlobalDomainRPS, 0),
MaxIDLengthLimit: dc.GetIntProperty(dynamicconfig.MaxIDLengthLimit, 1000),
MaxIDLengthWarnLimit: dc.GetIntProperty(dynamicconfig.MaxIDLengthWarnLimit, 150),
HistoryMgrNumConns: dc.GetIntProperty(dynamicconfig.FrontendHistoryMgrNumConns, 10),
MaxBadBinaries: dc.GetIntPropertyFilteredByDomain(dynamicconfig.FrontendMaxBadBinaries, domain.MaxBadBinaries),
EnableAdminProtection: dc.GetBoolProperty(dynamicconfig.EnableAdminProtection, false),
Expand Down
58 changes: 33 additions & 25 deletions service/frontend/workflowHandler.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -467,14 +467,14 @@ func (wh *WorkflowHandler) PollForActivityTask(
return nil, wh.error(errDomainNotSet, scope)
}

if len(pollRequest.GetDomain()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(pollRequest.GetDomain(), scope) {
return nil, wh.error(errDomainTooLong, scope)
}

if err := wh.validateTaskList(pollRequest.TaskList, scope); err != nil {
return nil, err
}
if len(pollRequest.GetIdentity()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(pollRequest.GetIdentity(), scope) {
return nil, wh.error(errIdentityTooLong, scope)
}

Expand Down Expand Up @@ -551,11 +551,11 @@ func (wh *WorkflowHandler) PollForDecisionTask(
if pollRequest.Domain == nil || pollRequest.GetDomain() == "" {
return nil, wh.error(errDomainNotSet, scope, tagsForErrorLog...)
}
if len(pollRequest.GetDomain()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(pollRequest.GetDomain(), scope) {
return nil, wh.error(errDomainTooLong, scope, tagsForErrorLog...)
}

if len(pollRequest.GetIdentity()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(pollRequest.GetIdentity(), scope) {
return nil, wh.error(errIdentityTooLong, scope, tagsForErrorLog...)
}

Expand Down Expand Up @@ -894,7 +894,7 @@ func (wh *WorkflowHandler) RespondActivityTaskCompleted(
if err != nil {
return wh.error(err, scope)
}
if len(completeRequest.GetIdentity()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(completeRequest.GetIdentity(), scope) {
return wh.error(errIdentityTooLong, scope)
}

Expand Down Expand Up @@ -994,7 +994,7 @@ func (wh *WorkflowHandler) RespondActivityTaskCompletedByID(
return wh.error(errActivityIDNotSet, scope)
}

if len(completeRequest.GetIdentity()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(completeRequest.GetIdentity(), scope) {
return wh.error(errIdentityTooLong, scope)
}

Expand Down Expand Up @@ -1112,7 +1112,7 @@ func (wh *WorkflowHandler) RespondActivityTaskFailed(
return errShuttingDown
}

if len(failedRequest.GetIdentity()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(failedRequest.GetIdentity(), scope) {
return wh.error(errIdentityTooLong, scope)
}

Expand Down Expand Up @@ -1187,7 +1187,7 @@ func (wh *WorkflowHandler) RespondActivityTaskFailedByID(
if activityID == "" {
return wh.error(errActivityIDNotSet, scope)
}
if len(failedRequest.GetIdentity()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(failedRequest.GetIdentity(), scope) {
return wh.error(errIdentityTooLong, scope)
}

Expand Down Expand Up @@ -1294,7 +1294,7 @@ func (wh *WorkflowHandler) RespondActivityTaskCanceled(
return errShuttingDown
}

if len(cancelRequest.GetIdentity()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(cancelRequest.GetIdentity(), scope) {
return wh.error(errIdentityTooLong, scope)
}

Expand Down Expand Up @@ -1381,7 +1381,7 @@ func (wh *WorkflowHandler) RespondActivityTaskCanceledByID(
if activityID == "" {
return wh.error(errActivityIDNotSet, scope)
}
if len(cancelRequest.GetIdentity()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(cancelRequest.GetIdentity(), scope) {
return wh.error(errIdentityTooLong, scope)
}

Expand Down Expand Up @@ -1507,7 +1507,7 @@ func (wh *WorkflowHandler) RespondDecisionTaskCompleted(
return nil, wh.error(err, scope)
}

if len(completeRequest.GetIdentity()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(completeRequest.GetIdentity(), scope) {
return nil, wh.error(errIdentityTooLong, scope)
}

Expand Down Expand Up @@ -1584,7 +1584,7 @@ func (wh *WorkflowHandler) RespondDecisionTaskFailed(
return errShuttingDown
}

if len(failedRequest.GetIdentity()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(failedRequest.GetIdentity(), scope) {
return wh.error(errIdentityTooLong, scope)
}

Expand Down Expand Up @@ -1736,15 +1736,15 @@ func (wh *WorkflowHandler) StartWorkflowExecution(
return nil, wh.error(errDomainNotSet, scope)
}

if len(domainName) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(domainName, scope) {
return nil, wh.error(errDomainTooLong, scope)
}

if startRequest.GetWorkflowId() == "" {
return nil, wh.error(errWorkflowIDNotSet, scope)
}

if len(startRequest.GetWorkflowId()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(startRequest.GetWorkflowId(), scope) {
return nil, wh.error(errWorkflowIDTooLong, scope)
}

Expand All @@ -1764,7 +1764,7 @@ func (wh *WorkflowHandler) StartWorkflowExecution(
return nil, wh.error(errWorkflowTypeNotSet, scope)
}

if len(startRequest.WorkflowType.GetName()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(startRequest.WorkflowType.GetName(), scope) {
return nil, wh.error(errWorkflowTypeTooLong, scope)
}

Expand All @@ -1784,7 +1784,7 @@ func (wh *WorkflowHandler) StartWorkflowExecution(
return nil, wh.error(errRequestIDNotSet, scope)
}

if len(startRequest.GetRequestId()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(startRequest.GetRequestId(), scope) {
return nil, wh.error(errRequestIDTooLong, scope)
}

Expand Down Expand Up @@ -2104,7 +2104,7 @@ func (wh *WorkflowHandler) SignalWorkflowExecution(
return wh.error(errDomainNotSet, scope, getWfIDRunIDTags(wfExecution)...)
}

if len(signalRequest.GetDomain()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(signalRequest.GetDomain(), scope) {
return wh.error(errDomainTooLong, scope, getWfIDRunIDTags(wfExecution)...)
}

Expand All @@ -2117,11 +2117,11 @@ func (wh *WorkflowHandler) SignalWorkflowExecution(
scope, getWfIDRunIDTags(wfExecution)...)
}

if len(signalRequest.GetSignalName()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(signalRequest.GetSignalName(), scope) {
return wh.error(errSignalNameTooLong, scope, getWfIDRunIDTags(wfExecution)...)
}

if len(signalRequest.GetRequestId()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(signalRequest.GetRequestId(), scope) {
return wh.error(errRequestIDTooLong, scope, getWfIDRunIDTags(wfExecution)...)
}

Expand Down Expand Up @@ -2196,7 +2196,7 @@ func (wh *WorkflowHandler) SignalWithStartWorkflowExecution(
return nil, wh.error(errDomainNotSet, scope, getWfIDRunIDTags(wfExecution)...)
}

if len(domainName) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(domainName, scope) {
return nil, wh.error(errDomainTooLong, scope, getWfIDRunIDTags(wfExecution)...)
}

Expand All @@ -2205,7 +2205,7 @@ func (wh *WorkflowHandler) SignalWithStartWorkflowExecution(
scope, getWfIDRunIDTags(wfExecution)...)
}

if len(signalWithStartRequest.GetWorkflowId()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(signalWithStartRequest.GetWorkflowId(), scope) {
return nil, wh.error(errWorkflowIDTooLong, scope, getWfIDRunIDTags(wfExecution)...)
}

Expand All @@ -2214,7 +2214,7 @@ func (wh *WorkflowHandler) SignalWithStartWorkflowExecution(
scope, getWfIDRunIDTags(wfExecution)...)
}

if len(signalWithStartRequest.GetSignalName()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(signalWithStartRequest.GetSignalName(), scope) {
return nil, wh.error(errSignalNameTooLong, scope, getWfIDRunIDTags(wfExecution)...)
}

Expand All @@ -2223,15 +2223,15 @@ func (wh *WorkflowHandler) SignalWithStartWorkflowExecution(
scope, getWfIDRunIDTags(wfExecution)...)
}

if len(signalWithStartRequest.WorkflowType.GetName()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(signalWithStartRequest.WorkflowType.GetName(), scope) {
return nil, wh.error(errWorkflowTypeTooLong, scope, getWfIDRunIDTags(wfExecution)...)
}

if err := wh.validateTaskList(signalWithStartRequest.TaskList, scope); err != nil {
return nil, err
}

if len(signalWithStartRequest.GetRequestId()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(signalWithStartRequest.GetRequestId(), scope) {
return nil, wh.error(errRequestIDTooLong, scope, getWfIDRunIDTags(wfExecution)...)
}

Expand Down Expand Up @@ -3517,7 +3517,7 @@ func (wh *WorkflowHandler) validateTaskList(t *gen.TaskList, scope metrics.Scope
if t == nil || t.Name == nil || t.GetName() == "" {
return wh.error(errTaskListNotSet, scope)
}
if len(t.GetName()) > wh.config.MaxIDLengthLimit() {
if !wh.validIDLength(t.GetName(), scope) {
return wh.error(errTaskListTooLong, scope)
}
return nil
Expand Down Expand Up @@ -3860,6 +3860,14 @@ func (wh *WorkflowHandler) getArchivedHistory(
}, nil
}

func (wh *WorkflowHandler) validIDLength(id string, scope metrics.Scope) bool {
valid := len(id) <= wh.config.MaxIDLengthLimit()
if len(id) > wh.config.MaxIDLengthWarnLimit() {
scope.IncCounter(metrics.CadenceErrIDLengthExceededWarnLimit)
}
return valid
}

func (wh *WorkflowHandler) convertIndexedKeyToThrift(keys map[string]interface{}) map[string]gen.IndexedValueType {
converted := make(map[string]gen.IndexedValueType)
for k, v := range keys {
Expand Down

0 comments on commit 3a15e73

Please sign in to comment.