Add gRPC filter

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- `opentelemetry-instrumentation-grpc` add supports to filter requests to instrument. ([#1241](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1241))
+
 ### Fixed
 
 - `opentelemetry-instrumentation-boto3sqs` Make propagation compatible with other SQS instrumentations, add 'messaging.url' span attribute, and fix missing package dependencies.
@@ -107,7 +111,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 - `opentelemetry-instrumentation-starlette` Capture custom request/response headers in span attributes
-  ([#1046])(https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1046)
+  ([#1046](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1046))
 
 ### Fixed
 - Prune autoinstrumentation sitecustomize module directory from PYTHONPATH immediately
@@ -130,35 +134,35 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - `opentelemetry-instrumentation-fastapi` Capture custom request/response headers in span attributes
-  ([#1032])(https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1032)
+  ([#1032](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1032))
 - `opentelemetry-instrumentation-django` Capture custom request/response headers in span attributes
-  ([#1024])(https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1024)
+  ([#1024](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1024))
 - `opentelemetry-instrumentation-asgi` Capture custom request/response headers in span attributes
-  ([#1004])(https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1004)
+  ([#1004](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1004))
 - `opentelemetry-instrumentation-psycopg2` extended the sql commenter support of dbapi into psycopg2
   ([#940](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/940))
 - `opentelemetry-instrumentation-falcon` Add support for falcon==1.4.1
-  ([#1000])(https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1000)
+  ([#1000](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1000))
 - `opentelemetry-instrumentation-falcon` Falcon: Capture custom request/response headers in span attributes
-  ([#1003])(https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1003)
+  ([#1003](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1003))
 - `opentelemetry-instrumentation-elasticsearch` no longer creates unique span names by including search target, replaces them with `<target>` and puts the value in attribute `elasticsearch.target`
   ([#1018](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1018))
 - `opentelemetry-instrumentation-pyramid` Handle non-HTTPException exceptions
   ([#1001](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1001))
 - `opentelemetry-instrumentation-system-metrics` restore `SystemMetrics` instrumentation as `SystemMetricsInstrumentor`
   ([#1012](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1012))
 - `opentelemetry-instrumentation-pyramid` Pyramid: Capture custom request/response headers in span attributes
-  ([#1022])(https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1022)
+  ([#1022](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1022))
 
 
 ## [1.10.0-0.29b0](https://github.com/open-telemetry/opentelemetry-python/releases/tag/v1.10.0-0.29b0) - 2022-03-10
 
 - `opentelemetry-instrumentation-wsgi` Capture custom request/response headers in span attributes
-  ([#925])(https://github.com/open-telemetry/opentelemetry-python-contrib/pull/925)
+  ([#925](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/925))
 - `opentelemetry-instrumentation-flask` Flask: Capture custom request/response headers in span attributes
-  ([#952])(https://github.com/open-telemetry/opentelemetry-python-contrib/pull/952)
+  ([#952](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/952))
 - `opentelemetry-instrumentation-tornado` Tornado: Capture custom request/response headers in span attributes
-  ([#950])(https://github.com/open-telemetry/opentelemetry-python-contrib/pull/950)
+  ([#950](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/950))
 
 ### Added
 
@@ -953,7 +957,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#572](https://github.com/open-telemetry/opentelemetry-python/pull/572))
 - `opentelemetry-ext-sqlite3` Initial release
 - `opentelemetry-ext-psycopg2` Implement instrumentor interface, enabling auto-instrumentation
-  ([#694]https://github.com/open-telemetry/opentelemetry-python/pull/694)
+  ([#694](https://github.com/open-telemetry/opentelemetry-python/pull/694))
 - `opentelemetry-ext-asgi` Add ASGI middleware
   ([#716](https://github.com/open-telemetry/opentelemetry-python/pull/716))
 - `opentelemetry-ext-django` Add exclude list for paths and hosts to prevent from tracing

instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/__init__.py

@@ -118,13 +118,63 @@ def serve():
     server = grpc.server(futures.ThreadPoolExecutor(),
                          interceptors = [server_interceptor()])
 
+Filters
+-------
+
+If you prefer to filter specific requests to be instrumented, you can specify
+the condition by assigning filters to instrumentors.
+
+You can write a global server instrumentor as follows:
+
+.. code-block::
+
+    from opentelemetry.instrumentation.grpc import filters, GrpcInstrumentorServer
+
+    grpc_server_instrumentor = GrpcInstrumentorServer(
+        filters = filters.any_of(
+            filters.method_name("SimpleMethod"),
+            filters.method_name("ComplexMethod"),
+        )
+    )
+    grpc_server_instrumentor.instrument()
+
+You can also use the filter to manual instrumentors:
+
+.. code-block::
+
+    my_interceptor = server_interceptor(
+        filters = filters.reverse(filters.method_name("TestMethod"))
+    )
+    server = grpc.server(futures.ThreadPoolExecutor(),
+                         interceptors = [my_interceptor])
+
+``filters`` option also applies to both global and manual client intrumentors.
+
+
+Environment variable
+--------------------
+
+If you'd like to exclude specific services for the instrumentations, you can use
+``OTEL_PYTHON_GRPC_EXCLUDED_SERVICES`` environment variables.
+
+For example, if you assign ``"GRPCTestServer,GRPCHealthServer"`` to the variable,
+then the global interceptor automatically add the filter to exclude requests to
+services ``GRPCTestServer`` and ``GRPCHealthServer``.
+
 """
-from typing import Collection
+import os
+from typing import Callable, Collection, List, Union
 
 import grpc  # pylint:disable=import-self
 from wrapt import wrap_function_wrapper as _wrap
 
 from opentelemetry import trace
+from opentelemetry.instrumentation.grpc.filters import (
+    all_of,
+    any_of,
+    reverse,
+    service_name,
+)
 from opentelemetry.instrumentation.grpc.grpcext import intercept_channel
 from opentelemetry.instrumentation.grpc.package import _instruments
 from opentelemetry.instrumentation.grpc.version import __version__
@@ -145,10 +195,26 @@ class GrpcInstrumentorServer(BaseInstrumentor):
         grpc_server_instrumentor = GrpcInstrumentorServer()
         grpc_server_instrumentor.instrument()
 
+        If you want to add filters that only intercept requests
+        to match the condition, assign filters option to GrpcInstrumentorServer.
+
+        grpc_server_instrumentor = GrpcInstrumentorServer(
+            filters=filters.method_prefix("SimpleMethod"))
+        grpc_server_instrumentor.instrument()
+
     """
 
     # pylint:disable=attribute-defined-outside-init, redefined-outer-name
 
+    def __init__(self, filters=None):
+        excluded_service_filter = _excluded_service_filter()
+        if excluded_service_filter is not None:
+            if filters is None:
+                filters = excluded_service_filter
+            else:
+                filters = all_of(filters, excluded_service_filter)
+        self._filters = filters
+
     def instrumentation_dependencies(self) -> Collection[str]:
         return _instruments
 
@@ -160,11 +226,16 @@ def server(*args, **kwargs):
             if "interceptors" in kwargs:
                 # add our interceptor as the first
                 kwargs["interceptors"].insert(
-                    0, server_interceptor(tracer_provider=tracer_provider)
+                    0,
+                    server_interceptor(
+                        tracer_provider=tracer_provider, filters=self._filters
+                    ),
                 )
             else:
                 kwargs["interceptors"] = [
-                    server_interceptor(tracer_provider=tracer_provider)
+                    server_interceptor(
+                        tracer_provider=tracer_provider, filters=self._filters
+                    )
                 ]
             return self._original_func(*args, **kwargs)
 
@@ -183,8 +254,25 @@ class GrpcInstrumentorClient(BaseInstrumentor):
         grpc_client_instrumentor = GrpcInstrumentorClient()
         grpc_client_instrumentor.instrument()
 
+        If you want to add filters that only intercept requests
+        to match the condition, assign filters option to GrpcInstrumentorClient.
+
+        grpc_client_instrumentor = GrpcInstrumentorClient(
+            filter=filters.reverse(filters.health_check())
+        )
+        grpc_client_instrumentor.instrument()
+
     """
 
+    def __init__(self, filters=None):
+        excluded_service_filter = _excluded_service_filter()
+        if excluded_service_filter is not None:
+            if filters is None:
+                filters = excluded_service_filter
+            else:
+                filters = all_of(filters, excluded_service_filter)
+        self._filters = filters
+
     # Figures out which channel type we need to wrap
     def _which_channel(self, kwargs):
         # handle legacy argument
@@ -221,37 +309,70 @@ def wrapper_fn(self, original_func, instance, args, kwargs):
         tracer_provider = kwargs.get("tracer_provider")
         return intercept_channel(
             channel,
-            client_interceptor(tracer_provider=tracer_provider),
+            client_interceptor(
+                tracer_provider=tracer_provider,
+                filters=self._filters,
+            ),
         )
 
 
-def client_interceptor(tracer_provider=None):
+def client_interceptor(tracer_provider=None, filters=None):
     """Create a gRPC client channel interceptor.
 
     Args:
         tracer: The tracer to use to create client-side spans.
 
+        filters: filter function that returns True if gRPC requests
+                 matches the condition. Default is None and intercept
+                 all requests.
+
     Returns:
         An invocation-side interceptor object.
     """
     from . import _client
 
     tracer = trace.get_tracer(__name__, __version__, tracer_provider)
 
-    return _client.OpenTelemetryClientInterceptor(tracer)
+    return _client.OpenTelemetryClientInterceptor(tracer, filters=filters)
 
 
-def server_interceptor(tracer_provider=None):
+def server_interceptor(tracer_provider=None, filters=None):
     """Create a gRPC server interceptor.
 
     Args:
         tracer: The tracer to use to create server-side spans.
 
+        filters: filter function that returns True if gRPC requests
+                 matches the condition. Default is None and intercept
+                 all requests.
+
     Returns:
         A service-side interceptor object.
     """
     from . import _server
 
     tracer = trace.get_tracer(__name__, __version__, tracer_provider)
 
-    return _server.OpenTelemetryServerInterceptor(tracer)
+    return _server.OpenTelemetryServerInterceptor(tracer, filters=filters)
+
+
+def _excluded_service_filter() -> Union[Callable[[object], bool], None]:
+    services = _parse_services(
+        os.environ.get("OTEL_PYTHON_GRPC_EXCLUDED_SERVICES", "")
+    )
+    if len(services) == 0:
+        return None
+    filters = []
+    for srv in services:
+        filters.append(service_name(srv))
+    return reverse(any_of(*filters))
+
+
+def _parse_services(excluded_services: str) -> List[str]:
+    if excluded_services != "":
+        excluded_service_list = [
+            s.strip() for s in excluded_services.split(",")
+        ]
+    else:
+        excluded_service_list = []
+    return excluded_service_list

instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_client.py

@@ -62,8 +62,9 @@ def callback(response_future):
 class OpenTelemetryClientInterceptor(
     grpcext.UnaryClientInterceptor, grpcext.StreamClientInterceptor
 ):
-    def __init__(self, tracer):
+    def __init__(self, tracer, filters=None):
         self._tracer = tracer
+        self._filter = filters
 
     def _start_span(self, method, **kwargs):
         service, meth = method.lstrip("/").split("/", 1)
@@ -148,6 +149,8 @@ def _intercept(self, request, metadata, client_info, invoker):
         return self._trace_result(span, rpc_info, result)
 
     def intercept_unary(self, request, metadata, client_info, invoker):
+        if self._filter is not None and not self._filter(client_info):
+            return invoker(request, metadata)
         return self._intercept(request, metadata, client_info, invoker)
 
     # For RPCs that stream responses, the result can be a generator. To record
@@ -188,6 +191,9 @@ def intercept_stream(
         if context.get_value(_SUPPRESS_INSTRUMENTATION_KEY):
             return invoker(request_or_iterator, metadata)
 
+        if self._filter is not None and not self._filter(client_info):
+            return invoker(request_or_iterator, metadata)
+
         if client_info.is_server_stream:
             return self._intercept_server_stream(
                 request_or_iterator, metadata, client_info, invoker

instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_server.py

@@ -172,9 +172,10 @@ class OpenTelemetryServerInterceptor(grpc.ServerInterceptor):
     Usage::
 
         tracer = some OpenTelemetry tracer
+        filter = filters.reverse(filters.method_name("service.Foo"))
 
         interceptors = [
-            OpenTelemetryServerInterceptor(tracer),
+            OpenTelemetryServerInterceptor(tracer, filter),
         ]
 
         server = grpc.server(
@@ -183,8 +184,9 @@ class OpenTelemetryServerInterceptor(grpc.ServerInterceptor):
 
     """
 
-    def __init__(self, tracer):
+    def __init__(self, tracer, filters=None):
         self._tracer = tracer
+        self._filter = filters
 
     @contextmanager
     def _set_remote_context(self, servicer_context):
@@ -259,6 +261,9 @@ def _start_span(
         )
 
     def intercept_service(self, continuation, handler_call_details):
+        if self._filter is not None and not self._filter(handler_call_details):
+            return continuation(handler_call_details)
+
         def telemetry_wrapper(behavior, request_streaming, response_streaming):
             def telemetry_interceptor(request_or_iterator, context):