Skip to content

feat(observability): auto-instrument httpx for outbound trace propagation #4148

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ecthelion77 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(observability): auto-instrument httpx for outbound trace propagation #4148

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 0 additions & 5 deletions .env.example
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2136,11 +2136,6 @@ OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317
# TOOL_CONCURRENT_LIMIT=10
# GATEWAY_TOOL_NAME_SEPARATOR=-

# Maximum length of response text returned for non-JSON REST API responses
# Longer responses are truncated to prevent exposing excessive sensitive data
# Default: 5000 characters, Range: 1000-100000
# REST_RESPONSE_TEXT_MAX_LENGTH=5000

# Prompt Configuration
# PROMPT_CACHE_SIZE=100
# MAX_PROMPT_SIZE=102400
Expand Down
17 changes: 17 additions & 0 deletions gunicorn.config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,23 @@ def post_fork(server, worker):
except ImportError:
pass

# Re-apply httpx instrumentation in each worker process.
# With preload_app=True, the master process sets _is_instrumented_by_opentelemetry=True
# and applies wrap_function_wrapper patches. After fork(), workers inherit the flag
# but NOT the monkey-patches. We must uninstrument (reset the flag) then re-instrument.
try:
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor

instrumentor = HTTPXClientInstrumentor()
if instrumentor.is_instrumented_by_opentelemetry:
instrumentor.uninstrument()
instrumentor.instrument()
server.log.info("Worker %s: httpx instrumentation applied", worker.pid)
except ImportError:
pass
except Exception as e:
server.log.warning("Worker %s: failed to instrument httpx: %s", worker.pid, e)


