add OpenTelemetry tracing with Jaeger (opea-project#54)

* add Jaeger support

Signed-off-by: Spycsh <[email protected]>

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix dep

Signed-off-by: Spycsh <[email protected]>

---------

Signed-off-by: Spycsh <[email protected]>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

comps/asr/Dockerfile

@@ -9,8 +9,7 @@ RUN apt-get update \
 COPY comps /home/comps
 
 RUN pip install --no-cache-dir --upgrade pip && \
-    pip install --no-cache-dir -r /home/comps/asr/requirements.txt && \
-    pip install --no-cache-dir -r /home/comps/cores/telemetry/requirements.txt
+    pip install --no-cache-dir -r /home/comps/asr/requirements.txt
 
 ENV PYTHONPATH=$PYTHONPATH:/home
 

comps/cores/telemetry/opea_telemetry.py

@@ -18,16 +18,21 @@
 
 from opentelemetry import trace
 from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter as HTTPSpanExporter
+from opentelemetry.sdk.resources import SERVICE_NAME, Resource
 from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import BatchSpanProcessor
 from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
 
-in_memory_exporter = InMemorySpanExporter()
-trace.set_tracer_provider(TracerProvider())
-trace.get_tracer_provider().add_span_processor(
+resource = Resource.create({SERVICE_NAME: "opea"})
+traceProvider = TracerProvider(resource=resource)
+
+traceProvider.add_span_processor(
     BatchSpanProcessor(HTTPSpanExporter(endpoint="http://10.165.57.68:4318/v1/traces"))  # change to ip
 )
-trace.get_tracer_provider().add_span_processor(BatchSpanProcessor(in_memory_exporter))
+in_memory_exporter = InMemorySpanExporter()
+traceProvider.add_span_processor(BatchSpanProcessor(in_memory_exporter))
+trace.set_tracer_provider(traceProvider)
+
 tracer = trace.get_tracer(__name__)
 
 

comps/tts/Dockerfile

@@ -11,8 +11,7 @@ RUN apt-get update \
 COPY comps /home/comps
 
 RUN pip install --no-cache-dir --upgrade pip && \
-    pip install --no-cache-dir -r /home/comps/tts/requirements.txt && \
-    pip install --no-cache-dir -r /home/comps/cores/telemetry/requirements.txt
+    pip install --no-cache-dir -r /home/comps/tts/requirements.txt
 
 ENV LD_PRELOAD=/usr/lib/x86_64-linux-gnu/libiomp5.so:/usr/lib/x86_64-linux-gnu/libtcmalloc.so.4
 ENV MALLOC_CONF="oversize_threshold:1,background_thread:true,metadata_thp:auto,dirty_decay_ms:9000000000,muzzy_decay_ms:9000000000"

tests/cores/telemetry/test_telemetry.py

@@ -13,60 +13,69 @@
 # limitations under the License.
 
 import asyncio
-import subprocess
 import time
 import unittest
 
 from comps import opea_telemetry
 from comps.cores.telemetry.opea_telemetry import in_memory_exporter
 
 
+@opea_telemetry
+def dummy_func_child():
+    return
+
+
 @opea_telemetry
 def dummy_func():
     time.sleep(1)
+    dummy_func_child()
+
+
+@opea_telemetry
+async def dummy_async_func_child():
+    return
 
 
 @opea_telemetry
 async def dummy_async_func():
     await asyncio.sleep(2)
+    await dummy_async_func_child()
 
 
 class TestTelemetry(unittest.TestCase):
-    def setUp(self):
-        self.p = subprocess.Popen(
-            "docker run -d -p 4317:4317 -p 4318:4318 --rm -v $(pwd)/collector-config.yaml:/etc/otelcol/config.yaml otel/opentelemetry-collector",
-            shell=True,
-            stdin=subprocess.PIPE,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-        )
-        raw_output, raw_err = self.p.communicate()
-        self.cid = raw_output.decode().replace("\n", "")
-
-    def tearDown(self):
-        rp = subprocess.Popen(
-            f"docker rm -f {self.cid}",
-            shell=True,
-            stdin=subprocess.PIPE,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-        )
-        _, _ = rp.communicate()
-        rp.kill()
-        self.p.kill()
 
     def test_time_tracing(self):
         dummy_func()
         asyncio.run(dummy_async_func())
         time.sleep(2)  # wait until all telemetries are finished
-        self.assertEqual(len(in_memory_exporter.get_finished_spans()), 2)
+        self.assertEqual(len(in_memory_exporter.get_finished_spans()), 4)
+
         for i in in_memory_exporter.get_finished_spans():
             if i.name == "dummy_func":
                 self.assertTrue(int((i.end_time - i.start_time) / 1e9) == 1)
             elif i.name == "dummy_async_func":
                 self.assertTrue(int((i.end_time - i.start_time) / 1e9) == 2)
+        in_memory_exporter.clear()
+
+    def test_call_tracing(self):
+        dummy_func()
+        asyncio.run(dummy_async_func())
+        time.sleep(2)  # wait until all telemetries are finished
+        self.assertEqual(len(in_memory_exporter.get_finished_spans()), 4)
+        for i in in_memory_exporter.get_finished_spans():
+            if i.name == "dummy_func":
+                dummy_func_trace_id = i.get_span_context().trace_id
+            elif i.name == "dummy_func_child":
+                dummy_func_child_parent_id = i.parent.trace_id
+            elif i.name == "dummy_async_func":
+                dummy_async_func_id = i.get_span_context().trace_id
+            elif i.name == "dummy_async_func_child":
+                dummy_async_func_child_parent_id = i.parent.trace_id
             else:
                 self.assertTrue(False)
+        self.assertEqual(dummy_func_trace_id, dummy_func_child_parent_id)
+        self.assertEqual(dummy_async_func_id, dummy_async_func_child_parent_id)
+        in_memory_exporter.clear()
 
 
 if __name__ == "__main__":