src/client.bend
bend-kube-runtime@0.1.1.0 · 680 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/utf8.bend as U
4import bend-kube-runtime@0.1.1.0/src/http.bend as H
5import bend-kube-runtime@0.1.1.0/src/core.bend as C
6import bend-kube-runtime@0.1.1.0/src/config.bend as K
7import bend-kube-runtime@0.1.1.0/src/net.bend as N
8import bend-kube-runtime@0.1.1.0/src/yaml.bend as Y
9
10# Client
11# ======
12#
13# kube-client in Bend: one verified TLS connection per request (Connection:
14# close), so no connection state outlives a call and a broken server cannot
15# poison a pool. Watches hold their connection for the stream's life; see
16# watcher.bend.
17#
18# Loops here take what just happened as a parameter (an event), and match
19# it at their head: Bend matches parameters only and forbids mutual
20# recursion, so a loop cannot branch inside a bind's continuation.
21
22def Res(-A: Type) -> Type:
23 Result<&1, &1, U32 & String, A>
24
25type KErr is Data:
26 KNet{code: U32, msg: String}
27 KApi{status: C.Status}
28 KDecode{msg: String}
29
30type Reply is Data:
31 Reply{status: U32, body: String}
32
33def KErr.show(e: KErr) -> String:
34 match e:
35 case KNet{c, m}:
36 "network: " ++ m
37 case KApi{s}:
38 "api: " ++ C.Status.show(s)
39 case KDecode{m}:
40 "decode: " ++ m
41
42# Whether a failed call is worth retrying: the network, 429 (API priority
43# and fairness), and 5xx. 4xx answers are final.
44def KErr.retryable(e: KErr) -> Bool:
45 match e:
46 case KNet{c, m}:
47 True{}
48 case KApi{C.Status{+c, r, m}}:
49 Bool.or(U32.is_eq(c, 429), U32.is_ge(c, 500))
50 case KDecode{m}:
51 False{}
52
53def KErr.code(e: KErr) -> U32:
54 match e:
55 case KApi{C.Status{c, r, m}}:
56 c
57 case _:
58 0
59
60# Files
61# -----
62
63def Fs.close_with(-A: Type, f: File, r: Res(A)) -> IO(Res(A)):
64 IO.bind(Unit, Res(A), File.close(f), u => IO.pure(Res(A), r))
65
66def Fs.read.got(p: File & Res(List<&2, U32>)) -> IO(Res(String)):
67 match p:
68 case (f, Done{bs}):
69 Fs.close_with(String, f, Done{U.Utf8.decode_all(bs)})
70 case (f, Fail{e}):
71 Fs.close_with(String, f, Fail{e})
72
73def Fs.read.sized.go(f: File, +n: U32, empty: Bool) -> IO(Res(String)):
74 match empty:
75 case True{}:
76 Fs.close_with(String, f, Done{""})
77 case False{}:
78 IO.bind(File & Res(List<&2, U32>), Res(String), File.read_bytes(f, n), Fs.read.got)
79
80def Fs.read.sized(p: File & Res(U32)) -> IO(Res(String)):
81 match p:
82 case (f, Done{+n}):
83 Fs.read.sized.go(f, n, U32.is_eq(n, 0))
84 case (f, Fail{e}):
85 Fs.close_with(String, f, Fail{e})
86
87def Fs.read.opened(r: Res(File)) -> IO(Res(String)):
88 match r:
89 case Done{f}:
90 IO.bind(File & Res(U32), Res(String), File.size(f), Fs.read.sized)
91 case Fail{e}:
92 IO.pure(Res(String), Fail{e})
93
94# A whole file as UTF-8 text.
95def Fs.read(path: String) -> IO(Res(String)):
96 IO.bind(Res(File), Res(String), File.open(path, "r"), Fs.read.opened)
97
98def Src.load.fin(+path: String, r: Res(String)) -> IO(Res(String)):
99 match r:
100 case Done{s}:
101 IO.pure(Res(String), Done{s})
102 case Fail{(c, m)}:
103 IO.pure(Res(String), Fail{(c, path ++ ": " ++ m)})
104
105def Strs.own(xs: List<&2, String>) -> List<String>:
106 match xs:
107 case Nil{}:
108 Nil{}
109 case Con{x, t}:
110 x <> Strs.own(t)
111
112def Exec.token.got(+cmd: String, t: J.Json) -> Res(String):
113 match t:
114 case J.JStr{tok}:
115 Done{tok}
116 case _:
117 Fail{(5, cmd ++ ": the ExecCredential has no status.token")}
118
119def Exec.token.json(+cmd: String, j: Result<&2, &2, String, J.Json>) -> Res(String):
120 match j:
121 case Done{+c}:
122 Exec.token.got(cmd, J.Json.at(c, ["status", "token"]))
123 case Fail{e}:
124 Fail{(5, cmd ++ ": bad ExecCredential: " ++ e)}
125
126# A credential plugin's answer (client-go's exec ExecCredential): its
127# status.token. The plugin runs on every connection, so it may cache and
128# refresh tokens as it sees fit.
129def Exec.token(+cmd: String, r: Res(U32 & (String & String))) -> IO(Res(String)):
130 match r:
131 case Done{(0, (out, err))}:
132 IO.pure(Res(String), Exec.token.json(cmd, J.Json.parse(out)))
133 case Done{(code, (out, err))}:
134 IO.pure(Res(String), Fail{(5, cmd ++ " failed: " ++ String.trim(err))})
135 case Fail{(c, m)}:
136 IO.pure(Res(String), Fail{(c, cmd ++ ": " ++ m)})
137
138def Src.load(s: K.Src) -> IO(Res(String)):
139 match s:
140 case K.Inline{t}:
141 IO.pure(Res(String), Done{t})
142 case K.FromFile{+p}:
143 IO.bind(Res(String), Res(String), Fs.read(p), r => Src.load.fin(p, r))
144 case K.FromExec{+cmd, args}:
145 IO.bind(Res(U32 & (String & String)), Res(String),
146 Process.run(cmd, Strs.own(args), "", 1048576, 30000), r => Exec.token(cmd, r))
147 case K.NoSrc{}:
148 IO.pure(Res(String), Done{""})
149
150# Credentials
151# -----------
152
153type Creds is Data:
154 Creds{ca: String, cert: String, key: String, token: String}
155
156def Creds.load.k(ca: String, cert: String, key: String, tok: Res(String)) -> IO(Res(Creds)):
157 match tok:
158 case Done{t}:
159 IO.pure(Res(Creds), Done{Creds{ca, cert, key, String.trim(t)}})
160 case Fail{e}:
161 IO.pure(Res(Creds), Fail{e})
162
163def Creds.load.c(ca: String, cert: String, key: Res(String), tok: K.Src) -> IO(Res(Creds)):
164 match key:
165 case Done{k}:
166 IO.bind(Res(String), Res(Creds), Src.load(tok), t => Creds.load.k(ca, cert, k, t))
167 case Fail{e}:
168 IO.pure(Res(Creds), Fail{e})
169
170def Creds.load.b(ca: String, cert: Res(String), key: K.Src, tok: K.Src) -> IO(Res(Creds)):
171 match cert:
172 case Done{c}:
173 IO.bind(Res(String), Res(Creds), Src.load(key), k => Creds.load.c(ca, c, k, tok))
174 case Fail{e}:
175 IO.pure(Res(Creds), Fail{e})
176
177def Creds.load.a(ca: Res(String), cert: K.Src, key: K.Src, tok: K.Src) -> IO(Res(Creds)):
178 match ca:
179 case Done{a}:
180 IO.bind(Res(String), Res(Creds), Src.load(cert), c => Creds.load.b(a, c, key, tok))
181 case Fail{e}:
182 IO.pure(Res(Creds), Fail{e})
183
184# Reads every secret a Config names; token files are read afresh each time.
185def Creds.load(c: K.Config) -> IO(Res(Creds)):
186 match c:
187 case K.Config{h, p, b, ca, cert, key, tok, ns, pl}:
188 IO.bind(Res(String), Res(Creds), Src.load(ca), a => Creds.load.a(a, cert, key, tok))
189
190def Conn.auth.go(token: String, none: Bool) -> List<&2, H.Header>:
191 match none:
192 case True{}:
193 Nil{}
194 case False{}:
195 [H.Header{"Authorization", "Bearer " ++ token}]
196
197# Connections
198# -----------
199
200def Conn.auth(+token: String) -> List<&2, H.Header>:
201 Conn.auth.go(token, String.is_empty(token))
202
203def Conn.user_agent() -> String:
204 "bend-kube/0.1"
205
206def Conn.keep(keep: Bool) -> String:
207 match keep:
208 case True{}:
209 "keep-alive"
210 case False{}:
211 "close"
212
213# The request's text, with the path under the server's base, our headers,
214# and Connection: keep-alive (pooled) or close.
215def Conn.render(req: H.Req, +base: String, +host: String, +token: String, keep: Bool) -> String:
216 match req:
217 case H.Req{m, path, hs, body}:
218 H.Http.render(H.Req{m, base ++ path,
219 List.append(&2, H.Header, hs, List.append(&2, H.Header, Conn.auth(token),
220 [H.Header{"User-Agent", Conn.user_agent()}, H.Header{"Connection", Conn.keep(keep)}])),
221 body}, host)
222
223def Conn.host_header.go(h: String, +p: U32, v6: Bool) -> String:
224 match v6:
225 case True{}:
226 "[" ++ h ++ "]:" ++ U32.show(p)
227 case False{}:
228 h ++ ":" ++ U32.show(p)
229
230def Conn.host_header(+h: String, +p: U32) -> String:
231 Conn.host_header.go(h, p, String.contains(h, ":"))
232
233# Opens a connection and sends req: the socket, ready to read the answer.
234def Conn.open.sent(p: Socket & Res(Unit)) -> IO(Res(N.Tls)):
235 match p:
236 case (s, Done{u}):
237 IO.pure(Res(N.Tls), Done{N.Tls{s}})
238 case (s, Fail{e}):
239 IO.bind(Unit, Res(N.Tls), N.Tls.close(s), u => IO.pure(Res(N.Tls), Fail{e}))
240
241def Conn.open.go(r: Res(Socket), text: String) -> IO(Res(N.Tls)):
242 match r:
243 case Done{s}:
244 IO.bind(Socket & Res(Unit), Res(N.Tls), N.Tls.send(s, text, 30000), Conn.open.sent)
245 case Fail{e}:
246 IO.pure(Res(N.Tls), Fail{e})
247
248def Conn.open.creds(+h: String, +p: U32, +base: String, req: H.Req, keep: Bool, cr: Res(Creds)) ->
249 IO(Res(N.Tls)):
250 match cr:
251 case Done{Creds{ca, cert, key, +tok}}:
252 IO.bind(Res(Socket), Res(N.Tls), N.Tls.open(h, p, ca, cert, key, 15000),
253 r => Conn.open.go(r, Conn.render(req, base, Conn.host_header(h, p), tok, keep)))
254 case Fail{e}:
255 IO.pure(Res(N.Tls), Fail{e})
256
257def Conn.open(+cfg: K.Config, req: H.Req, keep: Bool) -> IO(Res(N.Tls)):
258 match cfg:
259 case K.Config{+h, +p, +b, ca, cert, key, tok, ns, pl}:
260 IO.bind(Res(Creds), Res(N.Tls), Creds.load(cfg), cr => Conn.open.creds(h, p, b, req, keep, cr))
261
262# Reading a reply
263# ---------------
264
265# What the read loop just saw: bytes (or a failure) from a recv, or the
266# decoder's state after feeding them.
267type ReadEv is Type:
268 RGot{sock: Socket, r: Result<&1, &1, U32 & String, List<&2, U32>>}
269 RChk{sock: Socket, done: Bool, eof: Bool, bad: Bool}
270
271def ReadEv.got(p: Socket & Res(List<&2, U32>)) -> ReadEv:
272 match p:
273 case (s, r):
274 RGot{s, r}
275
276def Http.has_err(r: H.Resp) -> Bool:
277 Maybe.is_some(&2, String, H.Http.error(r))
278
279def Http.fail(msg: String) -> H.Resp:
280 H.Resp{H.PErr{msg}, None{}, Nil{}}
281
282# A failed recv: lost when no byte of a response came, else broken.
283def Http.lost(r: H.Resp, msg: String) -> H.Resp:
284 match r:
285 case H.Resp{H.PHead{Nil{}, t}, h, o}:
286 H.Resp{H.PLost{msg}, h, o}
287 case _:
288 Http.fail(msg)
289
290# Reads until the response is whole, the stream ends, or fuel runs out;
291# ms bounds each recv. Answers the socket and the decoder.
292def Conn.read(fuel: Nat, ev: ReadEv, +resp: H.Resp, +ms: U32) -> IO(Socket & H.Resp):
293 match fuel ev:
294 case 0n RGot{s, r}:
295 IO.pure(Socket & H.Resp, (s, Http.fail("response too long")))
296 case 0n RChk{s, d, e, b}:
297 IO.pure(Socket & H.Resp, (s, Http.fail("response too long")))
298 case 1n+f RGot{s, Done{+bs}}:
299 +next = H.Http.feed(bs, resp)
300 Conn.read(f, RChk{s, H.Http.is_done(next), List.is_empty(&2, U32, bs), Http.has_err(next)},
301 next, ms)
302 case 1n+f RGot{s, Fail{(c, m)}}:
303 IO.pure(Socket & H.Resp, (s, Http.lost(resp, "recv: " ++ m)))
304 case 1n+f RChk{s, True{}, e, b}:
305 IO.pure(Socket & H.Resp, (s, resp))
306 case 1n+f RChk{s, False{}, True{}, b}:
307 IO.pure(Socket & H.Resp, (s, H.Http.eof(resp)))
308 case 1n+f RChk{s, False{}, False{}, True{}}:
309 IO.pure(Socket & H.Resp, (s, resp))
310 case 1n+f RChk{s, False{}, False{}, False{}}:
311 IO.bind(Socket & Res(List<&2, U32>), Socket & H.Resp, N.Tls.recv(s, 65536, ms),
312 p => Conn.read(f, ReadEv.got(p), resp, ms))
313
314# Calls
315# -----
316
317def Reply.of(r: H.Resp & List<&2, U32>) -> Result<&2, &2, KErr, Reply>:
318 match r:
319 case (H.Resp{H.PDone{}, Some{H.Head{c, reason, hs}}, o}, body):
320 Done{Reply{c, U.Utf8.decode_all(body)}}
321 case (H.Resp{H.PErr{m}, h, o}, body):
322 Fail{KNet{0, m}}
323 case (H.Resp{H.PLost{m}, h, o}, body):
324 Fail{KNet{0, m}}
325 case _:
326 Fail{KNet{0, "incomplete response"}}
327
328def Call.read.fin(p: Socket & H.Resp) -> IO(Result<&2, &2, KErr, Reply>):
329 match p:
330 case (s, r):
331 IO.bind(Unit, Result<&2, &2, KErr, Reply>, N.Tls.close(s),
332 u => IO.pure(Result<&2, &2, KErr, Reply>, Reply.of(H.Http.take(r))))
333
334def Call.read(t: Res(N.Tls)) -> IO(Result<&2, &2, KErr, Reply>):
335 match t:
336 case Done{N.Tls{s}}:
337 IO.bind(Socket & H.Resp, Result<&2, &2, KErr, Reply>,
338 Conn.read(4294967295n, RChk{s, False{}, False{}, False{}}, H.Http.resp.init(), 60000),
339 Call.read.fin)
340 case Fail{(c, m)}:
341 IO.pure(Result<&2, &2, KErr, Reply>, Fail{KNet{c, m}})
342
343# Pooling
344# -------
345#
346# With a pool, a request first takes an idle connection; a connection goes
347# back only after a response whose end its framing marked (Content-Length
348# or chunked) and that did not say Connection: close. A taken connection
349# may have been closed by the server while idle: as Go's transport does,
350# the request is retried once on a fresh connection when it was not
351# written, or when nothing came back and its method is idempotent.
352
353# Sends req on an open (pooled) connection.
354def Conn.send_on.creds(s: Socket, +h: String, +p: U32, +base: String, req: H.Req, cr: Res(Creds)) ->
355 IO(Res(N.Tls)):
356 match cr:
357 case Done{Creds{ca, cert, key, +tok}}:
358 IO.bind(Socket & Res(Unit), Res(N.Tls),
359 N.Tls.send(s, Conn.render(req, base, Conn.host_header(h, p), tok, True{}), 30000),
360 Conn.open.sent)
361 case Fail{e}:
362 IO.bind(Unit, Res(N.Tls), N.Tls.close(s), u => IO.pure(Res(N.Tls), Fail{e}))
363
364def Conn.send_on(+cfg: K.Config, s: Socket, req: H.Req) -> IO(Res(N.Tls)):
365 match cfg:
366 case K.Config{+h, +p, +b, ca, cert, key, tok, ns, pl}:
367 IO.bind(Res(Creds), Res(N.Tls), Creds.load(cfg), cr => Conn.send_on.creds(s, h, p, b, req, cr))
368
369def Http.got_any(r: H.Resp) -> Bool:
370 match r:
371 case H.Resp{H.PHead{Nil{}, t}, h, o}:
372 False{}
373 case H.Resp{H.PLost{m}, h, o}:
374 False{}
375 case _:
376 True{}
377
378def Http.closes(r: H.Resp) -> Bool:
379 match r:
380 case H.Resp{p, Some{H.Head{c, reason, hs}}, o}:
381 String.contains(String.to_lower(Maybe.default(&2, String, H.Http.find(hs, "connection"), "")),
382 "close")
383 case _:
384 True{}
385
386def Req.idempotent.go(m: String) -> Bool:
387 match m:
388 case "GET":
389 True{}
390 case "HEAD":
391 True{}
392 case "PUT":
393 True{}
394 case "DELETE":
395 True{}
396 case _:
397 False{}
398
399def Req.idempotent(r: H.Req) -> Bool:
400 match r:
401 case H.Req{m, p, hs, b}:
402 Req.idempotent.go(m)
403
404def Pool.put.done(p: Poll<&1, &1, Result<&1, &1, Socket, Unit>, Socket>) -> IO(Unit):
405 match p:
406 case Ready{Done{u}}:
407 IO.pure(Unit, Unit{})
408 case Ready{Fail{s}}:
409 N.Tls.close(s)
410 case Wait{s}:
411 N.Tls.close(s)
412
413# Returns a connection to the pool; a full pool closes it instead.
414def Pool.put(+pool: Chan(Socket), s: Socket) -> IO(Unit):
415 IO.bind(Poll<&1, &1, Result<&1, &1, Socket, Unit>, Socket>, Unit,
416 Chan.try_send(Socket, pool, s, 0), Pool.put.done)
417
418type PEv is Type:
419 PTook{m: Poll<&1, &1, Maybe<&1, Socket>, Unit>}
420 PSent{reused: Bool, r: Result<&1, &1, U32 & String, N.Tls>}
421 PRead{reused: Bool, r: Socket & H.Resp}
422 PFin{sock: Socket, resp: H.Resp, keep: Bool, retry: Bool}
423
424def PEv.read(+reused: Bool, p: Socket & H.Resp) -> PEv:
425 PRead{reused, p}
426
427def PEv.fin(+reused: Bool, idem: Bool, p: Socket & H.Resp) -> PEv:
428 match p:
429 case (s, +r):
430 PFin{s, r, Bool.and(H.Http.is_done(r), Bool.not(Http.closes(r))),
431 Bool.and(reused, Bool.and(Bool.not(H.Http.is_done(r)), Bool.and(Bool.not(Http.got_any(r)), idem)))}
432
433def Pool.reply(r: H.Resp) -> Result<&2, &2, KErr, Reply>:
434 Reply.of(H.Http.take(r))
435
436def Pool.fail(msg: String) -> IO(Result<&2, &2, KErr, Reply>):
437 IO.pure(Result<&2, &2, KErr, Reply>, Fail{KNet{0, msg}})
438
439# One request through the pool (fuel bounds the retry).
440def Pool.run(fuel: Nat, ev: PEv, +cfg: K.Config, +pool: Chan(Socket), +req: H.Req) ->
441 IO(Result<&2, &2, KErr, Reply>):
442 match fuel ev:
443 case 0n PTook{Ready{Some{s}}}:
444 IO.bind(Unit, Result<&2, &2, KErr, Reply>, N.Tls.close(s), u => Pool.fail("pool: out of steps"))
445 case 0n PSent{u, Done{N.Tls{s}}}:
446 IO.bind(Unit, Result<&2, &2, KErr, Reply>, N.Tls.close(s), u2 => Pool.fail("pool: out of steps"))
447 case 0n PRead{u, (s, r)}:
448 IO.bind(Unit, Result<&2, &2, KErr, Reply>, N.Tls.close(s), u2 => Pool.fail("pool: out of steps"))
449 case 0n PFin{s, r, k, t}:
450 IO.bind(Unit, Result<&2, &2, KErr, Reply>, N.Tls.close(s), u2 => Pool.fail("pool: out of steps"))
451 case 0n e:
452 Pool.fail("pool: out of steps")
453 case 1n+f PTook{Ready{Some{s}}}:
454 IO.bind(Result<&1, &1, U32 & String, N.Tls>, Result<&2, &2, KErr, Reply>,
455 Conn.send_on(cfg, s, req), r => Pool.run(f, PSent{True{}, r}, cfg, pool, req))
456 case 1n+f PTook{m}:
457 IO.bind(Result<&1, &1, U32 & String, N.Tls>, Result<&2, &2, KErr, Reply>,
458 Conn.open(cfg, req, True{}), r => Pool.run(f, PSent{False{}, r}, cfg, pool, req))
459 case 1n+f PSent{+reused, Done{N.Tls{s}}}:
460 IO.bind(Socket & H.Resp, Result<&2, &2, KErr, Reply>,
461 Conn.read(4294967295n, RChk{s, False{}, False{}, False{}}, H.Http.resp.init(), 60000),
462 p => Pool.run(f, PEv.read(reused, p), cfg, pool, req))
463 case 1n+f PSent{True{}, Fail{e}}:
464 # a stale idle connection: the request was not written, so retry
465 Pool.run(f, PTook{Wait{Unit{}}}, cfg, pool, req)
466 case 1n+f PSent{False{}, Fail{(c, m)}}:
467 IO.pure(Result<&2, &2, KErr, Reply>, Fail{KNet{c, m}})
468 case 1n+f PRead{+reused, p}:
469 Pool.run(f, PEv.fin(reused, Req.idempotent(req), p), cfg, pool, req)
470 case 1n+f PFin{s, +r, True{}, t}:
471 IO.bind(Unit, Result<&2, &2, KErr, Reply>, Pool.put(pool, s),
472 u => IO.pure(Result<&2, &2, KErr, Reply>, Pool.reply(r)))
473 case 1n+f PFin{s, r, False{}, True{}}:
474 IO.bind(Unit, Result<&2, &2, KErr, Reply>, N.Tls.close(s),
475 u => Pool.run(f, PTook{Wait{Unit{}}}, cfg, pool, req))
476 case 1n+f PFin{s, +r, False{}, False{}}:
477 IO.bind(Unit, Result<&2, &2, KErr, Reply>, N.Tls.close(s),
478 u => IO.pure(Result<&2, &2, KErr, Reply>, Pool.reply(r)))
479
480def Client.send.with(+cfg: K.Config, req: H.Req, pool: Maybe<&2, Chan(Socket)>) ->
481 IO(Result<&2, &2, KErr, Reply>):
482 match pool:
483 case Some{+p}:
484 IO.bind(Poll<&1, &1, Maybe<&1, Socket>, Unit>, Result<&2, &2, KErr, Reply>,
485 Chan.try_recv(Socket, p, 0), m => Pool.run(10n, PTook{m}, cfg, p, req))
486 case None{}:
487 IO.bind(Res(N.Tls), Result<&2, &2, KErr, Reply>, Conn.open(cfg, req, False{}), Call.read)
488
489# Sends a request and reads the whole reply, whatever its status.
490def Client.send(+cfg: K.Config, req: H.Req) -> IO(Result<&2, &2, KErr, Reply>):
491 Client.send.with(cfg, req, K.Config.pool(cfg))
492
493def Client.json.status.fix.go(+c: U32, +k: U32, r: String, m: String, zero: Bool) -> C.Status:
494 match zero:
495 case True{}:
496 C.Status{c, r, m}
497 case False{}:
498 C.Status{k, r, m}
499
500# A Status without a code (some proxies) takes the HTTP status.
501def Client.json.status.fix(+c: U32, s: C.Status) -> C.Status:
502 match s:
503 case C.Status{+k, r, m}:
504 Client.json.status.fix.go(c, k, r, m, U32.is_eq(k, 0))
505
506def Client.json.status(+c: U32, j: Result<&2, &2, String, J.Json>, ok: Bool) ->
507 Result<&2, &2, KErr, J.Json>:
508 match j ok:
509 case Done{x} True{}:
510 Done{x}
511 case Fail{m} True{}:
512 Fail{KDecode{m}}
513 case Done{+x} False{}:
514 Fail{KApi{Client.json.status.fix(c, C.Status.of(x))}}
515 case Fail{m} False{}:
516 Fail{KApi{C.Status{c, "", "non-JSON error body"}}}
517
518# A 2xx reply's body as JSON; any other status as the Status it carries.
519def Client.json.of(r: Result<&2, &2, KErr, Reply>) -> Result<&2, &2, KErr, J.Json>:
520 match r:
521 case Done{Reply{+c, body}}:
522 Client.json.status(c, J.Json.parse(body), Bool.and(U32.is_ge(c, 200), U32.is_lt(c, 300)))
523 case Fail{e}:
524 Fail{e}
525
526def Client.json(+cfg: K.Config, req: H.Req) -> IO(Result<&2, &2, KErr, J.Json>):
527 IO.bind(Result<&2, &2, KErr, Reply>, Result<&2, &2, KErr, J.Json>, Client.send(cfg, req),
528 r => IO.pure(Result<&2, &2, KErr, J.Json>, Client.json.of(r)))
529
530# Verbs (kube-client's Api<K>)
531# -----
532
533def Client.get(+cfg: K.Config, a: C.Api, name: String) -> IO(Result<&2, &2, KErr, J.Json>):
534 Client.json(cfg, C.Req.read(a, name))
535
536def Client.list(+cfg: K.Config, a: C.Api, lp: C.ListParams) -> IO(Result<&2, &2, KErr, J.Json>):
537 Client.json(cfg, C.Req.list(a, lp))
538
539def Client.create(+cfg: K.Config, a: C.Api, obj: J.Json) -> IO(Result<&2, &2, KErr, J.Json>):
540 Client.json(cfg, C.Req.create(a, obj))
541
542def Client.replace(+cfg: K.Config, a: C.Api, name: String, obj: J.Json) ->
543 IO(Result<&2, &2, KErr, J.Json>):
544 Client.json(cfg, C.Req.replace(a, name, obj))
545
546def Client.delete(+cfg: K.Config, a: C.Api, name: String) -> IO(Result<&2, &2, KErr, J.Json>):
547 Client.json(cfg, C.Req.delete(a, name))
548
549def Client.patch(+cfg: K.Config, a: C.Api, name: String, p: C.PatchKind, body: J.Json) ->
550 IO(Result<&2, &2, KErr, J.Json>):
551 Client.json(cfg, C.Req.patch(a, name, "", p, body))
552
553def Client.patch_status(+cfg: K.Config, a: C.Api, name: String, p: C.PatchKind, body: J.Json) ->
554 IO(Result<&2, &2, KErr, J.Json>):
555 Client.json(cfg, C.Req.patch(a, name, "status", p, body))
556
557# Server-side apply, kube's recommended write: the server merges our fields
558# under our manager name and tracks ownership.
559def Client.apply(+cfg: K.Config, a: C.Api, name: String, manager: String, body: J.Json) ->
560 IO(Result<&2, &2, KErr, J.Json>):
561 Client.patch(cfg, a, name, C.PApply{manager, True{}}, body)
562
563def Kc.rebased(r: Result<&2, &2, String, J.Json>, +path: String) -> Result<&2, &2, String, J.Json>:
564 match r:
565 case Done{+kc}:
566 Done{K.Kube.rebase(kc, K.Kube.dir(path))}
567 case Fail{e}:
568 Fail{path ++ ": " ++ e}
569
570# Configs
571# -------
572
573# Reads one kubeconfig, JSON or YAML (yaml.bend), with its relative paths
574# made absolute against its own directory, as client-go does.
575def Kc.parse(+t: String, +path: String, json: Bool) -> Result<&2, &2, String, J.Json>:
576 match json:
577 case True{}:
578 Kc.rebased(J.Json.parse(t), path)
579 case False{}:
580 Done{K.Kube.rebase(Y.Yaml.parse(t), K.Kube.dir(path))}
581
582def Kc.read.got(r: Res(String), +path: String) -> Result<&2, &2, String, J.Json>:
583 match r:
584 case Done{+t}:
585 Kc.parse(t, path, String.starts_with(String.trim_start(t), "{"))
586 case Fail{(c, m)}:
587 Fail{m}
588
589def Kc.read(+path: String) -> IO(Result<&2, &2, String, J.Json>):
590 IO.bind(Res(String), Result<&2, &2, String, J.Json>, Fs.read(path),
591 r => IO.pure(Result<&2, &2, String, J.Json>, Kc.read.got(r, path)))
592
593# Merges the files of a KUBECONFIG list in order; a file that does not
594# exist is skipped, as client-go skips it.
595def Kc.merge.got(acc: Maybe<&2, J.Json>, r: Result<&2, &2, String, J.Json>) -> Maybe<&2, J.Json>:
596 match acc r:
597 case Some{+a} Done{+b}:
598 Some{K.Kube.merge(a, b)}
599 case None{} Done{b}:
600 Some{b}
601 case a Fail{e}:
602 a
603
604def Kc.merge(ps: List<&2, String>, acc: Maybe<&2, J.Json>) -> IO(Maybe<&2, J.Json>):
605 match ps:
606 case Nil{}:
607 IO.pure(Maybe<&2, J.Json>, acc)
608 case Con{+p, t}:
609 IO.bind(Result<&2, &2, String, J.Json>, Maybe<&2, J.Json>, Kc.read(p),
610 r => Kc.merge(t, Kc.merge.got(acc, r)))
611
612def Config.from_kubeconfig.got(r: Maybe<&2, J.Json>, ctx: String) -> Result<&2, &2, String, K.Config>:
613 match r:
614 case Some{+kc}:
615 K.Kube.config(kc, ctx)
616 case None{}:
617 Fail{"kubeconfig: no readable file"}
618
619# A kubeconfig (one path, or a KUBECONFIG list "a:b") at a context ("" for
620# its current one). JSON and YAML are both read natively.
621def Config.from_kubeconfig(+path: String, ctx: String) -> IO(Result<&2, &2, String, K.Config>):
622 IO.bind(Maybe<&2, J.Json>, Result<&2, &2, String, K.Config>,
623 Kc.merge(String.split(path, ':'), None{}),
624 m => IO.pure(Result<&2, &2, String, K.Config>, Config.from_kubeconfig.got(m, ctx)))
625
626def Config.in_cluster.ns(host: String, port: String, ns: Res(String)) ->
627 IO(Result<&2, &2, String, K.Config>):
628 match ns:
629 case Done{n}:
630 IO.pure(Result<&2, &2, String, K.Config>, Done{K.InCluster.config(host, port, n)})
631 case Fail{e}:
632 IO.pure(Result<&2, &2, String, K.Config>, Done{K.InCluster.config(host, port, "default")})
633
634def Config.in_cluster.port(host: String, port: Res(String)) -> IO(Result<&2, &2, String, K.Config>):
635 match port:
636 case Done{p}:
637 IO.bind(Res(String), Result<&2, &2, String, K.Config>,
638 Fs.read(K.InCluster.dir() ++ "/namespace"), n => Config.in_cluster.ns(host, p, n))
639 case Fail{e}:
640 IO.pure(Result<&2, &2, String, K.Config>, Fail{"KUBERNETES_SERVICE_PORT is not set"})
641
642def Config.in_cluster.host(host: Res(String)) -> IO(Result<&2, &2, String, K.Config>):
643 match host:
644 case Done{h}:
645 IO.bind(Res(String), Result<&2, &2, String, K.Config>,
646 IO.get_env("KUBERNETES_SERVICE_PORT"), p => Config.in_cluster.port(h, p))
647 case Fail{e}:
648 IO.pure(Result<&2, &2, String, K.Config>, Fail{"not in a cluster"})
649
650# The pod's own service account (client-go's InClusterConfig).
651def Config.in_cluster() -> IO(Result<&2, &2, String, K.Config>):
652 IO.bind(Res(String), Result<&2, &2, String, K.Config>, IO.get_env("KUBERNETES_SERVICE_HOST"),
653 Config.in_cluster.host)
654
655def Config.infer.home(r: Res(String)) -> IO(Result<&2, &2, String, K.Config>):
656 match r:
657 case Done{home}:
658 Config.from_kubeconfig(home ++ "/.kube/config", "")
659 case Fail{e}:
660 IO.pure(Result<&2, &2, String, K.Config>, Fail{"no KUBECONFIG, no cluster, no HOME"})
661
662def Config.infer.pod(r: Res(String)) -> IO(Result<&2, &2, String, K.Config>):
663 match r:
664 case Done{h}:
665 Config.in_cluster()
666 case Fail{e}:
667 IO.bind(Res(String), Result<&2, &2, String, K.Config>, IO.get_env("HOME"), Config.infer.home)
668
669def Config.infer.kc(r: Res(String)) -> IO(Result<&2, &2, String, K.Config>):
670 match r:
671 case Done{path}:
672 Config.from_kubeconfig(path, "")
673 case Fail{e}:
674 IO.bind(Res(String), Result<&2, &2, String, K.Config>, IO.get_env("KUBERNETES_SERVICE_HOST"),
675 Config.infer.pod)
676
677# As kube-rs's Config::infer: $KUBECONFIG when set, else the pod's service
678# account when in a cluster, else ~/.kube/config.
679def Config.infer() -> IO(Result<&2, &2, String, K.Config>):
680 IO.bind(Res(String), Result<&2, &2, String, K.Config>, IO.get_env("KUBECONFIG"), Config.infer.kc)