Automated fix for refs/heads/o11y_part1

BUILD

@@ -2254,6 +2254,28 @@ grpc_cc_library(
     ],
 )
 
+grpc_cc_library(
+    name = "grpc_rpc_encoding",
+    srcs = [
+        "src/cpp/ext/filters/census/rpc_encoding.cc",
+    ],
+    hdrs = [
+        "src/cpp/ext/filters/census/rpc_encoding.h",
+    ],
+    external_deps = [
+        "absl/base",
+        "absl/base:core_headers",
+        "absl/meta:type_traits",
+        "absl/status",
+        "absl/strings",
+        "absl/time",
+    ],
+    language = "c++",
+    tags = ["nofixdeps"],
+    visibility = ["@grpc:grpc_python_observability"],
+    deps = ["gpr_platform"],
+)
+
 grpc_cc_library(
     name = "grpc_opencensus_plugin",
     srcs = [

bazel/grpc_build_system.bzl

@@ -113,6 +113,7 @@ def _update_visibility(visibility):
         "tsi": PRIVATE,
         "xds": PRIVATE,
         "xds_client_core": PRIVATE,
+        "grpc_python_observability": PRIVATE,
     }
     final_visibility = []
     for rule in visibility:

requirements.bazel.txt

@@ -15,3 +15,4 @@ gevent==21.12.0
 zope.event==4.5.0
 setuptools==44.1.1
 xds-protos==0.0.11
+opencensus==0.10.0

setup.cfg

@@ -27,6 +27,7 @@ inputs =
     src/python/grpcio/grpc/experimental
     src/python/grpcio/grpc
     src/python/grpcio_tests/tests_aio
+    src/python/grpcio_observability/grpc_observability
     examples/python/auth
     examples/python/helloworld
 exclude =

src/cpp/ext/gcp/BUILD

@@ -81,7 +81,10 @@ grpc_cc_library(
         "absl/types:optional",
     ],
     language = "c++",
-    visibility = ["//test:__subpackages__"],
+    visibility = [
+        "//test:__subpackages__",
+        "@grpc:grpc_python_observability",
+    ],
     deps = [
         "//:gpr",
         "//:grpc_base",

src/python/grpcio/grpc/BUILD.bazel

@@ -94,6 +94,11 @@ py_library(
     srcs = ["_typing.py"],
 )
 
+py_library(
+    name = "_observability",
+    srcs = ["_observability.py"],
+)
+
 py_library(
     name = "grpcio",
     srcs = ["__init__.py"],
@@ -102,6 +107,7 @@ py_library(
     ],
     imports = ["../"],
     deps = [
+        ":_observability",
         ":_runtime_protos",
         ":_simple_stubs",
         ":_typing",

src/python/grpcio/grpc/__init__.py

@@ -2083,71 +2083,31 @@ class Compression(enum.IntEnum):
 ###################################  __all__  #################################
 
 __all__ = (
-    'FutureTimeoutError',
-    'FutureCancelledError',
-    'Future',
-    'ChannelConnectivity',
-    'StatusCode',
-    'Status',
-    'RpcError',
-    'RpcContext',
-    'Call',
-    'ChannelCredentials',
-    'CallCredentials',
-    'AuthMetadataContext',
-    'AuthMetadataPluginCallback',
-    'AuthMetadataPlugin',
-    'Compression',
-    'ClientCallDetails',
-    'ServerCertificateConfiguration',
-    'ServerCredentials',
-    'LocalConnectionType',
-    'UnaryUnaryMultiCallable',
-    'UnaryStreamMultiCallable',
-    'StreamUnaryMultiCallable',
-    'StreamStreamMultiCallable',
-    'UnaryUnaryClientInterceptor',
-    'UnaryStreamClientInterceptor',
-    'StreamUnaryClientInterceptor',
-    'StreamStreamClientInterceptor',
-    'Channel',
-    'ServicerContext',
-    'RpcMethodHandler',
-    'HandlerCallDetails',
-    'GenericRpcHandler',
-    'ServiceRpcHandler',
-    'Server',
-    'ServerInterceptor',
-    'unary_unary_rpc_method_handler',
-    'unary_stream_rpc_method_handler',
-    'stream_unary_rpc_method_handler',
-    'stream_stream_rpc_method_handler',
-    'method_handlers_generic_handler',
-    'ssl_channel_credentials',
-    'metadata_call_credentials',
-    'access_token_call_credentials',
-    'composite_call_credentials',
-    'composite_channel_credentials',
-    'compute_engine_channel_credentials',
-    'local_channel_credentials',
-    'local_server_credentials',
-    'alts_channel_credentials',
-    'alts_server_credentials',
-    'ssl_server_credentials',
-    'ssl_server_certificate_configuration',
-    'dynamic_ssl_server_credentials',
-    'channel_ready_future',
-    'insecure_channel',
-    'secure_channel',
-    'intercept_channel',
-    'server',
-    'protos',
-    'services',
-    'protos_and_services',
-    'xds_channel_credentials',
-    'xds_server_credentials',
-    'insecure_server_credentials',
-)
+    'FutureTimeoutError', 'FutureCancelledError', 'Future',
+    'ChannelConnectivity', 'StatusCode', 'Status', 'RpcError', 'RpcContext',
+    'Call', 'ChannelCredentials', 'CallCredentials', 'AuthMetadataContext',
+    'AuthMetadataPluginCallback', 'AuthMetadataPlugin', 'Compression',
+    'ClientCallDetails', 'ServerCertificateConfiguration', 'ServerCredentials',
+    'LocalConnectionType', 'UnaryUnaryMultiCallable',
+    'UnaryStreamMultiCallable', 'StreamUnaryMultiCallable',
+    'StreamStreamMultiCallable', 'UnaryUnaryClientInterceptor',
+    'UnaryStreamClientInterceptor', 'StreamUnaryClientInterceptor',
+    'StreamStreamClientInterceptor', 'Channel', 'ServicerContext',
+    'RpcMethodHandler', 'HandlerCallDetails', 'GenericRpcHandler',
+    'ServiceRpcHandler', 'Server', 'ServerInterceptor',
+    'unary_unary_rpc_method_handler', 'unary_stream_rpc_method_handler',
+    'stream_unary_rpc_method_handler', 'stream_stream_rpc_method_handler',
+    'method_handlers_generic_handler', 'ssl_channel_credentials',
+    'metadata_call_credentials', 'access_token_call_credentials',
+    'composite_call_credentials', 'composite_channel_credentials',
+    'compute_engine_channel_credentials', 'local_channel_credentials',
+    'local_server_credentials', 'alts_channel_credentials',
+    'alts_server_credentials', 'ssl_server_credentials',
+    'ssl_server_certificate_configuration', 'dynamic_ssl_server_credentials',
+    'channel_ready_future', 'insecure_channel', 'secure_channel',
+    'intercept_channel', 'server', 'protos', 'services', 'protos_and_services',
+    'xds_channel_credentials', 'xds_server_credentials',
+    'insecure_server_credentials')
 
 ############################### Extension Shims ################################
 

src/python/grpcio/grpc/_channel.py

@@ -14,6 +14,7 @@
 """Invocation-side implementation of gRPC Python."""
 
 import copy
+from datetime import datetime
 import functools
 import logging
 import os
@@ -28,6 +29,7 @@
 from grpc import _common  # pytype: disable=pyi-error
 from grpc import _compression  # pytype: disable=pyi-error
 from grpc import _grpcio_metadata  # pytype: disable=pyi-error
+from grpc import _observability  # pytype: disable=pyi-error
 from grpc._cython import cygrpc
 from grpc._typing import ChannelArgumentType
 from grpc._typing import DeserializingFunction
@@ -114,6 +116,9 @@ class _RPCState(object):
     cancelled: bool
     callbacks: List[NullaryCallbackType]
     fork_epoch: Optional[int]
+    rpc_start_time: Optional[datetime]
+    rpc_end_time: Optional[datetime]
+    method: Optional[str]
 
     def __init__(self, due: Sequence[cygrpc.OperationType],
                  initial_metadata: Optional[MetadataType],
@@ -136,6 +141,11 @@ def __init__(self, due: Sequence[cygrpc.OperationType],
         self.code = code
         self.details = details
         self.debug_error_string = None
+        # The following three fileds are used for observability.
+        # Updates to those fields do not trigger self.condition.
+        self.rpc_start_time = None
+        self.rpc_end_time = None
+        self.method = None
 
         # The semantics of grpc.Future.cancel and grpc.Future.cancelled are
         # slightly wonky, so they have to be tracked separately from the rest of the
@@ -191,6 +201,8 @@ def _handle_event(
                     state.code = code
                     state.details = batch_operation.details()
                     state.debug_error_string = batch_operation.error_string()
+            state.rpc_end_time = datetime.utcnow()
+            _observability.maybe_record_rpc_latency(state)
             callbacks.extend(state.callbacks)
             state.callbacks = None
     return callbacks
@@ -1014,6 +1026,8 @@ def _blocking(
                     operations,
                     None,
                 ),), self._context)
+            state.rpc_start_time = datetime.utcnow()
+            state.method = _common.decode(self._method)
             event = call.next_event()
             _handle_event(event, state, self._response_deserializer)
             return state, call
@@ -1062,6 +1076,8 @@ def future(
                 self._method, None, deadline, metadata,
                 None if credentials is None else credentials._credentials,
                 (operations,), event_handler, self._context)
+            state.rpc_start_time = datetime.utcnow()
+            state.method = _common.decode(self._method)
             return _MultiThreadedRendezvous(state, call,
                                             self._response_deserializer,
                                             deadline)
@@ -1120,6 +1136,8 @@ def __call__(  # pylint: disable=too-many-locals
             cygrpc.PropagationConstants.GRPC_PROPAGATE_DEFAULTS, self._method,
             None, _determine_deadline(deadline), metadata, call_credentials,
             operations_and_tags, self._context)
+        state.rpc_start_time = datetime.utcnow()
+        state.method = _common.decode(self._method)
         return _SingleThreadedRendezvous(state, call,
                                          self._response_deserializer, deadline)
 
@@ -1180,6 +1198,8 @@ def __call__(  # pylint: disable=too-many-locals
                 None if credentials is None else credentials._credentials,
                 operations, _event_handler(state, self._response_deserializer),
                 self._context)
+            state.rpc_start_time = datetime.utcnow()
+            state.method = _common.decode(self._method)
             return _MultiThreadedRendezvous(state, call,
                                             self._response_deserializer,
                                             deadline)
@@ -1223,6 +1243,8 @@ def _blocking(
             None if credentials is None else credentials._credentials,
             _stream_unary_invocation_operations_and_tags(
                 augmented_metadata, initial_metadata_flags), self._context)
+        state.rpc_start_time = datetime.utcnow()
+        state.method = _common.decode(self._method)
         _consume_request_iterator(request_iterator, state, call,
                                   self._request_serializer, None)
         while True:
@@ -1281,6 +1303,8 @@ def future(
             _stream_unary_invocation_operations(metadata,
                                                 initial_metadata_flags),
             event_handler, self._context)
+        state.rpc_start_time = datetime.utcnow()
+        state.method = _common.decode(self._method)
         _consume_request_iterator(request_iterator, state, call,
                                   self._request_serializer, event_handler)
         return _MultiThreadedRendezvous(state, call,
@@ -1338,6 +1362,8 @@ def __call__(
             None, _determine_deadline(deadline), augmented_metadata,
             None if credentials is None else credentials._credentials,
             operations, event_handler, self._context)
+        state.rpc_start_time = datetime.utcnow()
+        state.method = _common.decode(self._method)
         _consume_request_iterator(request_iterator, state, call,
                                   self._request_serializer, event_handler)
         return _MultiThreadedRendezvous(state, call,

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

@@ -17,7 +17,7 @@ cdef object _custom_op_on_c_call(int op, grpc_call *call):
   raise NotImplementedError("No custom hooks are implemented")
 
 def install_context_from_request_call_event(RequestCallEvent event):
-  pass
+  maybe_save_server_trace_context(event)
 
 def uninstall_context():
   pass

src/python/grpcio/grpc/_cython/_cygrpc/channel.pxd.pxi

@@ -26,6 +26,10 @@ cdef class _CallState:
 
   cdef grpc_call *c_call
   cdef set due
+  # call_tracer_capsule should have type of grpc._observability.ClientCallTracerCapsule
+  cdef object call_tracer_capsule
+  cdef void maybe_set_client_call_tracer_on_call(self, bytes method_name) except *
+  cdef void maybe_delete_call_tracer(self) except *
 
 
 cdef class _ChannelState: