simplified CB config

Author: nikhilsuri-db

src/databricks/sql/telemetry/circuit_breaker_manager.py

@@ -17,19 +17,15 @@
 logger = logging.getLogger(__name__)
 
 # Circuit Breaker Configuration Constants
-DEFAULT_FAILURE_THRESHOLD = 0.5
-DEFAULT_MINIMUM_CALLS = 20
-DEFAULT_TIMEOUT = 30
-DEFAULT_RESET_TIMEOUT = 30
-DEFAULT_EXPECTED_EXCEPTION = (Exception,)
-DEFAULT_NAME = "telemetry-circuit-breaker"
+MINIMUM_CALLS = 20
+RESET_TIMEOUT = 30
+CIRCUIT_BREAKER_NAME = "telemetry-circuit-breaker"
 
 # Circuit Breaker State Constants
 CIRCUIT_BREAKER_STATE_OPEN = "open"
 CIRCUIT_BREAKER_STATE_CLOSED = "closed"
 CIRCUIT_BREAKER_STATE_HALF_OPEN = "half-open"
 CIRCUIT_BREAKER_STATE_DISABLED = "disabled"
-CIRCUIT_BREAKER_STATE_NOT_INITIALIZED = "not_initialized"
 
 # Logging Message Constants
 LOG_CIRCUIT_BREAKER_STATE_CHANGED = "Circuit breaker state changed from %s to %s for %s"
@@ -76,56 +72,18 @@ def state_change(self, cb: CircuitBreaker, old_state, new_state) -> None:
             logger.info(LOG_CIRCUIT_BREAKER_HALF_OPEN, cb.name)
 
 
-@dataclass(frozen=True)
-class CircuitBreakerConfig:
-    """Configuration for circuit breaker behavior.
-
-    This class is immutable to prevent modification of circuit breaker settings.
-    All configuration values are set to constants defined at the module level.
-    """
-
-    # Failure threshold percentage (0.0 to 1.0)
-    failure_threshold: float = DEFAULT_FAILURE_THRESHOLD
-
-    # Minimum number of calls before circuit can open
-    minimum_calls: int = DEFAULT_MINIMUM_CALLS
-
-    # Time window for counting failures (in seconds)
-    timeout: int = DEFAULT_TIMEOUT
-
-    # Time to wait before trying to close circuit (in seconds)
-    reset_timeout: int = DEFAULT_RESET_TIMEOUT
-
-    # Expected exception types that should trigger circuit breaker
-    expected_exception: tuple = DEFAULT_EXPECTED_EXCEPTION
-
-    # Name for the circuit breaker (for logging)
-    name: str = DEFAULT_NAME
-
-
 class CircuitBreakerManager:
     """
     Manages circuit breaker instances for telemetry requests.
 
     This class provides a singleton pattern to manage circuit breaker instances
     per host, ensuring that telemetry failures don't impact main SQL operations.
+
+    Circuit breaker configuration is fixed and cannot be overridden.
     """
 
     _instances: Dict[str, CircuitBreaker] = {}
     _lock = threading.RLock()
-    _config: Optional[CircuitBreakerConfig] = None
-
-    @classmethod
-    def initialize(cls, config: CircuitBreakerConfig) -> None:
-        """
-        Initialize the circuit breaker manager with configuration.
-
-        Args:
-            config: Circuit breaker configuration
-        """
-        with cls._lock:
-            cls._config = config
-            logger.debug("CircuitBreakerManager initialized with config: %s", config)
 
     @classmethod
     def get_circuit_breaker(cls, host: str) -> CircuitBreaker:
@@ -138,10 +96,6 @@ def get_circuit_breaker(cls, host: str) -> CircuitBreaker:
         Returns:
             CircuitBreaker instance for the host
         """
-        if not cls._config:
-            # Return a no-op circuit breaker if not initialized
-            return cls._create_noop_circuit_breaker()
-
         with cls._lock:
             if host not in cls._instances:
                 cls._instances[host] = cls._create_circuit_breaker(host)
@@ -160,93 +114,16 @@ def _create_circuit_breaker(cls, host: str) -> CircuitBreaker:
         Returns:
             New CircuitBreaker instance
         """
