src/kube.bend

bend-kube-runtime@0.2.0.0 · 110 lines · 14 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/core.bend as C
4import bend-kube-runtime@0.2.0.0/src/config.bend as K
5import bend-kube-runtime@0.2.0.0/src/client.bend as Cl
6import bend-kube-runtime@0.2.0.0/src/queue.bend as Q
7import bend-kube-runtime@0.2.0.0/src/runtime.bend as R
8import bend-kube-runtime@0.2.0.0/src/serve.bend as V
9import bend-kube-runtime@0.2.0.0/src/leader.bend as E
10import bend-kube-runtime@0.2.0.0/src/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.
21type 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.
25def Opts.default() -> Opts:
26  Opts{8, 8080, None{}, ""}
27
28def 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
33def 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
38def 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
45def Kube.serve(+port: U32, +status: Chan(R.SMsg)) -> IO(Unit):
46  Kube.serve.go(port, status, U32.is_eq(port, 0))
47
48def 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
55def 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.
68def Kube.pool() -> IO(Chan(Socket)):
69  Chan.new(Socket, 8)
70
71def 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.
81def 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.
89def 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
99def 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
108def 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))