fix(logging): make LoggerAdapter OpenTelemetry-safe (Issue #837)

[harsh543]

Summary

This PR adds an OTel-safe logging mode for Temporal's Python LoggerAdapter classes, addressing Issue #837 where nested dicts in LogRecord.extra break OpenTelemetry logging pipelines.

Problem

Temporal's workflow.LoggerAdapter and activity.LoggerAdapter inject nested dictionaries into LogRecord.extra:

extra["temporal_workflow"] = {"workflow_id": "...", "run_id": "...", ...}

OpenTelemetry log attributes must be AnyValue types (scalars, not nested dicts). This causes Temporal context to be dropped or cause errors in OTel pipelines.

Solution

Add a temporal_extra_mode attribute to both LoggerAdapter classes with two modes:

Mode	Behavior	Use Case
"dict" (default)	Nested dict under temporal_workflow/temporal_activity	Backward compatibility
"flatten"	Scalar attributes with temporal.workflow.* / temporal.activity.* prefixes	OpenTelemetry

Usage

from temporalio import workflow, activity

# For OpenTelemetry compatibility:
workflow.logger.temporal_extra_mode = "flatten"
activity.logger.temporal_extra_mode = "flatten"

# Log records now have flat attributes:
# - temporal.workflow.workflow_id
# - temporal.workflow.workflow_type
# - temporal.workflow.run_id
# - temporal.activity.activity_id
# etc.

Changes

• New: temporalio/_log_utils.py - Shared helper functions and TemporalLogExtraMode type
• Modified: temporalio/workflow.py - Added temporal_extra_mode to LoggerAdapter
• Modified: temporalio/activity.py - Added temporal_extra_mode to LoggerAdapter
• New: tests/test_log_utils.py - Unit tests for all modes
• Modified: tests/worker/test_workflow.py - Parameterized integration tests for workflow logging modes
• Modified: tests/worker/test_activity.py - Parameterized integration tests for activity logging modes

Verification Checklist

• Default mode ("dict") preserves existing behavior - no breaking changes
• Flatten mode produces only scalar values (OTel-safe)
• Flatten mode uses temporal.workflow.* and temporal.activity.* prefixes
• Flattened keys don't conflict with LogRecord core attributes
• Non-primitive values converted to strings in flatten mode
• Global logger settings restored in finally blocks
• Type checking passes (pyright)
• Linting passes (ruff)
• Unit tests cover all modes and edge cases
• Integration tests verify workflow and activity adapters

Test Plan

Verify default behavior unchanged:

# Should still produce nested dict
assert "temporal_workflow" in record.__dict__
assert isinstance(record.__dict__["temporal_workflow"], dict)

Verify flatten mode is OTel-safe:

workflow.logger.temporal_extra_mode = "flatten"
# Should produce flat scalar attributes
assert "temporal.workflow.workflow_id" in record.__dict__
assert "temporal_workflow" not in record.__dict__

# All values are primitives
for key, value in record.__dict__.items():
    if key.startswith("temporal.workflow."):
        assert isinstance(value, (str, int, float, bool, type(None)))

Fixes #837

---

🤖 Generated with Claude Code

[CLAassistant]

All committers have signed the CLA.

[harsh543]

The PR is now:

• ✅ Tightly scoped to solving the OTel logging issue
• ✅ Free of unrelated README changes
• ✅ Simplified API with clear use cases
• ✅ Fully backward compatible by default
• ✅ Thoroughly tested with integration tests for both modes

Addressed feedback:

• Removed json mode (no use case identified)
• Reverted all README changes
• Parameterized duplicate tests (test_activity_logging[dict/flatten],
  test_workflow_logging[dict/flatten])

[harsh543]

Re: test changes

Good feedback — consolidated into a single parameterized test that validates all concerns:

What test_workflow_logging[dict/flatten] now tests:

1. Dict/flatten formatting — parameterized to verify LogRecord.extra structure for each mode
2. Replay suppression — worker restart mid-workflow triggers history replay; asserts old logs don't reappear
3. full_workflow_info_on_extra — verifies workflow.Info object is attached when enabled

Structure:

# --- First execution: logs should appear ---
await handle.signal(LoggingWorkflow.my_signal, "signal 1")
...
assert capturer.find_log("Signal: signal 1")

# --- Clear logs and continue execution (replay path) ---
capturer.log_queue.queue.clear()

# --- Replay execution: no duplicate logs ---
assert not capturer.find_log("Signal: signal 1")  # suppressed during replay
assert capturer.find_log("Signal: signal 3")       # new execution logs normally

Cleanup:

finally:
    workflow.logger.temporal_extra_mode = original_mode
    workflow.logger.full_workflow_info_on_extra = original_full_info

No separate replay test needed — one test, clear phases, all concerns covered.