1	import Base
2	import ./json.bend as J
3	import ./core.bend as C
4	import ./config.bend as K
5	import ./client.bend as Cl
6	import ./queue.bend as Q
7	import ./runtime.bend as R
8	import ./serve.bend as V
9	import ./leader.bend as E
10	import ./log.bend as L
11	
12	# Kube
13	# ====
14	#
15	# The entry point a controller binary calls (the kube crate's facade):
16	# infer the config, serve health and metrics, elect a leader when asked,
17	# and run the controller.
18	
19	# max: reconciles at once; port: health and metrics (0: off); leader: the
20	# Lease to hold while running; label: the primary watch's label selector.
21	type Opts is Data:
22	  Opts{max: U32, port: U32, leader: Maybe<&2, E.LOpts>, label: String}
23	
24	# 8 reconciles at once, health and metrics on :8080, no leader election.
25	def Opts.default() -> Opts:
26	  Opts{8, 8080, None{}, ""}
27	
28	def Opts.with_leader(o: Opts, ns: String, name: String, id: String) -> Opts:
29	  match o:
30	    case Opts{m, p, l, s}:
31	      Opts{m, p, Some{E.LOpts.default(ns, name, id)}, s}
32	
33	def Opts.with_label(o: Opts, label: String) -> Opts:
34	  match o:
35	    case Opts{m, p, l, s}:
36	      Opts{m, p, l, label}
37	
38	def Kube.serve.go(+port: U32, +status: Chan(R.SMsg), off: Bool) -> IO(Unit):
39	  match off:
40	    case True{}:
41	      IO.pure(Unit, Unit{})
42	    case False{}:
43	      IO.spawn(Unit, V.Serve.run(port, status))
44	
45	def Kube.serve(+port: U32, +status: Chan(R.SMsg)) -> IO(Unit):
46	  Kube.serve.go(port, status, U32.is_eq(port, 0))
47	
48	def Kube.elect(+cfg: K.Config, l: Maybe<&2, E.LOpts>, work: IO(Unit)) -> IO(Unit):
49	  match l:
50	    case Some{+lo}:
51	      E.Leader.elect(cfg, lo, work)
52	    case None{}:
53	      work
54	
55	def Kube.start(~rec: J.Json -> K.Config -> IO(Q.Action), +cfg: K.Config, +primary: C.Api,
56	  owned: List<&2, C.Api>, o: Opts) -> IO(Unit):
57	  match o:
58	    case Opts{+m, +p, l, sel}:
59	      IO.bind(Chan(R.Msg), Unit, R.Controller.inbox(),
60	        +inbox => IO.bind(Chan(R.SMsg), Unit, R.Keeper.start(),
61	          +status => IO.bind(Unit, Unit, Kube.serve(p, status),
62	            u => IO.bind(Unit, Unit, L.Log.info("starting controller",
63	              [L.Log.kv("kind", R.Api.kind(primary)), L.Log.kv("server", K.Config.host(cfg))]),
64	              u2 => Kube.elect(cfg, l,
65	                R.Controller.start(~rec, cfg, primary, sel, owned, m, inbox, status))))))
66	
67	# Up to 8 idle keep-alive connections shared by every request.
68	def Kube.pool() -> IO(Chan(Socket)):
69	  Chan.new(Socket, 8)
70	
71	def Kube.configured(~rec: J.Json -> K.Config -> IO(Q.Action), c: Result<&2, &2, String, K.Config>,
72	  primary: C.Api, owned: List<&2, C.Api>, o: Opts) -> IO(Unit):
73	  match c:
74	    case Done{cfg}:
75	      IO.bind(Chan(Socket), Unit, Kube.pool(),
76	        p => Kube.start(~rec, K.Config.with_pool(cfg, p), primary, owned, o))
77	    case Fail{e}:
78	      IO.die(Unit, 1, "bend-kube: no config: " ++ e)
79	
80	# Runs a controller for primary, owning owned, reconciling with rec.
81	def Kube.run(~rec: J.Json -> K.Config -> IO(Q.Action), primary: C.Api, owned: List<&2, C.Api>,
82	  o: Opts) -> IO(Unit):
83	  IO.bind(Result<&2, &2, String, K.Config>, Unit, Cl.Config.infer(),
84	    c => Kube.configured(~rec, c, primary, owned, o))
85	
86	# Several controllers in one process, sharing their watches (Manager):
87	# rec(i, obj, cfg) reconciles for the i-th controller of cs. Opts.label is
88	# unused here (each Ctrl has its own); max likewise.
89	def Kube.start_all(~rec: U32 -> J.Json -> K.Config -> IO(Q.Action), +cfg: K.Config, +cs: List<&2, R.Ctrl>,
90	  o: Opts) -> IO(Unit):
91	  match o:
92	    case Opts{m, +p, l, sel}:
93	      IO.bind(Chan(R.SMsg), Unit, R.Keeper.start(),
94	        +status => IO.bind(Unit, Unit, Kube.serve(p, status),
95	          u => IO.bind(Unit, Unit, L.Log.info("starting controllers",
96	            [L.Log.kv("count", Nat.show(List.length(&2, R.Ctrl, cs))), L.Log.kv("server", K.Config.host(cfg))]),
97	            u2 => Kube.elect(cfg, l, R.Manager.start(~rec, cfg, cs, status)))))
98	
99	def Kube.configured_all(~rec: U32 -> J.Json -> K.Config -> IO(Q.Action),
100	  c: Result<&2, &2, String, K.Config>, cs: List<&2, R.Ctrl>, o: Opts) -> IO(Unit):
101	  match c:
102	    case Done{cfg}:
103	      IO.bind(Chan(Socket), Unit, Kube.pool(),
104	        p => Kube.start_all(~rec, K.Config.with_pool(cfg, p), cs, o))
105	    case Fail{e}:
106	      IO.die(Unit, 1, "bend-kube: no config: " ++ e)
107	
108	def Kube.run_all(~rec: U32 -> J.Json -> K.Config -> IO(Q.Action), cs: List<&2, R.Ctrl>, o: Opts) -> IO(Unit):
109	  IO.bind(Result<&2, &2, String, K.Config>, Unit, Cl.Config.infer(),
110	    c => Kube.configured_all(~rec, c, cs, o))