Coverage for gco / services / leader_lease.py: 100.00%

1"""Single-writer election on a pre-created Kubernetes Lease. 

2 

3The platform services run with two or more replicas for availability, but 

4several of their side effects must happen exactly once per cluster — the 

5ALB-hostname write to SSM, and every webhook delivery. Both elect a single 

6writer through a ``coordination.k8s.io`` Lease with the same rules: 

7 

8* The Lease is **pre-created by the manifests** (``02-rbac.yaml``), so RBAC 

9 grants ``get``/``update`` on one named object instead of ``create`` on every 

10 Lease in the namespace. A missing Lease disables the side effect rather than 

11 letting two replicas perform it. 

12* Acquisition is a read followed by ``replace`` carrying the read's 

13 ``resourceVersion``; Kubernetes rejects a racing writer with HTTP 409, and 

14 the loser simply does not act this cycle. 

15* A holder whose ``renewTime`` is older than ``leaseDurationSeconds`` — or that 

16 never recorded one — is treated as gone and may be replaced. 

17* Every API or RBAC failure returns ``False``. Losing the side effect for one 

18 cycle is always safer than performing it twice. 

19 

20Callers re-run :func:`try_acquire_lease` on every cycle; a ``True`` result is 

21a renewal for the current holder and an acquisition for a new one. 

22""" 

23 

24from __future__ import annotations 

25 

26import logging 

27from dataclasses import dataclass 

28from datetime import UTC, datetime 

29from typing import Any 

30 

31from kubernetes.client.rest import ApiException 

32 

33logger = logging.getLogger(__name__) 

34 

35#: Shortest lease any caller may configure. Below this a slow API call or a 

36#: paused process could let a second replica win the Lease while the first 

37#: still believes it holds it. 

38LEASE_MIN_DURATION_SECONDS = 60 

39 

40#: (connect, read) timeouts for the two Lease calls, kept short so an 

41#: unreachable API server costs one cycle, not the whole loop. 

42LEASE_REQUEST_TIMEOUT: tuple[int, int] = (3, 10) 

43 

44 

45@dataclass(frozen=True) 

46class LeaseIdentity: 

47 """Which Lease to hold, as whom, and for how long.""" 

48 

49 name: str 

50 namespace: str 

51 holder: str 

52 duration_seconds: int 

53 

54 

55def lease_duration_from_env(value: str | None, *, label: str) -> int: 

56 """Parse a configured lease duration, enforcing the shared floor.""" 

57 configured = int(value) if value is not None else LEASE_MIN_DURATION_SECONDS 

58 if configured < LEASE_MIN_DURATION_SECONDS: 

59 logger.warning( 

60 "%s=%s is too short; enforcing %s seconds", 

61 label, 

62 configured, 

63 LEASE_MIN_DURATION_SECONDS, 

64 ) 

65 return LEASE_MIN_DURATION_SECONDS 

66 return configured 

67 

68 

69def try_acquire_lease( 

70 coordination_v1: Any, 

71 identity: LeaseIdentity, 

72 *, 

73 label: str, 

74 request_timeout: tuple[int, int] = LEASE_REQUEST_TIMEOUT, 

75) -> bool: 

76 """Acquire or renew ``identity`` for its holder; ``False`` means do not act. 

77 

78 ``label`` names the elected duty in log lines (for example ``"ALB-sync"`` 

79 or ``"webhook"``) so two elections in one process stay distinguishable. 

80 """ 

81 observed_at = datetime.now(UTC) 

82 

83 try: 

84 lease = coordination_v1.read_namespaced_lease( 

85 identity.name, 

86 identity.namespace, 

87 _request_timeout=request_timeout, 

88 ) 

89 spec = lease.spec 

90 current_holder = spec.holder_identity 

91 renew_time = spec.renew_time 

92 lease_duration = spec.lease_duration_seconds or identity.duration_seconds 

93 

94 expired = False 

95 if current_holder: 

96 if renew_time is None: 

97 # A holder without a renewal timestamp cannot prove it still 

98 # owns the lease. Treat it as expired so the named Lease 

99 # cannot remain wedged indefinitely. 

100 expired = True 

101 else: 

102 if renew_time.tzinfo is None: 

103 renew_time = renew_time.replace(tzinfo=UTC) 

104 expired = (observed_at - renew_time).total_seconds() >= lease_duration 

105 

106 if current_holder not in (None, "", identity.holder) and not expired: 

107 return False 

108 

109 acquiring = current_holder != identity.holder 

110 renewed_at = datetime.now(UTC) 

111 if acquiring: 

112 spec.holder_identity = identity.holder 

113 spec.acquire_time = renewed_at 

114 spec.lease_transitions = (spec.lease_transitions or 0) + 1 

115 spec.lease_duration_seconds = identity.duration_seconds 

116 spec.renew_time = renewed_at 

117 

118 try: 

119 coordination_v1.replace_namespaced_lease( 

120 identity.name, 

121 identity.namespace, 

122 lease, 

123 _request_timeout=request_timeout, 

124 ) 

125 except ApiException as exc: 

126 if exc.status == 409: 

127 logger.debug("Lost %s Lease race to another replica", label) 

128 return False 

129 raise 

130 

131 if acquiring: 

132 logger.info("Acquired %s leader Lease as %s", label, identity.holder) 

133 return True 

134 

135 except ApiException as exc: 

136 if exc.status == 404: 

137 logger.warning( 

138 "%s Lease %s/%s is missing; the elected duty is disabled until it is restored", 

139 label, 

140 identity.namespace, 

141 identity.name, 

142 ) 

143 else: 

144 logger.warning("%s Lease check failed (non-fatal): %s", label, exc) 

145 return False 

146 except Exception as exc: 

147 logger.warning("%s Lease check failed (non-fatal): %s", label, exc) 

148 return False