src/leader.bend

bend-kube-runtime@0.2.0.0 · 171 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
4import bend-kube-runtime@0.2.0.0/src/config.bend as K
5import bend-kube-runtime@0.2.0.0/src/client.bend as Cl
6import bend-kube-runtime@0.2.0.0/src/time.bend as T
7import bend-kube-runtime@0.2.0.0/src/sys.bend as Y
8import bend-kube-runtime@0.2.0.0/src/log.bend as L
9import bend-kube-runtime@0.2.0.0/src/lease.bend as E
10
11# Leader
12# ======
13#
14# client-go's leaderelection loop over lease.bend's decisions, with its
15# defaults: a 15s lease, renewed every 2s, given up when no renewal lands
16# for 10s. Losing the lease exits the process (status 1), as client-go's
17# usual OnStoppedLeading does, so two leaders never act at once; SIGTERM
18# hands the lease back (status 0).
19
20type LOpts is Data:
21  LOpts{ns: String, name: String, id: String, dur: U32, deadline: U32, retry: U32}
22
23def LOpts.default(ns: String, name: String, id: String) -> LOpts:
24  LOpts{ns, name, id, 15, 10000, 2000}
25
26def LOpts.api(o: LOpts) -> C.Api:
27  match o:
28    case LOpts{ns, n, i, d, dl, r}:
29      C.Api{C.Res.leases(), C.InNs{ns}}
30
31def LOpts.name(o: LOpts) -> String:
32  match o:
33    case LOpts{ns, n, i, d, dl, r}:
34      n
35
36def LOpts.retry(o: LOpts) -> U32:
37  match o:
38    case LOpts{ns, n, i, d, dl, r}:
39      r
40
41def LOpts.deadline(o: LOpts) -> U32:
42  match o:
43    case LOpts{ns, n, i, d, dl, r}:
44      dl
45
46def Leader.read.err(e: Cl.KErr, missing: Bool) -> Result<&2, &2, Cl.KErr, Maybe<&2, J.Json>>:
47  match missing:
48    case True{}:
49      Done{None{}}
50    case False{}:
51      Fail{e}
52
53# The Lease read: Some, None on 404, or the error.
54def Leader.read(r: Result<&2, &2, Cl.KErr, J.Json>) -> Result<&2, &2, Cl.KErr, Maybe<&2, J.Json>>:
55  match r:
56    case Done{l}:
57      Done{Some{l}}
58    case Fail{+e}:
59      Leader.read.err(e, U32.is_eq(Cl.KErr.code(e), 404))
60
61def Leader.decide(r: Result<&2, &2, Cl.KErr, Maybe<&2, J.Json>>, +o: LOpts, +obs: E.Obs,
62  +w: T.Wall) -> Result<&2, &2, Cl.KErr, E.Decision>:
63  match r o:
64    case Done{cur} LOpts{ns, n, id, d, dl, rt}:
65      Done{E.Lease.decide(cur, ns, n, id, obs, T.Wall.ms(w), d, T.Wall.micro(w))}
66    case Fail{e} _:
67      Fail{e}
68
69# What the loop just saw.
70type LEv is Type:
71  LTry{}
72  LRead{w: T.Wall, r: Result<&2, &2, Cl.KErr, J.Json>}
73  LDecided{now: U32, d: Result<&2, &2, Cl.KErr, E.Decision>}
74  LWrote{now: U32, ok: Bool, late: Bool}
75  LFail{stop: Bool}
76
77def Leader.wrote(r: Result<&2, &2, Cl.KErr, J.Json>) -> Bool:
78  Result.is_done(&2, &2, Cl.KErr, J.Json, r)
79
80def Leader.late(+now: U32, +last: U32, +deadline: U32) -> Bool:
81  U32.is_gt((now - last : U32), deadline)
82
83# On first acquiring: log and start the work.
84def Leader.started(leading: Bool, start: Maybe<&1, IO(Unit)>) -> IO(Unit):
85  match leading start:
86    case False{} Some{act}:
87      IO.bind(Unit, Unit, L.Log.info("became leader", Nil{}), u => IO.spawn(Unit, act))
88    case _ _:
89      IO.pure(Unit, Unit{})
90
91# A leader that cannot renew within the deadline stops at once (a
92# candidate that fails just tries again).
93def Leader.lost() -> IO(Unit):
94  IO.bind(Unit, Unit, L.Log.error("leader lease lost; exiting", Nil{}), u => IO.exit(1))
95
96# The election loop. start runs once, when the lease is first ours.
97def Leader.run(fuel: Nat, ev: LEv, +cfg: K.Config, +o: LOpts, +obs: E.Obs, +leading: Bool,
98  +last: U32, start: Maybe<&1, IO(Unit)>) -> IO(Unit):
99  match fuel ev:
100    case 0n e:
101      IO.pure(Unit, Unit{})
102    case 1n+f LTry{}:
103      IO.bind(U32 & (U32 & U32), Unit, Y.Sys.wall(),
104        w => IO.bind(Result<&2, &2, Cl.KErr, J.Json>, Unit,
105          Cl.Client.get(cfg, LOpts.api(o), LOpts.name(o)),
106          r => Leader.run(f, LRead{T.Wall.of(w), r}, cfg, o, obs, leading, last, start)))
107    case 1n+f LRead{+w, r}:
108      Leader.run(f, LDecided{T.Wall.ms(w), Leader.decide(Leader.read(r), o, obs, w)}, cfg, o, obs,
109        leading, last, start)
110    case 1n+f LDecided{+now, Done{E.Decision{E.LCreate{l}, ob}}}:
111      IO.bind(Result<&2, &2, Cl.KErr, J.Json>, Unit, Cl.Client.create(cfg, LOpts.api(o), l),
112        r => Leader.run(f, LWrote{now, Leader.wrote(r), Leader.late(now, last, LOpts.deadline(o))},
113          cfg, o, ob, leading, last, start))
114    case 1n+f LDecided{+now, Done{E.Decision{E.LUpdate{l}, ob}}}:
115      IO.bind(Result<&2, &2, Cl.KErr, J.Json>, Unit,
116        Cl.Client.replace(cfg, LOpts.api(o), LOpts.name(o), l),
117        r => Leader.run(f, LWrote{now, Leader.wrote(r), Leader.late(now, last, LOpts.deadline(o))},
118          cfg, o, ob, leading, last, start))
119    case 1n+f LDecided{+now, Done{E.Decision{E.LWait{}, ob}}}:
120      Leader.run(f, LWrote{now, False{}, Leader.late(now, last, LOpts.deadline(o))}, cfg, o, ob,
121        leading, last, start)
122    case 1n+f LDecided{+now, Fail{e}}:
123      IO.bind(Unit, Unit, L.Log.warn("lease read failed", [L.Log.kv("error", Cl.KErr.show(e))]),
124        u => Leader.run(f, LWrote{now, False{}, Leader.late(now, last, LOpts.deadline(o))}, cfg, o,
125          obs, leading, last, start))
126    case 1n+f LWrote{now, True{}, late}:
127      IO.bind(Unit, Unit, Leader.started(leading, start),
128        u => IO.bind(Unit, Unit, IO.sleep(LOpts.retry(o)),
129          u2 => Leader.run(f, LTry{}, cfg, o, obs, True{}, now, None{})))
130    case 1n+f LWrote{now, False{}, late}:
131      Leader.run(f, LFail{Bool.and(leading, late)}, cfg, o, obs, leading, last, start)
132    case 1n+f LFail{True{}}:
133      Leader.lost()
134    case 1n+f LFail{False{}}:
135      IO.bind(Unit, Unit, IO.sleep(LOpts.retry(o)),
136        u => Leader.run(f, LTry{}, cfg, o, obs, leading, last, start))
137
138def Leader.release.mine(+cfg: K.Config, +o: LOpts, l: J.Json, mine: Bool) -> IO(Unit):
139  match mine:
140    case True{}:
141      IO.bind(Result<&2, &2, Cl.KErr, J.Json>, Unit,
142        Cl.Client.replace(cfg, LOpts.api(o), LOpts.name(o), l),
143        r => L.Log.info("released leader lease", Nil{}))
144    case False{}:
145      IO.pure(Unit, Unit{})
146
147# Hands the lease back if it is ours.
148def Leader.release.got(r: Result<&2, &2, Cl.KErr, J.Json>, +cfg: K.Config, +o: LOpts,
149  w: T.Wall) -> IO(Unit):
150  match r o:
151    case Done{+l} LOpts{ns, n, +id, d, dl, rt}:
152      Leader.release.mine(cfg, LOpts{ns, n, id, d, dl, rt}, E.Lease.release(l, T.Wall.micro(w)),
153        String.eq(E.Lease.holder(l), id))
154    case _ _:
155      IO.pure(Unit, Unit{})
156
157def Leader.release(+cfg: K.Config, +o: LOpts) -> IO(Unit):
158  IO.bind(U32 & (U32 & U32), Unit, Y.Sys.wall(),
159    w => IO.bind(Result<&2, &2, Cl.KErr, J.Json>, Unit,
160      Cl.Client.get(cfg, LOpts.api(o), LOpts.name(o)), r => Leader.release.got(r, cfg, o, T.Wall.of(w))))
161
162# On SIGTERM or SIGINT: release the lease, then exit 0.
163def Leader.on_signal(+cfg: K.Config, +o: LOpts) -> IO(Unit):
164  IO.bind(U32, Unit, Y.Sys.signal(),
165    sig => IO.bind(Unit, Unit, L.Log.info("terminating", [L.Log.kv("signal", U32.show(sig))]),
166      u => IO.bind(Unit, Unit, Leader.release(cfg, o), u2 => IO.exit(0))))
167
168# Runs work only while we hold the lease (client-go's RunOrDie).
169def Leader.elect(+cfg: K.Config, +o: LOpts, work: IO(Unit)) -> IO(Unit):
170  IO.bind(Unit, Unit, IO.spawn(Unit, Leader.on_signal(cfg, o)),
171    u => Leader.run(4294967295n, LTry{}, cfg, o, E.Obs.none(), False{}, 0, Some{work}))