Support SparkIntegration activation after SparkContext created

sentry_sdk/integrations/spark/spark_driver.py

@@ -1,6 +1,7 @@
 import sentry_sdk
 from sentry_sdk.integrations import Integration
 from sentry_sdk.utils import capture_internal_exceptions, ensure_integration_enabled
+from sentry_sdk.scope import Scope
 
 from sentry_sdk._types import TYPE_CHECKING
 
@@ -17,7 +18,7 @@ class SparkIntegration(Integration):
     @staticmethod
     def setup_once():
         # type: () -> None
-        patch_spark_context_init()
+        _setup_sentry_tracing()
 
 
 def _set_app_properties():
@@ -49,7 +50,52 @@ def _start_sentry_listener(sc):
     sc._jsc.sc().addSparkListener(listener)
 
 
-def patch_spark_context_init():
+def _add_event_processor(sc):
+    # type: (Any) -> None
+    scope = sentry_sdk.get_isolation_scope()
+
+    @scope.add_event_processor
+    def process_event(event, hint):
+        # type: (Event, Hint) -> Optional[Event]
+        with capture_internal_exceptions():
+            if sentry_sdk.get_client().get_integration(SparkIntegration) is None:
+                return event
+
+            if sc._active_spark_context is None:
+                return event
+
+            event.setdefault("user", {}).setdefault("id", sc.sparkUser())
+
+            event.setdefault("tags", {}).setdefault(
+                "executor.id", sc._conf.get("spark.executor.id")
+            )
+            event["tags"].setdefault(
+                "spark-submit.deployMode",
+                sc._conf.get("spark.submit.deployMode"),
+            )
+            event["tags"].setdefault("driver.host", sc._conf.get("spark.driver.host"))
+            event["tags"].setdefault("driver.port", sc._conf.get("spark.driver.port"))
+            event["tags"].setdefault("spark_version", sc.version)
+            event["tags"].setdefault("app_name", sc.appName)
+            event["tags"].setdefault("application_id", sc.applicationId)
+            event["tags"].setdefault("master", sc.master)
+            event["tags"].setdefault("spark_home", sc.sparkHome)
+
+            event.setdefault("extra", {}).setdefault("web_url", sc.uiWebUrl)
+
+        return event
+
+
+def _activate_integration(sc):
+    # type: (SparkContext) -> None
+    from pyspark import SparkContext
+
+    _start_sentry_listener(sc)
+    _set_app_properties()
+    _add_event_processor(sc)
+
+
+def _patch_spark_context_init():
     # type: () -> None
     from pyspark import SparkContext
 
@@ -59,51 +105,22 @@ def patch_spark_context_init():
     def _sentry_patched_spark_context_init(self, *args, **kwargs):
         # type: (SparkContext, *Any, **Any) -> Optional[Any]
         rv = spark_context_init(self, *args, **kwargs)
-        _start_sentry_listener(self)
-        _set_app_properties()
-
-        scope = sentry_sdk.get_isolation_scope()
-
-        @scope.add_event_processor
-        def process_event(event, hint):
-            # type: (Event, Hint) -> Optional[Event]
-            with capture_internal_exceptions():
-                if sentry_sdk.get_client().get_integration(SparkIntegration) is None:
-                    return event
-
-                if self._active_spark_context is None:
-                    return event
-
-                event.setdefault("user", {}).setdefault("id", self.sparkUser())
-
-                event.setdefault("tags", {}).setdefault(
-                    "executor.id", self._conf.get("spark.executor.id")
-                )
-                event["tags"].setdefault(
-                    "spark-submit.deployMode",
-                    self._conf.get("spark.submit.deployMode"),
-                )
-                event["tags"].setdefault(
-                    "driver.host", self._conf.get("spark.driver.host")
-                )
-                event["tags"].setdefault(
-                    "driver.port", self._conf.get("spark.driver.port")
-                )
-                event["tags"].setdefault("spark_version", self.version)
-                event["tags"].setdefault("app_name", self.appName)
-                event["tags"].setdefault("application_id", self.applicationId)
-                event["tags"].setdefault("master", self.master)
-                event["tags"].setdefault("spark_home", self.sparkHome)
-
-                event.setdefault("extra", {}).setdefault("web_url", self.uiWebUrl)
-
-            return event
-
+        _activate_integration(self)
         return rv
 
     SparkContext._do_init = _sentry_patched_spark_context_init
 
 
+def _setup_sentry_tracing():
+    # type: () -> None
+    from pyspark import SparkContext
+
+    if SparkContext._active_spark_context is not None:
+        _activate_integration(SparkContext._active_spark_context)
+        return
+    _patch_spark_context_init()
+
+
 class SparkListener:
     def onApplicationEnd(self, applicationEnd):  # noqa: N802,N803
         # type: (Any) -> None
