feat(trace): support external trace id propagation (#22623)

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

api/core/app/apps/advanced_chat/app_generator.py

@@ -23,6 +23,7 @@ from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, InvokeFrom
 from core.app.entities.task_entities import ChatbotAppBlockingResponse, ChatbotAppStreamResponse
+from core.helper.trace_id_helper import extract_external_trace_id_from_args
 from core.model_runtime.errors.invoke import InvokeAuthorizationError
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.prompt.utils.get_thread_messages_length import get_thread_messages_length
@@ -112,7 +113,10 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         query = query.replace("\x00", "")
         inputs = args["inputs"]
 
-        extras = {"auto_generate_conversation_name": args.get("auto_generate_name", False)}
+        extras = {
+            "auto_generate_conversation_name": args.get("auto_generate_name", False),
+            **extract_external_trace_id_from_args(args),
+        }
 
         # get conversation
         conversation = None

api/core/app/apps/advanced_chat/generate_task_pipeline.py

@@ -559,6 +559,7 @@ class AdvancedChatAppGenerateTaskPipeline:
                 outputs=event.outputs,
                 conversation_id=self._conversation_id,
                 trace_manager=trace_manager,
+                external_trace_id=self._application_generate_entity.extras.get("external_trace_id"),
             )
             workflow_finish_resp = self._workflow_response_converter.workflow_finish_to_stream_response(
                 session=session,
@@ -590,6 +591,7 @@ class AdvancedChatAppGenerateTaskPipeline:
                 exceptions_count=event.exceptions_count,
                 conversation_id=None,
                 trace_manager=trace_manager,
+                external_trace_id=self._application_generate_entity.extras.get("external_trace_id"),
             )
             workflow_finish_resp = self._workflow_response_converter.workflow_finish_to_stream_response(
                 session=session,
@@ -622,6 +624,7 @@ class AdvancedChatAppGenerateTaskPipeline:
                 conversation_id=self._conversation_id,
                 trace_manager=trace_manager,
                 exceptions_count=event.exceptions_count,
+                external_trace_id=self._application_generate_entity.extras.get("external_trace_id"),
             )
             workflow_finish_resp = self._workflow_response_converter.workflow_finish_to_stream_response(
                 session=session,
@@ -653,6 +656,7 @@ class AdvancedChatAppGenerateTaskPipeline:
                     error_message=event.get_stop_reason(),
                     conversation_id=self._conversation_id,
                     trace_manager=trace_manager,
+                    external_trace_id=self._application_generate_entity.extras.get("external_trace_id"),
                 )
                 workflow_finish_resp = self._workflow_response_converter.workflow_finish_to_stream_response(
                     session=session,

api/core/app/apps/workflow/app_generator.py

@@ -22,6 +22,7 @@ from core.app.apps.workflow.generate_response_converter import WorkflowAppGenera
 from core.app.apps.workflow.generate_task_pipeline import WorkflowAppGenerateTaskPipeline
 from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity
 from core.app.entities.task_entities import WorkflowAppBlockingResponse, WorkflowAppStreamResponse
+from core.helper.trace_id_helper import extract_external_trace_id_from_args
 from core.model_runtime.errors.invoke import InvokeAuthorizationError
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.repositories import DifyCoreRepositoryFactory
@@ -123,6 +124,10 @@ class WorkflowAppGenerator(BaseAppGenerator):
         )
 
         inputs: Mapping[str, Any] = args["inputs"]
+
+        extras = {
+            **extract_external_trace_id_from_args(args),
+        }
         workflow_run_id = str(uuid.uuid4())
         # init application generate entity
         application_generate_entity = WorkflowAppGenerateEntity(
@@ -142,6 +147,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
             call_depth=call_depth,
             trace_manager=trace_manager,
             workflow_execution_id=workflow_run_id,
+            extras=extras,
         )
 
         contexts.plugin_tool_providers.set({})

api/core/app/apps/workflow/generate_task_pipeline.py

@@ -490,6 +490,7 @@ class WorkflowAppGenerateTaskPipeline:
                 outputs=event.outputs,
                 conversation_id=None,
                 trace_manager=trace_manager,
+                external_trace_id=self._application_generate_entity.extras.get("external_trace_id"),
             )
 
             # save workflow app log
@@ -524,6 +525,7 @@ class WorkflowAppGenerateTaskPipeline:
                 exceptions_count=event.exceptions_count,
                 conversation_id=None,
                 trace_manager=trace_manager,
+                external_trace_id=self._application_generate_entity.extras.get("external_trace_id"),
             )
 
             # save workflow app log
@@ -561,6 +563,7 @@ class WorkflowAppGenerateTaskPipeline:
                 conversation_id=None,
                 trace_manager=trace_manager,
                 exceptions_count=event.exceptions_count if isinstance(event, QueueWorkflowFailedEvent) else 0,
+                external_trace_id=self._application_generate_entity.extras.get("external_trace_id"),
             )
 
             # save workflow app log