Skip to content

Commit

Permalink
Fix DSC in celery tasks started by Celery Beat. (#3047)
Browse files Browse the repository at this point in the history
Only create the span for enqueuing the task when we are not currently running in a celery beat task. (because then there is only the span without a transaction and thus the baggage header can not be given to the child celery task.) Without the span the child celery task creates its own trace, this is what we want.
  • Loading branch information
antonpirker authored May 6, 2024
1 parent 0f0cde7 commit 5520bdb
Show file tree
Hide file tree
Showing 3 changed files with 41 additions and 12 deletions.
Original file line number Diff line number Diff line change
@@ -1,12 +1,12 @@
import boto3
import sentry_sdk
import sentry_sdk


monitor_slug = "python-sdk-aws-lambda-tests-cleanup"
monitor_config = {
"schedule": {
"type": "crontab",
"value": "0 12 * * 0", # 12 o'clock on Sunday
"value": "0 12 * * 0", # 12 o'clock on Sunday
},
"timezone": "UTC",
"checkin_margin": 2,
Expand All @@ -24,7 +24,7 @@ def delete_lambda_functions(prefix="test_"):
"""
client = boto3.client("lambda", region_name="us-east-1")
functions_deleted = 0

functions_paginator = client.get_paginator("list_functions")
for functions_page in functions_paginator.paginate():
for func in functions_page["Functions"]:
Expand All @@ -39,17 +39,17 @@ def delete_lambda_functions(prefix="test_"):
print(f"Got exception: {ex}")

return functions_deleted


def lambda_handler(event, context):
functions_deleted = delete_lambda_functions()

sentry_sdk.metrics.gauge(
key="num_aws_functions_deleted",
key="num_aws_functions_deleted",
value=functions_deleted,
)

return {
'statusCode': 200,
'body': f"{functions_deleted} AWS Lambda functions deleted successfully."
"statusCode": 200,
"body": f"{functions_deleted} AWS Lambda functions deleted successfully.",
}
25 changes: 22 additions & 3 deletions sentry_sdk/integrations/celery/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
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
Expand Down Expand Up @@ -223,6 +224,16 @@ def _update_celery_task_headers(original_headers, span, monitor_beat_tasks):
return updated_headers


class NoOpMgr:
def __enter__(self):
# type: () -> None
return None

def __exit__(self, exc_type, exc_value, traceback):
# type: (Any, Any, Any) -> None
return None


def _wrap_apply_async(f):
# type: (F) -> F
@wraps(f)
Expand All @@ -242,9 +253,17 @@ def apply_async(*args, **kwargs):

task = args[0]

with sentry_sdk.start_span(
op=OP.QUEUE_SUBMIT_CELERY, description=task.name
) as span:
task_started_from_beat = (
sentry_sdk.Scope.get_isolation_scope()._name == "celery-beat"
)

span_mgr = (
sentry_sdk.start_span(op=OP.QUEUE_SUBMIT_CELERY, description=task.name)
if not task_started_from_beat
else NoOpMgr()
) # type: Union[Span, NoOpMgr]

with span_mgr as span:
kwargs["headers"] = _update_celery_task_headers(
kwarg_headers, span, integration.monitor_beat_tasks
)
Expand Down
10 changes: 10 additions & 0 deletions sentry_sdk/tracing_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -421,6 +421,16 @@ def update(self, other_dict):
except AttributeError:
pass

def __repr__(self):
# type: (...) -> str
return "<PropagationContext _trace_id={} _span_id={} parent_span_id={} parent_sampled={} dynamic_sampling_context={}>".format(
self._trace_id,
self._span_id,
self.parent_span_id,
self.parent_sampled,
self.dynamic_sampling_context,
)


class Baggage:
"""
Expand Down

0 comments on commit 5520bdb

Please # to comment.