src/lease.bend

bend-kube-runtime@0.2.0.0 · 124 lines · 20 definitions

1import Base
2import bend-kube-runtime@0.2.0.0/src/json.bend as J
3import bend-kube-runtime@0.2.0.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)))