microsoft.opentelemetry.a365.core.exporters.utils module

Utilities for the Agent365 exporter.

Shared helpers used by the Agent365 exporter pipeline, including payload serialization, size enforcement, and token-resolution helpers.

microsoft.opentelemetry.a365.core.exporters.utils.hex_trace_id(value)

Convert a 128-bit trace ID to a 32-character hex string.

Return type:

str

microsoft.opentelemetry.a365.core.exporters.utils.hex_span_id(value)

Convert a 64-bit span ID to a 16-character hex string.

Return type:

str

microsoft.opentelemetry.a365.core.exporters.utils.kind_name(kind)

Return span kind name (enum name or numeric).

Return type:

str

microsoft.opentelemetry.a365.core.exporters.utils.status_name(code)

Return status code name.

Return type:

str

microsoft.opentelemetry.a365.core.exporters.utils.truncate_span(span_dict)

Truncate span attributes if the serialized span exceeds MAX_SPAN_SIZE_BYTES.

Removes the largest attributes first until the span fits within the limit.

Return type:

dict[str, Any]

microsoft.opentelemetry.a365.core.exporters.utils.filter_and_partition_by_identity(spans)

Filter export-eligible spans and partition them by (tenantId, agentId).

Only spans whose gen_ai.operation.name is in GEN_AI_OPERATION_NAMES are included; non-genAI spans (e.g. HTTP, DB) and spans with other operation names are filtered out. Spans without both tenant and agent identity are also skipped.

Return type:

dict[tuple[str, str], list[ReadableSpan]]

microsoft.opentelemetry.a365.core.exporters.utils.get_validated_domain_override()

Get and validate the domain override from environment variable.

Returns the validated domain override, or None if not set or invalid.

Return type:

str | None

microsoft.opentelemetry.a365.core.exporters.utils.build_export_url(endpoint, agent_id, tenant_id, use_s2s_endpoint=False)

Construct the full export URL from endpoint and agent ID.

Return type:

str

microsoft.opentelemetry.a365.core.exporters.utils.parse_retry_after(headers, now=None)

Parse the Retry-After header value.

Supports delta-seconds and HTTP-date values.

Return type:

float | None

microsoft.opentelemetry.a365.core.exporters.utils.is_agent365_exporter_enabled()

Check if Agent365 exporter is enabled via environment variable.

Return type:

bool

microsoft.opentelemetry.a365.core.exporters.utils.estimate_value_bytes(value)

Estimate the serialized byte size of a single attribute value in OTLP/HTTP JSON.

Return type:

int

microsoft.opentelemetry.a365.core.exporters.utils.estimate_span_bytes(span)

Heuristic estimator for the serialized size of an OTLP span in HTTP JSON.

Uses generous constants tuned to over-estimate by ~25-50%, providing headroom for JSON serializer variance (whitespace, enum representation, integer-as-string).

Return type:

int

microsoft.opentelemetry.a365.core.exporters.utils.chunk_by_size(items, get_size, max_chunk_bytes)

Split items into sub-batches whose cumulative estimated size stays under max_chunk_bytes.

Multi-item chunks are guaranteed to stay within the limit. A single item whose estimated size exceeds max_chunk_bytes forms its own one-item chunk (never silently dropped) even though that chunk exceeds the limit.

Invariants: - Input order is preserved across chunks. - Empty input produces empty output. - No item is ever dropped. - No chunk is ever empty.

Raises:

ValueError – If max_chunk_bytes is not positive, or if get_size returns a negative value for any item.

Return type:

list[list[TypeVar(T)]]

class microsoft.opentelemetry.a365.core.exporters.utils.A365Handlers(span_processors=<factory>)

Bases: object

Processors created for Agent365 export, mirroring OtlpHandlers.

span_processors: List[SpanProcessor]

microsoft.opentelemetry.a365.core.exporters.utils.is_a365_enabled(enable_a365=False)

Determine whether Agent365 export should be enabled.

Return type:

bool

microsoft.opentelemetry.a365.core.exporters.utils.coerce_storage_directory(value)

Convert nonblank paths, deferring validation to nonfatal storage initialization.

Return type:

Union[str, Path, None]

microsoft.opentelemetry.a365.core.exporters.utils.create_a365_components(token_resolver=None, contextual_token_resolver=None, disable_offline_storage=False, storage_directory=None)

Create Agent365 span processors ready to be added to a TracerProvider.

Parameters:

• token_resolver (Callable[[str, str], Optional[str]] | None) – Optional callable (agent_id, tenant_id) -> str | None. When provided, it is used instead of the default DefaultAzureCredential resolver. This allows callers to supply FIC-based or other custom tokens.

• contextual_token_resolver (Callable[[TokenResolverContext], Optional[str]] | None) – Optional callable (TokenResolverContext) -> str | None. Provides rich context including the agentic user ID. Takes precedence over token_resolver when set.

• disable_offline_storage (bool) – When True, disables durable delivery (no disk writes or replay). Defaults to False.

• storage_directory (Optional[str]) – Root for an Agent365-only a365 cache subdirectory. When None, Azure Monitor’s default path resolver selects an Agent365 namespace. Invalid configuration disables only offline delivery, with an error log. Use separate sibling roots for Agent365 and Azure Monitor; shared or overlapping cache paths are unsupported and are not rejected.

Return type:

A365Handlers

All other configuration is read from environment variables:

• ENABLE_A365_OBSERVABILITY_EXPORTER – must be true for the HTTP exporter

• A365_CLUSTER_CATEGORY – defaults to "prod"

• A365_USE_S2S_ENDPOINT – defaults to False

• A365_SUPPRESS_INVOKE_AGENT_INPUT – defaults to False