basic integratation of open telemetry

mashraf-222 · mashraf-222 · commit 77fd1a36f324 · 2025-11-05T02:14:53.000+02:00
diff --git a/env.example b/env.example
@@ -0,0 +1,33 @@
+# OpenTelemetry Configuration
+# Copy this file to .env and update values as needed
+
+# Enable/Disable OpenTelemetry SDK
+# Set to "true" to disable telemetry, "false" to enable (default: false)
+OTEL_SDK_DISABLED=false
+
+# Service Information
+# Service name (default: "optimize-me")
+OTEL_SERVICE_NAME=optimize-me
+
+# Service version (default: "0.1.0")
+OTEL_SERVICE_VERSION=0.1.0
+
+# Exporter Configuration
+# Type of exporter: "console" (development), "otlp" (production), or "jaeger"
+# (default: "console")
+OTEL_EXPORTER_TYPE=console
+
+# OTLP Exporter Endpoint
+# Endpoint for OTLP exporter (used when OTEL_EXPORTER_TYPE is "otlp" or "jaeger")
+# Format: http://host:port (default: "http://localhost:4318")
+OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318
+
+# Sampling Rate
+# Sampling rate for traces (0.0 to 1.0)
+# 1.0 = 100% sampling (all traces), 0.1 = 10% sampling (default: 1.0)
+OTEL_TRACES_SAMPLER_ARG=1.0
+
+# Log Level
+# Logging level for OpenTelemetry (DEBUG, INFO, WARNING, ERROR)
+# (default: "INFO")
+OTEL_LOG_LEVEL=INFO
diff --git a/examples/run_with_telemetry.py b/examples/run_with_telemetry.py
@@ -0,0 +1,43 @@
+import sys
+from pathlib import Path
+
+project_root = Path(__file__).parent.parent
+sys.path.insert(0, str(project_root))
+
+from src.telemetry import setup_telemetry
+from src.numerical.optimization import gradient_descent
+from src.algorithms.graph import graph_traversal, find_node_clusters
+from src.statistics.descriptive import describe
+import numpy as np
+import pandas as pd
+
+setup_telemetry(
+    service_name="optimize-me",
+    service_version="0.1.0",
+    exporter_type="console",
+)
+
+print("Running gradient descent...")
+X = np.array([[1, 2], [3, 4], [5, 6]])
+y = np.array([1, 2, 3])
+weights = gradient_descent(X, y, learning_rate=0.01, iterations=100)
+print(f"Final weights: {weights}\n")
+
+print("Running graph traversal...")
+graph = {1: {2, 3}, 2: {4}, 3: {4}, 4: {}}
+visited = graph_traversal(graph, 1)
+print(f"Visited nodes: {visited}\n")
+
+print("Running statistical describe...")
+series = pd.Series([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
+stats = describe(series)
+print(f"Statistics: {stats}\n")
+
+print("Running node clustering...")
+nodes = [{"id": 1}, {"id": 2}, {"id": 3}]
+edges = [{"source": 1, "target": 2}]
+clusters = find_node_clusters(nodes, edges)
+print(f"Clusters: {clusters}\n")
+
+print("Telemetry demonstration complete!")
+
diff --git a/pyproject.toml b/pyproject.toml
@@ -7,6 +7,10 @@ dependencies = [
     "codeflash>=0.17.3",
     "networkx>=3.5",
     "numpy>=2.3.3",
+    "opentelemetry-api>=1.38.0",
+    "opentelemetry-exporter-otlp-proto-grpc>=1.38.0",
+    "opentelemetry-instrumentation>=0.59b0",
+    "opentelemetry-sdk>=1.38.0",
     "pandas>=2.3.3",
 ]
 
diff --git a/requirements.txt b/requirements.txt
@@ -2,3 +2,7 @@ numpy
 pandas
 codeflash
 networkx
+opentelemetry-api>=1.38.0
+opentelemetry-exporter-otlp-proto-grpc>=1.38.0
+opentelemetry-instrumentation>=0.59b0
+opentelemetry-sdk>=1.38.0
diff --git a/src/algorithms/dynamic_programming.py b/src/algorithms/dynamic_programming.py
@@ -1,16 +1,21 @@
 from __future__ import annotations
 
+from src.telemetry.decorators import trace_function
 
+
+@trace_function
 def fibonacci(n):
     if n <= 1:
         return n
     return fibonacci(n - 1) + fibonacci(n - 2)
 
 
+@trace_function
 def matrix_sum(matrix: list[list[int]]) -> list[int]:
     return [sum(matrix[i]) for i in range(len(matrix)) if sum(matrix[i]) > 0]
 
 
+@trace_function
 def matrix_chain_order(matrices: list[tuple[int, int]]) -> int:
     """
     Find the minimum number of operations needed to multiply a chain of matrices.
@@ -42,6 +47,7 @@ def dp(i: int, j: int) -> int:
     return dp(0, n - 1)
 
 
+@trace_function
 def coin_change(coins: list[int], amount: int, index: int) -> int:
     if amount == 0:
         return 1
@@ -53,6 +59,7 @@ def coin_change(coins: list[int], amount: int, index: int) -> int:
     )
 
 
+@trace_function
 def knapsack(weights: list[int], values: list[int], capacity: int, n: int) -> int:
     if n == 0 or capacity == 0:
         return 0
diff --git a/src/algorithms/graph.py b/src/algorithms/graph.py
@@ -2,7 +2,10 @@
 
 import networkx as nx
 
+from src.telemetry.decorators import trace_function
 
+
+@trace_function
 def graph_traversal(graph: dict[int, dict[int]], node: int) -> dict[int]:
     visited = []
 
@@ -21,6 +24,7 @@ class PathFinder:
     def __init__(self, graph: dict[str, list[str]]):
         self.graph = graph
 
+    @trace_function
     def find_shortest_path(self, start: str, end: str) -> list[str]:
         if start not in self.graph or end not in self.graph:
             return []
@@ -102,6 +106,7 @@ def find_node_with_highest_degree(
     return max_degree_node
 
 
+@trace_function
 def find_node_clusters(nodes: list[dict], edges: list[dict]) -> list[list[dict]]:
     """Find connected components (clusters) in the graph."""
     # Create node ID to node mapping for easy lookup
@@ -149,6 +154,7 @@ def find_node_clusters(nodes: list[dict], edges: list[dict]) -> list[list[dict]]
     return clusters
 
 
+@trace_function
 def calculate_node_betweenness(
     nodes: list[str], edges: list[dict[str, str]]
 ) -> dict[str, float]:
diff --git a/src/data_processing/dataframe.py b/src/data_processing/dataframe.py
@@ -3,7 +3,10 @@
 import numpy as np
 import pandas as pd
 
+from src.telemetry.decorators import trace_function
 
+
+@trace_function
 def dataframe_filter(df: pd.DataFrame, column: str, value: Any) -> pd.DataFrame:
     indices = []
     for i in range(len(df)):
@@ -12,6 +15,7 @@ def dataframe_filter(df: pd.DataFrame, column: str, value: Any) -> pd.DataFrame:
     return df.iloc[indices].reset_index(drop=True)
 
 
+@trace_function
 def groupby_mean(df: pd.DataFrame, group_col: str, value_col: str) -> dict[Any, float]:
     sums = {}
     counts = {}
@@ -30,6 +34,7 @@ def groupby_mean(df: pd.DataFrame, group_col: str, value_col: str) -> dict[Any,
     return result
 
 
+@trace_function
 def dataframe_merge(
     left: pd.DataFrame, right: pd.DataFrame, left_on: str, right_on: str
 ) -> pd.DataFrame:
diff --git a/src/numerical/optimization.py b/src/numerical/optimization.py
@@ -1,6 +1,9 @@
 import numpy as np
 
+from src.telemetry.decorators import trace_function
 
+
+@trace_function(span_name="gradient_descent", capture_args=["iterations", "learning_rate"])
 def gradient_descent(
     X: np.ndarray, y: np.ndarray, learning_rate: float = 0.01, iterations: int = 1000
 ) -> np.ndarray:
diff --git a/src/statistics/descriptive.py b/src/statistics/descriptive.py
@@ -3,7 +3,10 @@
 import numpy as np
 import pandas as pd
 
+from src.telemetry.decorators import trace_function
 
+
+@trace_function
 def describe(series: pd.Series) -> dict[str, float]:
     values = [v for v in series if not pd.isna(v)]
     n = len(values)
@@ -41,6 +44,7 @@ def percentile(p):
     }
 
 
+@trace_function
 def correlation(df: pd.DataFrame) -> dict[Tuple[str, str], float]:
     numeric_columns = [
         col for col in df.columns if np.issubdtype(df[col].dtype, np.number)
diff --git a/src/telemetry/__init__.py b/src/telemetry/__init__.py
@@ -0,0 +1,9 @@
+from src.telemetry.setup import setup_telemetry
+from src.telemetry.decorators import trace_function, trace_function_with_metrics
+
+__all__ = [
+    "setup_telemetry",
+    "trace_function",
+    "trace_function_with_metrics",  # Placeholder for metrics implementation (currently behaves like trace_function)
+]
+
diff --git a/src/telemetry/config.py b/src/telemetry/config.py
@@ -0,0 +1,20 @@
+import os
+from typing import Optional
+
+
+class TelemetryConfig:
+    enabled: bool = os.getenv("OTEL_SDK_DISABLED", "false").lower() != "true"
+    service_name: str = os.getenv("OTEL_SERVICE_NAME", "optimize-me")
+    service_version: str = os.getenv("OTEL_SERVICE_VERSION", "0.1.0")
+    exporter_type: str = os.getenv("OTEL_EXPORTER_TYPE", "console")
+    exporter_endpoint: str = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4318")
+    sampling_rate: float = float(os.getenv("OTEL_TRACES_SAMPLER_ARG", "1.0"))
+    log_level: str = os.getenv("OTEL_LOG_LEVEL", "INFO")
+
+    @classmethod
+    def get_exporter_config(cls) -> dict:
+        config = {"type": cls.exporter_type}
+        if cls.exporter_type in ["otlp", "jaeger"]:
+            config["endpoint"] = cls.exporter_endpoint
+        return config
+
diff --git a/src/telemetry/decorators.py b/src/telemetry/decorators.py
@@ -0,0 +1,91 @@
+import functools
+import logging
+from typing import Any, Callable, Optional
+
+from opentelemetry import trace
+from opentelemetry.trace import Status, StatusCode
+
+from src.telemetry.config import TelemetryConfig
+
+logger = logging.getLogger(__name__)
+tracer = trace.get_tracer(__name__)
+
+
+def trace_function(
+    func: Optional[Callable] = None,
+    *,
+    span_name: Optional[str] = None,
+    capture_args: Optional[list[str]] = None,
+    capture_return: bool = False,
+) -> Callable:
+    def decorator(f: Callable) -> Callable:
+        if not TelemetryConfig.enabled:
+            return f
+
+        name = span_name or f"{f.__module__}.{f.__name__}"
+
+        @functools.wraps(f)
+        def wrapper(*args, **kwargs) -> Any:
+            with tracer.start_as_current_span(name) as span:
+                try:
+                    if capture_args:
+                        import inspect
+
+                        sig = inspect.signature(f)
+                        bound = sig.bind(*args, **kwargs)
+                        bound.apply_defaults()
+
+                        for arg_name in capture_args:
+                            if arg_name in bound.arguments:
+                                value = bound.arguments[arg_name]
+                                attr_value = str(value)
+                                if len(attr_value) > 250:
+                                    attr_value = attr_value[:250] + "..."
+                                span.set_attribute(f"function.{arg_name}", attr_value)
+
+                    result = f(*args, **kwargs)
+
+                    if capture_return:
+                        return_str = str(result)
+                        if len(return_str) > 250:
+                            return_str = return_str[:250] + "..."
+                        span.set_attribute("function.return", return_str)
+
+                    span.set_status(Status(StatusCode.OK))
+                    return result
+
+                except Exception as e:
+                    span.record_exception(e)
+                    span.set_status(Status(StatusCode.ERROR, str(e)))
+                    logger.error(
+                        f"Error in {name}: {e}",
+                        exc_info=True,
+                        extra={
+                            "file": f.__module__,
+                            "function": f.__name__,
+                            "error": str(e),
+                        },
+                    )
+                    raise
+
+        return wrapper
+
+    if func is None:
+        return decorator
+    else:
+        return decorator(func)
+
+# Placeholder for metrics implementation (currently behaves like trace_function)
+def trace_function_with_metrics(
+    func: Optional[Callable] = None,
+    *,
+    span_name: Optional[str] = None,
+    capture_args: Optional[list[str]] = None,
+) -> Callable:
+    return trace_function(
+        func,
+        span_name=span_name,
+        capture_args=capture_args,
+        capture_return=False,
+    )
+
diff --git a/src/telemetry/setup.py b/src/telemetry/setup.py
