1	import Base
2	import ./runtime.bend as R
3	import ./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	
17	def Serve.Res(-A: Type) -> Type:
18	  Result<&1, &1, U32 & String, A>
19	
20	def Serve.size(s: String) -> U32:
21	  U.Utf8.size(s)
22	
23	def 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	
32	def Serve.reason(+code: U32) -> String:
33	  Serve.reason.go(U32.is_eq(code, 200), U32.is_eq(code, 404))
34	
35	def 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	
40	def Serve.first_line.go(ls: List<&2, String>) -> String:
41	  match ls:
42	    case Con{l, t}:
43	      l
44	    case Nil{}:
45	      ""
46	
47	def Serve.first_line(s: String) -> String:
48	  Serve.first_line.go(String.lines(s))
49	
50	def 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.
58	def Serve.path(req: String) -> String:
59	  Serve.path.go(String.split(String.trim(Serve.first_line(req)), ' '))
60	
61	def 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).
70	def 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	
78	def 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	
85	def 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.
95	def Serve.ready(+ss: List<&2, R.Snap>) -> String:
96	  Serve.ready.go(List.is_empty(&2, R.Snap, ss), Serve.all_ready(ss))
97	
98	def Serve.bool(b: Bool) -> String:
99	  match b:
100	    case True{}:
101	      "1"
102	    case False{}:
103	      "0"
104	
105	def Serve.head(+name: String, kind: String, help: String) -> String:
106	  "# HELP " ++ name ++ " " ++ help ++ "\n# TYPE " ++ name ++ " " ++ kind ++ "\n"
107	
108	def Serve.snap.name(s: R.Snap) -> String:
109	  match s:
110	    case R.Snap{n, r, a, b, c, d, e}:
111	      n
112	
113	def 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	
129	def 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).
135	def 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.
141	def 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	
149	def 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.
154	def 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	
164	def 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.
168	def 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	
179	def 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	
185	def 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	
189	def 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	
197	def 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	
201	def 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	
208	type AEv is Type:
209	  AGot{l: Listener, r: Result<&1, &1, U32 & String, Socket>}
210	
211	def 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.
217	def 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	
226	def 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.
235	def 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))