Add gRPC filter (#1241)

CHANGELOG.md

@@ -8,6 +8,8 @@ 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))
 - Flask sqlalchemy psycopg2 integration
  ([#1224](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1224))
 - Add metric instrumentation in fastapi
@@ -125,7 +127,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
@@ -148,35 +150,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
 
@@ -971,7 +973,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,62 @@ 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(
+ filter_ = filters.any_of(
+ filters.method_name("SimpleMethod"),
+ filters.method_name("ComplexMethod"),
+ )
+ )
+ grpc_server_instrumentor.instrument()
+
+You can also use the filters directly on the provided interceptors:
+
+.. code-block::
+
+ my_interceptor = server_interceptor(
+ filter_ = filters.negate(filters.method_name("TestMethod"))
+ )
+ server = grpc.server(futures.ThreadPoolExecutor(),
+ interceptors = [my_interceptor])
+
+``filter_`` 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 adds the filters 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 (
+ any_of,
+ negate,
+ 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 +194,26 @@ class GrpcInstrumentorServer(BaseInstrumentor):
  grpc_server_instrumentor = GrpcInstrumentorServer()
  grpc_server_instrumentor.instrument()
 
+ If you want to add a filter that only intercept requests
+ to match the condition, pass ``filter_`` to GrpcInstrumentorServer.
+
+ grpc_server_instrumentor = GrpcInstrumentorServer(
+ filter_=filters.method_prefix("SimpleMethod"))
+ grpc_server_instrumentor.instrument()
+
  """
 
  # pylint:disable=attribute-defined-outside-init, redefined-outer-name
 
+ def __init__(self, filter_=None):
+ excluded_service_filter = _excluded_service_filter()
+ if excluded_service_filter is not None:
+ if filter_ is None:
+ filter_ = excluded_service_filter
+ else:
+ filter_ = any_of(filter_, excluded_service_filter)
+ self._filter = filter_
+
  def instrumentation_dependencies(self) -> Collection[str]:
  return _instruments
 
@@ -160,11 +225,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, filter_=self._filter
+ ),
  )
  else:
  kwargs["interceptors"] = [
- server_interceptor(tracer_provider=tracer_provider)
+ server_interceptor(
+ tracer_provider=tracer_provider, filter_=self._filter
+ )
  ]
  return self._original_func(*args, **kwargs)
 
@@ -183,8 +253,25 @@ class GrpcInstrumentorClient(BaseInstrumentor):
  grpc_client_instrumentor = GrpcInstrumentorClient()
  grpc_client_instrumentor.instrument()
 
+ If you want to add a filter that only intercept requests
+ to match the condition, pass ``filter_`` option to GrpcInstrumentorClient.
+
+ grpc_client_instrumentor = GrpcInstrumentorClient(
+ filter_=filters.negate(filters.health_check())
+ )
+ grpc_client_instrumentor.instrument()
+
  """
 
+ def __init__(self, filter_=None):
+ excluded_service_filter = _excluded_service_filter()
+ if excluded_service_filter is not None:
+ if filter_ is None:
+ filter_ = excluded_service_filter
+ else:
+ filter_ = any_of(filter_, excluded_service_filter)
+ self._filter = filter_
+
  # Figures out which channel type we need to wrap
  def _which_channel(self, kwargs):
  # handle legacy argument
@@ -221,37 +308,68 @@ 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,
+ filter_=self._filter,
+ ),
  )
 
 
-def client_interceptor(tracer_provider=None):
+def client_interceptor(tracer_provider=None, filter_=None):
  """Create a gRPC client channel interceptor.
 
  Args:
  tracer: The tracer to use to create client-side spans.
 
+ filter_: 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, filter_=filter_)
 
 
-def server_interceptor(tracer_provider=None):
+def server_interceptor(tracer_provider=None, filter_=None):
  """Create a gRPC server interceptor.
 
  Args:
  tracer: The tracer to use to create server-side spans.
 
+ filter_: 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, filter_=filter_)
+
+
+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 = (service_name(srv) for srv in services)
+ return negate(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, filter_=None):
  self._tracer = tracer
+ self._filter = filter_
 
  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

@@ -173,9 +173,10 @@ class OpenTelemetryServerInterceptor(grpc.ServerInterceptor):
  Usage::
 
  tracer = some OpenTelemetry tracer
+ filter = filters.negate(filters.method_name("service.Foo"))
 
  interceptors = [
- OpenTelemetryServerInterceptor(tracer),
+ OpenTelemetryServerInterceptor(tracer, filter),
  ]
 
  server = grpc.server(
@@ -184,8 +185,9 @@ class OpenTelemetryServerInterceptor(grpc.ServerInterceptor):
 
  """
 
- def __init__(self, tracer):
+ def __init__(self, tracer, filter_=None):
  self._tracer = tracer
+ self._filter = filter_
 
  @contextmanager
  def _set_remote_context(self, servicer_context):
@@ -261,6 +263,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):