feat: add detailed plugin violation information to OTEL spans

- Add violation.reason, violation.code, violation.description to spans
- Include optional http_status_code and mcp_error_code when present
- Sanitize violation.details to prevent PII leakage using existing redaction
- Mark spans as error when violations occur for better visibility
- Update both OTEL spans and internal observability spans

Closes #4267

Signed-off-by: Bob Shell <bob@example.com>
Signed-off-by: Vishu Bhatnagar <vishu.bhatnagar@ibm.com>

Author: vishu-bh

mcpgateway/alembic/versions/d3e4f5a6b7c8_add_binding_reference_id_to_tool_plugin_bindings.py

@@ -10,8 +10,8 @@
 from typing import Sequence, Union
 
 # Third-Party
-import sqlalchemy as sa
 from alembic import op
+import sqlalchemy as sa
 
 # revision identifiers, used by Alembic.
 revision: str = "d3e4f5a6b7c8"  # pragma: allowlist secret

mcpgateway/middleware/request_logging_middleware.py

@@ -62,8 +62,8 @@
 
 # First-Party
 from mcpgateway.auth import get_current_user
-from mcpgateway.config import settings
 from mcpgateway.common.validators import SecurityValidator
+from mcpgateway.config import settings
 from mcpgateway.middleware.path_filter import should_skip_request_logging
 from mcpgateway.services.logging_service import LoggingService
 from mcpgateway.services.structured_logger import get_structured_logger

mcpgateway/plugins/framework/manager.py

