src/serve.bend

bend-kube-runtime@0.2.0.0 · 236 lines · 35 definitions

1import Base
2import bend-kube-runtime@0.2.0.0/src/runtime.bend as R
3import bend-kube-runtime@0.2.0.0/src/utf8.bend as U
4
5# Serve
6# =====
7#
8# The health and metrics endpoints a Deployment's probes and Prometheus
9# scrape (controller-runtime's /healthz, /readyz and /metrics), over plain
10# HTTP/1.1. Each connection is served by its own computation, reads at
11# most 8KiB within 5s, and is closed after one answer.
12#
13# /healthz answers 200 while the process serves. /readyz and /metrics ask
14# the status keeper for the controller's latest snapshot; a replica that
15# is not the leader has none, and answers ready (standby).
16
17def Serve.Res(-A: Type) -> Type:
18  Result<&1, &1, U32 & String, A>
19
20def Serve.size(s: String) -> U32:
21  U.Utf8.size(s)
22
23def Serve.reason.go(ok: Bool, missing: Bool) -> String:
24  match ok missing:
25    case True{} _:
26      "OK"
27    case False{} True{}:
28      "Not Found"
29    case False{} False{}:
30      "Service Unavailable"
31
32def Serve.reason(+code: U32) -> String:
33  Serve.reason.go(U32.is_eq(code, 200), U32.is_eq(code, 404))
34
35def Serve.reply(+code: U32, ctype: String, +body: String) -> String:
36  "HTTP/1.1 " ++ U32.show(code) ++ " " ++ Serve.reason(code) ++ "\r\nContent-Type: " ++ ctype
37    ++ "\r\nContent-Length: " ++ U32.show(Serve.size(body)) ++ "\r\nConnection: close\r\n\r\n"
38    ++ body
39
40def Serve.first_line.go(ls: List<&2, String>) -> String:
41  match ls:
42    case Con{l, t}:
43      l
44    case Nil{}:
45      ""
46
47def Serve.first_line(s: String) -> String:
48  Serve.first_line.go(String.lines(s))
49
50def Serve.path.go(parts: List<&2, String>) -> String:
51  match parts:
52    case Con{m, Con{p, rest}}:
53      p
54    case _:
55      ""
56
57# The request line's path.
58def Serve.path(req: String) -> String:
59  Serve.path.go(String.split(String.trim(Serve.first_line(req)), ' '))
60
61def Serve.probe.answer(m: Poll<&1, &1, Maybe<&1, List<&2, R.Snap>>, Unit>) -> List<&2, R.Snap>:
62  match m:
63    case Ready{Some{x}}:
64      x
65    case _:
66      Nil{}
67
68# Every controller's snapshot from the status keeper, which answers at once
69# (none on a standby replica).
70def Serve.probe(+status: Chan(R.SMsg)) -> IO(List<&2, R.Snap>):
71  IO.bind(Chan(List<&2, R.Snap>), List<&2, R.Snap>, Chan.new(List<&2, R.Snap>, 1),
72    +c => IO.bind(Result<&1, &1, R.SMsg, Unit>, List<&2, R.Snap>,
73      Chan.send(R.SMsg, status, R.SGet{c}),
74      u => IO.bind(Poll<&1, &1, Maybe<&1, List<&2, R.Snap>>, Unit>, List<&2, R.Snap>,
75        Chan.try_recv(List<&2, R.Snap>, c, 500),
76        m => IO.pure(List<&2, R.Snap>, Serve.probe.answer(m)))))
77
78def Serve.all_ready(ss: List<&2, R.Snap>) -> Bool:
79  match ss:
80    case Nil{}:
81      True{}
82    case Con{R.Snap{nm, r, a, b, c, d, e}, t}:
83      Bool.and(r, Serve.all_ready(t))
84
85def Serve.ready.go(standby: Bool, ready: Bool) -> String:
86  match standby ready:
87    case True{} _:
88      Serve.reply(200, "text/plain", "standby\n")
89    case False{} True{}:
90      Serve.reply(200, "text/plain", "ok\n")
91    case False{} False{}:
92      Serve.reply(503, "text/plain", "store not synced\n")
93
94# Ready when every controller's store has listed; a standby is ready.
95def Serve.ready(+ss: List<&2, R.Snap>) -> String:
96  Serve.ready.go(List.is_empty(&2, R.Snap, ss), Serve.all_ready(ss))
97
98def Serve.bool(b: Bool) -> String:
99  match b:
100    case True{}:
101      "1"
102    case False{}:
103      "0"
104
105def Serve.head(+name: String, kind: String, help: String) -> String:
106  "# HELP " ++ name ++ " " ++ help ++ "\n# TYPE " ++ name ++ " " ++ kind ++ "\n"
107
108def Serve.snap.name(s: R.Snap) -> String:
109  match s:
110    case R.Snap{n, r, a, b, c, d, e}:
111      n
112
113def Serve.snap.pick.go(r: Bool, +a: U32, +b: U32, +c: U32, +d: U32, +e: U32, i0: Bool, i1: Bool,
114  i2: Bool, i3: Bool, i4: Bool) -> String:
115  match i0 i1 i2 i3 i4:
116    case True{} _ _ _ _:
117      Serve.bool(r)
118    case False{} True{} _ _ _:
119      U32.show(a)
120    case False{} False{} True{} _ _:
121      U32.show(b)
122    case False{} False{} False{} True{} _:
123      U32.show(c)
124    case False{} False{} False{} False{} True{}:
125      U32.show(d)
126    case _ _ _ _ _:
127      U32.show(e)
128
129def Serve.snap.pick(+i: U32, r: Bool, +a: U32, +b: U32, +c: U32, +d: U32, +e: U32) -> String:
130  Serve.snap.pick.go(r, a, b, c, d, e, U32.is_eq(i, 0), U32.is_eq(i, 1), U32.is_eq(i, 2),
131    U32.is_eq(i, 3), U32.is_eq(i, 4))
132
133# The field-th number of a snapshot (0 ready, 1 reconciles, 2 errors,
134# 3 events, 4 running, 5 objects).
135def Serve.snap.field(s: R.Snap, +i: U32) -> String:
136  match s:
137    case R.Snap{n, r, +a, +b, +c, +d, +e}:
138      Serve.snap.pick(i, r, a, b, c, d, e)
139
140# One sample per controller, labeled with its kind.
141def Serve.samples(ss: List<&2, R.Snap>, +name: String, +field: U32) -> String:
142  match ss:
143    case Nil{}:
144      ""
145    case Con{+x, t}:
146      name ++ "{controller=\"" ++ Serve.snap.name(x) ++ "\"} " ++ Serve.snap.field(x, field) ++ "\n"
147        ++ Serve.samples(t, name, field)
148
149def Serve.family(+ss: List<&2, R.Snap>, +name: String, kind: String, help: String, +field: U32) -> String:
150  Serve.head(name, kind, help) ++ Serve.samples(ss, name, field)
151
152# The Prometheus text format: one sample per controller; a standby
153# replica reports only that it leads nothing.
154def Serve.text(+ss: List<&2, R.Snap>) -> String:
155  Serve.head("bendkube_leader", "gauge", "Whether this replica runs the controllers.")
156    ++ "bendkube_leader " ++ Serve.bool(Bool.not(List.is_empty(&2, R.Snap, ss))) ++ "\n"
157    ++ Serve.family(ss, "bendkube_reconcile_total", "counter", "Reconciles finished.", 1)
158    ++ Serve.family(ss, "bendkube_reconcile_errors_total", "counter", "Reconciles that asked to retry.", 2)
159    ++ Serve.family(ss, "bendkube_watch_events_total", "counter", "Watcher outputs received.", 3)
160    ++ Serve.family(ss, "bendkube_reconcile_running", "gauge", "Reconciles running now.", 4)
161    ++ Serve.family(ss, "bendkube_store_objects", "gauge", "Objects in the primary store.", 5)
162    ++ Serve.family(ss, "bendkube_store_ready", "gauge", "Whether the primary store has listed.", 0)
163
164def Serve.metrics(+ss: List<&2, R.Snap>) -> String:
165  Serve.reply(200, "text/plain; version=0.0.4", Serve.text(ss))
166
167# The answer to a request.
168def Serve.route(path: String, +status: Chan(R.SMsg)) -> IO(String):
169  match path:
170    case "/healthz":
171      IO.pure(String, Serve.reply(200, "text/plain", "ok\n"))
172    case "/readyz":
173      IO.bind(List<&2, R.Snap>, String, Serve.probe(status), m => IO.pure(String, Serve.ready(m)))
174    case "/metrics":
175      IO.bind(List<&2, R.Snap>, String, Serve.probe(status), m => IO.pure(String, Serve.metrics(m)))
176    case _:
177      IO.pure(String, Serve.reply(404, "text/plain", "not found\n"))
178
179def Serve.answer.done(p: Socket & Poll<&1, &1, Result<&1, &1, (U32 & String) & List<&2, U32>,
180  Unit>, List<&2, U32>>) -> IO(Unit):
181  match p:
182    case (s, r):
183      Socket.close(s)
184
185def Serve.answer(s: Socket, text: String) -> IO(Unit):
186  IO.bind(Socket & Poll<&1, &1, Result<&1, &1, (U32 & String) & List<&2, U32>, Unit>,
187    List<&2, U32>>, Unit, TCP.try_send(s, text, 5000), Serve.answer.done)
188
189def Serve.handle.got(p: Socket & Poll<&1, &1, Serve.Res(Maybe<&1, String>), Unit>,
190  +status: Chan(R.SMsg)) -> IO(Unit):
191  match p:
192    case (s, Ready{Done{Some{req}}}):
193      IO.bind(String, Unit, Serve.route(Serve.path(req), status), t => Serve.answer(s, t))
194    case (s, _):
195      Socket.close(s)
196
197def Serve.handle(s: Socket, +status: Chan(R.SMsg)) -> IO(Unit):
198  IO.bind(Socket & Poll<&1, &1, Serve.Res(Maybe<&1, String>), Unit>, Unit,
199    TCP.try_recv(s, 8192, 5000), p => Serve.handle.got(p, status))
200
201def Serve.accepted(r: Serve.Res(Socket), +status: Chan(R.SMsg)) -> IO(Unit):
202  match r:
203    case Done{s}:
204      IO.spawn(Unit, Serve.handle(s, status))
205    case Fail{e}:
206      IO.pure(Unit, Unit{})
207
208type AEv is Type:
209  AGot{l: Listener, r: Result<&1, &1, U32 & String, Socket>}
210
211def AEv.got(p: Listener & Serve.Res(Socket)) -> AEv:
212  match p:
213    case (l, r):
214      AGot{l, r}
215
216# Accepts forever (fuel permitting), serving each connection on its own.
217def Serve.loop(fuel: Nat, ev: AEv, +status: Chan(R.SMsg)) -> IO(Unit):
218  match fuel ev:
219    case 0n AGot{l, r}:
220      Listener.close(l)
221    case 1n+f AGot{l, r}:
222      IO.bind(Unit, Unit, Serve.accepted(r, status),
223        u => IO.bind(Listener & Serve.Res(Socket), Unit, TCP.accept(l),
224          p => Serve.loop(f, AEv.got(p), status)))
225
226def Serve.listening(r: Serve.Res(Listener), +status: Chan(R.SMsg)) -> IO(Unit):
227  match r:
228    case Done{l}:
229      IO.bind(Listener & Serve.Res(Socket), Unit, TCP.accept(l),
230        p => Serve.loop(4294967295n, AEv.got(p), status))
231    case Fail{(c, m)}:
232      IO.print_err("health server: " ++ m)
233
234# Serves the endpoints on 0.0.0.0:port.
235def Serve.run(+port: U32, +status: Chan(R.SMsg)) -> IO(Unit):
236  IO.bind(Serve.Res(Listener), Unit, TCP.listen("0.0.0.0", port), r => Serve.listening(r, status))