src/helpers.bend
bend-kube-runtime@0.1.1.0 · 134 lines · 0 definitions
1import Base
2import bend-kube-runtime@0.1.1.0/src/json.bend as J
3import bend-kube-runtime@0.1.1.0/src/core.bend as C
4import bend-kube-runtime@0.1.1.0/src/config.bend as K
5import bend-kube-runtime@0.1.1.0/src/client.bend as Cl
6import bend-kube-runtime@0.1.1.0/src/queue.bend as Q
7import bend-kube-runtime@0.1.1.0/src/finalizer.bend as F
8import bend-kube-runtime@0.1.1.0/src/sys.bend as Y
9import bend-kube-runtime@0.1.1.0/src/time.bend as T
10import bend-kube-runtime@0.1.1.0/src/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
21def 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
30def 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.
40def 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
48def 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.
64def 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
72def 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
77def Event.kind(warn: Bool) -> String:
78 match warn:
79 case True{}:
80 "Warning"
81 case False{}:
82 "Normal"
83
84def 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".
92def 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>".
97def 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
107def 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
114type Stamp is Data:
115 Stamp{now: String, id: String}
116
117def Event.stamp(+w: T.Wall) -> Stamp:
118 Stamp{T.Wall.micro(w), T.Wall.id(w)}
119
120def 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.
131def 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))))