Merge pull request #842 from karrioapi/otel

Author: danh91

.env.grafana-test

@@ -0,0 +1,25 @@
+# OpenTelemetry configuration for Grafana testing
+OTEL_ENABLED=true
+OTEL_SERVICE_NAME=karrio-api-local
+OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317
+OTEL_EXPORTER_OTLP_PROTOCOL=grpc
+OTEL_ENVIRONMENT=local-development
+OTEL_RESOURCE_ATTRIBUTES=deployment.environment=local,service.namespace=karrio-dev,team=logistics
+
+# Basic Django settings for local testing
+SECRET_KEY=test-key-for-grafana-development
+DEBUG=True
+DATABASE_URL=sqlite:////tmp/karrio_grafana_test.db
+
+# Email settings to avoid errors
+EMAIL_FROM_ADDRESS=noreply@example.com
+EMAIL_HOST=smtp.example.com
+EMAIL_PORT=587
+EMAIL_HOST_USER=user@example.com
+EMAIL_HOST_PASSWORD=password
+DEFAULT_FROM_EMAIL=noreply@example.com
+
+# Disable some features for testing
+DETACHED_WORKER=false
+ADMIN_EMAIL=admin@example.com
+ADMIN_PASSWORD=demo

.env.sample

@@ -46,3 +46,33 @@ OIDC_RSA_PRIVATE_KEY="-----BEGIN RSA PRIVATE KEY-----\nMIIJKQIBAAKCAgEA92+93Tyg6
 
 # Plugin Configuration
 ENABLE_ALL_PLUGINS_BY_DEFAULT=True
+
+# =============================================================================
+# OPENTELEMETRY CONFIGURATION (OPTIONAL)
+# =============================================================================
+# Enable OpenTelemetry instrumentation
+# OTEL_ENABLED=false
+
+# Service name for traces and metrics
+# OTEL_SERVICE_NAME=karrio-api
+
+# OTLP Exporter endpoint (required if OTEL_ENABLED=true)
+# Examples:
+#   - Jaeger: http://localhost:4317 (gRPC) or http://localhost:4318 (HTTP)
+#   - Zipkin: http://localhost:9411/api/v2/spans
+#   - OTLP Collector: http://localhost:4317
+# OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317
+
+# Protocol for OTLP exporter (grpc or http)
+# OTEL_EXPORTER_OTLP_PROTOCOL=grpc
+
+# Optional headers for OTLP exporter (comma-separated key=value pairs)
+# Example: api-key=your-api-key,tenant-id=your-tenant
+# OTEL_EXPORTER_OTLP_HEADERS=
+
+# Environment name for telemetry data
+# OTEL_ENVIRONMENT=development
+
+# Additional resource attributes (comma-separated key=value pairs)
+# Example: team=backend,region=us-west-2
+# OTEL_RESOURCE_ATTRIBUTES=

.github/workflows/build.yml

@@ -46,7 +46,7 @@ jobs:
       - name: Set up Python
         uses: actions/setup-python@v4
         with:
-          python-version: '3.12'
+          python-version: "3.12"
 
       - name: Install build dependencies
         run: |

.gitignore

@@ -122,9 +122,11 @@ node_modules
 .vercel
 
 # karrio stuff:
+!apps/api/karrio/server/lib
 !apps/dashboard/src/lib
 !packages/lib
 
+
 .bash_history
 
 .idea/

CHANGELOG.md

@@ -1,3 +1,15 @@
+# Karrio 2025.5rc18
+
+## Changes
+
+### Feat
+
+- feat: add support for OpenTelemetry to Karrio server
+
+### Fix
+
+- fix: missing workflow-trigger table migrations
+
 # Karrio 2025.5rc17
 
 ## Changes

apps/api/karrio/server/VERSION

@@ -1 +1 @@
-2025.5rc17
+2025.5rc18

apps/api/karrio/server/lib/otel_huey.py

