add category to application error, do not log on benign errors for activities

Author: THardy98

temporalio/exceptions.py

@@ -69,6 +69,15 @@ def __init__(
         self.workflow_type = workflow_type
         self.run_id = run_id
 
+class ApplicationErrorCategory(IntEnum):
+    """Severity category for your application error. Maps to corresponding client-side logging/metrics behaviors"""
+
+    UNSPECIFIED = int(
+        temporalio.api.enums.v1.ApplicationErrorCategory.APPLICATION_ERROR_CATEGORY_UNSPECIFIED
+    )
+    BENIGN = int(
+            temporalio.api.enums.v1.ApplicationErrorCategory.APPLICATION_ERROR_CATEGORY_BENIGN
+    )
 
 class ApplicationError(FailureError):
     """Error raised during workflow/activity execution."""
@@ -80,6 +89,7 @@ def __init__(
         type: Optional[str] = None,
         non_retryable: bool = False,
         next_retry_delay: Optional[timedelta] = None,
+        category: ApplicationErrorCategory = ApplicationErrorCategory.UNSPECIFIED 
     ) -> None:
         """Initialize an application error."""
         super().__init__(
@@ -91,6 +101,7 @@ def __init__(
         self._type = type
         self._non_retryable = non_retryable
         self._next_retry_delay = next_retry_delay
+        self._category = category
 
     @property
     def details(self) -> Sequence[Any]:
@@ -121,6 +132,11 @@ def next_retry_delay(self) -> Optional[timedelta]:
         """
         return self._next_retry_delay
 
+    @property
+    def category(self) -> ApplicationErrorCategory:
+        """Severity category of the application error"""
+        return self._category
+
 
 class CancelledError(FailureError):
     """Error raised on workflow/activity cancellation."""

temporalio/worker/_activity.py

@@ -490,6 +490,17 @@ async def _run_activity(
                         temporalio.exceptions.CancelledError("Cancelled"),
                         completion.result.cancelled.failure,
                     )
+                elif (
+                    isinstance(
+                        err,
+                        temporalio.exceptions.ApplicationError,
+                    )
+                    and err.category == temporalio.exceptions.ApplicationErrorCategory.BENIGN
+                ):
+                    # Do not log for ApplicationError with BENIGN category.
+                    await self._data_converter.encode_failure(
+                        err, completion.result.failed.failure
+                    )
                 else:
                     temporalio.activity.logger.warning(
                         "Completing activity as failed", exc_info=True

tests/worker/test_workflow.py

@@ -86,6 +86,7 @@
 from temporalio.exceptions import (
     ActivityError,
     ApplicationError,
+    ApplicationErrorCategory,
     CancelledError,
     ChildWorkflowError,
     TemporalError,
@@ -7421,3 +7422,59 @@ async def test_workflow_dynamic_config_failure(client: Client):
         await assert_task_fail_eventually(
             handle, message_contains="Dynamic config failure"
         )
+
+@activity.defn
+async def raise_application_error(use_benign: bool):
+    if use_benign:
+        raise ApplicationError(
+            "This is a benign error",
+            category=ApplicationErrorCategory.BENIGN
+        )
+    else:
+        raise ApplicationError(
+            "This is a regular error",
+            category=ApplicationErrorCategory.UNSPECIFIED
+        )
+
+
+@workflow.defn
+class RaiseErrorWorkflow:
+    @workflow.run
+    async def run(self, use_benign: bool):
+        # Execute activity that will raise an error
+        await workflow.execute_activity(
+            raise_application_error,
+            args=[use_benign],
+            start_to_close_timeout=timedelta(seconds=5),
+            retry_policy=RetryPolicy(maximum_attempts=1)
+        )
+
+
+async def test_activity_benign_error_not_logged(client: Client):
+    with LogCapturer().logs_captured(
+        activity.logger.base_logger
+    ) as capturer:
+        async with new_worker(
+            client, RaiseErrorWorkflow, activities=[raise_application_error]
+        ) as worker:
+            # Run with benign error
+            with pytest.raises(WorkflowFailureError):
+                await client.execute_workflow(
+                    RaiseErrorWorkflow.run,
+                    args=[True],
+                    id=str(uuid.uuid4()),
+                    task_queue=worker.task_queue,
+                )
+            
+            assert capturer.find_log("Completing activity as failed") == None
+    
+            # Run with non-benign error
+            with pytest.raises(WorkflowFailureError):
+                await client.execute_workflow(
+                    RaiseErrorWorkflow.run,
+                    args=[False],
+                    id=str(uuid.uuid4()),
+                    task_queue=worker.task_queue,
+                )
+            
+            assert capturer.find_log("Completing activity as failed") != None