Merge branch 'master' into sentry-sdk-2.0

Author: sentrivana

CHANGELOG.md

@@ -84,6 +84,40 @@
 - Passing a function to `sentry_sdk.init`'s `transport` keyword argument has been deprecated. If you wish to provide a custom transport, please pass a `sentry_sdk.transport.Transport` instance or a subclass.
 - The parameter `propagate_hub` in `ThreadingIntegration()` was deprecated and renamed to `propagate_scope`.
 
+
+
+## 1.43.0
+
+### Various fixes & improvements
+
+- Add optional `keep_alive` (#2842) by @sentrivana
+
+  If you're experiencing frequent network issues between the SDK and Sentry,
+  you can try turning on TCP keep-alive:
+
+  ```python
+  import sentry_sdk
+
+  sentry_sdk.init(
+      # ...your usual settings...
+      keep_alive=True,
+  )
+  ```
+
+- Add support for Celery Redbeat cron tasks (#2643) by @kwigley
+
+  The SDK now supports the Redbeat scheduler in addition to the default
+  Celery Beat scheduler for auto instrumenting crons. See
+  [the docs](https://docs.sentry.io/platforms/python/integrations/celery/crons/)
+  for more information about how to set this up.
+
+- `aws_event` can be an empty list (#2849) by @sentrivana
+- Re-export `Event` in `types.py` (#2829) by @szokeasaurusrex
+- Small API docs improvement (#2828) by @antonpirker
+- Fixed OpenAI tests (#2834) by @antonpirker
+- Bump `checkouts/data-schemas` from `ed078ed` to `8232f17` (#2832) by @dependabot
+
+
 ## 1.42.0
 
 ### Various fixes & improvements

checkouts/data-schemas

@@ -274,6 +274,7 @@ def __init__(
         ignore_errors=[],  # type: Sequence[Union[type, str]]  # noqa: B006
         max_request_body_size="medium",  # type: str
         socket_options=None,  # type: Optional[List[Tuple[int, int, int | bytes]]]
+        keep_alive=False,  # type: bool
         before_send=None,  # type: Optional[EventProcessor]
         before_breadcrumb=None,  # type: Optional[BreadcrumbProcessor]
         debug=None,  # type: Optional[bool]

sentry_sdk/consts.py

@@ -80,7 +80,7 @@ def sentry_handler(aws_event, aws_context, *args, **kwargs):
         # will be the same for all events in the list, since they're all hitting
         # the lambda in the same request.)
 
-        if isinstance(aws_event, list):
+        if isinstance(aws_event, list) and len(aws_event) >= 1:
             request_data = aws_event[0]
             batch_size = len(aws_event)
         else:

sentry_sdk/integrations/aws_lambda.py

@@ -57,6 +57,11 @@
 except ImportError:
     raise DidNotEnable("Celery not installed")
 
+try:
+    from redbeat.schedulers import RedBeatScheduler  # type: ignore
+except ImportError:
+    RedBeatScheduler = None
+
 
 CELERY_CONTROL_FLOW_EXCEPTIONS = (Retry, Ignore, Reject)
 
@@ -77,6 +82,7 @@ def __init__(
 
         if monitor_beat_tasks:
             _patch_beat_apply_entry()
+            _patch_redbeat_maybe_due()
             _setup_celery_beat_signals()
 
     @staticmethod
@@ -525,6 +531,61 @@ def sentry_apply_entry(*args, **kwargs):
     Scheduler.apply_entry = sentry_apply_entry
 
 
+def _patch_redbeat_maybe_due():
+    # type: () -> None
+
+    if RedBeatScheduler is None:
+        return
+
+    original_maybe_due = RedBeatScheduler.maybe_due
+
+    def sentry_maybe_due(*args, **kwargs):
+        # type: (*Any, **Any) -> None
+        scheduler, schedule_entry = args
+        app = scheduler.app
+
+        celery_schedule = schedule_entry.schedule
+        monitor_name = schedule_entry.name
+
+        integration = sentry_sdk.get_client().get_integration(CeleryIntegration)
+        if integration is None:
+            return original_maybe_due(*args, **kwargs)
+
+        if match_regex_list(monitor_name, integration.exclude_beat_tasks):
+            return original_maybe_due(*args, **kwargs)
+
+        # When tasks are started from Celery Beat, make sure each task has its own trace.
+        scope = Scope.get_isolation_scope()
+        scope.set_new_propagation_context()
+
+        monitor_config = _get_monitor_config(celery_schedule, app, monitor_name)
+
+        is_supported_schedule = bool(monitor_config)
+        if is_supported_schedule:
+            headers = schedule_entry.options.pop("headers", {})
+            headers.update(
+                {
+                    "sentry-monitor-slug": monitor_name,
+                    "sentry-monitor-config": monitor_config,
+                }
+            )
+
+            check_in_id = capture_checkin(
+                monitor_slug=monitor_name,
+                monitor_config=monitor_config,
+                status=MonitorStatus.IN_PROGRESS,
+            )
+            headers.update({"sentry-monitor-check-in-id": check_in_id})
+
+            # Set the Sentry configuration in the options of the ScheduleEntry.
+            # Those will be picked up in `apply_async` and added to the headers.
+            schedule_entry.options["headers"] = headers
+
+        return original_maybe_due(*args, **kwargs)
+
+    RedBeatScheduler.maybe_due = sentry_maybe_due
+
+
 def _setup_celery_beat_signals():
     # type: () -> None
     task_success.connect(crons_task_success)

sentry_sdk/integrations/celery.py

@@ -1,26 +1,28 @@
 from abc import ABC, abstractmethod
 import io
-import warnings
-import urllib3
-import certifi
 import gzip
+import socket
 import time
+import warnings
 from datetime import datetime, timedelta, timezone
 from collections import defaultdict
 from urllib.request import getproxies
 
+import urllib3
+import certifi
+
 from sentry_sdk.consts import EndpointType
 from sentry_sdk.utils import Dsn, logger, capture_internal_exceptions
 from sentry_sdk.worker import BackgroundWorker
 from sentry_sdk.envelope import Envelope, Item, PayloadRef
-
 from sentry_sdk._types import TYPE_CHECKING
 
 if TYPE_CHECKING:
     from typing import Any
     from typing import Callable
     from typing import Dict
     from typing import Iterable
+    from typing import List
     from typing import Optional
     from typing import Tuple
     from typing import Type
@@ -35,6 +37,21 @@
     DataCategory = Optional[str]
 
 
+KEEP_ALIVE_SOCKET_OPTIONS = []
+for option in [
+    (socket.SOL_SOCKET, lambda: getattr(socket, "SO_KEEPALIVE"), 1),  # noqa: B009
+    (socket.SOL_TCP, lambda: getattr(socket, "TCP_KEEPIDLE"), 45),  # noqa: B009
+    (socket.SOL_TCP, lambda: getattr(socket, "TCP_KEEPINTVL"), 10),  # noqa: B009
+    (socket.SOL_TCP, lambda: getattr(socket, "TCP_KEEPCNT"), 6),  # noqa: B009
+]:
+    try:
+        KEEP_ALIVE_SOCKET_OPTIONS.append((option[0], option[1](), option[2]))
+    except AttributeError:
+        # a specific option might not be available on specific systems,
+        # e.g. TCP_KEEPIDLE doesn't exist on macOS
+        pass
+
+
 class Transport(ABC):
     """Baseclass for all transports.
 
@@ -424,8 +441,22 @@ def _get_pool_options(self, ca_certs):
             "ca_certs": ca_certs or certifi.where(),
         }
 
-        if self.options["socket_options"]:
-            options["socket_options"] = self.options["socket_options"]
+        socket_options = None  # type: Optional[List[Tuple[int, int, int | bytes]]]
+
+        if self.options["socket_options"] is not None:
+            socket_options = self.options["socket_options"]
+
+        if self.options["keep_alive"]:
+            if socket_options is None:
+                socket_options = []
+
+            used_options = {(o[0], o[1]) for o in socket_options}
+            for default_option in KEEP_ALIVE_SOCKET_OPTIONS:
+                if (default_option[0], default_option[1]) not in used_options:
+                    socket_options.append(default_option)
+
+        if socket_options is not None:
+            options["socket_options"] = socket_options
 
         return options
 

sentry_sdk/transport.py

@@ -49,6 +49,7 @@ def get_file_text(file_name):
         "beam": ["apache-beam>=2.12"],
         "bottle": ["bottle>=0.12.13"],
         "celery": ["celery>=3"],
+        "celery-redbeat": ["celery-redbeat>=2"],
         "chalice": ["chalice>=1.16.0"],
         "clickhouse-driver": ["clickhouse-driver>=0.2.0"],
         "django": ["django>=1.8"],

setup.py

@@ -481,6 +481,7 @@ def test_handler(event, context):
             True,
             2,
         ),
+        (b"[]", False, 1),
     ],
 )
 def test_non_dict_event(

tests/integrations/aws_lambda/test_aws.py

@@ -11,6 +11,7 @@
     _get_humanized_interval,
     _get_monitor_config,
     _patch_beat_apply_entry,
+    _patch_redbeat_maybe_due,
     crons_task_success,
     crons_task_failure,
     crons_task_retry,
@@ -440,3 +441,59 @@ def test_exclude_beat_tasks_option(
                     # The original Scheduler.apply_entry() is called, AND _get_monitor_config is called.
                     assert fake_apply_entry.call_count == 1
                     assert _get_monitor_config.call_count == 1
+
+
+@pytest.mark.parametrize(
+    "task_name,exclude_beat_tasks,task_in_excluded_beat_tasks",
+    [
+        ["some_task_name", ["xxx", "some_task.*"], True],
+        ["some_task_name", ["xxx", "some_other_task.*"], False],
+    ],
+)
+def test_exclude_redbeat_tasks_option(
+    task_name, exclude_beat_tasks, task_in_excluded_beat_tasks
+):
+    """
+    Test excluding Celery RedBeat tasks from automatic instrumentation.
+    """
+    fake_maybe_due = MagicMock()
+
+    fake_redbeat_scheduler = MagicMock()
+    fake_redbeat_scheduler.maybe_due = fake_maybe_due
+
+    fake_integration = MagicMock()
+    fake_integration.exclude_beat_tasks = exclude_beat_tasks
+
+    fake_client = MagicMock()
+    fake_client.get_integration.return_value = fake_integration
+
+    fake_schedule_entry = MagicMock()
+    fake_schedule_entry.name = task_name
+
+    fake_get_monitor_config = MagicMock()
+
+    with mock.patch(
+        "sentry_sdk.integrations.celery.RedBeatScheduler", fake_redbeat_scheduler
+    ) as RedBeatScheduler:  # noqa: N806
+        with mock.patch(
+            "sentry_sdk.integrations.celery.sentry_sdk.get_client",
+            return_value=fake_client,
+        ):
+            with mock.patch(
+                "sentry_sdk.integrations.celery._get_monitor_config",
+                fake_get_monitor_config,
+            ) as _get_monitor_config:
+                # Mimic CeleryIntegration patching of RedBeatScheduler.maybe_due()
+                _patch_redbeat_maybe_due()
+                # Mimic Celery RedBeat calling a task from the RedBeat schedule
+                RedBeatScheduler.maybe_due(fake_redbeat_scheduler, fake_schedule_entry)
+
+                if task_in_excluded_beat_tasks:
+                    # Only the original RedBeatScheduler.maybe_due() is called, _get_monitor_config is NOT called.
+                    assert fake_maybe_due.call_count == 1
+                    _get_monitor_config.assert_not_called()
+
+                else:
+                    # The original RedBeatScheduler.maybe_due() is called, AND _get_monitor_config is called.
+                    assert fake_maybe_due.call_count == 1
+                    assert _get_monitor_config.call_count == 1

tests/integrations/celery/test_celery_beat_crons.py

@@ -12,8 +12,8 @@
 from werkzeug.wrappers import Request, Response
 
 from sentry_sdk import Hub, Client, add_breadcrumb, capture_message, Scope
-from sentry_sdk.transport import _parse_rate_limits
 from sentry_sdk.envelope import Envelope, parse_json
+from sentry_sdk.transport import KEEP_ALIVE_SOCKET_OPTIONS, _parse_rate_limits
 from sentry_sdk.integrations.logging import LoggingIntegration, ignore_logger
 
 
@@ -164,6 +164,66 @@ def test_socket_options(make_client):
     assert options["socket_options"] == socket_options
 
 
+def test_keep_alive_true(make_client):
+    client = make_client(keep_alive=True)
+
+    options = client.transport._get_pool_options([])
+    assert options["socket_options"] == KEEP_ALIVE_SOCKET_OPTIONS
+
+
+def test_keep_alive_off_by_default(make_client):
+    client = make_client()
+    options = client.transport._get_pool_options([])
+    assert "socket_options" not in options
+
+
+def test_socket_options_override_keep_alive(make_client):
+    socket_options = [
+        (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
+        (socket.SOL_TCP, socket.TCP_KEEPINTVL, 10),
+        (socket.SOL_TCP, socket.TCP_KEEPCNT, 6),
+    ]
+
+    client = make_client(socket_options=socket_options, keep_alive=False)
+
+    options = client.transport._get_pool_options([])
+    assert options["socket_options"] == socket_options
+
+
+def test_socket_options_merge_with_keep_alive(make_client):
+    socket_options = [
+        (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 42),
+        (socket.SOL_TCP, socket.TCP_KEEPINTVL, 42),
+    ]
+
+    client = make_client(socket_options=socket_options, keep_alive=True)
+
+    options = client.transport._get_pool_options([])
+    try:
+        assert options["socket_options"] == [
+            (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 42),
+            (socket.SOL_TCP, socket.TCP_KEEPINTVL, 42),
+            (socket.SOL_TCP, socket.TCP_KEEPIDLE, 45),
+            (socket.SOL_TCP, socket.TCP_KEEPCNT, 6),
+        ]
+    except AttributeError:
+        assert options["socket_options"] == [
+            (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 42),
+            (socket.SOL_TCP, socket.TCP_KEEPINTVL, 42),
+            (socket.SOL_TCP, socket.TCP_KEEPCNT, 6),
+        ]
+
+
+def test_socket_options_override_defaults(make_client):
+    # If socket_options are set to [], this doesn't mean the user doesn't want
+    # any custom socket_options, but rather that they want to disable the urllib3
+    # socket option defaults, so we need to set this and not ignore it.
+    client = make_client(socket_options=[])
+
+    options = client.transport._get_pool_options([])
+    assert options["socket_options"] == []
+
+
 def test_transport_infinite_loop(capturing_server, request, make_client):
     client = make_client(
         debug=True,