@@ -118,13 +118,62 @@ def serve():
118118 server = grpc.server(futures.ThreadPoolExecutor(),
119119 interceptors = [server_interceptor()])
120120
121+ Filters
122+ -------
123+
124+ If you prefer to filter specific requests to be instrumented, you can specify
125+ the condition by assigning filters to instrumentors.
126+
127+ You can write a global server instrumentor as follows:
128+
129+ .. code-block::
130+
131+ from opentelemetry.instrumentation.grpc import filters, GrpcInstrumentorServer
132+
133+ grpc_server_instrumentor = GrpcInstrumentorServer(
134+ filter_ = filters.any_of(
135+ filters.method_name("SimpleMethod"),
136+ filters.method_name("ComplexMethod"),
137+ )
138+ )
139+ grpc_server_instrumentor.instrument()
140+
141+ You can also use the filters directly on the provided interceptors:
142+
143+ .. code-block::
144+
145+ my_interceptor = server_interceptor(
146+ filter_ = filters.negate(filters.method_name("TestMethod"))
147+ )
148+ server = grpc.server(futures.ThreadPoolExecutor(),
149+ interceptors = [my_interceptor])
150+
151+ ``filter_`` option also applies to both global and manual client intrumentors.
152+
153+
154+ Environment variable
155+ --------------------
156+
157+ If you'd like to exclude specific services for the instrumentations, you can use
158+ ``OTEL_PYTHON_GRPC_EXCLUDED_SERVICES`` environment variables.
159+
160+ For example, if you assign ``"GRPCTestServer,GRPCHealthServer"`` to the variable,
161+ then the global interceptor automatically adds the filters to exclude requests to
162+ services ``GRPCTestServer`` and ``GRPCHealthServer``.
163+
121164"""
122- from typing import Collection
165+ import os
166+ from typing import Callable , Collection , List , Union
123167
124168import grpc # pylint:disable=import-self
125169from wrapt import wrap_function_wrapper as _wrap
126170
127171from opentelemetry import trace
172+ from opentelemetry .instrumentation .grpc .filters import (
173+ any_of ,
174+ negate ,
175+ service_name ,
176+ )
128177from opentelemetry .instrumentation .grpc .grpcext import intercept_channel
129178from opentelemetry .instrumentation .grpc .package import _instruments
130179from opentelemetry .instrumentation .grpc .version import __version__
@@ -145,10 +194,26 @@ class GrpcInstrumentorServer(BaseInstrumentor):
145194 grpc_server_instrumentor = GrpcInstrumentorServer()
146195 grpc_server_instrumentor.instrument()
147196
197+ If you want to add a filter that only intercept requests
198+ to match the condition, pass ``filter_`` to GrpcInstrumentorServer.
199+
200+ grpc_server_instrumentor = GrpcInstrumentorServer(
201+ filter_=filters.method_prefix("SimpleMethod"))
202+ grpc_server_instrumentor.instrument()
203+
148204 """
149205
150206 # pylint:disable=attribute-defined-outside-init, redefined-outer-name
151207
208+ def __init__ (self , filter_ = None ):
209+ excluded_service_filter = _excluded_service_filter ()
210+ if excluded_service_filter is not None :
211+ if filter_ is None :
212+ filter_ = excluded_service_filter
213+ else :
214+ filter_ = any_of (filter_ , excluded_service_filter )
215+ self ._filter = filter_
216+
152217 def instrumentation_dependencies (self ) -> Collection [str ]:
153218 return _instruments
154219
@@ -160,11 +225,16 @@ def server(*args, **kwargs):
160225 if "interceptors" in kwargs :
161226 # add our interceptor as the first
162227 kwargs ["interceptors" ].insert (
163- 0 , server_interceptor (tracer_provider = tracer_provider )
228+ 0 ,
229+ server_interceptor (
230+ tracer_provider = tracer_provider , filter_ = self ._filter
231+ ),
164232 )
165233 else :
166234 kwargs ["interceptors" ] = [
167- server_interceptor (tracer_provider = tracer_provider )
235+ server_interceptor (
236+ tracer_provider = tracer_provider , filter_ = self ._filter
237+ )
168238 ]
169239 return self ._original_func (* args , ** kwargs )
170240
@@ -183,8 +253,25 @@ class GrpcInstrumentorClient(BaseInstrumentor):
183253 grpc_client_instrumentor = GrpcInstrumentorClient()
184254 grpc_client_instrumentor.instrument()
185255
256+ If you want to add a filter that only intercept requests
257+ to match the condition, pass ``filter_`` option to GrpcInstrumentorClient.
258+
259+ grpc_client_instrumentor = GrpcInstrumentorClient(
260+ filter_=filters.negate(filters.health_check())
261+ )
262+ grpc_client_instrumentor.instrument()
263+
186264 """
187265
266+ def __init__ (self , filter_ = None ):
267+ excluded_service_filter = _excluded_service_filter ()
268+ if excluded_service_filter is not None :
269+ if filter_ is None :
270+ filter_ = excluded_service_filter
271+ else :
272+ filter_ = any_of (filter_ , excluded_service_filter )
273+ self ._filter = filter_
274+
188275 # Figures out which channel type we need to wrap
189276 def _which_channel (self , kwargs ):
190277 # handle legacy argument
@@ -221,37 +308,68 @@ def wrapper_fn(self, original_func, instance, args, kwargs):
221308 tracer_provider = kwargs .get ("tracer_provider" )
222309 return intercept_channel (
223310 channel ,
224- client_interceptor (tracer_provider = tracer_provider ),
311+ client_interceptor (
312+ tracer_provider = tracer_provider ,
313+ filter_ = self ._filter ,
314+ ),
225315 )
226316
227317
228- def client_interceptor (tracer_provider = None ):
318+ def client_interceptor (tracer_provider = None , filter_ = None ):
229319 """Create a gRPC client channel interceptor.
230320
231321 Args:
232322 tracer: The tracer to use to create client-side spans.
233323
324+ filter_: filter function that returns True if gRPC requests
325+ matches the condition. Default is None and intercept
326+ all requests.
327+
234328 Returns:
235329 An invocation-side interceptor object.
236330 """
237331 from . import _client
238332
239333 tracer = trace .get_tracer (__name__ , __version__ , tracer_provider )
240334
241- return _client .OpenTelemetryClientInterceptor (tracer )
335+ return _client .OpenTelemetryClientInterceptor (tracer , filter_ = filter_ )
242336
243337
244- def server_interceptor (tracer_provider = None ):
338+ def server_interceptor (tracer_provider = None , filter_ = None ):
245339 """Create a gRPC server interceptor.
246340
247341 Args:
248342 tracer: The tracer to use to create server-side spans.
249343
344+ filter_: filter function that returns True if gRPC requests
345+ matches the condition. Default is None and intercept
346+ all requests.
347+
250348 Returns:
251349 A service-side interceptor object.
252350 """
253351 from . import _server
254352
255353 tracer = trace .get_tracer (__name__ , __version__ , tracer_provider )
256354
257- return _server .OpenTelemetryServerInterceptor (tracer )
355+ return _server .OpenTelemetryServerInterceptor (tracer , filter_ = filter_ )
356+
357+
358+ def _excluded_service_filter () -> Union [Callable [[object ], bool ], None ]:
359+ services = _parse_services (
360+ os .environ .get ("OTEL_PYTHON_GRPC_EXCLUDED_SERVICES" , "" )
361+ )
362+ if len (services ) == 0 :
363+ return None
364+ filters = (service_name (srv ) for srv in services )
365+ return negate (any_of (* filters ))
366+
367+
368+ def _parse_services (excluded_services : str ) -> List [str ]:
369+ if excluded_services != "" :
370+ excluded_service_list = [
371+ s .strip () for s in excluded_services .split ("," )
372+ ]
373+ else :
374+ excluded_service_list = []
375+ return excluded_service_list
0 commit comments