huey.py 5.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174
  1. import sys
  2. from datetime import datetime
  3. import sentry_sdk
  4. from sentry_sdk.api import continue_trace, get_baggage, get_traceparent
  5. from sentry_sdk.consts import OP, SPANSTATUS
  6. from sentry_sdk.integrations import DidNotEnable, Integration
  7. from sentry_sdk.scope import should_send_default_pii
  8. from sentry_sdk.tracing import (
  9. BAGGAGE_HEADER_NAME,
  10. SENTRY_TRACE_HEADER_NAME,
  11. TransactionSource,
  12. )
  13. from sentry_sdk.utils import (
  14. capture_internal_exceptions,
  15. ensure_integration_enabled,
  16. event_from_exception,
  17. SENSITIVE_DATA_SUBSTITUTE,
  18. reraise,
  19. )
  20. from typing import TYPE_CHECKING
  21. if TYPE_CHECKING:
  22. from typing import Any, Callable, Optional, Union, TypeVar
  23. from sentry_sdk._types import EventProcessor, Event, Hint
  24. from sentry_sdk.utils import ExcInfo
  25. F = TypeVar("F", bound=Callable[..., Any])
  26. try:
  27. from huey.api import Huey, Result, ResultGroup, Task, PeriodicTask
  28. from huey.exceptions import CancelExecution, RetryTask, TaskLockedException
  29. except ImportError:
  30. raise DidNotEnable("Huey is not installed")
  31. HUEY_CONTROL_FLOW_EXCEPTIONS = (CancelExecution, RetryTask, TaskLockedException)
  32. class HueyIntegration(Integration):
  33. identifier = "huey"
  34. origin = f"auto.queue.{identifier}"
  35. @staticmethod
  36. def setup_once():
  37. # type: () -> None
  38. patch_enqueue()
  39. patch_execute()
  40. def patch_enqueue():
  41. # type: () -> None
  42. old_enqueue = Huey.enqueue
  43. @ensure_integration_enabled(HueyIntegration, old_enqueue)
  44. def _sentry_enqueue(self, task):
  45. # type: (Huey, Task) -> Optional[Union[Result, ResultGroup]]
  46. with sentry_sdk.start_span(
  47. op=OP.QUEUE_SUBMIT_HUEY,
  48. name=task.name,
  49. origin=HueyIntegration.origin,
  50. ):
  51. if not isinstance(task, PeriodicTask):
  52. # Attach trace propagation data to task kwargs. We do
  53. # not do this for periodic tasks, as these don't
  54. # really have an originating transaction.
  55. task.kwargs["sentry_headers"] = {
  56. BAGGAGE_HEADER_NAME: get_baggage(),
  57. SENTRY_TRACE_HEADER_NAME: get_traceparent(),
  58. }
  59. return old_enqueue(self, task)
  60. Huey.enqueue = _sentry_enqueue
  61. def _make_event_processor(task):
  62. # type: (Any) -> EventProcessor
  63. def event_processor(event, hint):
  64. # type: (Event, Hint) -> Optional[Event]
  65. with capture_internal_exceptions():
  66. tags = event.setdefault("tags", {})
  67. tags["huey_task_id"] = task.id
  68. tags["huey_task_retry"] = task.default_retries > task.retries
  69. extra = event.setdefault("extra", {})
  70. extra["huey-job"] = {
  71. "task": task.name,
  72. "args": (
  73. task.args
  74. if should_send_default_pii()
  75. else SENSITIVE_DATA_SUBSTITUTE
  76. ),
  77. "kwargs": (
  78. task.kwargs
  79. if should_send_default_pii()
  80. else SENSITIVE_DATA_SUBSTITUTE
  81. ),
  82. "retry": (task.default_retries or 0) - task.retries,
  83. }
  84. return event
  85. return event_processor
  86. def _capture_exception(exc_info):
  87. # type: (ExcInfo) -> None
  88. scope = sentry_sdk.get_current_scope()
  89. if exc_info[0] in HUEY_CONTROL_FLOW_EXCEPTIONS:
  90. scope.transaction.set_status(SPANSTATUS.ABORTED)
  91. return
  92. scope.transaction.set_status(SPANSTATUS.INTERNAL_ERROR)
  93. event, hint = event_from_exception(
  94. exc_info,
  95. client_options=sentry_sdk.get_client().options,
  96. mechanism={"type": HueyIntegration.identifier, "handled": False},
  97. )
  98. scope.capture_event(event, hint=hint)
  99. def _wrap_task_execute(func):
  100. # type: (F) -> F
  101. @ensure_integration_enabled(HueyIntegration, func)
  102. def _sentry_execute(*args, **kwargs):
  103. # type: (*Any, **Any) -> Any
  104. try:
  105. result = func(*args, **kwargs)
  106. except Exception:
  107. exc_info = sys.exc_info()
  108. _capture_exception(exc_info)
  109. reraise(*exc_info)
  110. return result
  111. return _sentry_execute # type: ignore
  112. def patch_execute():
  113. # type: () -> None
  114. old_execute = Huey._execute
  115. @ensure_integration_enabled(HueyIntegration, old_execute)
  116. def _sentry_execute(self, task, timestamp=None):
  117. # type: (Huey, Task, Optional[datetime]) -> Any
  118. with sentry_sdk.isolation_scope() as scope:
  119. with capture_internal_exceptions():
  120. scope._name = "huey"
  121. scope.clear_breadcrumbs()
  122. scope.add_event_processor(_make_event_processor(task))
  123. sentry_headers = task.kwargs.pop("sentry_headers", None)
  124. transaction = continue_trace(
  125. sentry_headers or {},
  126. name=task.name,
  127. op=OP.QUEUE_TASK_HUEY,
  128. source=TransactionSource.TASK,
  129. origin=HueyIntegration.origin,
  130. )
  131. transaction.set_status(SPANSTATUS.OK)
  132. if not getattr(task, "_sentry_is_patched", False):
  133. task.execute = _wrap_task_execute(task.execute)
  134. task._sentry_is_patched = True
  135. with sentry_sdk.start_transaction(transaction):
  136. return old_execute(self, task, timestamp)
  137. Huey._execute = _sentry_execute