@@ -477,17 +477,68 @@ async def _execute_with_timeout(self, hook_ref: HookRef, payload: PluginPayload,
                 otel_span.set_attribute("plugin.continue_processing", result.continue_processing)
                 otel_span.set_attribute("plugin.stopped_chain", not result.continue_processing)
 
-            # End span with success
+                # Add detailed violation information when present
+                if result.violation:
+                    # First-Party
+                    from mcpgateway.observability import set_span_attribute, set_span_error
+
+                    # Add core violation fields (set_span_attribute handles sanitization)
+                    set_span_attribute(otel_span, "plugin.violation.reason", result.violation.reason)
+                    set_span_attribute(otel_span, "plugin.violation.code", result.violation.code)
+                    set_span_attribute(otel_span, "plugin.violation.description", result.violation.description)
+
+                    # Add HTTP status code if present (e.g., 429 for rate limiting)
+                    if result.violation.http_status_code:
+                        set_span_attribute(otel_span, "plugin.violation.http_status_code", result.violation.http_status_code)
+
+                    # Add MCP error code if present
+                    if result.violation.mcp_error_code:
+                        set_span_attribute(otel_span, "plugin.violation.mcp_error_code", result.violation.mcp_error_code)
+
+                    # Add violation details with sanitization to prevent PII leakage
+                    # Pass just the key name (not the full path) to sanitization for proper field matching
+                    if result.violation.details:
+                        # First-Party
+                        from mcpgateway.utils.trace_redaction import sanitize_trace_attribute_value
+
+                        for key, value in result.violation.details.items():
+                            # Sanitize using just the key name for proper redaction field matching
+                            sanitized_value = sanitize_trace_attribute_value(key, value)
+                            # Use _set_pre_sanitized_span_attribute to avoid double sanitization
+                            if otel_span and sanitized_value is not None:
+                                otel_span.set_attribute(f"plugin.violation.details.{key}", sanitized_value)
+
+                    # Mark span as error for better visibility in observability platforms
+                    set_span_error(otel_span, result.violation.description, record_exception=False)
+
+            # End span with success or error based on violation
             if span_id is not None:
                 try:
+                    span_attributes = {
+                        "plugin.had_violation": result.violation is not None,
+                        "plugin.modified_payload": result.modified_payload is not None,
+                        "plugin.continue_processing": result.continue_processing,
+                    }
+
+                    # Add violation details to internal observability span as well
+                    if result.violation:
+                        # First-Party
+                        from mcpgateway.utils.trace_redaction import sanitize_trace_attribute_value
+
+                        span_attributes["plugin.violation.reason"] = sanitize_trace_attribute_value("plugin.violation.reason", result.violation.reason)
+                        span_attributes["plugin.violation.code"] = sanitize_trace_attribute_value("plugin.violation.code", result.violation.code)
+                        span_attributes["plugin.violation.description"] = sanitize_trace_attribute_value("plugin.violation.description", result.violation.description)
+
+                        if result.violation.http_status_code:
+                            span_attributes["plugin.violation.http_status_code"] = result.violation.http_status_code
+
+                        if result.violation.mcp_error_code:
+                            span_attributes["plugin.violation.mcp_error_code"] = result.violation.mcp_error_code
+
                     self.observability.end_span(
                         span_id=span_id,
-                        status="ok",
-                        attributes={
-                            "plugin.had_violation": result.violation is not None,
-                            "plugin.modified_payload": result.modified_payload is not None,
-                            "plugin.continue_processing": result.continue_processing,
-                        },
+                        status="error" if result.violation else "ok",
+                        attributes=span_attributes,
                     )
                 except Exception as e:
                     logger.debug("Plugin observability end_span failed: %s", e)

mcpgateway/services/tool_plugin_binding_service.py

@@ -173,15 +173,9 @@ def upsert_bindings(
                         # different external references are claiming the same (team, tool, plugin)
                         # triple.  The new reference_id wins (last-caller-wins), but the old
                         # caller's DELETE by reference will now be a no-op.
-                        if (
-                            existing.binding_reference_id
-                            and policy.binding_reference_id
-                            and existing.binding_reference_id != policy.binding_reference_id
-                        ):
+                        if existing.binding_reference_id and policy.binding_reference_id and existing.binding_reference_id != policy.binding_reference_id:
                             logger.warning(
-                                "binding_reference_id ownership transfer: "
-                                "team=%s tool=%s plugin=%s old_ref=%s new_ref=%s — "
-                                "DELETE by old_ref will now be a no-op",
+                                "binding_reference_id ownership transfer: " "team=%s tool=%s plugin=%s old_ref=%s new_ref=%s — " "DELETE by old_ref will now be a no-op",
                                 team_id,
                                 tool_name,
                                 policy.plugin_id.value,
@@ -284,8 +278,7 @@ def list_bindings(
         if binding_reference_id:
             if team_id:
                 logger.warning(
-                    "Both team_id=%r and binding_reference_id=%r supplied to list_bindings; "
-                    "team_id will be ignored. Omit team_id when filtering by binding_reference_id.",
+                    "Both team_id=%r and binding_reference_id=%r supplied to list_bindings; " "team_id will be ignored. Omit team_id when filtering by binding_reference_id.",
                     team_id,
                     binding_reference_id,
                 )

tests/unit/mcpgateway/plugins/framework/test_observability.py

@@ -22,11 +22,13 @@
     GlobalContext,
     Plugin,
     PluginConfig,
+    PluginContext,
     PluginManager,
     PluginMode,
     PluginResult,
     PromptHookType,
     PromptPrehookPayload,
+    PromptPrehookResult,
 )
 from mcpgateway.plugins.framework.base import HookRef
 from mcpgateway.plugins.framework.manager import PluginExecutor
@@ -381,9 +383,206 @@ def record_span(name: str, attributes: Optional[Dict[str, Any]] = None):
     plugin_span = recorded[1]
     assert hook_chain_span.name == "plugin.hook.invoke"
     assert hook_chain_span.attributes["plugin.chain.stopped"] is True
-    assert hook_chain_span.attributes["plugin.chain.stopped_by"] == "BlockingPlugin"
-    assert plugin_span.name == "plugin.execute"
-    assert plugin_span.attributes["plugin.name"] == "BlockingPlugin"
+
+
+@pytest.mark.asyncio
+async def test_plugin_manager_captures_violation_details_in_otel_spans():
+    """Plugin manager should capture detailed violation information in OTEL spans with proper sanitization."""
+    manager = PluginManager("./tests/unit/mcpgateway/plugins/fixtures/configs/valid_no_plugin.yaml", observability=None)
+    await manager.initialize()
+
+    # Create a plugin that returns a violation with full details
+    config = PluginConfig(
+        name="ViolationPlugin",
+        description="Plugin that returns violations",
+        author="Test",
+        version="1.0",
+        tags=["test"],
+        kind="ViolationPlugin",
+        hooks=["prompt_pre_fetch"],
+        config={},
+        mode=PluginMode.ENFORCE,
+    )
+
+    class ViolationPlugin(Plugin):
+        async def prompt_pre_fetch(self, payload: PromptPrehookPayload, context: PluginContext) -> PromptPrehookResult:
+            from mcpgateway.plugins.framework import PluginViolation
+
+            violation = PluginViolation(
+                reason="Content policy violation",
+                description="Request contains prohibited content",
+                code="PROHIBITED_CONTENT",
+                details={
+                    "matched_pattern": "sensitive_keyword",
+                    "field": "user_input",
+                    "password": "secret123",  # Should be sanitized
+                    "api_key": "sk-test-key",  # Should be sanitized
+                },
+                http_status_code=403,
+                mcp_error_code=-32001,
+            )
+            return PromptPrehookResult(continue_processing=False, violation=violation)
+
+    plugin = ViolationPlugin(config)
+
+    class RecordingSpan:
+        def __init__(self, name: str, attributes: Optional[Dict[str, Any]] = None):
+            self.name = name
+            self.attributes = dict(attributes or {})
+            self.status = None
+            self.status_description = None
+
+        def set_attribute(self, key: str, value: Any) -> None:
+            self.attributes[key] = value
+
+        def set_status(self, status: Any) -> None:
+            self.status = status
+            if hasattr(status, "description"):
+                self.status_description = status.description
+
+    recorded: List[RecordingSpan] = []
+
+    @contextmanager
+    def record_span(name: str, attributes: Optional[Dict[str, Any]] = None):
+        span = RecordingSpan(name, attributes)
+        recorded.append(span)
+        yield span
+
+    with patch.object(manager._registry, "get_hook_refs_for_hook") as mock_get:
+        hook_ref = HookRef(PromptHookType.PROMPT_PRE_FETCH, PluginRef(plugin))
+        mock_get.return_value = [hook_ref]
+
+        payload = PromptPrehookPayload(prompt_id="test", args={"user": "test input"})
+        global_context = GlobalContext(request_id="req-violation-test")
+
+        with patch("mcpgateway.plugins.framework.manager.create_span", side_effect=record_span):
+            result, _ = await manager.invoke_hook(
+                PromptHookType.PROMPT_PRE_FETCH,
+                payload,
+                global_context=global_context,
+            )
+
+    # Verify violation was returned
+    assert result.continue_processing is False
+    assert result.violation is not None
+
+    # Find the plugin execution span
+    plugin_span = next((s for s in recorded if s.name == "plugin.execute"), None)
+    assert plugin_span is not None
+
+    # Verify core violation attributes are captured
+    assert plugin_span.attributes["plugin.had_violation"] is True
+    assert plugin_span.attributes["plugin.violation.reason"] == "Content policy violation"
+    assert plugin_span.attributes["plugin.violation.code"] == "PROHIBITED_CONTENT"
+    assert plugin_span.attributes["plugin.violation.description"] == "Request contains prohibited content"
+    assert plugin_span.attributes["plugin.violation.http_status_code"] == 403
+    assert plugin_span.attributes["plugin.violation.mcp_error_code"] == -32001
+
+    # Verify violation details are captured
+    assert "plugin.violation.details.matched_pattern" in plugin_span.attributes
+    assert plugin_span.attributes["plugin.violation.details.matched_pattern"] == "sensitive_keyword"
+    assert "plugin.violation.details.field" in plugin_span.attributes
+    assert plugin_span.attributes["plugin.violation.details.field"] == "user_input"
+
+    # Verify sensitive fields are sanitized (password and api_key should be redacted)
+    assert "plugin.violation.details.password" in plugin_span.attributes
+    assert plugin_span.attributes["plugin.violation.details.password"] == "***"
+    assert "plugin.violation.details.api_key" in plugin_span.attributes
+    assert plugin_span.attributes["plugin.violation.details.api_key"] == "***"
+
+    # Verify span is marked as error
+    assert plugin_span.status is not None
+    assert plugin_span.status_description == "Request contains prohibited content"
+
+    await manager.shutdown()
+
+
+@pytest.mark.asyncio
+async def test_plugin_manager_handles_violation_without_optional_fields():
+    """Plugin manager should handle violations that don't have optional fields (http_status_code, mcp_error_code, details)."""
+    manager = PluginManager("./tests/unit/mcpgateway/plugins/fixtures/configs/valid_no_plugin.yaml", observability=None)
+    await manager.initialize()
+
+    config = PluginConfig(
+        name="MinimalViolationPlugin",
+        description="Plugin with minimal violation",
+        author="Test",
+        version="1.0",
+        tags=["test"],
+        kind="MinimalViolationPlugin",
+        hooks=["prompt_pre_fetch"],
+        config={},
+        mode=PluginMode.ENFORCE,
+    )
+
+    class MinimalViolationPlugin(Plugin):
+        async def prompt_pre_fetch(self, payload: PromptPrehookPayload, context: PluginContext) -> PromptPrehookResult:
+            from mcpgateway.plugins.framework import PluginViolation
+
+            # Violation with only required fields
+            violation = PluginViolation(
+                reason="Rate limit exceeded",
+                description="Too many requests",
+                code="RATE_LIMIT",
+            )
+            return PromptPrehookResult(continue_processing=False, violation=violation)
+
+    plugin = MinimalViolationPlugin(config)
+
+    class RecordingSpan:
+        def __init__(self, name: str, attributes: Optional[Dict[str, Any]] = None):
+            self.name = name
+            self.attributes = dict(attributes or {})
+
+        def set_attribute(self, key: str, value: Any) -> None:
+            self.attributes[key] = value
+
+        def set_status(self, status: Any) -> None:
+            pass
+
+    recorded: List[RecordingSpan] = []
+
+    @contextmanager
+    def record_span(name: str, attributes: Optional[Dict[str, Any]] = None):
+        span = RecordingSpan(name, attributes)
+        recorded.append(span)
+        yield span
+
+    with patch.object(manager._registry, "get_hook_refs_for_hook") as mock_get:
+        hook_ref = HookRef(PromptHookType.PROMPT_PRE_FETCH, PluginRef(plugin))
+        mock_get.return_value = [hook_ref]
+
+        payload = PromptPrehookPayload(prompt_id="test", args={})
+        global_context = GlobalContext(request_id="req-minimal-violation")
+
+        with patch("mcpgateway.plugins.framework.manager.create_span", side_effect=record_span):
+            result, _ = await manager.invoke_hook(
+                PromptHookType.PROMPT_PRE_FETCH,
+                payload,
+                global_context=global_context,
+            )
+
+    # Verify violation was returned
+    assert result.continue_processing is False
+    assert result.violation is not None
+
+    # Find the plugin execution span
+    plugin_span = next((s for s in recorded if s.name == "plugin.execute"), None)
+    assert plugin_span is not None
+
+    # Verify core violation attributes are captured
+    assert plugin_span.attributes["plugin.had_violation"] is True
+    assert plugin_span.attributes["plugin.violation.reason"] == "Rate limit exceeded"
+    assert plugin_span.attributes["plugin.violation.code"] == "RATE_LIMIT"
+    assert plugin_span.attributes["plugin.violation.description"] == "Too many requests"
+
+    # Verify optional fields are not present when not set
+    assert "plugin.violation.http_status_code" not in plugin_span.attributes
+    assert "plugin.violation.mcp_error_code" not in plugin_span.attributes
+    # Details dict is empty, so no detail attributes should be present
+    assert not any(k.startswith("plugin.violation.details.") for k in plugin_span.attributes.keys())
+
+    await manager.shutdown()
 
     await manager.shutdown()