Implement thread safety during instantiation of TelemetryService (#1729)

* Implement thread safety during instantiation on TelemetryService

* Code and test cleanup

* Edited DESCRIPTION.md

* Update DESCRIPTION.md

Take Adam's suggestion

Co-authored-by: Adam Ling <[email protected]>

---------

Co-authored-by: Adam Ling <[email protected]>

Author: sfc-gh-yqiu

DESCRIPTION.md

@@ -8,6 +8,10 @@ Source code is also available at: https://github.com/snowflakedb/snowflake-conne
 
 # Release Notes
 
+- v3.2.1(TBD)
+
+  - Added thread safety in telemetry when instantiating multiple connections concurrently.
+
 - v3.2.0(September 06,2023)
 
   - Made the ``parser`` -> ``manager`` renaming more consistent in ``snowflake.connector.config_manager`` module.

src/snowflake/connector/telemetry_oob.py

@@ -11,6 +11,7 @@
 import uuid
 from collections import namedtuple
 from queue import Queue
+from threading import Lock
 from typing import Any
 
 from .compat import OK
@@ -152,13 +153,16 @@ def get_type(self) -> str:
 
 class TelemetryService:
     __instance = None
+    # prevents race condition from multiple threads creating Snowflake connections
+    __lock_init = Lock()
 
-    @staticmethod
-    def get_instance() -> TelemetryService:
+    @classmethod
+    def get_instance(cls) -> TelemetryService:
         """Static access method."""
-        if TelemetryService.__instance is None:
-            TelemetryService()
-        return TelemetryService.__instance
+        with cls.__lock_init:
+            if cls.__instance is None:
+                cls()
+        return cls.__instance
 
     def __init__(self) -> None:
         """Virtually private constructor."""

test/unit/test_telemetry_oob.py

@@ -6,6 +6,8 @@
 from __future__ import annotations
 
 import logging
+import time
+from concurrent.futures import ThreadPoolExecutor
 
 import pytest
 
@@ -26,6 +28,8 @@
     "password": "ShouldNotShowUp",
     "protocol": "http",
 }
+TEST_RACE_CONDITION_THREAD_COUNT = 2
+TEST_RACE_CONDITION_DELAY_SECONDS = 1
 telemetry_data = {}
 exception = RevocationCheckError("Test OCSP Revocation error")
 event_type = "Test OCSP Exception"
@@ -198,3 +202,25 @@ def test_generate_telemetry_with_driver_info():
         snowflake.connector.telemetry.TelemetryField.KEY_OOB_VERSION.value: "1.2.3",
         "key": "value",
     }
+
+
+class MockTelemetryService(TelemetryService):
+    """Mocks a delay in the __init__ of TelemetryService to simulate a race condition"""
+
+    def __init__(self, *args, **kwargs):
+        # this delay all but guarantees enough time to catch multiple threads entering __init__
+        time.sleep(TEST_RACE_CONDITION_DELAY_SECONDS)
+        super().__init__(*args, **kwargs)
+
+
+def test_get_instance_multithreaded():
+    """Tests thread safety of multithreaded calls to TelemetryService.get_instance()"""
+    TelemetryService._TelemetryService__instance = None
+    with ThreadPoolExecutor() as executor:
+        futures = [
+            executor.submit(MockTelemetryService.get_instance)
+            for _ in range(TEST_RACE_CONDITION_THREAD_COUNT)
+        ]
+        for future in futures:
+            # will error if singleton constraint violated
+            future.result()