fix(ray): job long running span could be never finished [backport 3.16] (#14902)

Backport a75ed3e from #14871 to 3.16.

If a job was failing during job_submission, the long_running_job span
was not finished causing the span to not leave driveline and creating
inconsistent behavior on the UI.

Co-authored-by: Louis Tricot <75956635+dubloom@users.noreply.github.com>

Author: dd-octo-sts[bot]

ddtrace/contrib/internal/ray/patch.py

@@ -208,22 +208,22 @@ def traced_submit_job(wrapped, instance, args, kwargs):
         else:
             job_name = DEFAULT_JOB_NAME
 
-    # Root span creation
     job_span = tracer.start_span("ray.job", service=job_name or DEFAULT_JOB_NAME, span_type=SpanTypes.RAY)
-    _inject_ray_span_tags_and_metrics(job_span)
-    job_span.set_tag_str(RAY_SUBMISSION_ID_TAG, submission_id)
-    if entrypoint:
-        job_span.set_tag_str(RAY_ENTRYPOINT, entrypoint)
+    try:
+        # Root span creation
+        _inject_ray_span_tags_and_metrics(job_span)
+        job_span.set_tag_str(RAY_SUBMISSION_ID_TAG, submission_id)
+        if entrypoint:
+            job_span.set_tag_str(RAY_ENTRYPOINT, entrypoint)
 
-    metadata = kwargs.get("metadata", {})
-    dot_paths = flatten_metadata_dict(metadata)
-    for k, v in dot_paths.items():
-        set_tag_or_truncate(job_span, k, v)
+        metadata = kwargs.get("metadata", {})
+        dot_paths = flatten_metadata_dict(metadata)
+        for k, v in dot_paths.items():
+            set_tag_or_truncate(job_span, k, v)
 
-    tracer.context_provider.activate(job_span)
-    start_long_running_job(job_span)
+        tracer.context_provider.activate(job_span)
+        start_long_running_job(job_span)
 
-    try:
         with tracer.trace(
             "ray.job.submit", service=job_name or DEFAULT_JOB_NAME, span_type=SpanTypes.RAY
         ) as submit_span:
@@ -254,7 +254,7 @@ def traced_submit_job(wrapped, instance, args, kwargs):
         job_span.set_tag_str(RAY_JOB_STATUS, RAY_STATUS_ERROR)
         job_span.error = 1
         job_span.set_exc_info(type(e), e, e.__traceback__)
-        job_span.finish()
+        stop_long_running_job(submission_id)
         raise e
 
 

ddtrace/contrib/internal/ray/span_manager.py

@@ -269,7 +269,7 @@ def start_long_running_job(job_span: Span) -> None:
     start_long_running_span(job_span)
 
 
-def stop_long_running_job(submission_id: str, job_info: Optional[JobInfo]) -> None:
+def stop_long_running_job(submission_id: str, job_info: Optional[JobInfo] = None) -> None:
     get_span_manager().stop_long_running_job(submission_id, job_info)
 
 

releasenotes/notes/ray-fix-stuck-job-span-e5c1a175c10968f9.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    ray: This fix resolves an issue where long-running job spans could remain unfinished when an exception occurred during job submission.