carlos / platform Public

Download this file

1	package host
2
3	import (
4	"context"
5	"errors"
6	"github.com/carlosframework/platform/internal/park"
7	"path/filepath"
8	"time"
9
10	"github.com/carlosframework/platform/internal/activator"
11	"github.com/carlosframework/platform/internal/cgroup"
12	"github.com/carlosframework/platform/internal/hostbox"
13	"github.com/carlosframework/platform/internal/hostclaim"
14	execlease "github.com/carlosframework/platform/internal/hostexec"
15	"github.com/carlosframework/platform/internal/hostguard"
16	"github.com/carlosframework/platform/internal/instancectl"
17	"github.com/carlosframework/platform/internal/provision"
18	)
19
20	// Installed only by root construction. An operation adapter must read its
21	// durable per-kind journal and validate the exact observed op, phase, holder
22	// generation, stamp and session. Agent CurrentAttempt caches are not readers.
23	// A missing adapter refuses that kind. No operation adapter ships in rev 2.
24	type claimLaunchAttemptReader interface {
25	ReadAttempt(context.Context, hostclaim.Observed) (hostclaim.WakeRetryAttempt, error)
26	}
27
28	func (h *ClaimLaunchRoot) authoritativeAttempt(ctx context.Context, c *hostclaim.Claim) (hostclaim.WakeRetryAttempt, error) {
29	if c.Op.Kind == hostclaim.KindServing {
30	retry := c.Holder.WakeRetry
31	if retry == nil || retry.State != "attempting" || retry.EpisodeID == "" || retry.AttemptsReserved < 1 || retry.AttemptsReserved > hostclaim.WakeRetryLimit || retry.Attempt == nil || !retry.NotBefore.IsZero() {
32	return hostclaim.WakeRetryAttempt{}, hostbox.ErrChanged
33	}
34	return *retry.Attempt, nil
35	}
36	switch c.Op.Kind {
37	case hostclaim.KindLanding, hostclaim.KindRelocating:
38	if reader := h.Attempts[c.Op.Kind]; reader != nil {
39	return reader.ReadAttempt(ctx, hostclaim.Observed{Host: c.Host, Claim: c})
40	}
41	}
42	return hostclaim.WakeRetryAttempt{}, hostbox.ErrUnsupported
43	}
44
45	func (h *ClaimLaunchRoot) launchSpec(q hostbox.RoleLaunchRequest) (instancectl.Launch, error) {
46	if q.Spec == nil || q.Reservation == nil {
47	return instancectl.Launch{}, hostbox.ErrRoleLaunchShape
48	}
49	t := q.Reservation.Target
50	if launch, bound, err := h.boundLaunchSpec(q); bound || err != nil {
51	return launch, err
52	}
53	paths, err := claimIsolatedPaths(map[string]string{"data": h.DataRoot}, t.Host, ClaimLaunchInputs{DBPath: provision.NestedDBPath(h.DataRoot, t.Host), Limit: cgroup.Limit{MemoryMax: q.MemoryMax}})
54	if err != nil {
55	return instancectl.Launch{}, err
56	}
57	if h.runRoot != "" {
58	paths.RunDir = filepath.Join(h.runRoot, provision.InstanceID(t.Host))
59	}
60	argv, _, _, err := activator.IsolatedArgv(t.Host, paths.LaunchDir, paths.RunDir, q.Spec.Args, q.Spec.Env)
61	if err != nil {
62	return instancectl.Launch{}, err
63	}
64	return instancectl.Launch{Schema: instancectl.LaunchSchema, Host: t.Host, Argv: argv, Env: append([]string(nil), q.Spec.Env...)}, nil
65	}
66
67	// authorize is always called under the host guard, including inside
68	// StartChecked after namespace/tier preparation and immediately before start.
69	func (h *ClaimLaunchRoot) authorize(ctx context.Context, peer StopPeer, q hostbox.RoleLaunchRequest, c *hostclaim.Claim) error {
70	t, r := q.Reservation.Target, *q.Reservation
71	if err := h.checkUnitBinding(ctx, t, false); err != nil {
72	return err
73	}
74	if err := ctx.Err(); err != nil {
75	return err
76	}
77	actor, err := h.Actor(ctx)
78	if err != nil {
79	return err
80	}
81	if !actor.Eligible || actor.Quarantined || actor.Box != t.Box || actor.Enrollment != t.Enrollment || actor.Region != t.Region || actor.Binding == "" || actor.Binding != t.DirectoryRevision {
82	return hostbox.ErrChanged
83	}
84	if c == nil || c.Host != t.Host {
85	return hostbox.ErrChanged
86	}
87	if err := claimLaunchAuthority(c, t, actor.Box); err != nil {
88	return err
89	}
90	a, err := h.authoritativeAttempt(ctx, c)
91	if err != nil {
92	return err
93	}
94	if a.ID == "" || a.InvocationID == "" || a.LaunchInputs == "" || r.ID != a.ID+":main" || r.InvocationID != a.InvocationID || r.Inputs != a.LaunchInputs || a.Session != t.Session || a.Box != t.Box || a.Enrollment != t.Enrollment || a.BootID != peer.BootID || a.ControllerID != peer.ControllerID {
95	return hostbox.ErrChanged
96	}
97	if q.Spec == nil || q.Spec.Identity.Config == "" || q.Spec.Identity.Release == "" || !claimHexDigest(q.Spec.Digest) {
98	return hostbox.ErrRoleLaunchShape
99	}
100	digest, err := q.Spec.Hash()
101	if err != nil || digest != a.LaunchInputs {
102	return hostbox.ErrChanged
103	}
104	if _, err := h.launchSpec(q); err != nil {
105	return err
106	}
107	if err := h.ValidateExecutable(t.Host, q.Executable, q.Spec.Digest); err != nil {
108	return err
109	}
110	if err := h.Kernel.VerifyDigest(ctx, q.Executable, r.Executable, q.Spec.Digest); err != nil {
111	return err
112	}
113	selection := hostbox.Selection{HomeRegion: actor.HomeRegion, FleetHomeRegion: actor.FleetHomeRegion, Host: t.Host, Action: claimSpawnAction(c), Box: t.Box, Region: t.Region, Captured: hostguard.Captured{Incarnation: t.Incarnation, Stamp: t.Stamp}, Kind: c.Op.Kind, Op: c.Op.ID, Phase: c.Op.Phase, Generation: c.Holder.Generation, Session: t.Session}
114	if err := h.Local.Check(ctx, actor, c, selection); err != nil {
115	return err
116	}
117	state, err := h.readExecution(ctx, c)
118	if err != nil {
119	return err
120	}
121	if c.Op.Kind != hostclaim.KindServing {
122	if state.Authority.Box != t.Box || state.Authority.Enrollment != t.Enrollment || state.Peer.KernelBootID != peer.BootID || state.Peer.PID != peer.PID || state.Peer.Start != peer.ProcessStart {
123	return execlease.ErrLost
124	}
125	}
126	now, boot, err := h.Clock.Now()
127	if err != nil {
128	return err
129	}
130	if c.Op.Kind != hostclaim.KindServing && boot >= state.SafeUntil {
131	return execlease.ErrLost
132	}
133	if until := park.QuietUntil(c.QuietAfter, actor.HomeRegion, t.Region); !until.IsZero() && !now.After(until) {
134	return hostbox.ErrChanged
135	}
136	return nil
137	}
138
139	// waitReady observes one start, never reserves or starts another attempt.
140	// Type=simple returns before namespace setup and the trampoline exec.
141	func (h *ClaimLaunchRoot) waitReady(ctx context.Context, host string, s instancectl.ClaimUnitStarted, want hostbox.LaunchFile) (ClaimLaunchObserved, error) {
142	budget, poll := h.ReadyBudget, h.ReadyPoll
143	if budget <= 0 {
144	budget = 5 * time.Second
145	}
146	if poll <= 0 {
147	poll = 25 * time.Millisecond
148	}
149	ctx, cancel := context.WithTimeout(ctx, budget)
150	defer cancel()
151	var pinned, tenantStart string
152	var tenantPID int
153	for {
154	if err := ctx.Err(); err != nil {
155	return ClaimLaunchObserved{}, err
156	}
157	if err := h.Units.Same(ctx, s); err != nil {
158	return ClaimLaunchObserved{}, err
159	}
160	got, err := h.Kernel.Observe(s)
161	if got.ProcessStart == "" || (pinned != "" && pinned != got.ProcessStart) {
162	return ClaimLaunchObserved{}, hostbox.ErrChanged
163	}
164	pinned = got.ProcessStart
165	if (got.TenantPID == 0) != (got.TenantStart == "") || (tenantPID != 0 && (tenantPID != got.TenantPID || tenantStart != got.TenantStart)) {
166	return ClaimLaunchObserved{}, hostbox.ErrChanged
167	}
168	if got.TenantPID != 0 {
169	tenantPID, tenantStart = got.TenantPID, got.TenantStart
170	}
171	if err != nil && !errors.Is(err, errClaimLaunchPending) {
172	return ClaimLaunchObserved{}, err
173	}
174	if err == nil && got.TenantPID > 0 && got.TenantStart != "" && got.Executable == want && got.Running == want {
175	if err := h.Units.Same(ctx, s); err != nil {
176	return ClaimLaunchObserved{}, err
177	}
178	if h.boundReady == nil || h.boundReady(ctx, host, s) == nil {
179	if err := h.Units.Same(ctx, s); err != nil {
180	return ClaimLaunchObserved{}, err
181	}
182	return got, nil
183	}
184	}
185	select {
186	case <-ctx.Done():
187	return ClaimLaunchObserved{}, ctx.Err()
188	case <-time.After(poll):
189	}
190	}
191	}
192
193	var errClaimLaunchPending = errors.New("claim launch: namespace or exec pending")
194