feat: Introduce logging context to Tangle

[morgan-wowk]

Issue
#67

Changes:

• Adds trace context helpers
• Add trace middleware to apply trace context to API requests
  • This helps logging formatters pull trace ids out of context to set them in logs
  • Sets the trace id on the response for client consumption
• Wraps API requests with trace context using middleware
• Passes trace id to pipeline executions using extras parameter
• Wraps pipeline execution with trace context (using trace id from extras parameter, or creating a new one if None)

Test pipeline notes:

• Requires ytest, starlette, httpx (e.g. pip3 install ...)

[Ark-kun]

Thank you for this PR.

I'm reviewing the changes and I think we might want to design this slightly differently.
I think that trace IDs are mostly relevant to API Server and API requests.
Orchestrator works autonomously, it's not triggered by API requests, so it does not have them or need them. Instead, the ExecutionNode.id and ContainerExecution.id already serve the same role.

On the other hand, API requests might benefit from trace IDs so that all logging messages that are generated when processing a single API call can be filtered and grouped together.

[morgan-wowk]

Thank you for this PR.

I'm reviewing the changes and I think we might want to design this slightly differently. I think that trace IDs are mostly relevant to API Server and API requests. Orchestrator works autonomously, it's not triggered by API requests, so it does not have them or need them. Instead, the ExecutionNode.id and ContainerExecution.id already serve the same role.

On the other hand, API requests might benefit from trace IDs so that all logging messages that are generated when processing a single API call can be filtered and grouped together.

This sounds good to me.

I will make some adjustments. Thanks!

[morgan-wowk]

Thank you for this PR.

I'm reviewing the changes and I think we might want to design this slightly differently. I think that trace IDs are mostly relevant to API Server and API requests. Orchestrator works autonomously, it's not triggered by API requests, so it does not have them or need them. Instead, the ExecutionNode.id and ContainerExecution.id already serve the same role.

On the other hand, API requests might benefit from trace IDs so that all logging messages that are generated when processing a single API call can be filtered and grouped together.

The work has been re-designed. Here is a summary:

Logging Context Improvements

Summary

Enhanced the Tangle orchestrator logging to automatically include execution context (execution_id, container_execution_id) in all log messages, making it easier to trace and filter logs for specific executions.

Changes Made

1. Added Logging Context to Orchestrator (orchestrator_sql.py)

Wrapped execution processing with logging_context.logging_context() to automatically attach execution IDs to all logs within that scope.

For Queued Executions:

# Set execution context for logging
with logging_context.logging_context(
    execution_id=queued_execution.id
):
    _logger.info("Before processing queued execution")
    # ... process execution ...
    _logger.info("After processing queued execution")

For Running Container Executions:

with logging_context.logging_context(
    execution_id=execution_id,
    container_execution_id=running_container_execution.id
):
    _logger.info("Before processing running container execution")
    # ... process container execution ...
    _logger.info("After processing running container execution")

2. Added Context-Aware Logging Configuration (start_local.py)

Created a custom formatter and filter that dynamically includes context fields in logs only when they're set.

class LoggingContextFilter(logging.Filter):
    """Adds contextual metadata to log records."""
    
    def filter(self, record: logging.LogRecord) -> bool:
        for key, value in get_all_context_metadata().items():
            if value is not None:
                setattr(record, key, value)
        return True


class ContextAwareFormatter(logging.Formatter):
    """Formatter that dynamically includes context fields only when they're set."""
    
    def format(self, record: logging.LogRecord) -> str:
        base_format = "%(asctime)s [%(levelname)s] %(name)s"
        
        # Collect context fields that are present
        context_parts = []
        context_metadata = get_all_context_metadata()
        for key, value in context_metadata.items():
            if value is not None and hasattr(record, key):
                context_parts.append(f"{key}={value}")
        
        # Add context to format if any exists
        if context_parts:
            base_format += " [" + " ".join(context_parts) + "]"
        
        base_format += ": %(message)s"
        
        formatter = logging.Formatter(base_format)
        return formatter.format(record)

3. Cleaned Up Redundant IDs in Log Messages

Removed execution IDs from log messages where they're already present in the logging context, eliminating duplication.

Examples of cleaned up logs:

# Before: ID appears twice (in context and message)
_logger.info(f"Before processing {queued_execution.id=}")

# After: ID only in context
_logger.info("Before processing queued execution")

# Before
_logger.info(f"Container execution {container_execution.id} is now in state {new_status}")

# After
_logger.info(f"Container execution is now in state {new_status}")

# Before
_logger.info(f"Terminating container execution {container_execution.id}.")

# After
_logger.info("Terminating container execution.")