@@ -208,10 +225,20 @@ class Java:
 
 
 class SentryListener(SparkListener):
+    def _add_breadcrumb(
+        self,
+        level,  # type: str
+        message,  # type: str
+        data=None,  # type: Optional[dict[str, Any]]
+    ):
+        # type: (...) -> None
+        Scope.set_isolation_scope(Scope.get_global_scope())
+        sentry_sdk.add_breadcrumb(level=level, message=message, data=data)
+
     def onJobStart(self, jobStart):  # noqa: N802,N803
         # type: (Any) -> None
         message = "Job {} Started".format(jobStart.jobId())
-        sentry_sdk.add_breadcrumb(level="info", message=message)
+        self._add_breadcrumb(level="info", message=message)
         _set_app_properties()
 
     def onJobEnd(self, jobEnd):  # noqa: N802,N803
@@ -227,14 +254,14 @@ def onJobEnd(self, jobEnd):  # noqa: N802,N803
             level = "warning"
             message = "Job {} Failed".format(jobEnd.jobId())
 
-        sentry_sdk.add_breadcrumb(level=level, message=message, data=data)
+        self._add_breadcrumb(level=level, message=message, data=data)
 
     def onStageSubmitted(self, stageSubmitted):  # noqa: N802,N803
         # type: (Any) -> None
         stage_info = stageSubmitted.stageInfo()
         message = "Stage {} Submitted".format(stage_info.stageId())
         data = {"attemptId": stage_info.attemptId(), "name": stage_info.name()}
-        sentry_sdk.add_breadcrumb(level="info", message=message, data=data)
+        self._add_breadcrumb(level="info", message=message, data=data)
         _set_app_properties()
 
     def onStageCompleted(self, stageCompleted):  # noqa: N802,N803
@@ -255,4 +282,4 @@ def onStageCompleted(self, stageCompleted):  # noqa: N802,N803
             message = "Stage {} Completed".format(stage_info.stageId())
             level = "info"
 
-        sentry_sdk.add_breadcrumb(level=level, message=message, data=data)
+        self._add_breadcrumb(level=level, message=message, data=data)

tests/integrations/spark/test_spark.py

@@ -1,6 +1,7 @@
 import pytest
 import sys
 from unittest.mock import patch
+
 from sentry_sdk.integrations.spark.spark_driver import (
     _set_app_properties,
     _start_sentry_listener,
@@ -18,8 +19,22 @@
 ################
 
 
-def test_set_app_properties():
-    spark_context = SparkContext(appName="Testing123")
+@pytest.fixture(scope="function")
+def sentry_init_with_reset(sentry_init):
+    from sentry_sdk.integrations import _processed_integrations
+
+    yield lambda: sentry_init(integrations=[SparkIntegration()])
+    _processed_integrations.remove("spark")
+
+
+@pytest.fixture(scope="function")
+def create_spark_context():
+    yield lambda: SparkContext(appName="Testing123")
+    SparkContext._active_spark_context.stop()
+
+
+def test_set_app_properties(create_spark_context):
+    spark_context = create_spark_context()
     _set_app_properties()
 
     assert spark_context.getLocalProperty("sentry_app_name") == "Testing123"
@@ -30,9 +45,8 @@ def test_set_app_properties():
     )
 
 
-def test_start_sentry_listener():
-    spark_context = SparkContext.getOrCreate()
-
+def test_start_sentry_listener(create_spark_context):
+    spark_context = create_spark_context()
     gateway = spark_context._gateway
     assert gateway._callback_server is None
 
@@ -41,9 +55,28 @@ def test_start_sentry_listener():
     assert gateway._callback_server is not None
 
 
-def test_initialize_spark_integration(sentry_init):
-    sentry_init(integrations=[SparkIntegration()])
-    SparkContext.getOrCreate()
+@patch("sentry_sdk.integrations.spark.spark_driver._patch_spark_context_init")
+def test_initialize_spark_integration_before_spark_context_init(
+    mock_patch_spark_context_init,
+    sentry_init_with_reset,
+    create_spark_context,
+):
+    sentry_init_with_reset()
+    create_spark_context()
+
+    mock_patch_spark_context_init.assert_called_once()
+
+
+@patch("sentry_sdk.integrations.spark.spark_driver._activate_integration")
+def test_initialize_spark_integration_after_spark_context_init(
+    mock_activate_integration,
+    create_spark_context,
+    sentry_init_with_reset,
+):
+    create_spark_context()
+    sentry_init_with_reset()
+
+    mock_activate_integration.assert_called_once()
 
 
 @pytest.fixture