src/watcher.bend

bend-kube-runtime@0.2.0.0 · 193 lines · 24 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# 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))