Feature: Support OTel Collector

README.md

@@ -55,21 +55,18 @@ Join our [slack channel](https://join.slack.com/t/hitsz-ids/shared_invite/zt-239
 - `Tracer` Support
  - [X] eBPF-based tracer
  - [X] Shell command tracer
- - [ ] Subprocess tracer
+ - [X] Subprocess tracer
 - `Filter` Support
  - [X] Pattern matching, based on regular expressions
-- `Collector` and `Analyzer` Support
- - [X] SQL database
- - [ ] OpenTelemetry
-- Analyzer Support
- - [X] SQL database
- - [ ] OpenTelemetry
+- Data Collection and Analysis
+ - [X] `Analyzer` Support SQL database
+ - [X] `Collector` Support SQL database and *OpenTelemetry(Experimental)*
 - User Interface
  - [X] CLI Tools
  - [X] PIP Service
  - [ ] Control Panel
 - Enhancements
- - [ ] Runc containers identification
+ - [ ] `RunC` containers identification
 
 The eBPF program requires kernel support, see [Kernel Support](./docs/kernel_config.md)
 

README_zh.md

@@ -59,16 +59,18 @@ duetector🔍是一个基于可扩展的的数据使用探测器，它可以在L
 - `Tracer`支持
  - [X] 基于eBPF的tracer
  - [X] 基于shell命令的tracer
- - [ ] 基于子进程的tracer
+ - [X] 基于子进程的tracer
 - `Filter`支持
  - [X] 支持正则的模式匹配
-- `Collector`和`Analyzer`支持
- - [X] SQL数据库
- - [ ] Opentelemetry
+- 遥测数据传输与分析
+ - [X] `Analyzer`支持SQL数据库
+ - [X] `Collector`支持SQL数据库和*Opentelemetry（实验性）*
 - 用户接口
  - [X] 命令行工具
  - [X] PIP服务
  - [ ] 控制平面
+- 增强功能
+ - [ ] `RunC`容器云原生支持
 
 eBPF程序需要内核支持，详见[内核支持](./docs/kernel_config.md)
 

dev-tools/start-docs-host.sh

@@ -6,6 +6,6 @@
 set -e
 docker run --rm \
 -it \
--p 8080:80 \
+-p 8910:80 \
 -v $(pwd)/../docs/build/html:/usr/share/nginx/html:ro \
 nginx

docs/source/collectors/index.rst

@@ -25,6 +25,7 @@ Avaliable Collector
  :maxdepth: 2
 
  DB Collectors <db>
+ OTel Collector <otel>
 
 
 Data Models

docs/source/collectors/otel.rst

@@ -0,0 +1,15 @@
+OTelCollector
+==================
+
+.. autoclass:: duetector.collectors.otel.OTelInitiator
+ :members:
+ :undoc-members:
+ :private-members:
+ :show-inheritance:
+
+
+.. autoclass:: duetector.collectors.otel.OTelCollector
+ :members:
+ :undoc-members:
+ :private-members:
+ :show-inheritance:

duetector/collectors/models.py

@@ -102,6 +102,16 @@ def from_namedtuple(tracer, data: NamedTuple) -> Tracking: # type: ignore
 
  return Tracking(**args)
 
+ def set_span(self, span):
+ for k in self.model_fields:
+ if k in ("tracer", "extended"):
+ continue
+ v = getattr(self, k)
+ if v is not None:
+ span.set_attribute(k, v)
+ for k, v in self.extended.items():
+ span.set_attribute(k, v)
+
 
 if __name__ == "__main__":
  Tracking(tracer="test", dt=datetime.now())

duetector/collectors/otel.py

@@ -0,0 +1,167 @@
+from typing import Any, Dict, Optional
+
+from opentelemetry import trace
+from opentelemetry.exporter.jaeger.proto.grpc import (
+ JaegerExporter as GRPCJaegerExporter,
+)
+from opentelemetry.exporter.jaeger.thrift import JaegerExporter as ThriftJaegerExporter
+from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
+ OTLPSpanExporter as GRPCOTLPSpanExporter,
+)
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
+ OTLPSpanExporter as HTTPOTLPSpanExporter,
+)
+from opentelemetry.exporter.zipkin.json import ZipkinExporter as JSONZipkinExporter
+from opentelemetry.exporter.zipkin.proto.http import (
+ ZipkinExporter as HTTPZipkinExporter,
+)
+from opentelemetry.sdk.resources import SERVICE_NAME, Resource
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
+
+from duetector.collectors.base import Collector
+from duetector.collectors.models import Tracking
+from duetector.extension.collector import hookimpl
+
+
+class OTelInitiator:
+ """
+ Host the OpenTelemetry SDK and initialize the provider and exporter.
+
+ Avaliable exporters:
+ - ``console``
+ - ``otlp-grpc``
+ - ``otlp-http``
+ - ``jaeger-thrift``
+ - ``jaeger-grpc``
+ - ``zipkin-http``
+ - ``zipkin-json``
+ - ``prometheus``
+
+ Example:
+
+ .. code-block:: python
+
+ otel = OTelInitiator()
+ trace = otel.initialize(
+ service_name="duetector",
+ exporter="console",
+ )
+
+ from opentelemetry import trace
+ tracer = trace.get_tracer(__name__)
+ with tracer.start_as_current_span("test") as span:
+ span.set_attribute("test", "test")
+
+ otel.shutdown()
+ """
+
+ exporter_cls = {
+ "console": ConsoleSpanExporter,
+ "otlp-grpc": GRPCOTLPSpanExporter,
+ "otlp-http": HTTPOTLPSpanExporter,
+ "jaeger-thrift": ThriftJaegerExporter,
+ "jaeger-grpc": GRPCJaegerExporter,
+ "zipkin-http": HTTPZipkinExporter,
+ "zipkin-json": JSONZipkinExporter,
+ # Prometheus only support metrics
+ # "prometheus": "TODO"
+ }
+
+ def __init__(self):
+ self._initialized = False
+ self.provider = None
+
+ def initialize(
+ self,
+ service_name="unknown-service",
+ resource_kwargs: Optional[Dict[str, Any]] = None,
+ provider_kwargs: Optional[Dict[str, Any]] = None,
+ exporter="console",
+ exporter_kwargs: Optional[Dict[str, Any]] = None,
+ ) -> None:
+ if self._initialized:
+ return
+
+ if not resource_kwargs:
+ resource_kwargs = {}
+ resource_kwargs.setdefault(SERVICE_NAME, service_name)
+ resource = Resource(attributes=resource_kwargs)
+
+ if not provider_kwargs:
+ provider_kwargs = {}
+ provider = TracerProvider(resource=resource, **provider_kwargs)
+ self.provider = provider
+
+ if not exporter_kwargs:
+ exporter_kwargs = {}
+ processor = BatchSpanProcessor(self.exporter_cls[exporter](**exporter_kwargs))
+
+ provider.add_span_processor(processor)
+ trace.set_tracer_provider(provider)
+ self._initialized = True
+
+ def shutdown(self):
+ if self._initialized and self.provider:
+ self.provider.shutdown()
+ self._initialized = False
+ self.provider = None
+
+
+class OTelCollector(Collector):
+ """
+ A collector using OpenTelemetry SDK.
+
+ Config:
+ - ``exporter``: One of ``console``, ``otlp-grpc``, ``otlp-http``, ``jaeger-thrift``, ``jaeger-grpc``, ``zipkin-http``, ``zipkin-json``, see :class:`OTelInitiator` for more details
+ - ``exporter_kwargs``: A dict of kwargs for exporter
+
+ Note:
+ Since v1.35, the Jaeger supports OTLP natively. Please use the OTLP exporter instead. Support for this exporter will end July 2023.
+
+ """
+
+ default_config = {
+ **Collector.default_config,
+ "disabled": True,
+ "exporter": "console",
+ "exporter_kwargs": {},
+ }
+
+ @property
+ def exporter(self) -> str:
+ return self.config.exporter
+
+ @property
+ def endpoint(self) -> Optional[str]:
+ return self.config.endpoint
+
+ @property
+ def exporter_kwargs(self) -> Dict[str, Any]:
+ return self.config.exporter_kwargs
+
+ def __init__(self, config: Optional[Dict[str, Any]] = None, *args, **kwargs):
+ super().__init__(config, *args, **kwargs)
+ self.otel = OTelInitiator()
+ self.otel.initialize(
+ service_name="duetector",
+ exporter=self.exporter,
+ exporter_kwargs=self.exporter_kwargs,
+ )
+
+ def _emit(self, t: Tracking):
+ tracer = trace.get_tracer(self.id)
+ with tracer.start_as_current_span(t.tracer) as span:
+ t.set_span(span)
+
+ def summary(self) -> Dict:
+ return {}
+
+ def shutdown(self):
+ super().shutdown()
+ self.otel.shutdown()
+
+
+@hookimpl
+def init_collector(config):
+ return OTelCollector(config)

