feat(instrumentation): add OpenTelemetry tracing and metrics with basic configurations

tests/integration/instrumentation/__init__.py

@@ -73,3 +73,35 @@ def get_services():
     response_json = response.json()
     services = response_json.get('data', []) or []
     return [service for service in services if service != 'jaeger-query']
+
+
+class ExecutorFailureWithTracing(Executor):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.failure_counter = 0
+
+    @requests(on='/index')
+    def empty(
+        self, docs: 'DocumentArray', tracing_context: Optional[Context], **kwargs
+    ):
+        if self.tracer:
+            with self.tracer.start_span('dummy', context=tracing_context) as span:
+                span.set_attribute('len_docs', len(docs))
+                if not self.failure_counter:
+                    self.failure_counter += 1
+                    raise NotImplementedError
+                else:
+                    return docs
+        else:
+            return docs
+
+
+def spans_with_error(spans):
+    error_spans = []
+    for span in spans:
+        for tag in span['tags']:
+            if 'otel.status_code' == tag.get('key', '') and 'ERROR' == tag.get(
+                'value', ''
+            ):
+                error_spans.append(span)
+    return error_spans

tests/integration/instrumentation/test_flow_instrumentation.py

@@ -4,11 +4,13 @@
 
 from jina import Flow
 from tests.integration.instrumentation import (
+    ExecutorFailureWithTracing,
     ExecutorTestWithTracing,
     get_services,
     get_trace_ids,
     get_traces,
     partition_spans_by_kind,
+    spans_with_error,
 )
 
 
@@ -61,3 +63,39 @@ def test_gateway_instrumentation(
 
     trace_ids = get_trace_ids(client_traces)
     assert len(trace_ids) == 1
+
+
+def test_executor_instrumentation(otlp_collector):
+    f = Flow(
+        tracing=True,
+        span_exporter_host='localhost',
+        span_exporter_port=4317,
+    ).add(uses=ExecutorFailureWithTracing)
+
+    with f:
+        from jina import DocumentArray
+
+        try:
+            f.post(
+                f'/index',
+                DocumentArray.empty(2),
+            )
+        except:
+            pass
+        # give some time for the tracing and metrics exporters to finish exporting.
+        # the client is slow to export the data
+        time.sleep(8)
+
+    client_type = 'GRPCClient'
+    client_traces = get_traces(client_type)
+    (server_spans, client_spans, internal_spans) = partition_spans_by_kind(
+        client_traces
+    )
+    assert len(spans_with_error(server_spans)) == 0
+    assert len(spans_with_error(client_spans)) == 0
+    assert len(internal_spans) == 2
+    # Errors reported by DataRequestHandler and request method level spans
+    assert len(spans_with_error(internal_spans)) == 2
+
+    trace_ids = get_trace_ids(client_traces)
+    assert len(trace_ids) == 1