src/stream.bend

bend-kube-runtime@0.2.0.0 · 151 lines · 17 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/net.bend as N
6import bend-kube-runtime@0.2.0.0/src/core.bend as C
7import bend-kube-runtime@0.2.0.0/src/config.bend as K
8import bend-kube-runtime@0.2.0.0/src/client.bend as Cl
9
10# Stream
11# ======
12#
13# A GET whose body arrives over time, such as a pod's log with
14# follow=true: each cut of body bytes is folded into a state by a callback
15# as it is decoded (chunked, sized, or read until close), until the body
16# ends. A status other than 200 is a KApi error carrying the body's Status
17# message. The connection is the stream's own, never pooled.
18
19def Res(-S: Type) -> Type:
20  Result<&1, &1, Cl.KErr, S>
21
22# What the loop just saw: bytes (or a failure) from a recv, or the decoder
23# after them with the body bytes it decoded, whether the server failed or
24# the stream is over, whether the answer is 200, and whether its head came.
25type SEv is Type:
26  SGot{sock: Socket, resp: H.Resp, r: Result<&1, &1, U32 & String, List<&2, U32>>}
27  SChk{sock: Socket, resp: H.Resp, body: List<&2, U32>, err: Maybe<&2, String>, ok: Bool,
28    headed: Bool, empty: Bool, fin: Bool, code: U32}
29
30def SEv.got(p: Socket & Result<&1, &1, U32 & String, List<&2, U32>>, resp: H.Resp) -> SEv:
31  match p:
32    case (s, r):
33      SGot{s, resp, r}
34
35def Stream.code(r: H.Resp) -> U32:
36  match r:
37    case H.Resp{p, Some{H.Head{c, reason, hs}}, out}:
38      c
39    case _:
40      0
41
42def Stream.check(s: Socket, +r: H.Resp, +body: List<&2, U32>, eof: Bool) -> SEv:
43  SChk{s, r, body, H.Http.error(r), U32.is_eq(Stream.code(r), 200),
44    Bool.not(U32.is_zero(Stream.code(r))), List.is_empty(&2, U32, body),
45    Bool.or(eof, H.Http.is_done(r)), Stream.code(r)}
46
47def Stream.taken(s: Socket, eof: Bool, rb: H.Resp & List<&2, U32>) -> SEv:
48  match rb:
49    case (+r, +body):
50      Stream.check(s, r, body, eof)
51
52def Stream.ended(eof: Bool, r: H.Resp) -> H.Resp:
53  match eof:
54    case True{}:
55      H.Http.eof(r)
56    case False{}:
57      r
58
59# Feeds a cut (empty at the end of the stream) and takes the body bytes it
60# decoded.
61def Stream.fed(s: Socket, +bs: List<&2, U32>, r: H.Resp) -> SEv:
62  Stream.taken(s, List.is_empty(&2, U32, bs),
63    H.Http.take(Stream.ended(List.is_empty(&2, U32, bs), H.Http.feed(bs, r))))
64
65def Stream.message(+text: String, j: Result<&2, &2, String, J.Json>) -> String:
66  match j:
67    case Done{s}:
68      J.Json.str_or(J.Json.at(s, ["message"]), text)
69    case Fail{m}:
70      text
71
72def Stream.refused(+code: U32, +bad: List<&2, U32>) -> Cl.KErr:
73  Cl.KApi{C.Status{code, "",
74    Stream.message(String.trim(U.Utf8.decode_all(bad)), J.Json.parse(U.Utf8.decode_all(bad)))}}
75
76def Stream.follow(on: Bool) -> String:
77  match on:
78    case True{}:
79      "&follow=true"
80    case False{}:
81      ""
82
83def Stream.recv(s: Socket, +ms: U32) ->
84  IO(Socket & Result<&1, &1, U32 & String, List<&2, U32>>):
85  N.Tls.recv(s, 65536, ms)
86
87# Reads until the body ends, folding the body into st; bad holds the body
88# of a refusal. Answers the socket and the outcome.
89def Stream.loop(~S: Type, ~on: S -> List<&2, U32> -> IO(S), fuel: Nat, ev: SEv, st: S,
90  +bad: List<&2, U32>, +ms: U32) -> IO(Socket & Res(S)):
91  match fuel ev:
92    case 0n SGot{s, rp, r}:
93      IO.pure(Socket & Res(S), (s, Fail{Cl.KNet{0, "stream: too long"}}))
94    case 0n SChk{s, r, b, er, ok, h, em, fin, c}:
95      IO.pure(Socket & Res(S), (s, Fail{Cl.KNet{0, "stream: too long"}}))
96    case 1n+f SGot{s, rp, Done{+bs}}:
97      Stream.loop(~S, ~on, f, Stream.fed(s, bs, rp), st, bad, ms)
98    case 1n+f SGot{s, rp, Fail{(c, m)}}:
99      IO.pure(Socket & Res(S), (s, Fail{Cl.KNet{c, "stream: recv: " ++ m}}))
100    case 1n+f SChk{s, r, b, Some{m}, ok, h, em, fin, c}:
101      IO.pure(Socket & Res(S), (s, Fail{Cl.KNet{0, "stream: " ++ m}}))
102    case 1n+f SChk{s, +r, b, None{}, True{}, h, False{}, fin, +c}:
103      IO.bind(S, Socket & Res(S), on(st, b),
104        st2 => Stream.loop(~S, ~on, f, SChk{s, r, Nil{}, None{}, True{}, True{}, True{}, fin, c},
105          st2, bad, ms))
106    case 1n+f SChk{s, r, b, None{}, True{}, h, True{}, True{}, c}:
107      IO.pure(Socket & Res(S), (s, Done{st}))
108    case 1n+f SChk{s, +r, b, None{}, True{}, h, True{}, False{}, c}:
109      IO.bind(Socket & Result<&1, &1, U32 & String, List<&2, U32>>, Socket & Res(S),
110        Stream.recv(s, ms), p => Stream.loop(~S, ~on, f, SEv.got(p, r), st, bad, ms))
111    case 1n+f SChk{s, r, b, None{}, False{}, True{}, em, True{}, +c}:
112      IO.pure(Socket & Res(S), (s, Fail{Stream.refused(c, List.append(&2, U32, bad, b))}))
113    case 1n+f SChk{s, +r, b, None{}, False{}, True{}, em, False{}, c}:
114      IO.bind(Socket & Result<&1, &1, U32 & String, List<&2, U32>>, Socket & Res(S),
115        Stream.recv(s, ms),
116        p => Stream.loop(~S, ~on, f, SEv.got(p, r), st, List.append(&2, U32, bad, b), ms))
117    case 1n+f SChk{s, r, b, None{}, False{}, False{}, em, True{}, c}:
118      IO.pure(Socket & Res(S), (s, Fail{Cl.KNet{0, "stream: connection closed before a response"}}))
119    case 1n+f SChk{s, +r, b, None{}, False{}, False{}, em, False{}, c}:
120      IO.bind(Socket & Result<&1, &1, U32 & String, List<&2, U32>>, Socket & Res(S),
121        Stream.recv(s, ms), p => Stream.loop(~S, ~on, f, SEv.got(p, r), st, bad, ms))
122
123def Stream.close(~S: Type, p: Socket & Res(S)) -> IO(Res(S)):
124  match p:
125    case (s, r):
126      IO.bind(Unit, Res(S), N.Tls.close(s), u => IO.pure(Res(S), r))
127
128def Stream.opened(~S: Type, ~on: S -> List<&2, U32> -> IO(S), st: S,
129  t: Result<&1, &1, U32 & String, N.Tls>, +ms: U32) -> IO(Res(S)):
130  match t:
131    case Done{N.Tls{s}}:
132      IO.bind(Socket & Res(S), Res(S),
133        Stream.loop(~S, ~on, 4294967295n,
134          SChk{s, H.Http.resp.init(), Nil{}, None{}, False{}, False{}, True{}, False{}, 0}, st,
135          Nil{}, ms), p => Stream.close(~S, p))
136    case Fail{(c, m)}:
137      IO.pure(Res(S), Fail{Cl.KNet{c, m}})
138
139# GETs path (with its query) and folds each cut of the body into st with
140# on, in order, until the body ends: Done with the last state. ms bounds
141# each wait for bytes (0: none; a followed log may stay quiet for long).
142def Stream.fold(~S: Type, ~on: S -> List<&2, U32> -> IO(S), st: S, +cfg: K.Config, path: String,
143  +ms: U32) -> IO(Res(S)):
144  IO.bind(Result<&1, &1, U32 & String, N.Tls>, Res(S),
145    Cl.Conn.open(cfg, H.Req{"GET", path, Nil{}, ""}, False{}), t => Stream.opened(~S, ~on, st, t, ms))
146
147# The log of a pod's container, followed while it runs (follow=true): the
148# stream ends when the container does.
149def Stream.log_path(ns: String, pod: String, container: String, follow: Bool) -> String:
150  "/api/v1/namespaces/" ++ H.Http.encode(ns) ++ "/pods/" ++ H.Http.encode(pod) ++
151    "/log?container=" ++ H.Http.encode(container) ++ Stream.follow(follow)