chore(tracing): adds out.host tag for celery [backport 1.14] (#6056)

Backport e876003 from #6006 to 1.14.

Adds `out.host` tag for celery as a precursor tag for `peer.service`
Testing is updated to check for new tag by calling hostname for local
machine

## Checklist

- [x] Change(s) are motivated and described in the PR description.
- [x] Testing strategy is described if automated tests are not included
in the PR.
- [x] Risk is outlined (performance impact, potential for breakage,
maintainability, etc).
- [x] Change is maintainable (easy to change, telemetry, documentation).

## Reviewer Checklist

- [x] Title is accurate.
- [x] No unnecessary changes are introduced.
- [x] Description motivates each change.
- [x] Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes unless absolutely necessary.
- [x] Testing strategy adequately addresses listed risk(s).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] Release note makes sense to a user of the library.
- [x] Reviewer has explicitly acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment.

Co-authored-by: Zarir Hamza <[email protected]>
Co-authored-by: Munir Abdinur <[email protected]>

Author: 3 people

ddtrace/contrib/celery/signals.py

@@ -1,4 +1,5 @@
 from celery import registry
+from celery.utils import nodenames
 
 from ddtrace import Pin
 from ddtrace import config
@@ -11,6 +12,7 @@
 from ...constants import SPAN_MEASURED_KEY
 from ...ext import SpanKind
 from ...ext import SpanTypes
+from ...ext import net
 from ...internal.logger import get_logger
 from ...propagation.http import HTTPPropagator
 from .utils import attach_span
@@ -124,6 +126,9 @@ def trace_before_publish(*args, **kwargs):
     span.set_tag_str(c.TASK_TAG_KEY, c.TASK_APPLY_ASYNC)
     span.set_tag_str("celery.id", task_id)
     set_tags_from_context(span, kwargs)
+    if kwargs.get("headers") is not None:
+        # required to extract hostname from origin header on `celery>=4.0`
+        set_tags_from_context(span, kwargs["headers"])
 
     # Note: adding tags from `traceback` or `state` calls will make an
     # API call to the backend for the properties so we should rely
@@ -156,6 +161,11 @@ def trace_after_publish(*args, **kwargs):
     if span is None:
         return
     else:
+        nodename = span.get_tag("celery.hostname")
+        if nodename is not None:
+            _, host = nodenames.nodesplit(nodename)
+            span.set_tag_str(net.TARGET_HOST, host)
+
         span.finish()
         detach_span(task, task_id, is_publish=True)
 

releasenotes/notes/celery-precursor-peer.service-d3b1bc05f1ca59ab.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    celery: Resolves an issue where hostname tags were not set in spans generated by ``celery>4.0``.

tests/contrib/celery/test_integration.py

@@ -1,5 +1,6 @@
 from collections import Counter
 import os
+import socket
 import subprocess
 from time import sleep
 
@@ -191,6 +192,7 @@ def fn_task_parameters(user, force_logout=False):
             assert async_span.get_tag("celery.routing_key") == "celery"
             assert async_span.get_tag("component") == "celery"
             assert async_span.get_tag("span.kind") == "producer"
+            assert async_span.get_tag("out.host") == socket.gethostname()
         else:
             assert 1 == len(traces)
             assert 1 == len(traces[0])
@@ -234,6 +236,7 @@ def fn_task_parameters(user, force_logout=False):
             assert async_span.get_tag("celery.routing_key") == "celery"
             assert async_span.get_tag("component") == "celery"
             assert async_span.get_tag("span.kind") == "producer"
+            assert async_span.get_tag("out.host") == socket.gethostname()
         else:
             assert 1 == len(traces)
             assert 1 == len(traces[0])
@@ -682,6 +685,7 @@ def fn_task_parameters(user, force_logout=False):
             assert async_span.get_tag("celery.routing_key") == "celery"
             assert async_span.get_tag("component") == "celery"
             assert async_span.get_tag("span.kind") == "producer"
+            assert async_span.get_tag("out.host") == socket.gethostname()
 
         run_span = self.find_span(name="celery.run")
         assert run_span.name == "celery.run"