put some backoff in loadgenerator when errors are created for demo

Author: greatestusername

src/shop-dc-shim/load-generator/shop_load_generator.py

@@ -14,6 +14,15 @@
 
 The generator creates realistic purchase patterns that would originate
 from datacenter-deployed point-of-sale systems calling cloud checkout.
+
+ERROR HANDLING & BACKOFF LOGIC:
+- Purchase requests are NEVER affected by error handling
+- Transaction status polling implements circuit breaker pattern
+- 500 errors on status checks trigger exponential backoff (1.5x multiplier)
+- After 10 consecutive status check failures, circuit breaker opens for 10 minutes
+- Purchase flow continues normally even when status checks are failing
+- This prevents overwhelming the service with failed status check requests
+- Designed to keep demo errors visible while preventing system overload
 """
 
 import json
@@ -64,6 +73,16 @@ def __init__(self, shop_service_url: str = "http://localhost:8070"):
         self.shop_service_url = shop_service_url
         self.session = self._create_session()
 
+        # Status check error tracking and backoff logic
+        self.status_check_failures = 0
+        self.consecutive_failures = 0
+        self.last_failure_time = 0
+        self.status_check_enabled = True
+        self.backoff_delay = 1  # Initial backoff delay in seconds
+        self.max_backoff_delay = 300  # Maximum backoff delay (5 minutes)
+        self.circuit_breaker_threshold = 10  # Number of consecutive failures before backing off more
+        self.circuit_breaker_reset_time = 600  # Time to wait before resetting circuit breaker (10 minutes)
+        
         # Store locations (simulating datacenter-deployed stores)
         self.stores = [
             StoreLocation("DC-NYC-01", "Manhattan Flagship", "New York", "NY", 
@@ -207,23 +226,80 @@ def submit_purchase(self, purchase_request: Dict[str, Any]) -> Dict[str, Any]:
             return {"success": False, "error": str(e)}
 
     def check_transaction_status(self, transaction_id: str) -> Dict[str, Any]:
-        """Check the status of a transaction"""
+        """Check the status of a transaction with backoff logic for 500 errors"""
+        # Check if circuit breaker is open and should be reset
+        current_time = time.time()
+        if not self.status_check_enabled and (current_time - self.last_failure_time) > self.circuit_breaker_reset_time:
+            logger.info("Resetting status check circuit breaker - attempting to resume normal operation")
+            self.status_check_enabled = True
+            self.consecutive_failures = 0
+            self.backoff_delay = 1
+        
+        # If circuit breaker is open, return immediately without making request
+        if not self.status_check_enabled:
+            time_until_reset = self.circuit_breaker_reset_time - (current_time - self.last_failure_time)
+            return {
+                "success": False, 
+                "error": f"Circuit breaker open - status checks disabled for {time_until_reset:.0f} more seconds",
+                "circuit_breaker_open": True
+            }
+        
+        # Apply current backoff delay before making request
+        if self.backoff_delay > 1:
+            logger.debug(f"Applying backoff delay of {self.backoff_delay:.1f}s before status check")
+            time.sleep(self.backoff_delay)
+        
         try:
             response = self.session.get(
                 f"{self.shop_service_url}/api/shop/transaction/{transaction_id}",
                 timeout=10
             )
 
             if response.status_code == 200:
+                # Success - reset failure counters
+                self.consecutive_failures = 0
+                self.backoff_delay = 1
                 return {"success": True, "data": response.json()}
+            elif response.status_code == 500:
+                # Handle 500 errors with backoff logic (but still allow them for demo)
+                self._handle_status_check_failure("500 Internal Server Error - Jackson serialization issue (demo)")
+                return {"success": False, "error": f"HTTP {response.status_code} (demo error)", "server_error": True}
+            elif response.status_code == 404:
+                # 404 is expected for new transactions, don't count as failure
+                return {"success": False, "error": f"HTTP {response.status_code} - Transaction not found"}
             else:
+                # Other error codes
+                self._handle_status_check_failure(f"HTTP {response.status_code}")
                 return {"success": False, "error": f"HTTP {response.status_code}"}
 
         except requests.exceptions.RequestException as e:
+            self._handle_status_check_failure(f"Request exception: {str(e)}")
             return {"success": False, "error": str(e)}
 
+    def _handle_status_check_failure(self, error_msg: str):
+        """Handle status check failures with exponential backoff and circuit breaker logic"""
+        self.status_check_failures += 1
+        self.consecutive_failures += 1
+        self.last_failure_time = time.time()
+        
+        # Implement exponential backoff
+        self.backoff_delay = min(self.backoff_delay * 1.5, self.max_backoff_delay)
+        
+        if self.consecutive_failures >= self.circuit_breaker_threshold:
+            self.status_check_enabled = False
+            logger.warning(f"Status check circuit breaker opened after {self.consecutive_failures} consecutive failures. "
+                         f"Last error: {error_msg}. Status checks disabled for {self.circuit_breaker_reset_time} seconds.")
+        else:
+            logger.debug(f"Status check failure #{self.consecutive_failures}: {error_msg}. "
+                        f"Next request will have {self.backoff_delay:.1f}s backoff delay.")
+    
     def run_single_transaction(self) -> Dict[str, Any]:
-        """Execute a single transaction (for use with thread pool)"""
+        """Execute a single transaction (for use with thread pool)
+        
+        NOTE: Purchase requests are NEVER affected by status check backoff logic.
+        Only transaction status polling is subject to circuit breaker protection.
+        This ensures purchases continue to flow even when status checks are failing.
+        """
         store = random.choice(self.stores)
         terminal = random.choice(store.terminals)
 
@@ -288,12 +364,20 @@ def run_continuous_load(self, transactions_per_minute: int = 10, duration_minute
                 if transaction_count % transactions_per_minute == 0:
                     elapsed_minutes = (time.time() - start_time) / 60
                     success_rate = (success_count / transaction_count) * 100 if transaction_count > 0 else 0
+                    stats = self.get_status_check_stats()
+                    
+                    status_info = ""
+                    if not stats["circuit_breaker_enabled"]:
+                        status_info = f" | Status checks: CIRCUIT BREAKER OPEN ({stats['consecutive_failures']} failures)"
+                    elif stats["total_failures"] > 0:
+                        status_info = f" | Status checks: {stats['total_failures']} total failures, {stats['backoff_delay']:.1f}s backoff"
+                    
                     if run_forever:
                         logger.info(f"Progress: {transaction_count} transactions in {elapsed_minutes:.1f} minutes "
-                                f"(Success rate: {success_rate:.1f}%) - Running indefinitely")
+                                f"(Success rate: {success_rate:.1f}%) - Running indefinitely{status_info}")
                     else:
                         logger.info(f"Progress: {transaction_count} transactions in {elapsed_minutes:.1f} minutes "
-                                f"(Success rate: {success_rate:.1f}%)")
+                                f"(Success rate: {success_rate:.1f}%){status_info}")
 
                 # Sleep to maintain target rate
                 elapsed = time.time() - loop_start
@@ -320,17 +404,27 @@ def run_continuous_load(self, transactions_per_minute: int = 10, duration_minute
             return
 
         final_success_rate = (success_count / transaction_count) * 100
+        stats = self.get_status_check_stats()
 
         logger.info(f"Load generation completed:")
         logger.info(f"  Total transactions: {transaction_count}")
         logger.info(f"  Successful submissions: {success_count}")
         logger.info(f"  Success rate: {final_success_rate:.1f}%")
         logger.info(f"  Total time: {total_time:.1f} seconds")
         logger.info(f"  Average TPM: {(transaction_count / (total_time / 60)):.1f}")
+        logger.info(f"Status check circuit breaker stats:")
+        logger.info(f"  Total status check failures: {stats['total_failures']}")
+        logger.info(f"  Circuit breaker status: {'OPEN' if not stats['circuit_breaker_enabled'] else 'CLOSED'}")
+        logger.info(f"  Final backoff delay: {stats['backoff_delay']:.1f}s")
+        
+        if stats["total_failures"] > 0:
+            logger.info("NOTE: Status check failures are expected for this demo (Jackson LocalDateTime serialization errors)")
+            logger.info("Purchase requests were never affected by status check failures - they continued successfully!")
 
     def _check_pending_transactions(self, pending_transactions: List[Dict[str, Any]]):
-        """Check status of pending transactions"""
+        """Check status of pending transactions with backoff logic"""
         completed_indices = []
+        status_check_skipped = 0
 
         for i, txn in enumerate(pending_transactions):
             # Only check transactions that are at least 30 seconds old
@@ -339,6 +433,11 @@ def _check_pending_transactions(self, pending_transactions: List[Dict[str, Any]]
 
             status_result = self.check_transaction_status(txn["transaction_id"])
 
+            # Handle circuit breaker case
+            if status_result.get("circuit_breaker_open"):
+                status_check_skipped += 1
+                continue
+            
             if status_result["success"]:
                 data = status_result["data"]
                 status = data.get("status")
@@ -347,10 +446,33 @@ def _check_pending_transactions(self, pending_transactions: List[Dict[str, Any]]
                 if status in ["COMPLETED", "FAILED"]:
                     logger.info(f"Transaction {txn['transaction_id'][:8]}... from {store} -> {status}")
                     completed_indices.append(i)
+            elif status_result.get("server_error"):
+                # 500 errors are expected for demo - log but continue tracking
+                logger.debug(f"Transaction {txn['transaction_id'][:8]}... status check failed with expected demo error: {status_result['error']}")
+            else:
+                # Other errors (404, network issues, etc.)
+                logger.debug(f"Transaction {txn['transaction_id'][:8]}... status check failed: {status_result['error']}")
 
         # Remove completed transactions from tracking
         for i in reversed(completed_indices):
             pending_transactions.pop(i)
+        
+        # Log circuit breaker status if any checks were skipped
+        if status_check_skipped > 0:
+            logger.info(f"Skipped {status_check_skipped} status checks due to circuit breaker protection")
+            logger.info(f"Status check circuit breaker stats: {self.consecutive_failures} consecutive failures, "
+                       f"next backoff delay: {self.backoff_delay:.1f}s")
+    
+    def get_status_check_stats(self) -> Dict[str, Any]:
+        """Get current status check statistics for monitoring"""
+        return {
+            "total_failures": self.status_check_failures,
+            "consecutive_failures": self.consecutive_failures,
+            "backoff_delay": self.backoff_delay,
+            "circuit_breaker_enabled": self.status_check_enabled,
+            "last_failure_time": self.last_failure_time,
+            "time_since_last_failure": time.time() - self.last_failure_time if self.last_failure_time > 0 else 0
+        }
 
     def run_burst_load(self, concurrent_transactions: int = 20, total_transactions: int = 100):
         """Run burst load with concurrent transactions"""