src/watcher.bend
bend-kube-runtime@0.1.1.0 · 193 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# Watcher
6# =======
7#
8# kube-runtime's watcher as a pure state machine, with client-go's
9# reflector deciding how the API behaves:
10#
11# - A watcher starts by listing in pages (limit + continue) and then watches
12# from the list's resourceVersion, with bookmarks, resuming from the last
13# version it saw whenever a watch ends (the server's timeoutSeconds).
14# - A 410 Gone (an expired version, on a watch or a continue token) relists
15# from scratch at once, as the reflector does; the store swaps in the new
16# list whole, so objects deleted meanwhile disappear.
17# - Other failures retry the same step after an exponential backoff
18# (800ms doubling to 30s, as both kube-rs and client-go default to);
19# success resets it.
20# - resourceVersion is opaque: it is only ever copied, never compared.
21#
22# The IO driver (runtime.bend) performs the action a state asks for, feeds
23# the outcome back as an input, and hands the outputs to the store.
24
25# What to do next: list (from a continue token, "" for the first page) or
26# watch from a version.
27type Phase is Data:
28 WList{cont: String}
29 WWatch{rv: String}
30
31# fails counts failures since the last success; first says the next page
32# starts a fresh list.
33type WState is Data:
34 WState{phase: Phase, fails: U32, first: Bool}
35
36type Input is Data:
37 IPage{page: C.Page}
38 IListErr{code: U32}
39 IEvent{ev: C.WEvent}
40 IWatchEnd{}
41 IWatchErr{code: U32}
42
43# What the store and controller see (kube-runtime's watcher::Event).
44type Out is Data:
45 OInit{}
46 OInitApply{obj: J.Json}
47 OInitDone{}
48 OApply{obj: J.Json}
49 ODelete{obj: J.Json}
50
51# A step's result: the next state, its outputs in order, and whether the
52# driver must back off before the next action.
53type Step is Data:
54 Step{state: WState, outs: List<&2, Out>, wait: Bool}
55
56def WState.init() -> WState:
57 WState{WList{""}, 0, True{}}
58
59def Backoff.delay.go(+fails: U32, capped: Bool) -> U32:
60 match capped:
61 case True{}:
62 30000
63 case False{}:
64 U32.min((800 << U32.to_nat((fails - 1 : U32)) : U32), 30000)
65
66# Backoff
67# -------
68
69# The delay before the next try after fails failures: 800ms doubling, at
70# most 30s. The driver adds jitter.
71def Backoff.delay(+fails: U32) -> U32:
72 Backoff.delay.go(fails, U32.is_ge(fails, 6))
73
74# Steps
75# -----
76
77def Watcher.page_outs(xs: List<&2, J.Json>, tail: List<&2, Out>) -> List<&2, Out>:
78 match xs:
79 case Nil{}:
80 tail
81 case Con{x, t}:
82 OInitApply{x} <> Watcher.page_outs(t, tail)
83
84def Watcher.init_out(first: Bool, rest: List<&2, Out>) -> List<&2, Out>:
85 match first:
86 case True{}:
87 OInit{} <> rest
88 case False{}:
89 rest
90
91def Watcher.done_out(last: Bool) -> List<&2, Out>:
92 match last:
93 case True{}:
94 [OInitDone{}]
95 case False{}:
96 Nil{}
97
98def Watcher.page_phase(rv: String, +cont: String, last: Bool) -> Phase:
99 match last:
100 case True{}:
101 WWatch{rv}
102 case False{}:
103 WList{cont}
104
105def Watcher.page.go(first: Bool, items: List<&2, J.Json>, rv: String, +cont: String,
106 +last: Bool) -> Step:
107 Step{WState{Watcher.page_phase(rv, cont, last), 0, last},
108 Watcher.init_out(first, Watcher.page_outs(items, Watcher.done_out(last))), False{}}
109
110# A page of the list: the first page opens a relist (Init), the last one
111# (no continue token) closes it (InitDone) and starts the watch.
112def Watcher.page(first: Bool, p: C.Page) -> Step:
113 match p:
114 case C.Page{items, rv, +cont}:
115 Watcher.page.go(first, items, rv, cont, String.is_empty(cont))
116
117def Watcher.is_gone(+code: U32) -> Bool:
118 U32.is_eq(code, 410)
119
120def Watcher.relist(+fails: U32) -> Step:
121 Step{WState{WList{""}, fails, True{}}, Nil{}, False{}}
122
123def Watcher.retry(p: Phase, +fails: U32, first: Bool) -> Step:
124 Step{WState{p, (fails + 1 : U32), first}, Nil{}, True{}}
125
126# A failed list: an expired continue token relists; else the same page is
127# retried after a backoff.
128def Watcher.list_err(cont: String, +fails: U32, first: Bool, gone: Bool) -> Step:
129 match gone:
130 case True{}:
131 Watcher.relist(fails)
132 case False{}:
133 Watcher.retry(WList{cont}, fails, first)
134
135def Watcher.watch_err(rv: String, +fails: U32, gone: Bool) -> Step:
136 match gone:
137 case True{}:
138 Watcher.relist(fails)
139 case False{}:
140 Watcher.retry(WWatch{rv}, fails, False{})
141
142# One watch event while watching from rv.
143def Watcher.event(rv: String, +fails: U32, ev: C.WEvent) -> Step:
144 match ev:
145 case C.WAdded{+o}:
146 Step{WState{WWatch{C.Obj.rv(o)}, 0, False{}}, [OApply{o}], False{}}
147 case C.WModified{+o}:
148 Step{WState{WWatch{C.Obj.rv(o)}, 0, False{}}, [OApply{o}], False{}}
149 case C.WDeleted{+o}:
150 Step{WState{WWatch{C.Obj.rv(o)}, 0, False{}}, [ODelete{o}], False{}}
151 case C.WBookmark{v}:
152 Step{WState{WWatch{v}, 0, False{}}, Nil{}, False{}}
153 case C.WError{C.Status{+code, r, m}}:
154 Watcher.watch_err(rv, fails, Watcher.is_gone(code))
155 case C.WBad{m}:
156 Watcher.retry(WWatch{rv}, fails, False{})
157
158# The state machine.
159def Watcher.step(st: WState, i: Input) -> Step:
160 match st i:
161 case WState{WList{c}, f, first} IPage{p}:
162 Watcher.page(first, p)
163 case WState{WList{c}, +f, first} IListErr{+code}:
164 Watcher.list_err(c, f, first, Watcher.is_gone(code))
165 case WState{WWatch{rv}, +f, first} IEvent{ev}:
166 Watcher.event(rv, f, ev)
167 case WState{WWatch{rv}, f, first} IWatchEnd{}:
168 Step{WState{WWatch{rv}, 0, False{}}, Nil{}, False{}}
169 case WState{WWatch{rv}, +f, first} IWatchErr{+code}:
170 Watcher.watch_err(rv, f, Watcher.is_gone(code))
171 case s _:
172 # an input the phase did not ask for changes nothing
173 Step{s, Nil{}, False{}}
174
175def Watcher.merge.go(outs: List<&2, Out>, w: Bool, s: Step) -> Step:
176 match s:
177 case Step{st, more, w2}:
178 Step{st, List.append(&2, Out, outs, more), Bool.or(w, w2)}
179
180# Feeds a run of inputs (the events of one recv), joining the outputs. A
181# step that asks to back off ends the run's effect on the stream: the
182# driver reconnects.
183def Watcher.merge(acc: Step, x: Input) -> Step:
184 match acc:
185 case Step{st, outs, w}:
186 Watcher.merge.go(outs, w, Watcher.step(st, x))
187
188def Watcher.steps(xs: List<&2, Input>, acc: Step) -> Step:
189 match xs:
190 case Nil{}:
191 acc
192 case Con{x, t}:
193 Watcher.steps(t, Watcher.merge(acc, x))