carlos / platform Public

Download this file

1	package hostops
2
3	import (
4	"context"
5	"errors"
6	"fmt"
7	"github.com/carlosframework/platform/internal/alerting"
8	"os"
9	"strings"
10	"time"
11
12	"github.com/carlosframework/platform/internal/hostclaim"
13	"github.com/carlosframework/platform/internal/release"
14	)
15
16	// ListClaims is discovery only. Every action subsequently reads its own key.
17	func ListClaims(ctx context.Context, s release.Store) ([]string, error) {
18	keys, err := s.List(ctx, "control/hosts/")
19	if err != nil {
20	return nil, err
21	}
22	var hosts []string
23	for _, key := range keys {
24	tail, ok := strings.CutPrefix(key, "control/hosts/")
25	if !ok {
26	return nil, ErrIdentity
27	}
28	if strings.Contains(tail, "/") || !strings.HasSuffix(tail, ".json") {
29	continue
30	}
31	host := strings.TrimSuffix(tail, ".json")
32	if _, err := hostclaim.Key(host); err != nil {
33	return nil, err
34	}
35	hosts = append(hosts, host)
36	}
37	return hosts, nil
38	}
39
40	type Recovery struct {
41	// ScheduleLossPages triggers a separate worker after materialization.
42	ScheduleLossPages func(context.Context)
43	ReportAutomaticLosses func(context.Context, string) error
44	MoveFenceAuthority hostclaim.FenceAuthority
45	Moves func(context.Context, *MoveRecoveryGrant) error
46	// MoveLogs, when set, runs once a pass, after every claim: the console's
47	// delivery of the move log entries its move pass queued, through the
48	// pass's MoveLogWriter (move_log_writer.go).
49	MoveLogs func(context.Context, *MoveLogWriter) error
50	SweepAdmissions func(context.Context) error
51	Controller Controller
52	// These are required integrations, not no-op defaults.
53	Purge func(context.Context, hostclaim.Observed) error
54	Placement func(context.Context, hostclaim.Observed) error
55	SweepRecords func(context.Context) error
56	SweepPurges func(context.Context) error
57	}
58
59	func (r Recovery) Pass(ctx context.Context) error {
60	c := r.Controller
61	if err := c.ready(); err != nil {
62	return err
63	}
64	if r.Purge == nil || r.Placement == nil || r.SweepRecords == nil || r.SweepPurges == nil || r.SweepAdmissions == nil {
65	return ErrContract
66	}
67	hosts, err := ListClaims(ctx, c.Store)
68	if err != nil {
69	return err
70	}
71	var all []error
72	for _, host := range hosts {
73	if err := ctx.Err(); err != nil {
74	return errors.Join(append(all, err)...)
75	}
76	obs, err := hostclaim.Read(ctx, c.Store, host)
77	if err != nil {
78	all = append(all, err)
79	continue
80	}
81	if obs.Absent {
82	continue
83	}
84	if r.ReportAutomaticLosses != nil && len(obs.Claim.EvidenceRefs) > 0 {
85	if err := r.ReportAutomaticLosses(ctx, host); err != nil {
86	all = append(all, err)
87	if pageErr := alerting.Page(ctx, alerting.LevelWarning, hostclaim.ErrLossRecord.Error(), fmt.Sprintf("%s: %v", host, err)); pageErr != nil {
88	all = append(all, pageErr)
89	}
90	}
91	obs, err = hostclaim.Read(ctx, c.Store, host)
92	if err != nil {
93	all = append(all, err)
94	continue
95	}
96	}
97	if obs.Absent || obs.Claim == nil {
98	continue
99	}
100	op := obs.Claim.Op
101	// Zero time is unknown age, never old enough. Freeing empty parking is safe
102	// independently of age; this includes Plan 4's holderless sweep retries.
103	if moveRecoveryKind(op.Kind) {
104	if r.Moves == nil {
105	continue
106	}
107	err = runMoveRecovery(ctx, obs, r.Moves, r.MoveFenceAuthority)
108	} else if op.Kind == hostclaim.KindParking && len(obs.Claim.Obligations) == 0 {
109	err = c.Free(ctx, host, op.ID)
110	} else if op.Kind != "" && op.Kind != hostclaim.KindServing && !op.StartedAt.IsZero() && c.Now().After(op.StartedAt.Add(5*time.Minute)) {
111	switch op.Kind {
112	case hostclaim.KindCreating:
113	err = c.RecoverCreate(ctx, host)
114	case hostclaim.KindDeleting:
115	err = c.Retire(ctx, host, op.ID)
116	case hostclaim.KindPurging:
117	err = r.Purge(ctx, obs)
118	case hostclaim.KindPlacing:
119	err = r.Placement(ctx, obs)
120	}
121	}
122	if err != nil && !errors.Is(err, ErrPending) {
123	all = append(all, err)
124	}
125	}
126	if r.ScheduleLossPages != nil {
127	r.ScheduleLossPages(ctx)
128	}
129	if r.MoveLogs != nil {
130	if err := runMoveLogDrain(ctx, r.MoveLogs); err != nil && !errors.Is(err, ErrPending) {
131	all = append(all, err)
132	}
133	}
134	// Records have their own uncapped scan: late orphans outlive the one-hour
135	// cancellation hint; records without claims need the shared adoption path.
136	if err := r.SweepRecords(ctx); err != nil {
137	all = append(all, err)
138	}
139	// Reservations can exist without any claim (crash just after reserve).
140	if err := r.SweepPurges(ctx); err != nil {
141	all = append(all, err)
142	}
143	if err := r.SweepAdmissions(ctx); err != nil {
144	all = append(all, err)
145	}
146	return errors.Join(all...)
147	}
148
149	type App struct{ Account, Name string }
150
151	func Apps(ctx context.Context, s release.Store) ([]App, error) {
152	accounts, err := s.ListDirs(ctx, "apps/")
153	if err != nil {
154	return nil, err
155	}
156	var out []App
157	for _, prefix := range accounts {
158	account := strings.TrimSuffix(strings.TrimPrefix(prefix, "apps/"), "/")
159	if prefix != "apps/"+account+"/" {
160	return nil, ErrIdentity
161	}
162	// The retained legacy layout is apps/<app>/..., so its first segment is
163	// an app name, not necessarily an account. Skip names that are valid apps
164	// but invalid accounts before treating this as a qualified account root.
165	// A letters-only legacy name can pass both validators, so preserve its
166	// children as candidates and let the exact record and purge parsers
167	// validate them. The path alone cannot tell that case from a real account
168	// containing apps with those legal names; never infer deletion authority
169	// over retained legacy content from this prefix shape.
170	if !release.ValidAccount(account) {
171	if release.ValidApp(account) && prefix == release.LegacyAppPrefix(account) {
172	continue
173	}
174	return nil, ErrIdentity
175	}
176	apps, err := s.ListDirs(ctx, prefix)
177	if err != nil {
178	return nil, err
179	}
180	for _, ap := range apps {
181	app := strings.TrimSuffix(strings.TrimPrefix(ap, prefix), "/")
182	if _, err := PurgeKey(account, app); err != nil || ap != prefix+app+"/" {
183	return nil, ErrIdentity
184	}
185	out = append(out, App{Account: account, Name: app})
186	}
187	}
188	return out, nil
189	}
190
191	func (c Controller) SweepAllRecords(ctx context.Context) error {
192	if err := c.ready(); err != nil {
193	return err
194	}
195	apps, err := Apps(ctx, c.Store)
196	if err != nil {
197	return err
198	}
199	var all []error
200	for _, app := range apps {
201	records, err := Records(ctx, c.Store, app.Account, app.Name)
202	if err != nil {
203	all = append(all, err)
204	continue
205	}
206	for _, listed := range records {
207	record, err := ReadRecord(ctx, c.Store, listed.Holder.Key)
208	if errors.Is(err, os.ErrNotExist) {
209	continue
210	}
211	if err != nil {
212	all = append(all, err)
213	continue
214	}
215	obs, err := hostclaim.Read(ctx, c.Store, record.Holder.Host)
216	if err != nil {
217	all = append(all, err)
218	continue
219	}
220	if obs.Absent {
221	err = ErrPending
222	} else {
223	_, err = RemoveOrphan(ctx, c.Store, record)
224	}
225	if err != nil {
226	all = append(all, err)
227	}
228	}
229	}
230	return errors.Join(all...)
231	}
232
233	func SweepAllPurges(ctx context.Context, s release.Store, replay func(context.Context, App, Purge) error) error {
234	if replay == nil {
235	return ErrContract
236	}
237	apps, err := Apps(ctx, s)
238	if err != nil {
239	return err
240	}
241	var all []error
242	for _, app := range apps {
243	p, _, err := ReadPurge(ctx, s, app.Account, app.Name)
244	if errors.Is(err, os.ErrNotExist) {
245	continue
246	}
247	if err != nil {
248	all = append(all, err)
249	continue
250	}
251	if err := replay(ctx, app, p); err != nil {
252	all = append(all, err)
253	}
254	}
255	return errors.Join(all...)
256	}
257