src/client.bend

bend-kube-runtime@0.2.0.0 · 689 lines · 90 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/utf8.bend as U
4import bend-kube-runtime@0.2.0.0/src/http.bend as H
5import bend-kube-runtime@0.2.0.0/src/core.bend as C
6import bend-kube-runtime@0.2.0.0/src/config.bend as K
7import bend-kube-runtime@0.2.0.0/src/net.bend as N
8import bend-kube-runtime@0.2.0.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
213def Conn.connection(keep: Bool, own: Bool) -> List<&2, H.Header>:
214  match own:
215    case True{}:
216      Nil{}
217    case False{}:
218      [H.Header{"Connection", Conn.keep(keep)}]
219
220# The request's text, with the path under the server's base, our headers,
221# and Connection: keep-alive (pooled) or close, unless the request names
222# its own (an Upgrade).
223def Conn.render(req: H.Req, +base: String, +host: String, +token: String, keep: Bool) -> String:
224  match req:
225    case H.Req{m, path, +hs, body}:
226      H.Http.render(H.Req{m, base ++ path,
227        List.append(&2, H.Header, hs, List.append(&2, H.Header, Conn.auth(token),
228          Con{H.Header{"User-Agent", Conn.user_agent()},
229            Conn.connection(keep, Maybe.is_some(&2, String, H.Http.find(hs, "connection")))})),
230        body}, host)
231
232def Conn.host_header.go(h: String, +p: U32, v6: Bool) -> String:
233  match v6:
234    case True{}:
235      "[" ++ h ++ "]:" ++ U32.show(p)
236    case False{}:
237      h ++ ":" ++ U32.show(p)
238
239def Conn.host_header(+h: String, +p: U32) -> String:
240  Conn.host_header.go(h, p, String.contains(h, ":"))
241
242# Opens a connection and sends req: the socket, ready to read the answer.
243def Conn.open.sent(p: Socket & Res(Unit)) -> IO(Res(N.Tls)):
244  match p:
245    case (s, Done{u}):
246      IO.pure(Res(N.Tls), Done{N.Tls{s}})
247    case (s, Fail{e}):
248      IO.bind(Unit, Res(N.Tls), N.Tls.close(s), u => IO.pure(Res(N.Tls), Fail{e}))
249
250def Conn.open.go(r: Res(Socket), text: String) -> IO(Res(N.Tls)):
251  match r:
252    case Done{s}:
253      IO.bind(Socket & Res(Unit), Res(N.Tls), N.Tls.send(s, text, 30000), Conn.open.sent)
254    case Fail{e}:
255      IO.pure(Res(N.Tls), Fail{e})
256
257def Conn.open.creds(+h: String, +p: U32, +base: String, req: H.Req, keep: Bool, cr: Res(Creds)) ->
258  IO(Res(N.Tls)):
259  match cr:
260    case Done{Creds{ca, cert, key, +tok}}:
261      IO.bind(Res(Socket), Res(N.Tls), N.Tls.open(h, p, ca, cert, key, 15000),
262        r => Conn.open.go(r, Conn.render(req, base, Conn.host_header(h, p), tok, keep)))
263    case Fail{e}:
264      IO.pure(Res(N.Tls), Fail{e})
265
266def Conn.open(+cfg: K.Config, req: H.Req, keep: Bool) -> IO(Res(N.Tls)):
267  match cfg:
268    case K.Config{+h, +p, +b, ca, cert, key, tok, ns, pl}:
269      IO.bind(Res(Creds), Res(N.Tls), Creds.load(cfg), cr => Conn.open.creds(h, p, b, req, keep, cr))
270
271# Reading a reply
272# ---------------
273
274# What the read loop just saw: bytes (or a failure) from a recv, or the
275# decoder's state after feeding them.
276type ReadEv is Type:
277  RGot{sock: Socket, r: Result<&1, &1, U32 & String, List<&2, U32>>}
278  RChk{sock: Socket, done: Bool, eof: Bool, bad: Bool}
279
280def ReadEv.got(p: Socket & Res(List<&2, U32>)) -> ReadEv:
281  match p:
282    case (s, r):
283      RGot{s, r}
284
285def Http.has_err(r: H.Resp) -> Bool:
286  Maybe.is_some(&2, String, H.Http.error(r))
287
288def Http.fail(msg: String) -> H.Resp:
289  H.Resp{H.PErr{msg}, None{}, Nil{}}
290
291# A failed recv: lost when no byte of a response came, else broken.
292def Http.lost(r: H.Resp, msg: String) -> H.Resp:
293  match r:
294    case H.Resp{H.PHead{Nil{}, t}, h, o}:
295      H.Resp{H.PLost{msg}, h, o}
296    case _:
297      Http.fail(msg)
298
299# Reads until the response is whole, the stream ends, or fuel runs out;
300# ms bounds each recv. Answers the socket and the decoder.
301def Conn.read(fuel: Nat, ev: ReadEv, +resp: H.Resp, +ms: U32) -> IO(Socket & H.Resp):
302  match fuel ev:
303    case 0n RGot{s, r}:
304      IO.pure(Socket & H.Resp, (s, Http.fail("response too long")))
305    case 0n RChk{s, d, e, b}:
306      IO.pure(Socket & H.Resp, (s, Http.fail("response too long")))
307    case 1n+f RGot{s, Done{+bs}}:
308      +next = H.Http.feed(bs, resp)
309      Conn.read(f, RChk{s, H.Http.is_done(next), List.is_empty(&2, U32, bs), Http.has_err(next)},
310        next, ms)
311    case 1n+f RGot{s, Fail{(c, m)}}:
312      IO.pure(Socket & H.Resp, (s, Http.lost(resp, "recv: " ++ m)))
313    case 1n+f RChk{s, True{}, e, b}:
314      IO.pure(Socket & H.Resp, (s, resp))
315    case 1n+f RChk{s, False{}, True{}, b}:
316      IO.pure(Socket & H.Resp, (s, H.Http.eof(resp)))
317    case 1n+f RChk{s, False{}, False{}, True{}}:
318      IO.pure(Socket & H.Resp, (s, resp))
319    case 1n+f RChk{s, False{}, False{}, False{}}:
320      IO.bind(Socket & Res(List<&2, U32>), Socket & H.Resp, N.Tls.recv(s, 65536, ms),
321        p => Conn.read(f, ReadEv.got(p), resp, ms))
322
323# Calls
324# -----
325
326def Reply.of(r: H.Resp & List<&2, U32>) -> Result<&2, &2, KErr, Reply>:
327  match r:
328    case (H.Resp{H.PDone{}, Some{H.Head{c, reason, hs}}, o}, body):
329      Done{Reply{c, U.Utf8.decode_all(body)}}
330    case (H.Resp{H.PErr{m}, h, o}, body):
331      Fail{KNet{0, m}}
332    case (H.Resp{H.PLost{m}, h, o}, body):
333      Fail{KNet{0, m}}
334    case _:
335      Fail{KNet{0, "incomplete response"}}
336
337def Call.read.fin(p: Socket & H.Resp) -> IO(Result<&2, &2, KErr, Reply>):
338  match p:
339    case (s, r):
340      IO.bind(Unit, Result<&2, &2, KErr, Reply>, N.Tls.close(s),
341        u => IO.pure(Result<&2, &2, KErr, Reply>, Reply.of(H.Http.take(r))))
342
343def Call.read(t: Res(N.Tls)) -> IO(Result<&2, &2, KErr, Reply>):
344  match t:
345    case Done{N.Tls{s}}:
346      IO.bind(Socket & H.Resp, Result<&2, &2, KErr, Reply>,
347        Conn.read(4294967295n, RChk{s, False{}, False{}, False{}}, H.Http.resp.init(), 60000),
348        Call.read.fin)
349    case Fail{(c, m)}:
350      IO.pure(Result<&2, &2, KErr, Reply>, Fail{KNet{c, m}})
351
352# Pooling
353# -------
354#
355# With a pool, a request first takes an idle connection; a connection goes
356# back only after a response whose end its framing marked (Content-Length
357# or chunked) and that did not say Connection: close. A taken connection
358# may have been closed by the server while idle: as Go's transport does,
359# the request is retried once on a fresh connection when it was not
360# written, or when nothing came back and its method is idempotent.
361
362# Sends req on an open (pooled) connection.
363def Conn.send_on.creds(s: Socket, +h: String, +p: U32, +base: String, req: H.Req, cr: Res(Creds)) ->
364  IO(Res(N.Tls)):
365  match cr:
366    case Done{Creds{ca, cert, key, +tok}}:
367      IO.bind(Socket & Res(Unit), Res(N.Tls),
368        N.Tls.send(s, Conn.render(req, base, Conn.host_header(h, p), tok, True{}), 30000),
369        Conn.open.sent)
370    case Fail{e}:
371      IO.bind(Unit, Res(N.Tls), N.Tls.close(s), u => IO.pure(Res(N.Tls), Fail{e}))
372
373def Conn.send_on(+cfg: K.Config, s: Socket, req: H.Req) -> IO(Res(N.Tls)):
374  match cfg:
375    case K.Config{+h, +p, +b, ca, cert, key, tok, ns, pl}:
376      IO.bind(Res(Creds), Res(N.Tls), Creds.load(cfg), cr => Conn.send_on.creds(s, h, p, b, req, cr))
377
378def Http.got_any(r: H.Resp) -> Bool:
379  match r:
380    case H.Resp{H.PHead{Nil{}, t}, h, o}:
381      False{}
382    case H.Resp{H.PLost{m}, h, o}:
383      False{}
384    case _:
385      True{}
386
387def Http.closes(r: H.Resp) -> Bool:
388  match r:
389    case H.Resp{p, Some{H.Head{c, reason, hs}}, o}:
390      String.contains(String.to_lower(Maybe.default(&2, String, H.Http.find(hs, "connection"), "")),
391        "close")
392    case _:
393      True{}
394
395def Req.idempotent.go(m: String) -> Bool:
396  match m:
397    case "GET":
398      True{}
399    case "HEAD":
400      True{}
401    case "PUT":
402      True{}
403    case "DELETE":
404      True{}
405    case _:
406      False{}
407
408def Req.idempotent(r: H.Req) -> Bool:
409  match r:
410    case H.Req{m, p, hs, b}:
411      Req.idempotent.go(m)
412
413def Pool.put.done(p: Poll<&1, &1, Result<&1, &1, Socket, Unit>, Socket>) -> IO(Unit):
414  match p:
415    case Ready{Done{u}}:
416      IO.pure(Unit, Unit{})
417    case Ready{Fail{s}}:
418      N.Tls.close(s)
419    case Wait{s}:
420      N.Tls.close(s)
421
422# Returns a connection to the pool; a full pool closes it instead.
423def Pool.put(+pool: Chan(Socket), s: Socket) -> IO(Unit):
424  IO.bind(Poll<&1, &1, Result<&1, &1, Socket, Unit>, Socket>, Unit,
425    Chan.try_send(Socket, pool, s, 0), Pool.put.done)
426
427type PEv is Type:
428  PTook{m: Poll<&1, &1, Maybe<&1, Socket>, Unit>}
429  PSent{reused: Bool, r: Result<&1, &1, U32 & String, N.Tls>}
430  PRead{reused: Bool, r: Socket & H.Resp}
431  PFin{sock: Socket, resp: H.Resp, keep: Bool, retry: Bool}
432
433def PEv.read(+reused: Bool, p: Socket & H.Resp) -> PEv:
434  PRead{reused, p}
435
436def PEv.fin(+reused: Bool, idem: Bool, p: Socket & H.Resp) -> PEv:
437  match p:
438    case (s, +r):
439      PFin{s, r, Bool.and(H.Http.is_done(r), Bool.not(Http.closes(r))),
440        Bool.and(reused, Bool.and(Bool.not(H.Http.is_done(r)), Bool.and(Bool.not(Http.got_any(r)), idem)))}
441
442def Pool.reply(r: H.Resp) -> Result<&2, &2, KErr, Reply>:
443  Reply.of(H.Http.take(r))
444
445def Pool.fail(msg: String) -> IO(Result<&2, &2, KErr, Reply>):
446  IO.pure(Result<&2, &2, KErr, Reply>, Fail{KNet{0, msg}})
447
448# One request through the pool (fuel bounds the retry).
449def Pool.run(fuel: Nat, ev: PEv, +cfg: K.Config, +pool: Chan(Socket), +req: H.Req) ->
450  IO(Result<&2, &2, KErr, Reply>):
451  match fuel ev:
452    case 0n PTook{Ready{Some{s}}}:
453      IO.bind(Unit, Result<&2, &2, KErr, Reply>, N.Tls.close(s), u => Pool.fail("pool: out of steps"))
454    case 0n PSent{u, Done{N.Tls{s}}}:
455      IO.bind(Unit, Result<&2, &2, KErr, Reply>, N.Tls.close(s), u2 => Pool.fail("pool: out of steps"))
456    case 0n PRead{u, (s, r)}:
457      IO.bind(Unit, Result<&2, &2, KErr, Reply>, N.Tls.close(s), u2 => Pool.fail("pool: out of steps"))
458    case 0n PFin{s, r, k, t}:
459      IO.bind(Unit, Result<&2, &2, KErr, Reply>, N.Tls.close(s), u2 => Pool.fail("pool: out of steps"))
460    case 0n e:
461      Pool.fail("pool: out of steps")
462    case 1n+f PTook{Ready{Some{s}}}:
463      IO.bind(Result<&1, &1, U32 & String, N.Tls>, Result<&2, &2, KErr, Reply>,
464        Conn.send_on(cfg, s, req), r => Pool.run(f, PSent{True{}, r}, cfg, pool, req))
465    case 1n+f PTook{m}:
466      IO.bind(Result<&1, &1, U32 & String, N.Tls>, Result<&2, &2, KErr, Reply>,
467        Conn.open(cfg, req, True{}), r => Pool.run(f, PSent{False{}, r}, cfg, pool, req))
468    case 1n+f PSent{+reused, Done{N.Tls{s}}}:
469      IO.bind(Socket & H.Resp, Result<&2, &2, KErr, Reply>,
470        Conn.read(4294967295n, RChk{s, False{}, False{}, False{}}, H.Http.resp.init(), 60000),
471        p => Pool.run(f, PEv.read(reused, p), cfg, pool, req))
472    case 1n+f PSent{True{}, Fail{e}}:
473      # a stale idle connection: the request was not written, so retry
474      Pool.run(f, PTook{Wait{Unit{}}}, cfg, pool, req)
475    case 1n+f PSent{False{}, Fail{(c, m)}}:
476      IO.pure(Result<&2, &2, KErr, Reply>, Fail{KNet{c, m}})
477    case 1n+f PRead{+reused, p}:
478      Pool.run(f, PEv.fin(reused, Req.idempotent(req), p), cfg, pool, req)
479    case 1n+f PFin{s, +r, True{}, t}:
480      IO.bind(Unit, Result<&2, &2, KErr, Reply>, Pool.put(pool, s),
481        u => IO.pure(Result<&2, &2, KErr, Reply>, Pool.reply(r)))
482    case 1n+f PFin{s, r, False{}, True{}}:
483      IO.bind(Unit, Result<&2, &2, KErr, Reply>, N.Tls.close(s),
484        u => Pool.run(f, PTook{Wait{Unit{}}}, cfg, pool, req))
485    case 1n+f PFin{s, +r, False{}, False{}}:
486      IO.bind(Unit, Result<&2, &2, KErr, Reply>, N.Tls.close(s),
487        u => IO.pure(Result<&2, &2, KErr, Reply>, Pool.reply(r)))
488
489def Client.send.with(+cfg: K.Config, req: H.Req, pool: Maybe<&2, Chan(Socket)>) ->
490  IO(Result<&2, &2, KErr, Reply>):
491  match pool:
492    case Some{+p}:
493      IO.bind(Poll<&1, &1, Maybe<&1, Socket>, Unit>, Result<&2, &2, KErr, Reply>,
494        Chan.try_recv(Socket, p, 0), m => Pool.run(10n, PTook{m}, cfg, p, req))
495    case None{}:
496      IO.bind(Res(N.Tls), Result<&2, &2, KErr, Reply>, Conn.open(cfg, req, False{}), Call.read)
497
498# Sends a request and reads the whole reply, whatever its status.
499def Client.send(+cfg: K.Config, req: H.Req) -> IO(Result<&2, &2, KErr, Reply>):
500  Client.send.with(cfg, req, K.Config.pool(cfg))
501
502def Client.json.status.fix.go(+c: U32, +k: U32, r: String, m: String, zero: Bool) -> C.Status:
503  match zero:
504    case True{}:
505      C.Status{c, r, m}
506    case False{}:
507      C.Status{k, r, m}
508
509# A Status without a code (some proxies) takes the HTTP status.
510def Client.json.status.fix(+c: U32, s: C.Status) -> C.Status:
511  match s:
512    case C.Status{+k, r, m}:
513      Client.json.status.fix.go(c, k, r, m, U32.is_eq(k, 0))
514
515def Client.json.status(+c: U32, j: Result<&2, &2, String, J.Json>, ok: Bool) ->
516  Result<&2, &2, KErr, J.Json>:
517  match j ok:
518    case Done{x} True{}:
519      Done{x}
520    case Fail{m} True{}:
521      Fail{KDecode{m}}
522    case Done{+x} False{}:
523      Fail{KApi{Client.json.status.fix(c, C.Status.of(x))}}
524    case Fail{m} False{}:
525      Fail{KApi{C.Status{c, "", "non-JSON error body"}}}
526
527# A 2xx reply's body as JSON; any other status as the Status it carries.
528def Client.json.of(r: Result<&2, &2, KErr, Reply>) -> Result<&2, &2, KErr, J.Json>:
529  match r:
530    case Done{Reply{+c, body}}:
531      Client.json.status(c, J.Json.parse(body), Bool.and(U32.is_ge(c, 200), U32.is_lt(c, 300)))
532    case Fail{e}:
533      Fail{e}
534
535def Client.json(+cfg: K.Config, req: H.Req) -> IO(Result<&2, &2, KErr, J.Json>):
536  IO.bind(Result<&2, &2, KErr, Reply>, Result<&2, &2, KErr, J.Json>, Client.send(cfg, req),
537    r => IO.pure(Result<&2, &2, KErr, J.Json>, Client.json.of(r)))
538
539# Verbs (kube-client's Api<K>)
540# -----
541
542def Client.get(+cfg: K.Config, a: C.Api, name: String) -> IO(Result<&2, &2, KErr, J.Json>):
543  Client.json(cfg, C.Req.read(a, name))
544
545def Client.list(+cfg: K.Config, a: C.Api, lp: C.ListParams) -> IO(Result<&2, &2, KErr, J.Json>):
546  Client.json(cfg, C.Req.list(a, lp))
547
548def Client.create(+cfg: K.Config, a: C.Api, obj: J.Json) -> IO(Result<&2, &2, KErr, J.Json>):
549  Client.json(cfg, C.Req.create(a, obj))
550
551def Client.replace(+cfg: K.Config, a: C.Api, name: String, obj: J.Json) ->
552  IO(Result<&2, &2, KErr, J.Json>):
553  Client.json(cfg, C.Req.replace(a, name, obj))
554
555def Client.delete(+cfg: K.Config, a: C.Api, name: String) -> IO(Result<&2, &2, KErr, J.Json>):
556  Client.json(cfg, C.Req.delete(a, name))
557
558def Client.patch(+cfg: K.Config, a: C.Api, name: String, p: C.PatchKind, body: J.Json) ->
559  IO(Result<&2, &2, KErr, J.Json>):
560  Client.json(cfg, C.Req.patch(a, name, "", p, body))
561
562def Client.patch_status(+cfg: K.Config, a: C.Api, name: String, p: C.PatchKind, body: J.Json) ->
563  IO(Result<&2, &2, KErr, J.Json>):
564  Client.json(cfg, C.Req.patch(a, name, "status", p, body))
565
566# Server-side apply, kube's recommended write: the server merges our fields
567# under our manager name and tracks ownership.
568def Client.apply(+cfg: K.Config, a: C.Api, name: String, manager: String, body: J.Json) ->
569  IO(Result<&2, &2, KErr, J.Json>):
570  Client.patch(cfg, a, name, C.PApply{manager, True{}}, body)
571
572def Kc.rebased(r: Result<&2, &2, String, J.Json>, +path: String) -> Result<&2, &2, String, J.Json>:
573  match r:
574    case Done{+kc}:
575      Done{K.Kube.rebase(kc, K.Kube.dir(path))}
576    case Fail{e}:
577      Fail{path ++ ": " ++ e}
578
579# Configs
580# -------
581
582# Reads one kubeconfig, JSON or YAML (yaml.bend), with its relative paths
583# made absolute against its own directory, as client-go does.
584def Kc.parse(+t: String, +path: String, json: Bool) -> Result<&2, &2, String, J.Json>:
585  match json:
586    case True{}:
587      Kc.rebased(J.Json.parse(t), path)
588    case False{}:
589      Done{K.Kube.rebase(Y.Yaml.parse(t), K.Kube.dir(path))}
590
591def Kc.read.got(r: Res(String), +path: String) -> Result<&2, &2, String, J.Json>:
592  match r:
593    case Done{+t}:
594      Kc.parse(t, path, String.starts_with(String.trim_start(t), "{"))
595    case Fail{(c, m)}:
596      Fail{m}
597
598def Kc.read(+path: String) -> IO(Result<&2, &2, String, J.Json>):
599  IO.bind(Res(String), Result<&2, &2, String, J.Json>, Fs.read(path),
600    r => IO.pure(Result<&2, &2, String, J.Json>, Kc.read.got(r, path)))
601
602# Merges the files of a KUBECONFIG list in order; a file that does not
603# exist is skipped, as client-go skips it.
604def Kc.merge.got(acc: Maybe<&2, J.Json>, r: Result<&2, &2, String, J.Json>) -> Maybe<&2, J.Json>:
605  match acc r:
606    case Some{+a} Done{+b}:
607      Some{K.Kube.merge(a, b)}
608    case None{} Done{b}:
609      Some{b}
610    case a Fail{e}:
611      a
612
613def Kc.merge(ps: List<&2, String>, acc: Maybe<&2, J.Json>) -> IO(Maybe<&2, J.Json>):
614  match ps:
615    case Nil{}:
616      IO.pure(Maybe<&2, J.Json>, acc)
617    case Con{+p, t}:
618      IO.bind(Result<&2, &2, String, J.Json>, Maybe<&2, J.Json>, Kc.read(p),
619        r => Kc.merge(t, Kc.merge.got(acc, r)))
620
621def Config.from_kubeconfig.got(r: Maybe<&2, J.Json>, ctx: String) -> Result<&2, &2, String, K.Config>:
622  match r:
623    case Some{+kc}:
624      K.Kube.config(kc, ctx)
625    case None{}:
626      Fail{"kubeconfig: no readable file"}
627
628# A kubeconfig (one path, or a KUBECONFIG list "a:b") at a context ("" for
629# its current one). JSON and YAML are both read natively.
630def Config.from_kubeconfig(+path: String, ctx: String) -> IO(Result<&2, &2, String, K.Config>):
631  IO.bind(Maybe<&2, J.Json>, Result<&2, &2, String, K.Config>,
632    Kc.merge(String.split(path, ':'), None{}),
633    m => IO.pure(Result<&2, &2, String, K.Config>, Config.from_kubeconfig.got(m, ctx)))
634
635def Config.in_cluster.ns(host: String, port: String, ns: Res(String)) ->
636  IO(Result<&2, &2, String, K.Config>):
637  match ns:
638    case Done{n}:
639      IO.pure(Result<&2, &2, String, K.Config>, Done{K.InCluster.config(host, port, n)})
640    case Fail{e}:
641      IO.pure(Result<&2, &2, String, K.Config>, Done{K.InCluster.config(host, port, "default")})
642
643def Config.in_cluster.port(host: String, port: Res(String)) -> IO(Result<&2, &2, String, K.Config>):
644  match port:
645    case Done{p}:
646      IO.bind(Res(String), Result<&2, &2, String, K.Config>,
647        Fs.read(K.InCluster.dir() ++ "/namespace"), n => Config.in_cluster.ns(host, p, n))
648    case Fail{e}:
649      IO.pure(Result<&2, &2, String, K.Config>, Fail{"KUBERNETES_SERVICE_PORT is not set"})
650
651def Config.in_cluster.host(host: Res(String)) -> IO(Result<&2, &2, String, K.Config>):
652  match host:
653    case Done{h}:
654      IO.bind(Res(String), Result<&2, &2, String, K.Config>,
655        IO.get_env("KUBERNETES_SERVICE_PORT"), p => Config.in_cluster.port(h, p))
656    case Fail{e}:
657      IO.pure(Result<&2, &2, String, K.Config>, Fail{"not in a cluster"})
658
659# The pod's own service account (client-go's InClusterConfig).
660def Config.in_cluster() -> IO(Result<&2, &2, String, K.Config>):
661  IO.bind(Res(String), Result<&2, &2, String, K.Config>, IO.get_env("KUBERNETES_SERVICE_HOST"),
662    Config.in_cluster.host)
663
664def Config.infer.home(r: Res(String)) -> IO(Result<&2, &2, String, K.Config>):
665  match r:
666    case Done{home}:
667      Config.from_kubeconfig(home ++ "/.kube/config", "")
668    case Fail{e}:
669      IO.pure(Result<&2, &2, String, K.Config>, Fail{"no KUBECONFIG, no cluster, no HOME"})
670
671def Config.infer.pod(r: Res(String)) -> IO(Result<&2, &2, String, K.Config>):
672  match r:
673    case Done{h}:
674      Config.in_cluster()
675    case Fail{e}:
676      IO.bind(Res(String), Result<&2, &2, String, K.Config>, IO.get_env("HOME"), Config.infer.home)
677
678def Config.infer.kc(r: Res(String)) -> IO(Result<&2, &2, String, K.Config>):
679  match r:
680    case Done{path}:
681      Config.from_kubeconfig(path, "")
682    case Fail{e}:
683      IO.bind(Res(String), Result<&2, &2, String, K.Config>, IO.get_env("KUBERNETES_SERVICE_HOST"),
684        Config.infer.pod)
685
686# As kube-rs's Config::infer: $KUBECONFIG when set, else the pod's service
687# account when in a cluster, else ~/.kube/config.
688def Config.infer() -> IO(Result<&2, &2, String, K.Config>):
689  IO.bind(Res(String), Result<&2, &2, String, K.Config>, IO.get_env("KUBECONFIG"), Config.infer.kc)