Overview
The Telemetry Common module provides shared utility functions for OpenTelemetry context management, span retrieval, data serialization, sensitive data redaction, and user attribute enrichment used across the Guardrails telemetry subsystem.
Description
This module contains foundational utilities used by both the guard tracing and open inference telemetry modules:
Context and Span Management: get_current_context safely retrieves the current OpenTelemetry context. get_span returns the active span from a provided span or the current context, with graceful fallback to None.
Serialization: serialize converts arbitrary values (objects with to_dict, __dict__, dicts, lists) to JSON strings. to_dict converts values to dictionaries using similar heuristics.
Context Propagation: wrap_with_otel_context creates a wrapper function that attaches a specified OpenTelemetry context before executing the wrapped function and detaches it afterward, ensuring trace context is preserved across execution boundaries (e.g., thread pools, async frameworks).
User Attributes: add_user_attributes reads OpenTelemetry baggage values (client IP, user agent, referrer, user ID, organization, app) and sets them as span attributes.
Sensitive Data Handling: redact masks all but the last four characters of a string with asterisks. ismatchingkey checks if a key name contains sensitive patterns (key, token, password). can_convert_to_dict tests if a string is valid JSON. recursive_key_operation recursively traverses nested data structures (dicts, lists, JSON strings) and applies an operation to values whose keys match sensitive patterns.
Usage
Use this module's functions when building telemetry integrations for Guardrails. The serialization and redaction utilities are used by the open inference and guard tracing modules to safely log span attributes without exposing sensitive credentials.
Code Reference
Source Location
- Repository: Guardrails
- File:
guardrails/telemetry/common.py
Signature
def get_current_context() -> Union[Context, None]
def get_span(span: Optional[Span] = None) -> Optional[Span]
def serialize(val: Any) -> Optional[str]
def to_dict(val: Any) -> Dict
def wrap_with_otel_context(
outer_scope_otel_context: Context,
func: Callable[..., Any],
) -> Callable[..., Any]
def add_user_attributes(span: Span)
def redact(value: str) -> str
def ismatchingkey(
target_key: str,
keys_to_match: tuple[str, ...] = ("key", "token", "password"),
) -> bool
def can_convert_to_dict(s: str) -> bool
def recursive_key_operation(
data: Optional[Union[Dict[str, Any], List[Any], str]],
operation: Callable[[str], str],
keys_to_match: List[str] = ["key", "token", "password"],
) -> Optional[Union[Dict[str, Any], List[Any], str]]
Import
from guardrails.telemetry.common import (
get_current_context,
get_span,
serialize,
to_dict,
wrap_with_otel_context,
add_user_attributes,
redact,
ismatchingkey,
can_convert_to_dict,
recursive_key_operation,
)
I/O Contract
get_current_context
| Returns |
Type |
Description
|
| Context |
Union[Context, None] |
The current OpenTelemetry context, or None if unavailable
|
get_span
| Parameter |
Type |
Description
|
span |
Optional[Span] |
An explicit span to use; if None, retrieves from current context
|
| Returns |
Type |
Description
|
| Span |
Optional[Span] |
The active span, or None if no valid span is available
|
serialize
| Parameter |
Type |
Description
|
val |
Any |
The value to serialize
|
| Returns |
Type |
Description
|
| JSON string |
Optional[str] |
JSON string representation, or None on failure or None input
|
wrap_with_otel_context
| Parameter |
Type |
Description
|
outer_scope_otel_context |
Context |
The OpenTelemetry context to attach during function execution
|
func |
Callable[..., Any] |
The function to wrap
|
| Returns |
Type |
Description
|
| Wrapped function |
Callable[..., Any] |
A function that executes with the specified context attached
|
redact
| Parameter |
Type |
Description
|
value |
str |
The string to redact
|
| Returns |
Type |
Description
|
| Redacted string |
str |
String with all but the last 4 characters replaced by asterisks
|
recursive_key_operation
| Parameter |
Type |
Description
|
data |
Optional[Union[Dict, List, str]] |
The data structure to traverse
|
operation |
Callable[[str], str] |
Function to apply to matched key values
|
keys_to_match |
List[str] |
Keys to match (default: ["key", "token", "password"])
|
| Returns |
Type |
Description
|
| Modified data |
Optional[Union[Dict, List, str]] |
The data structure with the operation applied to values of matched keys
|
Usage Examples
from guardrails.telemetry.common import (
serialize, redact, recursive_key_operation, wrap_with_otel_context
)
# Serialize an object to JSON
data = {"name": "Alice", "score": 0.95}
json_str = serialize(data)
# '{"name": "Alice", "score": 0.95}'
# Redact a sensitive value
api_key = "sk-1234567890abcdef"
redacted = redact(api_key)
# "**************cdef"
# Recursively redact sensitive keys in nested data
config = {
"model": "gpt-4",
"api_key": "sk-secret123456",
"nested": {
"token": "tok-abcdef7890",
}
}
safe_config = recursive_key_operation(config, redact)
# {"model": "gpt-4", "api_key": "********3456", "nested": {"token": "******7890"}}
# Wrap a function with OpenTelemetry context
from opentelemetry import context
current_ctx = context.get_current()
wrapped_fn = wrap_with_otel_context(current_ctx, my_function)
result = wrapped_fn(arg1, arg2)
Related Pages