duetector/collectors/register.py

@@ -1,4 +1,4 @@
 # Expose for plugin system
-from . import base, db
+from . import base, db, otel
 
-registers = [base, db]
+registers = [base, db, otel]

duetector/static/config.toml

@@ -63,6 +63,16 @@ poll_timeout = 10
 disabled = false
 include_extension = true
 
+[collector.otelcollector]
+disabled = true
+statis_id = ""
+exporter = "console"
+
+[collector.otelcollector.backend_args]
+max_workers = 10
+
+[collector.otelcollector.exporter_kwargs]
+
 [collector.dbcollector]
 disabled = false
 statis_id = ""

duetector/utils.py

@@ -1,3 +1,4 @@
+import threading
 from datetime import datetime, timedelta
 
 try:
@@ -8,10 +9,13 @@
 
 class Singleton(type):
  _instances = {}
+ _lock = threading.Lock()
 
  def __call__(cls, *args, **kwargs):
  if cls not in cls._instances:
- cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
+ with cls._lock:
+ if cls not in cls._instances:
+ cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
  return cls._instances[cls]
 
 

pyproject.toml

@@ -16,10 +16,19 @@ dependencies = [
  "SQLAlchemy>=2",
  "click",
  "psutil",
- # Following are for web server
+ # OTel
+ "opentelemetry-sdk",
+ "opentelemetry-api",
+ "opentelemetry-semantic-conventions",
+ "opentelemetry-exporter-otlp-proto-grpc",
+ "opentelemetry-exporter-otlp-proto-http",
+ "opentelemetry-exporter-jaeger",
+ "opentelemetry-exporter-zipkin-proto-http",
+ "opentelemetry-exporter-zipkin-json",
+ # Web server
  "fastapi",
  "uvicorn[standard]",
- "anyio"
+ "anyio",
 ]
 dynamic = ["version"]
 classifiers = [

tests/config.toml

@@ -55,6 +55,16 @@ poll_timeout = 10
 disabled = false
 include_extension = true
 
+[collector.otelcollector]
+disabled = false
+statis_id = ""
+exporter = "console"
+
+[collector.otelcollector.backend_args]
+max_workers = 10
+
+[collector.otelcollector.exporter_kwargs]
+
 [collector.dbcollector]
 disabled = false
 id = "unittest"