Coverage for qreward / utils / scheduler / overload.py: 100.00%

1"""Overload detection for identifying server-side overload 

2conditions.""" 

3 

4from http import HTTPStatus (empty)

5from typing import Set (empty)

6 

7from qreward.globals import ( (empty)

8 LIBRARY_OVERLOAD_EXCEPTIONS_MAPPING, 

9 OVERLOAD_EXCEPTIONS, 

10 OVERLOAD_KEYWORDS, 

11 SYSTEM_OVERLOAD_INDICATORS, 

12) 

13 

14 

15MAX_EXCEPTION_CHAIN_DEPTH = 50 (empty)

16 

17 

18class OverloadChecker: (empty)

19 """Checker for detecting server-side overload conditions from 

20 exceptions.""" 

21 

22 @staticmethod (empty)

23 def check(exception: BaseException) -> bool: (empty)

24 """Check if exception indicates server overload. 

25 

26 Iteratively walks the exception chain (via __cause__ and 

27 __context__) 

28 up to MAX_EXCEPTION_CHAIN_DEPTH levels to detect overload signals. 

29 

30 Args: 

31 exception: Exception to check 

32 

33 Returns: 

34 True if server is overloaded 

35 """ 

36 pending = [exception] (empty)

37 visited: Set[int] = set() (empty)

38 depth = 0 (empty)

39 

40 while pending and depth < MAX_EXCEPTION_CHAIN_DEPTH: (empty)

41 current = pending.pop(0) (empty)

42 current_id = id(current) (empty)

43 

44 if current_id in visited: (empty)

45 continue (empty)

46 visited.add(current_id) (empty)

47 depth += 1 (empty)

48 

49 if OverloadChecker._check_single(current): (empty)

50 return True (empty)

51 

52 for attr_name in ("__cause__", "__context__"): (empty)

53 chained = getattr(current, attr_name, None) (empty)

54 if chained is not None: (empty)

55 pending.append(chained) (empty)

56 

57 return False (empty)

58 

59 @staticmethod (empty)

60 def _check_single(exception: BaseException) -> bool: (empty)

61 """Check a single exception (without following chains) for 

62 overload signals. 

63 

64 Args: 

65 exception: Exception to check 

66 

67 Returns: 

68 True if this specific exception indicates server overload 

69 """ 

70 # 1. HTTP status code check 

71 if hasattr(exception, "status_code"): (empty)

72 status_code = exception.status_code (empty)

73 if status_code in ( (empty)

74 HTTPStatus.SERVICE_UNAVAILABLE.value, 

75 HTTPStatus.TOO_MANY_REQUESTS.value, 

76 HTTPStatus.BAD_GATEWAY.value, 

77 HTTPStatus.GATEWAY_TIMEOUT.value, 

78 ): 

79 return True (empty)

80 

81 # 2. Exception type check (full module path) 

82 exception_type_full = ( (empty)

83 f"{type(exception).__module__}.{type(exception).__name__}" 

84 ) 

85 exception_type_name = type(exception).__name__ (empty)

86 if ( (empty)

87 exception_type_full in OVERLOAD_EXCEPTIONS 

88 or exception_type_name in OVERLOAD_EXCEPTIONS 

89 ): 

90 return True (empty)

91 

92 # 3. Library-specific exception handling 

93 for lib_name, exceptions in ( (empty)

94 LIBRARY_OVERLOAD_EXCEPTIONS_MAPPING.items() 

95 ): 

96 if lib_name in exception_type_full and any( (empty)

97 exc in exception_type_name for exc in exceptions 

98 ): 

99 return True (empty)

100 

101 # 4. Exception message content check 

102 error_message = str(exception).lower() (empty)

103 for keyword in OVERLOAD_KEYWORDS: (empty)

104 if keyword in error_message: (empty)

105 return True (empty)

106 

107 # 5. System-level check 

108 for indicator in SYSTEM_OVERLOAD_INDICATORS: (empty)

109 if indicator in error_message: (empty)

110 return True (empty)

111 

112 # 6. Check exception args 

113 if hasattr(exception, "args") and exception.args: (empty)

114 if any( (empty)

115 isinstance(arg, str) 

116 and any( 

117 keyword in arg.lower() for keyword in OVERLOAD_KEYWORDS 

118 ) 

119 for arg in exception.args 

120 ): 

121 return True (empty)

122 

123 return False (empty)