Restore original behavior by always creating a span (#3005)

In the original implementation of celery trace propagation we had code to only create a span for the task if it was NOT started by Celery Beat (because there is no transaction created in the beat process, so also not span should be created).

See this code: https://github.com/getsentry/sentry-python/blob/master/sentry_sdk/integrations/celery.py#L187-L200

Turns out this has never worked and task_started_from_beat has always been False meaning a span was ALWAYS created.

(This did never break anything or caused any troube. When looking into a transaction less future this is also absolutely fine.)

So this PR restores now the original behavior by always creating a span.

sentry_sdk/integrations/celery/__init__.py

@@ -11,7 +11,7 @@
     _patch_redbeat_maybe_due,
     _setup_celery_beat_signals,
 )
-from sentry_sdk.integrations.celery.utils import NoOpMgr, _now_seconds_since_epoch
+from sentry_sdk.integrations.celery.utils import _now_seconds_since_epoch
 from sentry_sdk.integrations.logging import ignore_logger
 from sentry_sdk.tracing import BAGGAGE_HEADER_NAME, TRANSACTION_SOURCE_TASK
 from sentry_sdk._types import TYPE_CHECKING
@@ -30,7 +30,6 @@
     from typing import List
     from typing import Optional
     from typing import TypeVar
-    from typing import Union
 
     from sentry_sdk._types import EventProcessor, Event, Hint, ExcInfo
     from sentry_sdk.tracing import Span
@@ -243,15 +242,9 @@ def apply_async(*args, **kwargs):
 
         task = args[0]
 
-        # Do not create a span when the task is a Celery Beat task
-        # (Because we do not have a transaction in that case)
-        span_mgr = (
-            sentry_sdk.start_span(op=OP.QUEUE_SUBMIT_CELERY, description=task.name)
-            if not Scope.get_isolation_scope()._name == "celery-beat"
-            else NoOpMgr()
-        )  # type: Union[Span, NoOpMgr]
-
-        with span_mgr as span:
+        with sentry_sdk.start_span(
+            op=OP.QUEUE_SUBMIT_CELERY, description=task.name
+        ) as span:
             kwargs["headers"] = _update_celery_task_headers(
                 kwarg_headers, span, integration.monitor_beat_tasks
             )