@@ -0,0 +1,146 @@
+"""
+OpenTelemetry instrumentation for Huey task queue.
+
+This module provides tracing support for Huey tasks, enabling distributed tracing
+across API requests and background tasks.
+"""
+import functools
+import logging
+from typing import Any, Callable, Dict, Optional
+
+from opentelemetry import trace, context, propagate
+from opentelemetry.trace import Status, StatusCode
+from opentelemetry.semconv.trace import SpanAttributes
+
+logger = logging.getLogger(__name__)
+
+
+class HueyInstrumentor:
+    """Instrumentation for Huey task queue."""
+    
+    _instance = None
+    _instrumented = False
+    
+    def __new__(cls):
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+        return cls._instance
+    
+    def instrument(self, huey_instance=None):
+        """
+        Instrument Huey for OpenTelemetry tracing.
+        
+        Args:
+            huey_instance: The Huey instance to instrument. If None, will try to
+                          import from Django settings.
+        """
+        if self._instrumented:
+            logger.debug("Huey already instrumented")
+            return
+        
+        try:
+            if huey_instance is None:
+                from django.conf import settings
+                huey_instance = settings.HUEY
+            
+            # Wrap the task decorator
+            original_task = huey_instance.task
+            huey_instance.task = self._wrap_task_decorator(original_task, huey_instance)
+            
+            # Wrap periodic tasks
+            if hasattr(huey_instance, 'periodic_task'):
+                original_periodic = huey_instance.periodic_task
+                huey_instance.periodic_task = self._wrap_task_decorator(original_periodic, huey_instance)
+            
+            self._instrumented = True
+            logger.info("Huey instrumented for OpenTelemetry")
+            
+        except Exception as e:
+            logger.warning(f"Failed to instrument Huey: {e}")
+    
+    def _wrap_task_decorator(self, original_decorator: Callable, huey_instance) -> Callable:
+        """Wrap the Huey task decorator to add tracing."""
+        
+        @functools.wraps(original_decorator)
+        def wrapped_decorator(*args, **kwargs):
+            decorated = original_decorator(*args, **kwargs)
+            
+            def task_wrapper(fn):
+                task_fn = decorated(fn)
+                
+                @functools.wraps(task_fn)
+                def traced_task(*task_args, **task_kwargs):
+                    tracer = trace.get_tracer(__name__)
+                    
+                    # Extract trace context from task kwargs if present
+                    trace_context = task_kwargs.pop('_otel_context', None)
+                    if trace_context:
+                        ctx = propagate.extract(trace_context)
+                        token = context.attach(ctx)
+                    else:
+                        token = None
+                    
+                    # Start span for the task
+                    task_name = fn.__name__
+                    with tracer.start_as_current_span(
+                        f"huey.task.{task_name}",
+                        kind=trace.SpanKind.CONSUMER,
+                    ) as span:
+                        try:
+                            # Set span attributes
+                            span.set_attribute("messaging.system", "huey")
+                            span.set_attribute("messaging.destination", task_name)
+                            span.set_attribute("messaging.operation", "process")
+                            span.set_attribute("task.name", task_name)
+                            
+                            # Execute the task
+                            result = task_fn(*task_args, **task_kwargs)
+                            span.set_status(Status(StatusCode.OK))
+                            return result
+                            
+                        except Exception as e:
+                            span.set_status(Status(StatusCode.ERROR, str(e)))
+                            span.record_exception(e)
+                            raise
+                        finally:
+                            if token:
+                                context.detach(token)
+                
+                # Preserve original attributes
+                traced_task.task = task_fn.task if hasattr(task_fn, 'task') else task_fn
+                if hasattr(task_fn, '__name__'):
+                    traced_task.__name__ = task_fn.__name__
+                if hasattr(task_fn, '__module__'):
+                    traced_task.__module__ = task_fn.__module__
+                
+                return traced_task
+            
+            return task_wrapper
+        
+        return wrapped_decorator
+
+
+def inject_trace_context(task_kwargs: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Inject current trace context into task kwargs for propagation.
+    
+    This should be called when enqueuing a task to propagate the trace context
+    to the background worker.
+    
+    Args:
+        task_kwargs: The kwargs to be passed to the task
+        
+    Returns:
+        Updated kwargs with trace context
+    """
+    carrier = {}
+    propagate.inject(carrier)
+    if carrier:
+        task_kwargs['_otel_context'] = carrier
+    return task_kwargs
+
+
+def instrument_huey():
+    """Convenience function to instrument Huey."""
+    instrumentor = HueyInstrumentor()
+    instrumentor.instrument()

apps/api/karrio/server/settings/apm.py

@@ -56,3 +56,128 @@
         # django.contrib.auth) you may enable sending PII data.
         send_default_pii=True,
     )
+
+
+# OpenTelemetry Configuration
+OTEL_ENABLED = config("OTEL_ENABLED", default=False, cast=bool)
+OTEL_SERVICE_NAME = config("OTEL_SERVICE_NAME", default="karrio-api")
+OTEL_EXPORTER_OTLP_ENDPOINT = config("OTEL_EXPORTER_OTLP_ENDPOINT", default=None)
+OTEL_EXPORTER_OTLP_PROTOCOL = config("OTEL_EXPORTER_OTLP_PROTOCOL", default="grpc")
+OTEL_EXPORTER_OTLP_HEADERS = config("OTEL_EXPORTER_OTLP_HEADERS", default="")
+OTEL_TRACES_EXPORTER = config("OTEL_TRACES_EXPORTER", default="otlp")
+OTEL_METRICS_EXPORTER = config("OTEL_METRICS_EXPORTER", default="otlp")
+OTEL_LOGS_EXPORTER = config("OTEL_LOGS_EXPORTER", default="otlp")
+OTEL_RESOURCE_ATTRIBUTES = config("OTEL_RESOURCE_ATTRIBUTES", default="")
+OTEL_ENVIRONMENT = config("OTEL_ENVIRONMENT", default=config("ENV", default="production"))
+
+# Only initialize OpenTelemetry if enabled and endpoint is configured
+if OTEL_ENABLED and OTEL_EXPORTER_OTLP_ENDPOINT:
+    import logging
+    from opentelemetry import trace, metrics
+    from opentelemetry.sdk.trace import TracerProvider
+    from opentelemetry.sdk.trace.export import BatchSpanProcessor
+    from opentelemetry.sdk.metrics import MeterProvider
+    from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
+    from opentelemetry.sdk.resources import Resource, SERVICE_NAME, SERVICE_VERSION
+    from opentelemetry.instrumentation.django import DjangoInstrumentor
+    from opentelemetry.instrumentation.requests import RequestsInstrumentor
+    from opentelemetry.instrumentation.logging import LoggingInstrumentor
+    from opentelemetry.instrumentation.psycopg2 import Psycopg2Instrumentor
+    from opentelemetry.instrumentation.redis import RedisInstrumentor
+    
+    # Import appropriate exporter based on protocol
+    if OTEL_EXPORTER_OTLP_PROTOCOL == "grpc":
+        from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
+        from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
+    else:  # http/protobuf
+        from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+        from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
+    
+    # Parse headers if provided
+    headers = {}
+    if OTEL_EXPORTER_OTLP_HEADERS:
+        for header_pair in OTEL_EXPORTER_OTLP_HEADERS.split(","):
+            if "=" in header_pair:
+                key, value = header_pair.split("=", 1)
+                headers[key.strip()] = value.strip()
+    
+    # Parse resource attributes
+    resource_attributes = {
+        SERVICE_NAME: OTEL_SERVICE_NAME,
+        SERVICE_VERSION: config("VERSION", default="unknown"),
+        "environment": OTEL_ENVIRONMENT,
+        "deployment.environment": OTEL_ENVIRONMENT,
+    }
+    
+    if OTEL_RESOURCE_ATTRIBUTES:
+        for attr_pair in OTEL_RESOURCE_ATTRIBUTES.split(","):
+            if "=" in attr_pair:
+                key, value = attr_pair.split("=", 1)
+                resource_attributes[key.strip()] = value.strip()
+    
+    # Create resource
+    resource = Resource(attributes=resource_attributes)
+    
+    # Configure Trace Provider
+    trace_provider = TracerProvider(resource=resource)
+    trace.set_tracer_provider(trace_provider)
+    
+    # Configure span exporter
+    span_exporter = OTLPSpanExporter(
+        endpoint=OTEL_EXPORTER_OTLP_ENDPOINT,
+        headers=headers if headers else None,
+    )
+    span_processor = BatchSpanProcessor(span_exporter)
+    trace_provider.add_span_processor(span_processor)
+    
+    # Configure Metrics Provider
+    metric_exporter = OTLPMetricExporter(
+        endpoint=OTEL_EXPORTER_OTLP_ENDPOINT,
+        headers=headers if headers else None,
+    )
+    metric_reader = PeriodicExportingMetricReader(
+        exporter=metric_exporter,
+        export_interval_millis=30000,  # Export metrics every 30 seconds
+    )
+    meter_provider = MeterProvider(
+        resource=resource,
+        metric_readers=[metric_reader],
+    )
+    metrics.set_meter_provider(meter_provider)
+    
+    # Instrument Django
+    DjangoInstrumentor().instrument(
+        is_sql_commentor_enabled=True,  # Add trace context to SQL queries
+        request_hook=lambda span, request: span.set_attribute("http.client_ip", request.META.get("REMOTE_ADDR", "")),
+        response_hook=lambda span, request, response: span.set_attribute("http.response.size", len(response.content) if hasattr(response, 'content') else 0),
+    )
+    
+    # Instrument other libraries
+    RequestsInstrumentor().instrument()  # HTTP client requests
+    LoggingInstrumentor().instrument(set_logging_format=True)  # Add trace context to logs
+    
+    # Instrument database if PostgreSQL is used
+    if config("DATABASE_ENGINE", default="").endswith("postgresql"):
+        try:
+            Psycopg2Instrumentor().instrument()
+        except Exception:
+            pass  # Psycopg2 might not be installed
+    
+    # Instrument Redis if configured
+    if config("REDIS_HOST", default=None):
+        try:
+            RedisInstrumentor().instrument()
+        except Exception:
+            pass  # Redis might not be installed
+    
+    # Instrument Huey task queue (temporarily disabled due to compatibility issues)
+    # try:
+    #     from karrio.server.lib.otel_huey import instrument_huey
+    #     instrument_huey()
+    # except Exception as e:
+    #     logger = logging.getLogger(__name__)
+    #     logger.warning(f"Failed to instrument Huey: {e}")
+    
+    # Log that OpenTelemetry is enabled
+    logger = logging.getLogger(__name__)
+    logger.info(f"OpenTelemetry enabled: Service={OTEL_SERVICE_NAME}, Endpoint={OTEL_EXPORTER_OTLP_ENDPOINT}")

apps/api/karrio/server/settings/workers.py

@@ -51,3 +51,19 @@
         filename=WORKER_DB_FILE_NAME,
         **({"immediate": WORKER_IMMEDIATE_MODE} if WORKER_IMMEDIATE_MODE else {}),
     )
+
+
+# Apply OpenTelemetry instrumentation to Huey if enabled
+OTEL_ENABLED = decouple.config("OTEL_ENABLED", default=False, cast=bool)
+OTEL_EXPORTER_OTLP_ENDPOINT = decouple.config("OTEL_EXPORTER_OTLP_ENDPOINT", default=None)
+
+if OTEL_ENABLED and OTEL_EXPORTER_OTLP_ENDPOINT:
+    try:
+        # Import and apply instrumentation to the Huey instance
+        from karrio.server.lib.otel_huey import HueyInstrumentor
+        instrumentor = HueyInstrumentor()
+        instrumentor.instrument(HUEY)
+    except Exception as e:
+        import logging
+        logger = logging.getLogger(__name__)
+        logger.warning(f"Failed to instrument Huey in worker settings: {e}")