Support grpc secure channel

duetector/analyzer/jaeger/analyzer.py

@@ -5,6 +5,8 @@
 
 import grpc
 
+from duetector.utils import get_grpc_cred_from_path
+
 try:
     from functools import cache
 except ImportError:
@@ -35,7 +37,8 @@ class JaegerAnalyzer(Analyzer):
     default_config = {
         "disabled": True,
         # TODO: Support secure channel
-        # "secure_channel": False,
+        "secure": False,
+        "creds_file_path": "",
         "host": "localhost",
         "port": 16685,
     }
@@ -56,8 +59,13 @@ def channel(self) -> ChannelInitializer:
                 print(response)
 
         """
-        target_func = grpc.aio.insecure_channel
-        kwargs = {"target": f"{self.config.host}:{self.config.port}"}
+        kwargs = {}
+        if self.config.secure:
+            target_func = grpc.aio.secure_channel
+            kwargs["credentials"] = get_grpc_cred_from_path(self.config.creds_file_path)
+        else:
+            target_func = grpc.aio.insecure_channel
+        kwargs["target"] = f"{self.config.host}:{self.config.port}"
 
         return functools.partial(target_func, **kwargs)
 

duetector/collectors/otel.py

@@ -22,7 +22,7 @@
 from duetector.collectors.base import Collector
 from duetector.collectors.models import Tracking
 from duetector.extension.collector import hookimpl
-from duetector.utils import Singleton
+from duetector.utils import Singleton, get_grpc_cred_from_path
 
 
 class OTelInitiator(metaclass=Singleton):
@@ -130,6 +130,10 @@ class OTelCollector(Collector):
         "disabled": True,
         "exporter": "console",
         "exporter_kwargs": {},
+        "grpc_exporter_kwargs": {
+            "secure": False,
+            "creds_file_path": "",
+        },
     }
 
     @property
@@ -142,19 +146,36 @@ def endpoint(self) -> Optional[str]:
 
     @property
     def exporter_kwargs(self) -> Dict[str, Any]:
-        return self.config.exporter_kwargs
+        return self.config.exporter_kwargs._config_dict
 
     @property
     def service_name(self) -> str:
         return self.service_sep.join([f"{self.service_prefix}", f"{self.id}"])
 
+    @property
+    def grpc_exporter_kwargs(self) -> Dict[str, Any]:
+        kwargs = self.config.grpc_exporter_kwargs._config_dict
+        wrapped_kwargs = {}
+        if kwargs.get("secure"):
+            creds = get_grpc_cred_from_path(kwargs.get("creds_file_path"))
+            wrapped_kwargs = {
+                "insecure": False,
+                "credentials": creds,
+            }
+
+        return wrapped_kwargs
+
     def __init__(self, config: Optional[Dict[str, Any]] = None, *args, **kwargs):
         super().__init__(config, *args, **kwargs)
+
+        if "grpc" in self.exporter:
+            self.exporter_kwargs.update(self.grpc_exporter_kwargs)
+
         self.otel = OTelInitiator()
         self.otel.initialize(
             service_name=self.service_name,
             exporter=self.exporter,
-            exporter_kwargs=self.exporter_kwargs._config_dict,
+            exporter_kwargs=self.exporter_kwargs,
         )
 
     def _emit(self, t: Tracking):

duetector/static/config.toml

@@ -73,6 +73,10 @@ max_workers = 10
 
 [collector.otelcollector.exporter_kwargs]
 
+[collector.otelcollector.grpc_exporter_kwargs]
+secure = false
+creds_file_path = ""
+
 [collector.dbcollector]
 disabled = false
 statis_id = ""
@@ -100,6 +104,8 @@ include_extension = true
 
 [analyzer.jaegeranalyzer]
 disabled = true
+secure = false
+creds_file_path = ""
 host = "localhost"
 port = 16685
 

duetector/utils.py

@@ -1,11 +1,17 @@
 import threading
 from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Optional, Union
 
 try:
     from functools import cache
 except ImportError:
     from functools import lru_cache as cache
 
+import grpc
+
+from duetector.log import logger
+
 
 class Singleton(type):
     _instances = {}
@@ -43,6 +49,20 @@ def get_boot_time_duration_ns(ns) -> datetime:
     return get_boot_time() + timedelta(microseconds=ns / 1000)
 
 
+def get_grpc_cred_from_path(path: Optional[Union[str, Path]]) -> grpc.ChannelCredentials:
+    if not path:
+        logger.info("No creds file provided, using system certs.")
+        return grpc.ssl_channel_credentials()
+
+    if isinstance(path, str):
+        path = Path(path)
+    if not path.exists():
+        logger.warning(f"Could not find creds file: {path}, using system certs.")
+        return grpc.ssl_channel_credentials()
+    with path.open("rb") as f:
+        return grpc.ssl_channel_credentials(f.read())
+
+
 if __name__ == "__main__":
     print(get_boot_time())
     print(get_boot_time_duration_ns("13205215231927"))

pytest.ini

@@ -1,4 +1,4 @@
 [pytest]
 asyncio_mode = auto
 
-timeout = 60
+timeout = 180