src/lease.bend
bend-kube-runtime@0.1.1.0 · 124 lines · 0 definitions
1import Base
2import bend-kube-runtime@0.1.1.0/src/json.bend as J
3import bend-kube-runtime@0.1.1.0/src/core.bend as C
4
5# Lease
6# =====
7#
8# Leader election over a coordination.k8s.io/v1 Lease, deciding as
9# client-go's LeaderElector.tryAcquireOrRenew does:
10#
11# - No Lease: create one naming us.
12# - A Lease held by someone else whose record we saw change less than
13# leaseDuration ago (by our own clock: remote timestamps are never
14# compared, so clock skew cannot steal a lease): wait.
15# - Otherwise (ours, unheld, or expired): write it naming us, through the
16# resourceVersion the read returned, so of two contenders only one
17# write lands (optimistic concurrency).
18#
19# Taking a lease from another holder counts a transition and resets
20# acquireTime; renewing our own keeps both.
21
22# What we last saw of the record, and when (our clock, in ms).
23type Obs is Data:
24 Obs{holder: String, renew: String, at: U32}
25
26type LeaseAct is Data:
27 LCreate{lease: J.Json}
28 LUpdate{lease: J.Json}
29 LWait{}
30
31type Decision is Data:
32 Decision{act: LeaseAct, obs: Obs}
33
34def Obs.none() -> Obs:
35 Obs{"", "", 0}
36
37def Lease.spec(+l: J.Json, field: String) -> J.Json:
38 J.Json.get(J.Json.get(l, "spec"), field)
39
40def Lease.holder(+l: J.Json) -> String:
41 J.Json.str_or(Lease.spec(l, "holderIdentity"), "")
42
43def Lease.renew_time(+l: J.Json) -> String:
44 J.Json.str_or(Lease.spec(l, "renewTime"), "")
45
46def Lease.transitions(+l: J.Json) -> U32:
47 Maybe.default(&2, U32, J.Json.as_u32(Lease.spec(l, "leaseTransitions")), 0)
48
49def Lease.new_spec(id: String, +dur: U32, acquire: String, renew: String, +trans: U32) -> J.Json:
50 J.Json.obj([J.KV{"holderIdentity", J.JStr{id}}, J.KV{"leaseDurationSeconds", J.Json.u32(dur)},
51 J.KV{"acquireTime", J.JStr{acquire}}, J.KV{"renewTime", J.JStr{renew}},
52 J.KV{"leaseTransitions", J.Json.u32(trans)}])
53
54# A fresh Lease naming us.
55def Lease.fresh(ns: String, name: String, id: String, +dur: U32, +now: String) -> J.Json:
56 J.Json.obj([J.KV{"apiVersion", J.JStr{"coordination.k8s.io/v1"}}, J.KV{"kind", J.JStr{"Lease"}},
57 J.KV{"metadata", J.Json.obj([J.KV{"name", J.JStr{name}}, J.KV{"namespace", J.JStr{ns}}])},
58 J.KV{"spec", Lease.new_spec(id, dur, now, now, 0)}])
59
60def Lease.observe.go(h: String, r: String, o: Obs, +now: U32, same: Bool) -> Obs:
61 match same:
62 case True{}:
63 o
64 case False{}:
65 Obs{h, r, now}
66
67# The record seen now: the old observation if nothing changed, else a new
68# one stamped now.
69def Lease.observe(+l: J.Json, o: Obs, +now: U32) -> Obs:
70 match o:
71 case Obs{+h, +r, at}:
72 Lease.observe.go(Lease.holder(l), Lease.renew_time(l), Obs{h, r, at}, now,
73 Bool.and(String.eq(h, Lease.holder(l)), String.eq(r, Lease.renew_time(l))))
74
75# Whether another holder's lease is still live by our clock.
76def Lease.held(+l: J.Json, +me: String, o: Obs, +now: U32, +dur: U32) -> Bool:
77 match o:
78 case Obs{h, r, +at}:
79 Bool.and(Bool.not(String.is_empty(Lease.holder(l))),
80 Bool.and(Bool.not(String.eq(Lease.holder(l), me)),
81 U32.is_lt((now - at : U32), (dur * 1000 : U32))))
82
83def Lease.take.go(+l: J.Json, me: String, +dur: U32, +now: String, ours: Bool) -> J.Json:
84 match ours:
85 case True{}:
86 J.Json.set(l, "spec", Lease.new_spec(me, dur,
87 J.Json.str_or(Lease.spec(l, "acquireTime"), now), now, Lease.transitions(l)))
88 case False{}:
89 J.Json.set(l, "spec", Lease.new_spec(me, dur, now, now, (Lease.transitions(l) + 1 : U32)))
90
91# The Lease rewritten to name us, keeping its metadata (resourceVersion).
92def Lease.take(+l: J.Json, +me: String, +dur: U32, +now: String) -> J.Json:
93 Lease.take.go(l, me, dur, now, String.eq(Lease.holder(l), me))
94
95def Lease.decide.fin(+l: J.Json, +me: String, o: Obs, +dur: U32, +now: String, held: Bool) ->
96 Decision:
97 match held:
98 case True{}:
99 Decision{LWait{}, o}
100 case False{}:
101 Decision{LUpdate{Lease.take(l, me, dur, now)}, o}
102
103def Lease.decide.held(+l: J.Json, +me: String, +o: Obs, +now_ms: U32, +dur: U32, +now: String) ->
104 Decision:
105 Lease.decide.fin(l, me, o, dur, now, Lease.held(l, me, o, now_ms, dur))
106
107def Lease.decide.got(+l: J.Json, +me: String, o: Obs, +now_ms: U32, +dur: U32, +now: String) ->
108 Decision:
109 Lease.decide.held(l, me, Lease.observe(l, o, now_ms), now_ms, dur, now)
110
111# One round: the Lease read (None when absent), us, what we saw before,
112# our clock in ms, the lease duration in seconds, and the time to stamp.
113def Lease.decide(cur: Maybe<&2, J.Json>, ns: String, name: String, +me: String, o: Obs,
114 +now_ms: U32, +dur: U32, +now: String) -> Decision:
115 match cur:
116 case None{}:
117 Decision{LCreate{Lease.fresh(ns, name, me, dur, now)}, Obs{me, now, now_ms}}
118 case Some{+l}:
119 Lease.decide.got(l, me, o, now_ms, dur, now)
120
121# The Lease handed back: no holder, a one-second duration, so the next
122# candidate takes it at once (client-go's ReleaseOnCancel).
123def Lease.release(+l: J.Json, +now: String) -> J.Json:
124 J.Json.set(l, "spec", Lease.new_spec("", 1, now, now, Lease.transitions(l)))