def post_worker_init(worker):
worker.log.info("worker initialization completed")
Expand Down
2 changes: 0 additions & 2 deletions mcpgateway/admin_ui/admin.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -485,7 +485,6 @@ Admin.showUsageStatsModal = showUsageStatsModal;
import {
editTool,
initToolSelect,
invokeTool,
testTool,
enrichTool,
generateToolTestCases,
Expand All @@ -497,7 +496,6 @@ import {

Admin.editTool = editTool;
Admin.initToolSelect = initToolSelect;
Admin.invokeTool = invokeTool;
Admin.testTool = testTool;
Admin.enrichTool = enrichTool;
Admin.generateToolTestCases = generateToolTestCases;
Expand Down
199 changes: 6 additions & 193 deletions mcpgateway/admin_ui/tools.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2916,11 +2916,8 @@ export const runToolValidation = async function (testIndex) {
const payload = {
jsonrpc: "2.0",
id: Date.now(),
method: "tools/call",
params: {
name: AppState.currentTestTool.name,
arguments: params,
},
method: AppState.currentTestTool.name,
params,
};

// Parse custom headers from the passthrough headers field
Expand Down Expand Up @@ -3202,16 +3199,14 @@ export const runToolTest = async function () {
const runButton = document.querySelector('button[onclick="runToolTest()"]');

if (!form || !AppState.currentTestTool) {
console.error("Tool test form or current tool not found", {
form: !!form,
currentTestTool: AppState.currentTestTool,
});
console.error("Tool test form or current tool not found");
showErrorMessage("Tool test form not available");
return;
}

// Prevent multiple concurrent test runs
if (runButton && runButton.disabled) {
console.log("Tool test already running");
return;
}

Expand Down Expand Up @@ -3334,11 +3329,8 @@ export const runToolTest = async function () {
const payload = {
jsonrpc: "2.0",
id: Date.now(),
method: "tools/call",
params: {
name: AppState.currentTestTool.name,
arguments: params,
},
method: AppState.currentTestTool.name,
params,
};

// Parse custom headers from the passthrough headers field
Expand Down Expand Up @@ -3510,182 +3502,3 @@ export const cleanupToolTestModal = function () {
console.error("Error cleaning up tool test modal:", error);
}
};

// ===================================================================
// TOOL INVOCATION (opens test modal by tool name)
// ===================================================================

/**
* Fetch tool details from the API by name.
* @param {string} toolName - The name of the tool to fetch
* @returns {Promise<Object>} The tool object
*/
export async function fetchToolDetails(toolName) {
const response = await fetchWithTimeout(
`${window.ROOT_PATH}/admin/tools/${encodeURIComponent(toolName)}`,
{
headers: {
Accept: "application/json",
"Content-Type": "application/json",
},
}
);

if (!response.ok) {
const errorText = await response.text();
throw new Error(
`Failed to fetch tool details (${response.status}): ${errorText}`
);
}

return await response.json();
}

/**
* Create a form input field based on schema.
* @param {string} key - The field name
* @param {Object} schema - The field schema
* @param {boolean} isRequired - Whether the field is required
* @returns {HTMLElement} The input element
*/
export function createFormInput(key, schema, isRequired) {
let input;
const baseInputClass =
"mt-1 block w-full rounded-md border-gray-300 dark:border-gray-600 shadow-sm focus:border-indigo-500 focus:ring-indigo-500 dark:bg-gray-700 dark:text-gray-200";

if (schema.enum) {
input = document.createElement("select");
input.className = baseInputClass;
schema.enum.forEach((option) => {
const opt = document.createElement("option");
opt.value = option;
opt.textContent = option;
if (option === schema.default) {
opt.selected = true;
}
input.appendChild(opt);
});
} else if (schema.type === "boolean") {
input = document.createElement("input");
input.type = "checkbox";
input.className =
"h-4 w-4 text-indigo-600 focus:ring-indigo-500 border-gray-300 rounded dark:bg-gray-700";
if (schema.default === true) {
input.checked = true;
}
} else if (schema.type === "number" || schema.type === "integer") {
input = document.createElement("input");
input.type = "number";
input.className = baseInputClass;
if (schema.default !== undefined) {
input.value = schema.default;
}
} else {
input = document.createElement("input");
input.type = "text";
input.className = baseInputClass;
if (schema.default !== undefined) {
input.value = schema.default;
}
}

input.name = key;
if (isRequired) {
input.required = true;
}

return input;
}

/**
* Generate form fields from tool input schema.
* @param {Object} tool - The tool object with input_schema
*/
export function generateToolFormFields(tool) {
const formFields = safeGetElement("tool-test-form-fields");
if (!formFields) return;

// Clear existing fields safely
while (formFields.firstChild) {
formFields.removeChild(formFields.firstChild);
}

if (!tool.input_schema || !tool.input_schema.properties) {
const noParams = document.createElement("p");
noParams.className = "text-sm text-gray-500 dark:text-gray-400";
noParams.textContent = "This tool has no input parameters.";
formFields.appendChild(noParams);
return;
}

const properties = tool.input_schema.properties;
const required = tool.input_schema.required || [];

for (const [key, schema] of Object.entries(properties)) {
const isRequired = required.includes(key);
const fieldDiv = document.createElement("div");

const label = document.createElement("label");
label.className =
"block text-sm font-medium text-gray-700 dark:text-gray-300 mb-1";
label.textContent = `${key}${isRequired ? " *" : ""}`;
fieldDiv.appendChild(label);

if (schema.description) {
const desc = document.createElement("p");
desc.className = "text-xs text-gray-500 dark:text-gray-400 mb-1";
desc.textContent = schema.description;
fieldDiv.appendChild(desc);
}

const input = createFormInput(key, schema, isRequired);
fieldDiv.appendChild(input);
formFields.appendChild(fieldDiv);
}
}

/**
* Open tool test modal and fetch tool details from API.
* Called by the "Invoke" button in the Tools table.
* @param {string} toolName - The name of the tool to test
*/
export const invokeTool = async function (toolName) {
try {
const tool = await fetchToolDetails(toolName);

// Store tool details in AppState for runToolTest to access
AppState.currentTestTool = tool;

// Populate modal title and description
const titleEl = safeGetElement("tool-test-modal-title");
const descEl = safeGetElement("tool-test-modal-description");
if (titleEl) {
titleEl.textContent = `Test Tool: ${tool.displayName || tool.name}`;
}
if (descEl) {
descEl.textContent = tool.description || "";
}

// Clear previous results
const resultContainer = safeGetElement("tool-test-result");
if (resultContainer) {
resultContainer.textContent = "";
}

// Show the modal
const modal = safeGetElement("tool-test-modal");
if (modal) {
modal.classList.remove("hidden");
}

// Generate form fields based on input schema
if (typeof window.renderToolTestForm === "function") {
window.renderToolTestForm(tool);
} else {
generateToolFormFields(tool);
}
} catch (error) {
console.error("Error invoking tool:", error);
showErrorMessage("Failed to open tool test modal: " + error.message);
}
};
8 changes: 0 additions & 8 deletions mcpgateway/config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1603,14 +1603,6 @@ def parse_issuers(cls, v: Any) -> list[str]:
max_tool_retries: int = 3
tool_rate_limit: int = 100 # requests per minute
tool_concurrent_limit: int = 10
rest_response_text_max_length: int = Field(
default=5000,
ge=1000,
le=100000,
description="Maximum length of response text to return for non-JSON REST API responses. "
"Longer responses are truncated to prevent exposing excessive sensitive data. "
"Default: 5000 characters. Range: 1000-100000.",
)

# Content Security - Size Limits
content_max_resource_size: int = Field(default=102400, ge=1024, le=10485760, description="Maximum size in bytes for resource content (default: 100KB)") # 100KB # Minimum 1KB # Maximum 10MB
Expand Down
19 changes: 16 additions & 3 deletions mcpgateway/observability.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,7 +113,6 @@ class _ConsoleSpanExporterStub: # pragma: no cover - test patch replaces this
logging.getLogger(__name__).debug("Skipping OpenTelemetry shim setup: %s", exc)

# First-Party
from mcpgateway import __version__ # noqa: E402 # pylint: disable=wrong-import-position
from mcpgateway.config import get_settings # noqa: E402 # pylint: disable=wrong-import-position
from mcpgateway.utils.correlation_id import get_correlation_id # noqa: E402 # pylint: disable=wrong-import-position
from mcpgateway.utils.log_sanitizer import sanitize_for_log # noqa: E402 # pylint: disable=wrong-import-position
Expand Down Expand Up @@ -844,7 +843,7 @@ def init_telemetry() -> Optional[Any]:
# Create resource attributes
resource_attributes: Dict[str, Any] = {
"service.name": cfg.otel_service_name,
"service.version": __version__,
"service.version": "1.0.0-RC-2",
"deployment.environment": _get_deployment_environment(),
}

Expand Down Expand Up @@ -989,7 +988,7 @@ def on_end(self, span):
# Get tracer
# Obtain a tracer if trace API available; otherwise create a no-op tracer
if trace is not None and hasattr(trace, "get_tracer"):
_TRACER = cast(Any, trace).get_tracer("mcp-gateway", __version__, schema_url="https://opentelemetry.io/schemas/1.11.0")
_TRACER = cast(Any, trace).get_tracer("mcp-gateway", "1.0.0-RC-2", schema_url="https://opentelemetry.io/schemas/1.11.0")
else:

class _NoopTracer:
Expand All @@ -1008,6 +1007,20 @@ def start_as_current_span(self, _name: str): # type: ignore[override]

_TRACER = _NoopTracer()

# Auto-instrument httpx for outbound trace context propagation
# This ensures traceparent headers are injected into all HTTP requests
# made to backend MCP servers, enabling distributed tracing across the gateway.
try:
# Third-Party
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor # type: ignore[import-untyped]

HTTPXClientInstrumentor().instrument()
logger.info(" httpx instrumentation enabled (outbound trace propagation)")
except ImportError:
logger.debug("opentelemetry-instrumentation-httpx not installed — outbound propagation disabled")
except Exception as inst_err:
logger.warning("Failed to instrument httpx: %s", inst_err)

logger.info(f"✅ OpenTelemetry initialized with {exporter_type} exporter")
if exporter_type == "otlp":
logger.info(f" Endpoint: {_resolve_otlp_endpoint()}")
Expand Down
Loading