amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package host
2
3import (
4 "context"
5 "crypto/ed25519"
6 "errors"
7 "log"
8 "path/filepath"
9 "strings"
10 "sync"
11 "time"
12
13 "github.com/carlosframework/platform/internal/activator"
14 "github.com/carlosframework/platform/internal/appconfig"
15 "github.com/carlosframework/platform/internal/boxchoice"
16 "github.com/carlosframework/platform/internal/boxvolumes"
17 "github.com/carlosframework/platform/internal/edgesfile"
18 "github.com/carlosframework/platform/internal/genselect"
19 "github.com/carlosframework/platform/internal/hostactors"
20 "github.com/carlosframework/platform/internal/hostbox"
21 "github.com/carlosframework/platform/internal/hostclaim"
22 "github.com/carlosframework/platform/internal/hostfence"
23 "github.com/carlosframework/platform/internal/hostfloor"
24 "github.com/carlosframework/platform/internal/hostops"
25 "github.com/carlosframework/platform/internal/hostrelocate"
26 "github.com/carlosframework/platform/internal/hostwrite"
27 "github.com/carlosframework/platform/internal/lease"
28 "github.com/carlosframework/platform/internal/provision"
29 "github.com/carlosframework/platform/internal/registry"
30 "github.com/carlosframework/platform/internal/release"
31 "github.com/carlosframework/platform/internal/termfence"
32)
33
34// RelocationRuntime is one agent's relocation: its worker, participant and
35// effects, built once the claim assembly is bound. It replays relocations;
36// it starts none (Worker.Start is the operator's and the placer's, not
37// this pass's).
38type RelocationRuntime struct {
39 Worker relocationStepper
40 Participant relocationStepper
41 Box string // this box: its alerts, and the completions it finishes
42 // Convert is J3 for a landing claim (convertFencedLanding); Finish is
43 // RelocationEffects.Finish for a completion this box has not yet
44 // marked discharged (ruling 80).
45 Convert func(context.Context, release.Store, hostclaim.Observed) error
46 Finish func(ctx context.Context, host, op string) error
47 // Hosts, when set, is the hosts a pass reads instead of every claim
48 // in the bucket. The completion-only runtime's (ruling 82) are this
49 // box's registry rows and every host its own op journals name
50 // (relocate/ops/*/{probe-fence,attempts}.json), so a completion whose
51 // row is gone is still visited and its refusal said (Task 10 review
52 // M1). An error beside a non-nil list is said, and the listed hosts
53 // are still read.
54 Hosts func() ([]string, error)
55 // Self, when Box is empty, answers this box: the completion-only
56 // runtime has no roster read of its own, so it asks the bound
57 // assembly's verified stop identity. It is asked at most once a pass,
58 // and only once a serving claim with an undischarged completion is
59 // found (review M2), so a box with nothing to finish makes no identity
60 // read. A pass that cannot answer it finishes nothing (carry 8: a
61 // completion is finished only on the box it serves on).
62 Self func(context.Context) (string, error)
63 DataDir string // the managed data root: relocate/ journals (N9)
64 alerts relocationAlerts
65 // upgrade, when set, builds the full pass this completion-only runtime
66 // stands in for while region HA is on. The first pass, bounded like
67 // every pass, tries it (review M9); a pass whose build fails tries it
68 // again, until it builds or answers that the owners are not bound. A
69 // read that failed once is never taken as "no relocation on this box".
70 upgrade func(context.Context) (*RelocationRuntime, error)
71 // config is the assembly config the full pass was built with
72 // (NewRelocationRuntime); zero for the completion-only runtime. No
73 // production code reads it: it is the seam Task 10's review M7 test
74 // asserts the held-stage evidence (ReleasePub, Ratchet) through
75 // (relocatepass_carries_test.go), kept for that reading alone.
76 config RelocationAssemblyConfig
77}
78
79// relocationStepper is hostrelocate.Worker's and Participant's Step.
80type relocationStepper interface {
81 Step(context.Context, string) (hostclaim.Observed, error)
82}
83
84// relocationStores is rulings 50 and 51: the data bucket as a release.Store
85// of its own (production's lease.VersionedStore is not one), and the AWS
86// signing region kept apart from the logical placement region.
87type relocationStores struct {
88 Files release.Store
89 Versions lease.VersionedStore
90 Bucket string
91 Region string // logical: genselect, pins, the worker, the chooser
92 LeaseRegion string // AWS: lease.Acquire, lease.NewVersionedS3
93}
94
95func openRelocationStores(ctx context.Context, self hostactors.Box, dataBucket string, files func(context.Context, string) (release.Store, error), versioned func(context.Context, string, string) (lease.VersionedStore, error)) (relocationStores, error) {
96 if dataBucket == "" || self.Region == "" || self.Cloud.Region == "" || files == nil || versioned == nil {
97 return relocationStores{}, hostwrite.ErrUnavailable
98 }
99 f, err := files(ctx, dataBucket)
100 if err != nil {
101 return relocationStores{}, err
102 }
103 v, err := versioned(ctx, dataBucket, self.Cloud.Region)
104 if err != nil {
105 return relocationStores{}, err
106 }
107 return relocationStores{Files: f, Versions: v, Bucket: dataBucket, Region: self.Region, LeaseRegion: self.Cloud.Region}, nil
108}
109
110// relocationLeases is the worker's lease provider (ruling 51): the replica
111// lease on a holder's location, signed for the cloud region.
112func relocationLeases(s relocationStores) func(context.Context, string) (hostrelocate.Lease, error) {
113 return func(ctx context.Context, location string) (hostrelocate.Lease, error) {
114 held, err := lease.Acquire(ctx, s.Bucket, location, s.LeaseRegion)
115 if err != nil {
116 return nil, err
117 }
118 return held, nil
119 }
120}
121
122// relocationAssemblyConfig fills Task 8a's config from the bound assembly,
123// the stores and main's participation discharge.
124func relocationAssemblyConfig(a *ClaimAssembly, providers ClaimProviders, s relocationStores, reg *registry.Registry, act RelocationActivator, d hostactors.Directory, optDir string, discharge func(context.Context, hostclaim.Observed) error) (RelocationAssemblyConfig, error) {
125 src, ok := providers.Selector.(hostbox.WorkSource)
126 if !ok || a == nil || reg == nil || act == nil || discharge == nil || optDir == "" || a.Guard.DataDir == "" {
127 return RelocationAssemblyConfig{}, hostwrite.ErrUnavailable
128 }
129 dd := a.Guard.DataDir
130 return RelocationAssemblyConfig{Directory: d, Data: s.Files, Versions: s.Versions, Bucket: s.Bucket, Region: s.Region, LeaseRegion: s.LeaseRegion,
131 Registry: reg, DataDir: dd, OptDir: optDir, WorkDir: filepath.Join(dd, "relocate", "build"), ReleaseDir: filepath.Join(dd, "relocate", "release"),
132 Activator: act, Source: src, Discharge: discharge, Now: time.Now}, nil
133}
134
135// relocationRuntimeConfig is relocationAssemblyConfig with what the agent
136// binds beyond it. ReleasePub and Ratchet are the held stage's evidence
137// (carry 4): the agent's release key (CARLOS_RELEASE_PUBKEY, the key the
138// held-stage pass verifies under) and the edge's pointer digest ratchet,
139// pointerRatchet(false, ...), exactly as moves' held stage takes it in the
140// agent (movespass.go, heldReleaseReady); never root's. The secrets key,
141// when the box has one, renders the held config.
142func relocationRuntimeConfig(getenv func(string) string, a *ClaimAssembly, providers ClaimProviders, s relocationStores, reg *registry.Registry, act RelocationActivator, d hostactors.Directory, in relocationRuntimeInputs, discharge func(context.Context, hostclaim.Observed) error) (RelocationAssemblyConfig, error) {
143 if getenv == nil || len(in.Pub) != ed25519.PublicKeySize {
144 return RelocationAssemblyConfig{}, hostwrite.ErrUnavailable
145 }
146 cfg, err := relocationAssemblyConfig(a, providers, s, reg, act, d, in.OptDir, discharge)
147 if err != nil {
148 return RelocationAssemblyConfig{}, err
149 }
150 cfg.ReleasePub = in.Pub
151 cfg.Ratchet = pointerRatchet(false, cfg.OptDir, cfg.DataDir, getenv)
152 if raw := getenv("CARLOS_SECRETS_KEY"); raw != "" {
153 key, err := appconfig.ParsePrivateKey(raw, "CARLOS_SECRETS_KEY")
154 if err != nil {
155 return RelocationAssemblyConfig{}, err
156 }
157 cfg.SecretsKey = key
158 }
159 return cfg, nil
160}
161
162// channelRelease is ruling 30: the host's channel pointer in the control
163// bucket, by the holder's account and app and the record's channel.
164func channelRelease(store release.Store, pub ed25519.PublicKey) func(context.Context, string) (string, error) {
165 return func(ctx context.Context, host string) (string, error) {
166 obs, err := hostclaim.Read(ctx, store, host)
167 if err != nil || obs.Absent || obs.Claim.Holder == nil {
168 return "", errors.Join(err, hostrelocate.ErrRefused)
169 }
170 rec, err := provision.GetRecord(ctx, store, obs.Claim.Holder.Account, obs.Claim.Holder.App, host)
171 if err != nil {
172 return "", err
173 }
174 p, err := release.Channel(ctx, store, pub, obs.Claim.Holder.Account, obs.Claim.Holder.App, rec.Channel)
175 if err != nil {
176 return "", err
177 }
178 return p.Version, nil
179 }
180}
181
182// volumeDischarge is ruling 3 with enrolled_at in the roster: unqualified
183// without it.
184func volumeDischarge(store release.Store) func(context.Context, hostactors.Box, termfence.Record) *hostclaim.LocalDischarge {
185 return func(ctx context.Context, b hostactors.Box, f termfence.Record) *hostclaim.LocalDischarge {
186 if b.EnrolledAt.IsZero() {
187 return &hostclaim.LocalDischarge{Reason: "the roster records no enrollment time"}
188 }
189 at, err := time.Parse(time.RFC3339Nano, f.ObservedAt)
190 if err != nil {
191 return &hostclaim.LocalDischarge{Reason: err.Error()}
192 }
193 records, err := boxvolumes.List(ctx, store, b.Enrollment)
194 if err != nil {
195 return &hostclaim.LocalDischarge{Reason: err.Error()}
196 }
197 v := boxvolumes.Discharge(records, boxvolumes.Entry{Box: b.BoxID, Enrollment: b.Enrollment, Cloud: b.Cloud}, b.EnrolledAt, at)
198 return &hostclaim.LocalDischarge{Qualified: v.Discharge, Reason: v.Reason}
199 }
200}
201
202// relocationRelocatable is Worker.Relocatable over relocationRecord (K24):
203// only a hibernating, path-deriving instance record that mints nested
204// relocates (ruling 45).
205func relocationRelocatable(dataDir string, store release.Store) func(context.Context, hostclaim.Observed) error {
206 return func(ctx context.Context, obs hostclaim.Observed) error {
207 _, err := relocationRecord(ctx, dataDir, store, obs)
208 return err
209 }
210}
211
212// errRelocationNotStarted is the pass's answer to Worker.Start's two
213// providers: this runtime replays relocations and never starts one. The
214// app admission and the census are the starter's (the operator's carlos
215// ops host relocate and spec 3's placer), with their own bindings.
216var errRelocationNotStarted = errors.New("host: the relocation pass starts no relocation")
217
218type relocationStartRefused struct{}
219
220func (relocationStartRefused) Register(context.Context, hostclaim.Observed, hostrelocate.Owner, string) (string, error) {
221 return "", errRelocationNotStarted
222}
223
224func (relocationStartRefused) Recheck(context.Context, string) error { return errRelocationNotStarted }
225
226// relocationRuntimeInputs is what NewRelocationRuntime binds beyond the
227// claim's own: the box's enrollment, the release key, the bucket
228// constructors (production's S3; tests' fakes).
229type relocationRuntimeInputs struct {
230 Enrollment hostfloor.Enrollment
231 Pub ed25519.PublicKey
232 Files func(context.Context, string) (release.Store, error)
233 Versioned func(context.Context, string, string) (lease.VersionedStore, error)
234 OptDir string
235}
236
237// productionRelocationRuntimeInputs reads the box's enrollment and the
238// release key, and opens the data bucket with release.NewS3Store and
239// lease.NewVersionedS3.
240func productionRelocationRuntimeInputs(getenv func(string) string, optDir string) (relocationRuntimeInputs, error) {
241 pub, err := release.ParsePublicKey(getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")
242 if err != nil {
243 return relocationRuntimeInputs{}, err
244 }
245 enrollment, err := hostfloor.ReadEnrollment("/etc/carlos/host-actor.json")
246 if err != nil {
247 return relocationRuntimeInputs{}, err
248 }
249 return relocationRuntimeInputs{Enrollment: enrollment, Pub: pub, OptDir: optDir,
250 Files: func(ctx context.Context, bucket string) (release.Store, error) {
251 s, err := release.NewS3Store(ctx, bucket)
252 if err != nil {
253 return nil, err
254 }
255 return s, nil
256 },
257 Versioned: func(ctx context.Context, bucket, region string) (lease.VersionedStore, error) {
258 return lease.NewVersionedS3(ctx, map[string]string{bucket: region})
259 }}, nil
260}
261
262// NewRelocationRuntime builds the agent's relocation runtime from the bound
263// claim assembly. It returns (nil, nil) unless CARLOS_HOST_CLAIM is on.
264// With CARLOS_REGION_HA off, or the relocation owners not bound into the
265// wake's dispatcher (ruling 53: they need moves' C8 provider), it is the
266// completion-only runtime (ruling 82): a relocation that already completed
267// on this box is still finished. Only with both on and the owners bound is
268// it the full pass.
269func NewRelocationRuntime(ctx context.Context, getenv func(string) string, store release.Store, local *productionClaimLocal, providers ClaimProviders, reg *registry.Registry, act RelocationActivator, in relocationRuntimeInputs) (*RelocationRuntime, error) {
270 if !hostwrite.Enabled(getenv) || local == nil {
271 return nil, nil
272 }
273 if !regionHAEnabled(getenv) {
274 return newCompletionRuntime(local, reg, act)
275 }
276 a := local.boundAssembly()
277 if a == nil {
278 return nil, hostwrite.ErrUnavailable
279 }
280 if d, ok := wakeDispatch(a); !ok || d.Relocation == nil {
281 return newCompletionRuntime(local, reg, act)
282 }
283 if a.RetirementAuthority == nil || store == nil || reg == nil || act == nil || len(in.Pub) != ed25519.PublicKeySize || in.Enrollment.Box == "" {
284 return nil, hostwrite.ErrUnavailable
285 }
286 d := a.RetirementAuthority.Directory
287 roster, err := d.Read(ctx)
288 if err != nil {
289 return nil, err
290 }
291 self, ok := roster.Roster.Boxes[in.Enrollment.Box]
292 if !ok || self.Enrollment != in.Enrollment.Enrollment {
293 return nil, hostbox.ErrChanged
294 }
295 stores, err := openRelocationStores(ctx, self, getenv("CARLOS_HIBERNATE_BUCKET"), in.Files, in.Versioned)
296 if err != nil {
297 return nil, err
298 }
299 cfg, err := relocationRuntimeConfig(getenv, a, providers, stores, reg, act, d, in, local.DischargeCompleted)
300 if err != nil {
301 return nil, err
302 }
303 effects := a.RelocationEffects(cfg, local.Acknowledged)
304 chooser := hostrelocate.RegionChooser{Records: store, Namespace: self.Namespace, Region: stores.Region,
305 Executors: hostops.ExecutorChooser{Directory: d, Namespace: self.Namespace, Pool: self.Pool, HomeName: stores.Region, Now: time.Now,
306 Edges: func(ctx context.Context) (boxchoice.Snapshot, error) {
307 return edgesfile.LiveSnapshot(ctx, store, time.Now())
308 },
309 Loads: func(ctx context.Context) (map[string]boxchoice.Load, error) { return hostops.ClaimLoads(ctx, store) },
310 // The fingerprint this box's own edges file publishes: a target
311 // must hold the same secrets key to render the held config.
312 SecretsKeyFP: edgesFileSecretsFP(getenv)}}
313 worker := hostrelocate.Worker{Claims: store, Directory: d,
314 Fences: hostfence.ClaimFenceAuthority{Directory: d, Now: time.Now},
315 Releases: StopReceiptReleases{Receipts: RelocationReleaseReceipts{Store: store, Directory: d}},
316 Admission: relocationStartRefused{},
317 Census: func(context.Context, string, []string) error { return errRelocationNotStarted },
318 Choose: chooser.Choose, Eligible: chooser.Eligible,
319 Relocatable: relocationRelocatable(cfg.DataDir, store),
320 Leases: relocationLeases(stores),
321 Pins: hostrelocate.StorePinner{Claims: store, Data: stores.Files, Bucket: stores.Bucket, Region: stores.Region, HomeRegion: cfg.Source.HomeRegion,
322 SelectDB: func(ctx context.Context, r genselect.Request) (*genselect.Plan, error) {
323 return genselect.Select(ctx, stores.Versions, r)
324 },
325 Release: channelRelease(store, in.Pub)},
326 Prepared: PreparedFromInstallManifest{Claims: store, Files: stores.Files, Data: stores.Versions, Bucket: stores.Bucket},
327 Discharge: volumeDischarge(store),
328 Notify: func(ctx context.Context) { _ = release.BumpSerial(ctx, store, time.Now()) },
329 Self: termfence.WrittenBy{Box: in.Enrollment.Box, Enrollment: in.Enrollment.Enrollment},
330 Region: stores.Region, HomeRegion: cfg.Source.HomeRegion, Controller: newClaimID(), NewID: newClaimID, Now: time.Now}
331 participant := hostrelocate.Participant{Claims: store, Box: in.Enrollment.Box, Enrollment: in.Enrollment.Enrollment, Effects: effects, Now: time.Now}
332 return &RelocationRuntime{Worker: worker, Participant: participant, Box: in.Enrollment.Box, Convert: convertFencedLanding(worker, store), Finish: effects.Finish, DataDir: cfg.DataDir, config: cfg}, nil
333}
334
335// newCompletionRuntime is ruling 82: the smallest runtime that finishes a
336// relocation already completed on this box (discharge, hold release,
337// owned-fence lift, mark; relocationReleaseHold) and does nothing else. It
338// steps no relocation, converts no landing and starts nothing, and it
339// reads only the claims of this box's own registry rows, so a box that
340// never completed a relocation reads its own hosts' claims and writes
341// nothing.
342func newCompletionRuntime(local *productionClaimLocal, reg *registry.Registry, act RelocationActivator) (*RelocationRuntime, error) {
343 a := local.boundAssembly()
344 if a == nil || reg == nil || act == nil || a.Guard.DataDir == "" {
345 return nil, hostwrite.ErrUnavailable
346 }
347 cfg := RelocationAssemblyConfig{Registry: reg, Activator: act, DataDir: a.Guard.DataDir, Discharge: local.DischargeCompleted, Now: time.Now}
348 dataDir := a.Guard.DataDir
349 hosts := func() ([]string, error) {
350 rows, err := reg.List()
351 if err != nil {
352 return nil, err
353 }
354 seen := make(map[string]bool, len(rows))
355 out := make([]string, 0, len(rows))
356 for _, row := range rows {
357 if !seen[row.Host] {
358 seen[row.Host] = true
359 out = append(out, row.Host)
360 }
361 }
362 named, err := relocationJournalHosts(dataDir)
363 for _, host := range named {
364 if !seen[host] {
365 seen[host] = true
366 out = append(out, host)
367 }
368 }
369 return out, err
370 }
371 // Carry 8: the completion-only runtime has no roster read, so this box
372 // is the bound assembly's own verified stop identity, asked each pass.
373 self := func(ctx context.Context) (string, error) {
374 l, err := a.localStopIdentity(ctx)
375 if err != nil {
376 return "", err
377 }
378 return l.Actor.Box, nil
379 }
380 return &RelocationRuntime{Finish: a.relocationReleaseHold(cfg), Hosts: hosts, Self: self, DataDir: a.Guard.DataDir}, nil
381}
382
383// wakeDispatch is the wake's owner dispatcher, as Build bound it.
384func wakeDispatch(a *ClaimAssembly) (ClaimOwnerDispatch, bool) {
385 if a == nil || a.Wake == nil {
386 return ClaimOwnerDispatch{}, false
387 }
388 d, ok := a.Wake.Inputs.(ClaimOwnerDispatch)
389 return d, ok
390}
391
392// Run is one pass every 30 seconds, each bounded at two minutes, and one
393// more whenever serial is signalled. The agent gives no serial channel
394// (launchRelocationPass): the control serial's one wake channel is the
395// regional loop's (agent.go, serialClaims), so the pass runs on its ticker
396// alone and a BumpSerial wakes it no sooner than the next tick.
397func (r *RelocationRuntime) Run(ctx context.Context, store release.Store, serial <-chan struct{}) {
398 tick := time.NewTicker(30 * time.Second)
399 defer tick.Stop()
400 for {
401 pass, cancel := context.WithTimeout(ctx, 2*time.Minute)
402 r.once(pass, store)
403 cancel()
404 select {
405 case <-ctx.Done():
406 return
407 case <-tick.C:
408 case <-serial:
409 }
410 }
411}
412
413// once is one pass. Per host, in this order, so a passed probe completes
414// in one pass (ruling 49): the participant (its probe stop), the worker
415// (the serving CAS), then the completion this box has not yet finished,
416// found from the claim's record (ruling 80). A landing whose target is
417// fenced is converted when its move committed (J3). A claim is read once
418// a pass, and once more only after a step that may have written it.
419func (r *RelocationRuntime) once(ctx context.Context, store release.Store) {
420 r.tryUpgrade(ctx)
421 hosts, err := r.hosts(ctx, store)
422 r.result("", "list hosts", err)
423 if err != nil && hosts == nil {
424 return
425 }
426 var self passSelf
427 read := make(map[string]bool, len(hosts))
428 for _, host := range hosts {
429 read[host] = true
430 obs, err := hostclaim.Read(ctx, store, host)
431 // A claim that cannot be read is not "nothing to do": said once,
432 // and read again next pass.
433 if r.result(host, "read", err); err != nil || obs.Absent {
434 continue
435 }
436 stepped := false
437 switch obs.Claim.Op.Kind {
438 case hostclaim.KindLanding:
439 if r.Convert != nil {
440 stepped = true
441 r.result(host, "convert", r.Convert(ctx, store, obs))
442 }
443 case hostclaim.KindRelocating:
444 if r.Participant != nil {
445 stepped = true
446 _, err := r.Participant.Step(ctx, host)
447 r.result(host, "participant", notYetIsNoError(err))
448 }
449 if r.Worker != nil {
450 stepped = true
451 _, err := r.Worker.Step(ctx, host)
452 r.result(host, "worker", notYetIsNoError(err))
453 }
454 }
455 if stepped {
456 // A step may have just made the serving CAS (ruling 49).
457 obs, err = hostclaim.Read(ctx, store, host)
458 if r.result(host, "read", err); err != nil {
459 continue
460 }
461 }
462 r.finishCompletion(ctx, host, obs, &self)
463 }
464 r.alerts.keep(read)
465 // Ruling 61: journals of operations that are over go, once per pass,
466 // after the pass's own steps and never beside a journal write (carry 3:
467 // sweepRelocationJournals holds relocationJournals exclusively).
468 if r.DataDir != "" {
469 r.result("", "sweep journals", sweepRelocationJournals(ctx, r.DataDir, store))
470 }
471}
472
473func notYetIsNoError(err error) error {
474 if errors.Is(err, hostrelocate.ErrNotYet) {
475 return nil
476 }
477 return err
478}
479
480// tryUpgrade replaces a completion-only stand-in with the full pass once
481// it builds (upgrade). A build that answers no relocation owners ends the
482// tries: completions are then all this box runs, as NewRelocationRuntime
483// answers when the owners are not bound.
484func (r *RelocationRuntime) tryUpgrade(ctx context.Context) {
485 if r.upgrade == nil {
486 return
487 }
488 full, err := r.upgrade(ctx)
489 if r.result("", "region HA: relocation pass not built, finishing completions only", err); err != nil {
490 return
491 }
492 r.upgrade = nil
493 if full == nil || full.Worker == nil {
494 log.Printf("agent: region HA: the relocation owners are not bound, so this box finishes completions only")
495 return
496 }
497 r.Worker, r.Participant, r.Box, r.Convert, r.Finish, r.Hosts, r.Self, r.DataDir, r.config = full.Worker, full.Participant, full.Box, full.Convert, full.Finish, full.Hosts, full.Self, full.DataDir, full.config
498 log.Printf("agent: region HA: relocation pass started")
499}
500
501// hosts is every claim in the bucket, or Hosts when it is set.
502func (r *RelocationRuntime) hosts(ctx context.Context, store release.Store) ([]string, error) {
503 if r.Hosts != nil {
504 return r.Hosts()
505 }
506 keys, err := store.List(ctx, "control/hosts/")
507 if err != nil {
508 return nil, err
509 }
510 var out []string
511 for _, key := range keys {
512 host, ok := strings.CutSuffix(strings.TrimPrefix(key, "control/hosts/"), ".json")
513 if ok && !strings.Contains(host, "/") {
514 out = append(out, host)
515 }
516 }
517 return out, nil
518}
519
520// passSelf is this box as one pass knows it: asked at most once.
521type passSelf struct {
522 asked bool
523 box string
524}
525
526// thisBox is Box, or Self's answer, asked once a pass and only by a
527// pending completion (review M2). No answer is no box.
528func (r *RelocationRuntime) thisBox(ctx context.Context, s *passSelf) string {
529 if r.Box != "" {
530 return r.Box
531 }
532 if !s.asked {
533 s.asked = true
534 err := errors.New("no identity to answer which box this is")
535 if r.Self != nil {
536 s.box, err = r.Self(ctx)
537 }
538 if r.result("", "this box", err); err != nil {
539 s.box = ""
540 }
541 }
542 return s.box
543}
544
545// finishCompletion is ruling 80: a claim serving on this box whose
546// completed relocation names this box and is not yet marked discharged is
547// handed to Finish (the hold release, relocationReleaseHold) on every
548// pass, from the pass's latest read of it, so a serving CAS made this pass
549// is finished in the same pass. The work is found from the claim's record,
550// never from the row: a row already released, a lift that never ran or a
551// mark whose CAS failed is retried until the mark is written.
552//
553// A claim with no undischarged completion asks nothing more, not even
554// which box this is (review M2). A claim that does not serve on this box is
555// skipped before Finish is called (carry 8). A record that cannot be read
556// on a claim serving here is said and retried, never read as "no
557// completion" (carry 1), and so is every refusal of Finish, a probe-fence
558// owner it cannot find (carry 6) and an ErrChanged after the serving-here
559// check (review M4) among them: the record stays unmarked.
560func (r *RelocationRuntime) finishCompletion(ctx context.Context, host string, obs hostclaim.Observed, self *passSelf) {
561 if r.Finish == nil || obs.Absent || obs.Claim == nil || obs.Claim.Op.Kind != hostclaim.KindServing {
562 return
563 }
564 c := obs.Claim
565 done, ok, err := hostclaim.ReadCompletedRelocation(c)
566 if err == nil && (!ok || done.Discharged) {
567 r.alerts.clear(host, "finish")
568 return
569 }
570 box := r.thisBox(ctx, self)
571 if box == "" || c.Op.Exec != box {
572 return
573 }
574 if err != nil {
575 r.alert(host, "finish", err)
576 return
577 }
578 if done.Stop.Box != box {
579 return
580 }
581 r.result(host, "finish", r.Finish(ctx, host, done.Op))
582}
583
584// convertFencedLanding is J3 through the worker's own authorization
585// (hostrelocate.Worker.ConvertLanding): a landing whose op.exec box has a
586// termination fence becomes a failover relocation when its move committed
587// (CommittedLandingRecords). An uncommitted one is moves' case and answers
588// nil (ErrWrongState); any other refusal, a caller that is not an active
589// worker of the landing's region included, is returned for the pass to
590// alert once.
591func convertFencedLanding(w hostrelocate.Worker, store release.Store) func(context.Context, release.Store, hostclaim.Observed) error {
592 return func(ctx context.Context, _ release.Store, obs hostclaim.Observed) error {
593 _, err := w.ConvertLanding(ctx, obs, CommittedLandingRecords{Records: MovesRecordsFrom(store)})
594 if errors.Is(err, hostclaim.ErrWrongState) {
595 return nil
596 }
597 return err
598 }
599}
600
601func (r *RelocationRuntime) alert(host, step string, err error) { r.alerts.say(host, step, err) }
602
603// result says err for host's step, or, with none, forgets the step's last
604// cause, so the cause is said again if it returns.
605func (r *RelocationRuntime) result(host, step string, err error) {
606 if err != nil {
607 r.alerts.say(host, step, err)
608 return
609 }
610 r.alerts.clear(host, step)
611}
612
613// relocationAlerts is rulings 35 and 36's one alert per host, step and
614// cause (review M8): a cause is said when it is first seen for that host
615// and step, and again only after that step has succeeded since, or after
616// another cause. It holds one entry per host and step failing now; each
617// pass forgets the hosts it no longer reads, so it is bounded by them.
618type relocationAlerts struct {
619 mu sync.Mutex
620 last map[[2]string]string // {host, step} -> the cause last said
621}
622
623func (a *relocationAlerts) say(host, step string, err error) {
624 a.mu.Lock()
625 defer a.mu.Unlock()
626 if a.last == nil {
627 a.last = map[[2]string]string{}
628 }
629 k := [2]string{host, step}
630 if cause, said := a.last[k]; said && cause == err.Error() {
631 return
632 }
633 a.last[k] = err.Error()
634 if host == "" {
635 log.Printf("relocate: %s: %v", step, err)
636 return
637 }
638 log.Printf("relocate %s: %s: %v", host, step, err)
639}
640
641func (a *relocationAlerts) clear(host, step string) {
642 a.mu.Lock()
643 defer a.mu.Unlock()
644 delete(a.last, [2]string{host, step})
645}
646
647// keep forgets every host's causes but those of hosts, and the box-wide
648// ones (host "").
649func (a *relocationAlerts) keep(hosts map[string]bool) {
650 a.mu.Lock()
651 defer a.mu.Unlock()
652 for k := range a.last {
653 if k[0] != "" && !hosts[k[0]] {
654 delete(a.last, k)
655 }
656 }
657}
658
659// size is the number of causes held (tests).
660func (a *relocationAlerts) size() int {
661 a.mu.Lock()
662 defer a.mu.Unlock()
663 return len(a.last)
664}
665
666// registerRelocationSource registers region HA's owner source with the
667// roster directory the claim's own owners read (claimproduction.go). The
668// release key and the deployment bucket are read from the claim entry's
669// environment, the one the flags and the claim store came from (final
670// review minor 5: one reading of the environment).
671func registerRelocationSource(claimCtx context.Context, dataDir string) error {
672 store, getenv := claimStore(claimCtx), claimEntryEnv(claimCtx)
673 pub, err := release.ParsePublicKey(getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")
674 if err != nil || store == nil {
675 return errors.Join(hostwrite.ErrUnavailable, err)
676 }
677 return RegisterProductionClaimRelocationSource(&RelocationSource{DataDir: dataDir,
678 Directory: hostactors.Directory{Store: store, PublicKey: pub, Bucket: getenv("CARLOS_DEPLOYMENT_BUCKET")},
679 FilesStore: func(ctx context.Context, bucket string) (release.Store, error) {
680 s, err := release.NewS3Store(ctx, bucket)
681 if err != nil {
682 return nil, err
683 }
684 return s, nil
685 }})
686}
687
688// startRelocationPass builds the relocation runtime from the claim entry's
689// assembly and starts its loop. With claim mode on it always starts one:
690// the completion-only pass (ruling 82), which with region HA on builds
691// the full pass on its first, bounded, pass (review M9) and becomes it
692// once the owners are bound. Nothing with claim mode off; with it on,
693// every way of not starting the pass is said.
694func startRelocationPass(agentCtx, claimCtx context.Context, claims *ClaimWiring, reg *registry.Registry, svc *activator.Service) {
695 getenv := claimEntryEnv(claimCtx)
696 if !hostwrite.Enabled(getenv) {
697 return
698 }
699 entry, _ := claimCtx.Value(claimEntryKey{}).(claimEntry)
700 // Claim mode is on, so a completion recorded on this box must be
701 // finished here (ruling 82).
702 if claims == nil || svc == nil || entry.Assembly == nil {
703 log.Printf("agent: relocation completion pass not started: claim mode is on but the claim wiring, the activator or the claim assembly is missing")
704 return
705 }
706 local, ok := entry.Assembly.Local.(*productionClaimLocal)
707 if !ok {
708 log.Printf("agent: relocation completion pass not started: the assembly has no production participation owner")
709 return
710 }
711 var full func(context.Context) (*RelocationRuntime, error)
712 ha := regionHAEnabled(getenv)
713 if ha {
714 full = func(ctx context.Context) (*RelocationRuntime, error) {
715 in, err := productionRelocationRuntimeInputs(getenv, release.OptDir())
716 if err != nil {
717 return nil, err
718 }
719 return NewRelocationRuntime(ctx, getenv, entry.Store, local, claims.Providers, reg, svc, in)
720 }
721 }
722 runtime, err := relocationPassFor(local, reg, svc, full)
723 if err != nil {
724 log.Printf("agent: relocation completion pass not started: %v", err)
725 return
726 }
727 launchRelocationPass(agentCtx, runtime, entry.Store, ha)
728}
729
730// relocationPassFor is the runtime the agent runs in claim mode: the
731// completion-only one (ruling 82), standing in for the full pass while
732// full (region HA on) builds it. Nothing is built here: the build reads
733// the roster and the data bucket, so it runs inside the first pass, under
734// the pass's bound, never on the agent's startup path (review M9). A build
735// that fails is not "no relocation on this box": each pass tries it again.
736func relocationPassFor(local *productionClaimLocal, reg *registry.Registry, act RelocationActivator, full func(context.Context) (*RelocationRuntime, error)) (*RelocationRuntime, error) {
737 r, err := newCompletionRuntime(local, reg, act)
738 if err != nil {
739 return nil, err
740 }
741 r.upgrade = full
742 return r, nil
743}
744
745// launchRelocationPass says the pass started and runs it. Everything it
746// says is decided before the loop starts: the loop owns the runtime's
747// fields from then on, and a pass may replace them (tryUpgrade; review
748// M3). The channel closes when the loop has returned.
749func launchRelocationPass(ctx context.Context, r *RelocationRuntime, store release.Store, regionHA bool) <-chan struct{} {
750 log.Printf("agent: relocation pass started (completions only: %v)", !regionHA)
751 done := make(chan struct{})
752 go func() {
753 defer close(done)
754 r.Run(ctx, store, nil)
755 }()
756 return done
757}
758