add support for benign application error category

Author: THardy98

internal/error.go

@@ -119,6 +119,9 @@ Workflow consumers will get an instance of *WorkflowExecutionError. This error w
 */
 
 type (
+	// Category of the error. Maps to logging/metrics behaviours.
+	ApplicationErrorCategory string
+
 	// ApplicationErrorOptions represents a combination of error attributes and additional requests.
 	// All fields are optional, providing flexibility in error customization.
 	//
@@ -137,6 +140,7 @@ type (
 		//
 		// NOTE: This option is supported by Temporal Server >= v1.24.2 older version will ignore this value.
 		NextRetryDelay time.Duration
+		Category       ApplicationErrorCategory
 	}
 
 	// ApplicationError returned from activity implementations with message and optional details.
@@ -150,6 +154,7 @@ type (
 		cause          error
 		details        converter.EncodedValues
 		nextRetryDelay time.Duration
+		category       ApplicationErrorCategory
 	}
 
 	// TimeoutError returned when activity or child workflow timed out.
@@ -380,6 +385,11 @@ var (
 	ErrMissingWorkflowID = errors.New("workflow ID is unset for Nexus operation")
 )
 
+const (
+	// ErrorCategoryBenign indicates an error that is expected under normal operation and should not trigger alerts.
+	ErrorCategoryBenign ApplicationErrorCategory = "benign"
+)
+
 // NewApplicationError create new instance of *ApplicationError with message, type, and optional details.
 func NewApplicationError(msg string, errType string, nonRetryable bool, cause error, details ...interface{}) error {
 	return NewApplicationErrorWithOptions(
@@ -397,6 +407,7 @@ func NewApplicationErrorWithOptions(msg string, errType string, options Applicat
 		cause:          options.Cause,
 		nonRetryable:   options.NonRetryable,
 		nextRetryDelay: options.NextRetryDelay,
+		category:       options.Category,
 	}
 	// When return error to user, use EncodedValues as details and data is ready to be decoded by calling Get
 	details := options.Details
@@ -661,6 +672,11 @@ func (e *ApplicationError) Unwrap() error {
 // a zero value means to use the activities retry policy.
 func (e *ApplicationError) NextRetryDelay() time.Duration { return e.nextRetryDelay }
 
+// Category returns the ApplicationErrorCategory of the error.
+func (e *ApplicationError) Category() ApplicationErrorCategory {
+	return e.category
+}
+
 // Error from error interface
 func (e *TimeoutError) Error() string {
 	msg := fmt.Sprintf("%s (type: %s)", e.message(), e.timeoutType)
@@ -1029,3 +1045,42 @@ func getErrType(err error) string {
 
 	return t.Name()
 }
+
+func applicationErrorCategoryToProto(category ApplicationErrorCategory) enumspb.ApplicationErrorCategory {
+	switch category {
+	case ErrorCategoryBenign:
+		return enumspb.APPLICATION_ERROR_CATEGORY_BENIGN
+	case "":
+		// Zero value maps to unspecified
+		return enumspb.APPLICATION_ERROR_CATEGORY_UNSPECIFIED
+	default:
+		// Fallback to unspecified if unknown case
+		return enumspb.APPLICATION_ERROR_CATEGORY_UNSPECIFIED
+	}
+}
+
+func applicationErrorCategoryFromProto(category enumspb.ApplicationErrorCategory) ApplicationErrorCategory {
+	switch category {
+	case enumspb.APPLICATION_ERROR_CATEGORY_BENIGN:
+		return ErrorCategoryBenign
+	case enumspb.APPLICATION_ERROR_CATEGORY_UNSPECIFIED:
+		// Unspecified maps to zero value
+		return ""
+	default:
+		// Fallback to zero value if unknown case
+		return ""
+	}
+}
+
+func IsBenignApplicationError(err error) bool {
+	var appError *ApplicationError
+	return errors.As(err, &appError) && appError.Category() == ErrorCategoryBenign
+}
+
+func isBenignProtoApplicationFailure(failure *failurepb.Failure) bool {
+	if failure == nil {
+		return false
+	}
+	appFailureInfo := failure.GetApplicationFailureInfo()
+	return appFailureInfo != nil && appFailureInfo.GetCategory() == enumspb.APPLICATION_ERROR_CATEGORY_BENIGN
+}

internal/error_test.go

@@ -723,6 +723,7 @@ func Test_convertErrorToFailure_ApplicationErrorWithExtraRequests(t *testing.T)
 			NonRetryable: true,
 			Cause:        errors.New("cause error"),
 			Details:      []interface{}{"details", 2208},
+			Category:     ErrorCategoryBenign,
 		},
 	)
 	f := fc.ErrorToFailure(err)
@@ -734,15 +735,27 @@ func Test_convertErrorToFailure_ApplicationErrorWithExtraRequests(t *testing.T)
 	require.Equal("cause error", f.GetCause().GetMessage())
 	require.Equal("", f.GetCause().GetApplicationFailureInfo().GetType())
 	require.Nil(f.GetCause().GetCause())
+	require.Equal(enumspb.APPLICATION_ERROR_CATEGORY_BENIGN, f.GetApplicationFailureInfo().GetCategory())
 
 	err2 := fc.FailureToError(f)
 	var applicationErr *ApplicationError
 	require.True(errors.As(err2, &applicationErr))
 	require.Equal("message (type: customType, retryable: false): cause error", applicationErr.Error())
+	require.Equal(ErrorCategoryBenign, applicationErr.Category())
 
 	err2 = errors.Unwrap(err2)
 	require.True(errors.As(err2, &applicationErr))
 	require.Equal("cause error", applicationErr.Error())
+
+	err := NewApplicationErrorWithOptions(
+		"another message",
+		"another customType",
+		ApplicationErrorOptions{
+			Category: "",
+		},
+	)
+	f := fc.ErrorToFailure(err)
+	require.Equal(enumspb.APPLICATION_ERROR_CATEGORY_UNSPECIFIED, f.GetApplicationFailureInfo().GetCategory())
 }
 
 func Test_convertErrorToFailure_EncodeMessage(t *testing.T) {
@@ -1104,6 +1117,7 @@ func Test_convertFailureToError_ApplicationFailure(t *testing.T) {
 			Type:         "MyCoolType",
 			NonRetryable: true,
 			Details:      details,
+			Category:     enumspb.APPLICATION_ERROR_CATEGORY_BENIGN,
 		}},
 		Cause: &failurepb.Failure{
 			Message: "cause message",
@@ -1120,6 +1134,7 @@ func Test_convertFailureToError_ApplicationFailure(t *testing.T) {
 	require.Equal("message (type: MyCoolType, retryable: false): cause message (type: UnknownType, retryable: true)", applicationErr.Error())
 	require.Equal("MyCoolType", applicationErr.Type())
 	require.Equal(true, applicationErr.NonRetryable())
+	require.Equal(ErrorCategoryBenign, applicationErr.Category())
 	var str string
 	var n int
 	require.NoError(applicationErr.Details(&str, &n))
@@ -1149,8 +1164,9 @@ func Test_convertFailureToError_ApplicationFailure(t *testing.T) {
 	f = &failurepb.Failure{
 		Message: "message",
 		FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{
-			Type:    "CoolError",
-			Details: details,
+			Type:     "CoolError",
+			Details:  details,
+			Category: enumspb.APPLICATION_ERROR_CATEGORY_UNSPECIFIED,
 		}},
 	}
 
@@ -1160,6 +1176,7 @@ func Test_convertFailureToError_ApplicationFailure(t *testing.T) {
 	require.Equal("message (type: CoolError, retryable: true)", coolErr.Error())
 	require.Equal("CoolError", coolErr.Type())
 	require.Equal(false, coolErr.NonRetryable())
+	require.Equal("", coolErr.Category())
 }
 
 func Test_convertFailureToError_CanceledFailure(t *testing.T) {

internal/failure_converter.go

@@ -115,6 +115,7 @@ func (dfc *DefaultFailureConverter) ErrorToFailure(err error) *failurepb.Failure
 			NonRetryable:   err.NonRetryable(),
 			Details:        convertErrDetailsToPayloads(err.details, dfc.dataConverter),
 			NextRetryDelay: delay,
+			Category:       applicationErrorCategoryToProto(err.Category()),
 		}
 		failure.FailureInfo = &failurepb.Failure_ApplicationFailureInfo{ApplicationFailureInfo: failureInfo}
 	case *CanceledError:
@@ -250,6 +251,7 @@ func (dfc *DefaultFailureConverter) FailureToError(failure *failurepb.Failure) e
 					Cause:          dfc.FailureToError(failure.GetCause()),
 					Details:        []interface{}{details},
 					NextRetryDelay: nextRetryDelay,
+					Category:       applicationErrorCategoryFromProto(applicationFailureInfo.GetCategory()),
 				},
 			)
 		}

internal/internal_task_handlers.go

@@ -2311,7 +2311,11 @@ func (ath *activityTaskHandlerImpl) Execute(taskQueue string, t *workflowservice
 		return nil, ctx.Err()
 	}
 	if err != nil && err != ErrActivityResultPending {
-		ath.logger.Error("Activity error.",
+		logFunc := ath.logger.Error // Default to Error
+		if IsBenignApplicationError(err) {
+			logFunc = ath.logger.Debug // Use Debug for benign application errors
+		}
+		logFunc("Activity error.",
 			tagWorkflowID, t.WorkflowExecution.GetWorkflowId(),
 			tagRunID, t.WorkflowExecution.GetRunId(),
 			tagActivityType, activityType,

internal/internal_task_pollers.go

@@ -474,19 +474,26 @@ func (wtp *workflowTaskPoller) RespondTaskCompletedWithMetrics(
 ) (response *workflowservice.RespondWorkflowTaskCompletedResponse, err error) {
 	metricsHandler := wtp.metricsHandler.WithTags(metrics.WorkflowTags(task.WorkflowType.GetName()))
 	if taskErr != nil {
-		wtp.logger.Warn("Failed to process workflow task.",
-			tagWorkflowType, task.WorkflowType.GetName(),
-			tagWorkflowID, task.WorkflowExecution.GetWorkflowId(),
-			tagRunID, task.WorkflowExecution.GetRunId(),
-			tagAttempt, task.Attempt,
-			tagError, taskErr)
 		failWorkflowTask := wtp.errorToFailWorkflowTask(task.TaskToken, taskErr)
+		completedRequest = failWorkflowTask
+
 		failureReason := "WorkflowError"
 		if failWorkflowTask.Cause == enumspb.WORKFLOW_TASK_FAILED_CAUSE_NON_DETERMINISTIC_ERROR {
 			failureReason = "NonDeterminismError"
 		}
-		incrementWorkflowTaskFailureCounter(metricsHandler, failureReason)
-		completedRequest = failWorkflowTask
+
+		logFunc := wtp.logger.Warn
+		if IsBenignApplicationError(taskErr) {
+			logFunc = wtp.logger.Debug
+		} else {
+			incrementWorkflowTaskFailureCounter(metricsHandler, failureReason)
+		}
+		logFunc("Failed to process workflow task.",
+			tagWorkflowType, task.WorkflowType.GetName(),
+			tagWorkflowID, task.WorkflowExecution.GetWorkflowId(),
+			tagRunID, task.WorkflowExecution.GetRunId(),
+			tagAttempt, task.Attempt,
+			tagError, taskErr)
 	}
 
 	metricsHandler.Timer(metrics.WorkflowTaskExecutionLatency).Record(time.Since(startTime))
@@ -705,7 +712,7 @@ func (lath *localActivityTaskHandler) executeLocalActivityTask(task *localActivi
 				metricsHandler.Counter(metrics.LocalActivityErrorCounter).Inc(1)
 				err = newPanicError(p, st)
 			}
-			if err != nil {
+			if err != nil && !IsBenignApplicationError(err) {
 				metricsHandler.Counter(metrics.LocalActivityFailedCounter).Inc(1)
 				metricsHandler.Counter(metrics.LocalActivityExecutionFailedCounter).Inc(1)
 			}
@@ -1104,8 +1111,10 @@ func (atp *activityTaskPoller) ProcessTask(task interface{}) error {
 		return err
 	}
 	// in case if activity execution failed, request should be of type RespondActivityTaskFailedRequest
-	if _, ok := request.(*workflowservice.RespondActivityTaskFailedRequest); ok {
-		activityMetricsHandler.Counter(metrics.ActivityExecutionFailedCounter).Inc(1)
+	if req, ok := request.(*workflowservice.RespondActivityTaskFailedRequest); ok {
+		if !isBenignProtoApplicationFailure(req.Failure) {
+			activityMetricsHandler.Counter(metrics.ActivityExecutionFailedCounter).Inc(1)
+		}
 	}
 	activityMetricsHandler.Timer(metrics.ActivityExecutionLatency).Record(time.Since(executionStartTime))