-        config = cls._config
-        if config is None:
-            raise RuntimeError("CircuitBreakerManager not initialized")
-
-        # Create circuit breaker with configuration
+        # Create circuit breaker with fixed configuration
         breaker = CircuitBreaker(
-            fail_max=config.minimum_calls,  # Number of failures before circuit opens
-            reset_timeout=config.reset_timeout,
-            name=f"{config.name}-{host}",
+            fail_max=MINIMUM_CALLS,
+            reset_timeout=RESET_TIMEOUT,
+            name=f"{CIRCUIT_BREAKER_NAME}-{host}",
         )
-
-        # Add state change listeners for logging
         breaker.add_listener(CircuitBreakerStateListener())
 
         return breaker
 
-    @classmethod
-    def _create_noop_circuit_breaker(cls) -> CircuitBreaker:
-        """
-        Create a no-op circuit breaker that always allows calls.
-
-        Returns:
-            CircuitBreaker that never opens
-        """
-        # Create a circuit breaker with very high thresholds so it never opens
-        breaker = CircuitBreaker(
-            fail_max=1000000,  # Very high threshold
-            reset_timeout=1,  # Short reset time
-            name="noop-circuit-breaker",
-        )
-        return breaker
-
-    @classmethod
-    def get_circuit_breaker_state(cls, host: str) -> str:
-        """
-        Get the current state of the circuit breaker for a host.
-
-        Args:
-            host: The hostname
-
-        Returns:
-            Current state of the circuit breaker
-        """
-        if not cls._config:
-            return CIRCUIT_BREAKER_STATE_DISABLED
-
-        with cls._lock:
-            if host not in cls._instances:
-                return CIRCUIT_BREAKER_STATE_NOT_INITIALIZED
-
-            breaker = cls._instances[host]
-            return breaker.current_state
-
-    @classmethod
-    def reset_circuit_breaker(cls, host: str) -> None:
-        """
-        Reset the circuit breaker for a host to closed state.
-
-        Args:
-            host: The hostname
-        """
-        with cls._lock:
-            if host in cls._instances:
-                # pybreaker doesn't have a reset method, we need to recreate the breaker
-                del cls._instances[host]
-                logger.info("Reset circuit breaker for host: %s", host)
-
-    @classmethod
-    def clear_circuit_breaker(cls, host: str) -> None:
-        """
-        Remove the circuit breaker instance for a host.
-
-        Args:
-            host: The hostname
-        """
-        with cls._lock:
-            if host in cls._instances:
-                del cls._instances[host]
-                logger.debug("Cleared circuit breaker for host: %s", host)
-
-    @classmethod
-    def clear_all_circuit_breakers(cls) -> None:
-        """Clear all circuit breaker instances."""
-        with cls._lock:
-            cls._instances.clear()
-            logger.debug("Cleared all circuit breakers")
-
 
 def is_circuit_breaker_error(exception: Exception) -> bool:
     """

src/databricks/sql/telemetry/telemetry_client.py

@@ -47,7 +47,6 @@
     CircuitBreakerTelemetryPushClient,
 )
 from databricks.sql.telemetry.circuit_breaker_manager import (
-    CircuitBreakerConfig,
     is_circuit_breaker_error,
 )
 
@@ -200,34 +199,15 @@ def __init__(
 
         # Create telemetry push client based on circuit breaker enabled flag
         if client_context.telemetry_circuit_breaker_enabled:
-            # Create circuit breaker configuration from client context or use defaults
-            self._circuit_breaker_config = CircuitBreakerConfig(
-                failure_threshold=getattr(
-                    client_context, "telemetry_circuit_breaker_failure_threshold", 0.5
-                ),
-                minimum_calls=getattr(
-                    client_context, "telemetry_circuit_breaker_minimum_calls", 20
-                ),
-                timeout=getattr(
-                    client_context, "telemetry_circuit_breaker_timeout", 30
-                ),
-                reset_timeout=getattr(
-                    client_context, "telemetry_circuit_breaker_reset_timeout", 30
-                ),
-                name=f"telemetry-circuit-breaker-{session_id_hex}",
-            )
-
-            # Create circuit breaker telemetry push client
+            # Create circuit breaker telemetry push client with fixed configuration
             self._telemetry_push_client: ITelemetryPushClient = (
                 CircuitBreakerTelemetryPushClient(
                     TelemetryPushClient(self._http_client),
                     host_url,
-                    self._circuit_breaker_config,
                 )
             )
         else:
             # Circuit breaker disabled - use direct telemetry push client
-            self._circuit_breaker_config = None
             self._telemetry_push_client: ITelemetryPushClient = TelemetryPushClient(
                 self._http_client
             )
@@ -410,18 +390,6 @@ def close(self):
         logger.debug("Closing TelemetryClient for connection %s", self._session_id_hex)
         self._flush()
 
-    def get_circuit_breaker_state(self) -> str:
-        """Get the current state of the circuit breaker."""
-        return self._telemetry_push_client.get_circuit_breaker_state()
-
-    def is_circuit_breaker_open(self) -> bool:
-        """Check if the circuit breaker is currently open."""
-        return self._telemetry_push_client.is_circuit_breaker_open()
-
-    def reset_circuit_breaker(self) -> None:
-        """Reset the circuit breaker."""
-        self._telemetry_push_client.reset_circuit_breaker()
-
 
 class TelemetryClientFactory:
     """

