[AbortError] And and check AbortError while abort (#33969)

Fix: https://github.com/grpc/grpc/issues/33890

In case of abort, currently we don't log anything, this change added an
`AbortError` as the default error for abort, if any other error
happened, we'll log the error so user will be aware of the other error.

Related gRFC: https://github.com/grpc/proposal/pull/388

### Testing
* Tested locally, after this change, any non-AbortError will be logged,
sample log:
```
ERROR:grpc._server:Exception happened while abort: Other exception happened!
Traceback (most recent call last):
  File "/usr/local/google/home/xuanwn/.cache/bazel/_bazel_xuanwn/da3828576aa39e99a5c826cc2e2e22fb/sandbox/linux-sandbox/9/execroot/com_github_grpc_grpc/bazel-out/k8-fastbuild/bin/src/python/grpcio_tests/tests/unit/_abort_test.native.runfiles/com_github_grpc_grpc/src/python/grpcio/grpc/_server.py", line 553, in _call_behavior
    response_or_iterator = behavior(argument, context)
  File "/usr/local/google/home/xuanwn/.cache/bazel/_bazel_xuanwn/da3828576aa39e99a5c826cc2e2e22fb/sandbox/linux-sandbox/9/execroot/com_github_grpc_grpc/bazel-out/k8-fastbuild/bin/src/python/grpcio_tests/tests/unit/_abort_test.native.runfiles/com_github_grpc_grpc/src/python/grpcio_tests/tests/unit/_abort_test.py", line 95, in abort_with_status_unary_unary_raise_additional_exception
    raise Exception("Other exception happened!")
Exception: Other exception happened!
```

src/python/grpcio/grpc/__init__.py

@@ -21,6 +21,8 @@ import sys
 
 from grpc import _compression
 from grpc._cython import cygrpc as _cygrpc
+from grpc._cython.cygrpc import AbortError
+from grpc._cython.cygrpc import BaseError
 from grpc._runtime_protos import protos
 from grpc._runtime_protos import protos_and_services
 from grpc._runtime_protos import services
@@ -310,7 +312,7 @@ class Status(abc.ABC):
 #############################  gRPC Exceptions  ################################
 
 
-class RpcError(Exception):
+class RpcError(BaseError):
     """Raised by the gRPC library to indicate non-OK-status RPC termination."""
 
 
@@ -2242,6 +2244,8 @@ __all__ = (
     "ServiceRpcHandler",
     "Server",
     "ServerInterceptor",
+    "AbortError",
+    "BaseError",
     "unary_unary_rpc_method_handler",
     "unary_stream_rpc_method_handler",
     "stream_unary_rpc_method_handler",

src/python/grpcio/grpc/_cython/_cygrpc/aio/common.pyx.pxi

@@ -80,30 +80,6 @@ _COMPRESSION_METADATA_STRING_MAPPING = {
     CompressionAlgorithm.gzip: 'gzip',
 }
 
-class BaseError(Exception):
-    """The base class for exceptions generated by gRPC AsyncIO stack."""
-
-
-class UsageError(BaseError):
-    """Raised when the usage of API by applications is inappropriate.
-
-    For example, trying to invoke RPC on a closed channel, mixing two styles
-    of streaming API on the client side. This exception should not be
-    suppressed.
-    """
-
-
-class AbortError(BaseError):
-    """Raised when calling abort in servicer methods.
-
-    This exception should not be suppressed. Applications may catch it to
-    perform certain clean-up logic, and then re-raise it.
-    """
-
-
-class InternalError(BaseError):
-    """Raised upon unexpected errors in native code."""
-
 
 def schedule_coro_threadsafe(object coro, object loop):
     try:

src/python/grpcio/grpc/_cython/_cygrpc/common.pyx.pxi

@@ -0,0 +1,40 @@
+# Copyright 2023 The gRPC Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+class BaseError(Exception):
+    """
+    The base class for exceptions generated by gRPC.
+    """
+
+class UsageError(BaseError):
+    """
+    Raised when the usage of API by applications is inappropriate.
+
+    For example, trying to invoke RPC on a closed channel, mixing two styles
+    of streaming API on the client side. This exception should not be
+    suppressed.
+    """
+
+class AbortError(BaseError):
+    """
+    Raised when calling abort in servicer methods.
+
+    This exception should not be suppressed. Applications may catch it to
+    perform certain clean-up logic, and then re-raise it.
+    """
+
+class InternalError(BaseError):
+    """
+    Raised upon unexpected errors in native code.
+    """

src/python/grpcio/grpc/_cython/cygrpc.pyx

@@ -55,6 +55,7 @@ include "_cygrpc/tag.pyx.pxi"
 include "_cygrpc/time.pyx.pxi"
 include "_cygrpc/vtable.pyx.pxi"
 include "_cygrpc/_hooks.pyx.pxi"
+include "_cygrpc/common.pyx.pxi"
 include "_cygrpc/observability.pyx.pxi"
 
 include "_cygrpc/grpc_gevent.pyx.pxi"

src/python/grpcio/grpc/_server.py

@@ -41,6 +41,7 @@ from grpc import _common  # pytype: disable=pyi-error
 from grpc import _compression  # pytype: disable=pyi-error
 from grpc import _interceptor  # pytype: disable=pyi-error
 from grpc._cython import cygrpc
+from grpc._cython.cygrpc import AbortError
 from grpc._typing import ArityAgnosticMethodHandler
 from grpc._typing import ChannelArgumentType
 from grpc._typing import DeserializingFunction
@@ -401,7 +402,7 @@ class _Context(grpc.ServicerContext):
             self._state.code = code
             self._state.details = _common.encode(details)
             self._state.aborted = True
-            raise Exception()
+            raise AbortError()
 
     def abort_with_status(self, status: grpc.Status) -> None:
         self._state.trailing_metadata = status.trailing_metadata
@@ -554,6 +555,15 @@ def _call_behavior(
         except Exception as exception:  # pylint: disable=broad-except
             with state.condition:
                 if state.aborted:
+                    if not isinstance(exception, AbortError):
+                        try:
+                            details = f"Exception happened while aborting: {exception}"
+                        except Exception:  # pylint: disable=broad-except
+                            details = (
+                                "Calling abort raised unprintable Exception!"
+                            )
+                            traceback.print_exc()
+                        _LOGGER.exception(details)
                     _abort(
                         state,
                         rpc_event.call,

src/python/grpcio_tests/tests/unit/_api_test.py

@@ -57,6 +57,8 @@ class AllTest(unittest.TestCase):
             "ServiceRpcHandler",
             "Server",
             "ServerInterceptor",
+            "AbortError",
+            "BaseError",
             "LocalConnectionType",
             "local_channel_credentials",
             "local_server_credentials",