amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package instancectl
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "os"
8 "path"
9 "path/filepath"
10 "strconv"
11 "strings"
12 "time"
13
14 "github.com/carlosframework/platform/internal/cgroup"
15 "github.com/carlosframework/platform/internal/provision"
16)
17
18// ClaimUnits runs a reserved main role through main's isolated run unit for
19// the claim root launch protocol (claim Plan 11, R10). It is the same
20// template, slice, tier and build view /start uses; what differs is that the
21// unit's /opt/carlos/app source is the exact build root verified, not the
22// host's live link, and that it refuses rather than joins a unit that is not
23// idle. Root calls it only after validating the reservation under the host
24// guard; it adds no request surface of its own.
25type ClaimUnits struct {
26 BoundHost, BoundUnit string
27 BoundAddr string
28 Systemctl func(context.Context, ...string) (string, error)
29 CgroupRoot string
30 ExeRoot string
31 UnitRoot string
32 DataRoot string
33 // Room is the disk floor check; nil skips it, as a nil Statfs does.
34 Room func() error
35 WaitBudget time.Duration
36 Poll time.Duration
37}
38
39// ClaimUnit is one start: the host, the verified build and the slice tier.
40type ClaimUnit struct {
41 Host string
42 Executable string
43 MemoryMax int64
44}
45
46// ClaimUnitStarted is what systemd reported after the start. ControlGroup is
47// relative to the cgroup root.
48type ClaimUnitStarted struct {
49 Unit, InvocationID, ControlGroup string
50 MainPID int
51}
52
53var (
54 // ErrClaimUnitBusy refuses a start while the host's run unit is active,
55 // changing state, or its cgroup still holds a process.
56 ErrClaimUnitBusy = errors.New("instance control: the claim unit is not idle")
57 // ErrClaimUnit refuses a start whose request or outcome it cannot vouch for.
58 ErrClaimUnit = errors.New("instance control: the claim unit did not start as requested")
59)
60
61// ClaimUnitControlGroup is the host's run unit cgroup, relative to the
62// cgroup root.
63func ClaimUnitControlGroup(host string) string {
64 return path.Join(parentSlices, SliceName(host), UnitName(RoleRun, host))
65}
66
67func claimTier(n int64) (cgroup.Limit, bool) {
68 switch n {
69 case cgroup.TierDefault.MemoryMax, cgroup.TierChildren.MemoryMax:
70 return cgroup.Limit{MemoryMax: n}, true
71 }
72 return cgroup.Limit{}, false
73}
74
75func (u ClaimUnits) budget() time.Duration {
76 if u.WaitBudget > 0 {
77 return u.WaitBudget
78 }
79 return 30 * time.Second
80}
81
82func (u ClaimUnits) poll() time.Duration {
83 if u.Poll > 0 {
84 return u.Poll
85 }
86 return 50 * time.Millisecond
87}
88
89func (u ClaimUnits) show(ctx context.Context, unit string, props ...string) (map[string]string, error) {
90 out, err := u.Systemctl(ctx, "show", "-p", strings.Join(props, ","), unit)
91 if err != nil {
92 return nil, fmt.Errorf("show %s: %w", unit, err)
93 }
94 return parseShow(out), nil
95}
96
97// cgroupEmpty reads the unit's cgroup.events: an absent group is empty.
98func (u ClaimUnits) cgroupEmpty(host string) error {
99 _, group, err := u.Placement(host)
100 if err != nil {
101 return err
102 }
103 events := filepath.Join(u.CgroupRoot, group, "cgroup.events")
104 b, err := os.ReadFile(events)
105 if errors.Is(err, os.ErrNotExist) {
106 return nil
107 }
108 if err != nil {
109 return err
110 }
111 if strings.Contains(string(b), "populated 0") {
112 return nil
113 }
114 return ErrClaimUnitBusy
115}
116
117func idleState(state string) bool { return state == "inactive" || state == "failed" }
118
119// Start starts host's run unit on the verified build. It refuses unless the
120// unit is idle, and stops any invocation it then cannot vouch for: one that
121// systemd did not newly create, or that runs outside the host's run cgroup.
122func (u ClaimUnits) Start(ctx context.Context, in ClaimUnit) (ClaimUnitStarted, error) {
123 return u.start(ctx, in, nil, nil, nil)
124}
125
126// StartChecked runs the root check after all preparation, immediately before
127// systemctl start. It binds root's verified input snapshot over the mutable
128// tenant launch file. The ordinary Start path keeps its existing inputs.
129func (u ClaimUnits) StartChecked(ctx context.Context, in ClaimUnit, launch Launch, check func() error) (ClaimUnitStarted, error) {
130 if check == nil {
131 return ClaimUnitStarted{}, ErrClaimUnit
132 }
133 return u.start(ctx, in, &launch, check, nil)
134}
135
136func (u ClaimUnits) start(ctx context.Context, in ClaimUnit, launch *Launch, check func() error, permit *ClaimExecPermit) (ClaimUnitStarted, error) {
137 tier, ok := claimTier(in.MemoryMax)
138 if u.Systemctl == nil || u.CgroupRoot == "" || u.ExeRoot == "" || !ok || !provision.ValidHostShape(in.Host) ||
139 !filepath.IsAbs(in.Executable) || filepath.Clean(in.Executable) != in.Executable || in.Executable == SystemBinary || strings.HasPrefix(in.Executable, LiveDir+"/") {
140 return ClaimUnitStarted{}, ErrClaimUnit
141 }
142 unit, group, err := u.Placement(in.Host)
143 if err != nil {
144 return ClaimUnitStarted{}, err
145 }
146 if u.BoundUnit != "" && (permit == nil || launch == nil || check == nil) {
147 return ClaimUnitStarted{}, ErrClaimUnit
148 }
149 before, err := u.show(ctx, unit, "ActiveState", "InvocationID")
150 if err != nil {
151 return ClaimUnitStarted{}, err
152 }
153 if !idleState(before["ActiveState"]) {
154 return ClaimUnitStarted{}, ErrClaimUnitBusy
155 }
156 if err := u.cgroupEmpty(in.Host); err != nil {
157 return ClaimUnitStarted{}, err
158 }
159 if u.Room != nil {
160 if err := u.Room(); err != nil {
161 return ClaimUnitStarted{}, err
162 }
163 }
164 if err := ensureExe(u.ExeRoot, provision.InstanceID(in.Host), in.Executable, in.Host); err != nil {
165 return ClaimUnitStarted{}, err
166 }
167 if err := sliceTier(ctx, u.Systemctl, in.Host, tier); err != nil {
168 return ClaimUnitStarted{}, err
169 }
170 var invocation string
171 refuse := func(cause error) (ClaimUnitStarted, error) {
172 cleanup, cancel := context.WithTimeout(context.WithoutCancel(ctx), u.budget())
173 defer cancel()
174 if u.BoundUnit != "" {
175 return ClaimUnitStarted{}, errors.Join(cause, u.StopInvocation(cleanup, in.Host, invocation))
176 }
177 return ClaimUnitStarted{}, errors.Join(cause, u.Stop(cleanup, in.Host))
178 }
179 if launch != nil {
180 if err := u.freezeLaunch(ctx, in.Host, *launch); err != nil {
181 return ClaimUnitStarted{}, err
182 }
183 }
184 var gate *claimExecGate
185 if permit != nil {
186 gate, err = u.prepareExecGate(ctx, in.Host, *permit)
187 if err != nil {
188 return ClaimUnitStarted{}, errors.Join(err, u.clearLaunch(ctx, in.Host))
189 }
190 defer gate.close()
191 }
192 if check != nil {
193 if err := check(); err != nil {
194 return ClaimUnitStarted{}, errors.Join(err, u.clearLaunch(ctx, in.Host))
195 }
196 }
197 if out, err := u.Systemctl(ctx, "start", unit); err != nil {
198 return refuse(fmt.Errorf("start %s: %s: %w", unit, strings.TrimSpace(out), err))
199 }
200 ready, cancel := context.WithTimeout(ctx, u.budget())
201 defer cancel()
202 for {
203 after, err := u.show(ready, unit, "InvocationID", "MainPID", "ControlGroup", "ActiveState")
204 if err != nil {
205 return refuse(err)
206 }
207 id := after["InvocationID"]
208 if idleState(after["ActiveState"]) || (id != "" && id == before["InvocationID"]) || (invocation != "" && id != invocation) {
209 return refuse(ErrClaimUnit)
210 }
211 if id != "" {
212 invocation = id
213 }
214 pid, _ := strconv.Atoi(after["MainPID"])
215 if id != "" && pid > 0 && after["ActiveState"] == "active" {
216 if after["ControlGroup"] != "/"+group {
217 return refuse(ErrClaimUnit)
218 }
219 started := ClaimUnitStarted{Unit: unit, InvocationID: id, ControlGroup: group, MainPID: pid}
220 if gate != nil {
221 if err := gate.authorize(ready, u, &started, check); err != nil {
222 cleanup, cancel := context.WithTimeout(context.WithoutCancel(ctx), u.budget())
223 defer cancel()
224 if sameErr := u.Same(cleanup, started); sameErr != nil {
225 return ClaimUnitStarted{}, errors.Join(err, u.Gone(cleanup, in.Host, started.InvocationID))
226 }
227 return refuse(err)
228 }
229 }
230 return started, nil
231 }
232 select {
233 case <-ready.Done():
234 return refuse(ready.Err())
235 case <-time.After(u.poll()):
236 }
237 }
238
239}
240
241// Stop stops host's run unit and returns once systemd reports it inactive or
242// failed and its cgroup is empty, bounded by ctx and WaitBudget.
243func (u ClaimUnits) Stop(ctx context.Context, host string) error {
244 if u.Systemctl == nil || u.CgroupRoot == "" || !provision.ValidHostShape(host) {
245 return ErrClaimUnit
246 }
247 unit, _, err := u.Placement(host)
248 if err != nil {
249 return err
250 }
251 if out, err := u.Systemctl(ctx, "stop", unit); err != nil {
252 return fmt.Errorf("stop %s: %s: %w", unit, strings.TrimSpace(out), err)
253 }
254 ctx, cancel := context.WithTimeout(ctx, u.budget())
255 defer cancel()
256 for {
257 state, err := u.show(ctx, unit, "ActiveState")
258 if err == nil && idleState(state["ActiveState"]) {
259 break
260 }
261 select {
262 case <-ctx.Done():
263 return fmt.Errorf("%s: still %q: %w", unit, state["ActiveState"], ErrClaimUnitBusy)
264 case <-time.After(u.poll()):
265 }
266 }
267 if err := WaitCgroupEmpty(ctx, u.CgroupRoot, SliceName(host), unit, u.poll()); err != nil {
268 return err
269 }
270 return u.clearLaunch(ctx, host)
271}
272
273// Gone proves an invocation of host's run unit has ended: systemd reports a
274// different invocation, or the unit idle with an empty cgroup. An empty
275// invocation asks only the second question.
276func (u ClaimUnits) Gone(ctx context.Context, host, invocation string) error {
277 if u.Systemctl == nil || u.CgroupRoot == "" || !provision.ValidHostShape(host) {
278 return ErrClaimUnit
279 }
280 unit, _, err := u.Placement(host)
281 if err != nil {
282 return err
283 }
284 state, err := u.show(ctx, unit, "ActiveState", "InvocationID")
285 if err != nil {
286 return err
287 }
288 if u.BoundUnit == "" && invocation != "" && state["InvocationID"] != "" && state["InvocationID"] != invocation {
289 return nil
290 }
291 if !idleState(state["ActiveState"]) {
292 return ErrClaimUnitBusy
293 }
294 if err := u.cgroupEmpty(host); err != nil {
295 return err
296 }
297 return u.clearLaunch(ctx, host)
298}
299
300// DiskRoom is the start floor /start applies (#214 item 1), for the claim
301// launch: statfs of dataRoot, refused below DiskFloor.
302func DiskRoom(dataRoot string) func() error {
303 return func() error {
304 avail, total, err := statfsBytes(dataRoot)
305 if err != nil {
306 return fmt.Errorf("statfs %s: %w", dataRoot, err)
307 }
308 if floor := DiskFloor(total); avail < floor {
309 return fmt.Errorf("%w: %s free, below the %s floor", ErrDiskFull, humanBytes(avail), humanBytes(floor))
310 }
311 return nil
312 }
313}
314
315// Same refuses replacement invocations and PID changes throughout readiness.
316func (u ClaimUnits) Same(ctx context.Context, s ClaimUnitStarted) error {
317 if u.Systemctl == nil || s.Unit == "" || s.InvocationID == "" || s.MainPID <= 0 {
318 return ErrClaimUnit
319 }
320 if u.BoundUnit != "" {
321 unit, group, err := u.Placement(u.BoundHost)
322 if err != nil || s.Unit != unit || s.ControlGroup != group {
323 return ErrClaimUnit
324 }
325 }
326 state, err := u.show(ctx, s.Unit, "InvocationID", "MainPID", "ControlGroup", "ActiveState")
327 if err != nil {
328 return err
329 }
330 if state["InvocationID"] != s.InvocationID || state["MainPID"] != strconv.Itoa(s.MainPID) || state["ControlGroup"] != "/"+s.ControlGroup || state["ActiveState"] != "active" {
331 return ErrClaimUnit
332 }
333 return nil
334}
335