src/databricks/sql/telemetry/telemetry_push_client.py

@@ -20,10 +20,8 @@
 from databricks.sql.common.unified_http_client import UnifiedHttpClient
 from databricks.sql.common.http import HttpMethod
 from databricks.sql.telemetry.circuit_breaker_manager import (
-    CircuitBreakerConfig,
     CircuitBreakerManager,
     is_circuit_breaker_error,
-    CIRCUIT_BREAKER_STATE_OPEN,
 )
 
 logger = logging.getLogger(__name__)
@@ -55,21 +53,6 @@ def request_context(
         """Context manager for making HTTP requests."""
         pass
 
-    @abstractmethod
-    def get_circuit_breaker_state(self) -> str:
-        """Get the current state of the circuit breaker."""
-        pass
-
-    @abstractmethod
-    def is_circuit_breaker_open(self) -> bool:
-        """Check if the circuit breaker is currently open."""
-        pass
-
-    @abstractmethod
-    def reset_circuit_breaker(self) -> None:
-        """Reset the circuit breaker to closed state."""
-        pass
-
 
 class TelemetryPushClient(ITelemetryPushClient):
     """Direct HTTP client implementation for telemetry requests."""
@@ -108,47 +91,27 @@ def request_context(
         ) as response:
             yield response
 
-    def get_circuit_breaker_state(self) -> str:
-        """Circuit breaker is not available in direct implementation."""
-        return "not_available"
-
-    def is_circuit_breaker_open(self) -> bool:
-        """Circuit breaker is not available in direct implementation."""
-        return False
-
-    def reset_circuit_breaker(self) -> None:
-        """Circuit breaker is not available in direct implementation."""
-        pass
-
 
 class CircuitBreakerTelemetryPushClient(ITelemetryPushClient):
     """Circuit breaker wrapper implementation for telemetry requests."""
 
-    def __init__(
-        self, delegate: ITelemetryPushClient, host: str, config: CircuitBreakerConfig
-    ):
+    def __init__(self, delegate: ITelemetryPushClient, host: str):
         """
         Initialize the circuit breaker telemetry push client.
 
         Args:
             delegate: The underlying telemetry push client to wrap
             host: The hostname for circuit breaker identification
-            config: Circuit breaker configuration
         """
         self._delegate = delegate
         self._host = host
-        self._config = config
 
-        # Initialize circuit breaker manager with config
-        CircuitBreakerManager.initialize(config)
-
-        # Get circuit breaker for this host
+        # Get circuit breaker for this host (creates if doesn't exist)
         self._circuit_breaker = CircuitBreakerManager.get_circuit_breaker(host)
 
         logger.debug(
-            "CircuitBreakerTelemetryPushClient initialized for host %s with config: %s",
+            "CircuitBreakerTelemetryPushClient initialized for host %s",
             host,
-            config,
         )
 
     def request(
@@ -208,15 +171,3 @@ def _make_request():
             # Re-raise non-circuit breaker exceptions
             logger.debug("Telemetry request failed for host %s: %s", self._host, e)
             raise
-
-    def get_circuit_breaker_state(self) -> str:
-        """Get the current state of the circuit breaker."""
-        return CircuitBreakerManager.get_circuit_breaker_state(self._host)
-
-    def is_circuit_breaker_open(self) -> bool:
-        """Check if the circuit breaker is currently open."""
-        return self.get_circuit_breaker_state() == CIRCUIT_BREAKER_STATE_OPEN
-
-    def reset_circuit_breaker(self) -> None:
-        """Reset the circuit breaker to closed state."""
-        CircuitBreakerManager.reset_circuit_breaker(self._host)