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 ./finalizer.bend as F
8	import ./sys.bend as Y
9	import ./time.bend as T
10	import ./log.bend as L
11	
12	# Helpers
13	# =======
14	#
15	# What a reconciler reaches for: kube-runtime's finalizer() and an Event
16	# recorder (client-go's record.EventRecorder, on events.k8s.io/v1).
17	
18	# Finalizers
19	# ----------
20	
21	def Fin.added(r: Result<&2, &2, Cl.KErr, J.Json>) -> IO(Q.Action):
22	  match r:
23	    case Done{o}:
24	      # the patch itself triggers the next reconcile
25	      IO.pure(Q.Action, Q.AwaitChange{})
26	    case Fail{+e}:
27	      IO.bind(Unit, Q.Action, L.Log.warn("adding finalizer failed", [L.Log.kv("error", Cl.KErr.show(e))]),
28	        u => IO.pure(Q.Action, Q.Retry{Cl.KErr.show(e)}))
29	
30	def Fin.removed(a: Q.Action, r: Result<&2, &2, Cl.KErr, J.Json>) -> IO(Q.Action):
31	  match r:
32	    case Done{o}:
33	      IO.pure(Q.Action, a)
34	    case Fail{+e}:
35	      IO.bind(Unit, Q.Action, L.Log.warn("removing finalizer failed", [L.Log.kv("error", Cl.KErr.show(e))]),
36	        u => IO.pure(Q.Action, Q.Retry{Cl.KErr.show(e)}))
37	
38	# After a cleanup: a failed one keeps the finalizer (and retries); a
39	# finished one removes it, through a patch that tests it is still ours.
40	def Fin.cleaned(+cfg: K.Config, api: C.Api, name: String, patch: J.Json, a: Q.Action) -> IO(Q.Action):
41	  match a:
42	    case Q.Retry{m}:
43	      IO.pure(Q.Action, Q.Retry{m})
44	    case +x:
45	      IO.bind(Result<&2, &2, Cl.KErr, J.Json>, Q.Action, Cl.Client.patch(cfg, api, name, C.PJson{}, patch),
46	        r => Fin.removed(x, r))
47	
48	def Fin.run.go(~apply: J.Json -> K.Config -> IO(Q.Action), ~cleanup: J.Json -> K.Config -> IO(Q.Action),
49	  +cfg: K.Config, +api: C.Api, +o: J.Json, s: F.FState) -> IO(Q.Action):
50	  match s:
51	    case F.FApply{}:
52	      apply(o, cfg)
53	    case F.FAdd{p}:
54	      IO.bind(Result<&2, &2, Cl.KErr, J.Json>, Q.Action, Cl.Client.patch(cfg, api, C.Obj.name(o), C.PJson{}, p),
55	        Fin.added)
56	    case F.FCleanup{p}:
57	      IO.bind(Q.Action, Q.Action, cleanup(o, cfg), a => Fin.cleaned(cfg, api, C.Obj.name(o), p, a))
58	    case F.FDone{}:
59	      IO.pure(Q.Action, Q.AwaitChange{})
60	
61	# Reconciles o under the finalizer called name (kube-runtime's finalizer):
62	# apply while it lives, cleanup once it is deleted. Both must be
63	# idempotent; a cleanup that answers Retry keeps the object alive.
64	def Finalizer.run(~apply: J.Json -> K.Config -> IO(Q.Action),
65	  ~cleanup: J.Json -> K.Config -> IO(Q.Action), +cfg: K.Config, +api: C.Api, name: String,
66	  +o: J.Json) -> IO(Q.Action):
67	  Fin.run.go(~apply, ~cleanup, cfg, api, o, F.Fin.state(o, name))
68	
69	# Events
70	# ------
71	
72	def Event.regarding(+o: J.Json, +r: C.Res) -> J.Json:
73	  J.Json.obj([J.KV{"apiVersion", J.JStr{C.Res.api_version(r)}}, J.KV{"kind", J.JStr{C.Res.kind(r)}},
74	    J.KV{"name", J.JStr{C.Obj.name(o)}}, J.KV{"namespace", J.JStr{C.Obj.namespace(o)}},
75	    J.KV{"uid", J.JStr{C.Obj.uid(o)}}, J.KV{"resourceVersion", J.JStr{C.Obj.rv(o)}}])
76	
77	def Event.kind(warn: Bool) -> String:
78	  match warn:
79	    case True{}:
80	      "Warning"
81	    case False{}:
82	      "Normal"
83	
84	def Event.ns.go(ns: String, empty: Bool) -> String:
85	  match empty:
86	    case True{}:
87	      "default"
88	    case False{}:
89	      ns
90	
91	# Cluster-scoped objects record their events in "default".
92	def Event.ns(+ns: String) -> String:
93	  Event.ns.go(ns, String.is_empty(ns))
94	
95	# The Event is named "<object>.<unix microseconds>", as client-go names
96	# them "<object>.<unix nanoseconds in hex>".
97	def Event.body(+o: J.Json, +r: C.Res, warn: Bool, reason: String, action: String, note: String,
98	  +controller: String, +now: String, id: String) -> J.Json:
99	  J.Json.obj([J.KV{"apiVersion", J.JStr{"events.k8s.io/v1"}}, J.KV{"kind", J.JStr{"Event"}},
100	    J.KV{"metadata", J.Json.obj([J.KV{"name", J.JStr{C.Obj.name(o) ++ "." ++ id}},
101	      J.KV{"namespace", J.JStr{Event.ns(C.Obj.namespace(o))}}])},
102	    J.KV{"eventTime", J.JStr{now}}, J.KV{"reportingController", J.JStr{controller}},
103	    J.KV{"reportingInstance", J.JStr{controller}}, J.KV{"action", J.JStr{action}},
104	    J.KV{"reason", J.JStr{reason}}, J.KV{"type", J.JStr{Event.kind(warn)}}, J.KV{"note", J.JStr{note}},
105	    J.KV{"regarding", Event.regarding(o, r)}])
106	
107	def Event.sent(r: Result<&2, &2, Cl.KErr, J.Json>) -> IO(Unit):
108	  match r:
109	    case Done{e}:
110	      IO.pure(Unit, Unit{})
111	    case Fail{e}:
112	      L.Log.warn("publishing event failed", [L.Log.kv("error", Cl.KErr.show(e))])
113	
114	type Stamp is Data:
115	  Stamp{now: String, id: String}
116	
117	def Event.stamp(+w: T.Wall) -> Stamp:
118	  Stamp{T.Wall.micro(w), T.Wall.id(w)}
119	
120	def Event.post(+cfg: K.Config, +o: J.Json, +r: C.Res, warn: Bool, reason: String, action: String,
121	  note: String, controller: String, st: Stamp) -> IO(Unit):
122	  match st:
123	    case Stamp{now, id}:
124	      IO.bind(Result<&2, &2, Cl.KErr, J.Json>, Unit,
125	        Cl.Client.create(cfg, C.Api{C.Res.events(), C.InNs{Event.ns(C.Obj.namespace(o))}},
126	          Event.body(o, r, warn, reason, action, note, controller, now, id)),
127	        Event.sent)
128	
129	# Records an Event about o (of resource r). A failure is logged, never
130	# raised: events are best effort, as client-go's are.
131	def Event.publish(+cfg: K.Config, +o: J.Json, +r: C.Res, warn: Bool, reason: String, action: String,
132	  note: String, controller: String) -> IO(Unit):
133	  IO.bind(U32 & (U32 & U32), Unit, Y.Sys.wall(),
134	    w => Event.post(cfg, o, r, warn, reason, action, note, controller, Event.stamp(T.Wall.of(w))))