Log Output Examples

Without Context (non-orchestrator logs)

2026-01-19 14:40:52,611 [INFO] start_local: Starting the orchestrator
2026-01-19 14:40:52,620 [INFO] uvicorn.error: Application startup complete.

With Execution Context Only

2026-01-19 14:30:13,650 [INFO] cloud_pipelines_backend.orchestrator_sql [execution_id=019bd8612bb80ea96aba]: Before processing queued execution
2026-01-19 14:30:13,674 [INFO] cloud_pipelines_backend.orchestrator_sql [execution_id=019bd8612bb80ea96aba]: Execution will reuse the old_execution.id='019b3428a0f1e29c82d4'
2026-01-19 14:30:13,681 [INFO] cloud_pipelines_backend.orchestrator_sql [execution_id=019bd8612bb80ea96aba]: After processing queued execution

With Both Execution and Container Execution Context

2026-01-19 14:30:14,687 [INFO] cloud_pipelines_backend.orchestrator_sql [execution_id=019bd8612bb888cc61d4 container_execution_id=019b342ab120726fdee2]: Before processing running container execution
2026-01-19 14:30:14,701 [INFO] cloud_pipelines_backend.orchestrator_sql [execution_id=019bd8612bb888cc61d4 container_execution_id=019b342ab120726fdee2]: Container execution is now in state RUNNING (was PENDING)
2026-01-19 14:30:14,713 [INFO] cloud_pipelines_backend.orchestrator_sql [execution_id=019bd8612bb888cc61d4 container_execution_id=019b342ab120726fdee2]: After processing running container execution

Benefits

1. Easy Log Filtering: Filter logs by execution_id or container_execution_id in log aggregation tools
2. Better Traceability: Follow an execution through its entire lifecycle
3. No Duplication: IDs appear once in context, not repeated in every message
4. Dynamic Context: Formatter automatically includes any context fields without code changes
5. Cleaner Messages: Log messages are more readable without ID clutter

Extensibility

The context system is fully extensible. You can add any arbitrary metadata:

with logging_context.logging_context(
    execution_id=execution.id,
    container_execution_id=container.id,
    pipeline_run_id=pipeline.id,  # Would work if added
    user_id=user.id,               # Would work if added
    custom_field="value"           # Any custom field works
):
    _logger.info("Processing with custom context")

All fields automatically appear in logs:

[INFO] module [execution_id=123 container_execution_id=456 pipeline_run_id=789 user_id=admin custom_field=value]: Processing with custom context

Files Modified

• tangle/cloud_pipelines_backend/orchestrator_sql.py - Added logging context to execution processing
• tangle/start_local.py - Added LoggingContextFilter and ContextAwareFormatter

Backward Compatibility

• ✅ All changes are backward compatible
• ✅ Logs without context still work (no context fields shown)
• ✅ Existing log parsing tools continue to work
• ✅ All tests pass

[Ark-kun]

Thank you. This looks pretty good.
I've left some comments and approved it.

[Ark-kun]

Hmm. Are you sure we no longer need to log processing times?

[morgan-wowk]

I restored duration. It should have been removed yet. Maybe in the future when we have a replacement solution

[morgan-wowk]

Thanks for pointing this out

[Ark-kun]

Note that you're slightly changing the logic here by making SQL queries outside the try/except context. If an exception occurs, it will go up the stack.

You might want to swap try and with contexts.

Also, AFAIK, the code now always queries the ExecutionNodes DB table whereas before it would only do that when needed. This is likely not affecting perf too much but I'd like you to understand the changes you're making (basically, you add and extra DB query to extract more information from the DB for the purpose of logging it.).

[morgan-wowk]

I will review this area again. My editor during development of this portion was not behaving very well, telling me the code was wrong (showing warnings) and I remember it was difficult for me satisfy. Maybe, this lead to this difference in behaviour.

[morgan-wowk]

Also, AFAIK, the code now always queries the ExecutionNodes DB table whereas before it would only do that when needed. This is likely not affecting perf too much but I'd like you to understand the changes you're making (basically, you add and extra DB query to extract more information from the DB for the purpose of logging it.).

Thank you. I do understand this.

I'm curious, would you prefer to take any other approach such as passing metadata around differently to avoid a DB call? Open to any better ideas.

[morgan-wowk]

Thanks for pairing. I have implemented the changes we discussed:

• Singular with block

          with contextual_logging.logging_context(
              container_execution_id=running_container_execution.id,
              execution_node_ids=execution_node_ids,
          ):

• Moving the execution_node_ids select we wrote above / outside the try catch

Also:

• Restored duration tracking