| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "flag" |
| 7 | "io" |
| 8 | "log" |
| 9 | "net/http" |
| 10 | "os" |
| 11 | "os/signal" |
| 12 | "path/filepath" |
| 13 | "runtime" |
| 14 | "strings" |
| 15 | "syscall" |
| 16 | "time" |
| 17 | |
| 18 | "github.com/carlosframework/platform/internal/hostwrite" |
| 19 | "github.com/carlosframework/platform/internal/lbhealth" |
| 20 | "github.com/carlosframework/platform/internal/status" |
| 21 | "github.com/carlosframework/platform/internal/version" |
| 22 | ) |
| 23 | |
| 24 | // The agent's joining mode (region HA spec 2 §5 step 4) and the gate that |
| 25 | // chooses it (joinmode.go). |
| 26 | // |
| 27 | // A joining box publishes its signed identity record (boxRecordLoop) and, |
| 28 | // once its guard is armed, its guard-ready report: the private status |
| 29 | // report every hosting box publishes, carrying the guard's facts |
| 30 | // (claimReport). It never publishes an edges file: a box an edges file |
| 31 | // names is referenced, and automatic enrollment refuses a referenced box |
| 32 | // (hostactors references.go, enroll.go). It opens no registry, starts no |
| 33 | // activator and serves no instance. Its only listener is the load |
| 34 | // balancer's health check on the edge's HTTP address, which answers 503: |
| 35 | // lbhealth.Default is never marked hosting here. |
| 36 | // |
| 37 | // When the box is invited it asks root to arm the guard by touching |
| 38 | // joinRequestMarker (carlos-system-join.path). Once a start check finds it |
| 39 | // enrolled and armed, it returns, and RunAgent ends the process so systemd |
| 40 | // starts the agent again, now as the claim agent through the guard. |
| 41 | |
| 42 | // joinEvery is how often joining checks its mode. |
| 43 | const joinEvery = 15 * time.Second |
| 44 | |
| 45 | // joinNudgeEvery is how often joining re-publishes its report or re-asks |
| 46 | // root, at most: the report stays well inside reportFreshness, and root |
| 47 | // is not asked again while it is still arming. While the box stays |
| 48 | // invited with its guard not armed, root is asked again at this interval, |
| 49 | // once a minute: carlos-system-join.path fires only when the marker |
| 50 | // changes, so a carlos system join that failed (an IMDS or store read, |
| 51 | // say) is retried by the next touch, and by nothing else. |
| 52 | const joinNudgeEvery = time.Minute |
| 53 | |
| 54 | // joinRequestMarker is the file carlos-system-join.path watches, under the |
| 55 | // data dir. |
| 56 | const joinRequestMarker = "system-join-requested" |
| 57 | |
| 58 | // joining is one joining run's steps; production's are productionJoining. |
| 59 | type joining struct { |
| 60 | facts modeFacts |
| 61 | every time.Duration |
| 62 | // serve answers the load balancer until ctx ends. |
| 63 | serve func(ctx context.Context) error |
| 64 | // record publishes the box's identity record, retrying until it is in |
| 65 | // place. |
| 66 | record func(ctx context.Context) |
| 67 | // report publishes the guard-ready report. |
| 68 | report func(ctx context.Context) error |
| 69 | // requestArm asks root to arm the guard. |
| 70 | requestArm func() error |
| 71 | logf func(format string, args ...any) |
| 72 | } |
| 73 | |
| 74 | // run joins until the box can host (the decision is returned) or ctx |
| 75 | // ends (the last decision, and ctx's error). |
| 76 | func (j joining) run(ctx context.Context) (modeDecision, error) { |
| 77 | lbhealth.Default.SetHosting(false) |
| 78 | go func() { |
| 79 | if err := j.serve(ctx); err != nil && !errors.Is(err, http.ErrServerClosed) { |
| 80 | j.logf("agent: joining: the load balancer health check is not answered: %v", err) |
| 81 | } |
| 82 | }() |
| 83 | go j.record(ctx) |
| 84 | var last modeDecision |
| 85 | var lastErr string |
| 86 | var nudged time.Time |
| 87 | for { |
| 88 | d, err := selectMode(ctx, j.facts) |
| 89 | now := j.facts.now() |
| 90 | switch { |
| 91 | case err != nil: |
| 92 | if err.Error() != lastErr { |
| 93 | j.logf("agent: joining: %v; checking again", err) |
| 94 | lastErr = err.Error() |
| 95 | } |
| 96 | case d.Mode == modeHosting: |
| 97 | j.logf("agent: joining: %s is enrolled and its guard is armed; restarting as the claim agent", d.Box) |
| 98 | return d, nil |
| 99 | default: |
| 100 | lastErr = "" |
| 101 | if d.Stage != last.Stage || d.Why != last.Why { |
| 102 | j.logf("agent: joining: %s: %s %s", d.Box, d.Stage, d.Why) |
| 103 | nudged = time.Time{} |
| 104 | } |
| 105 | last = d |
| 106 | if now.Sub(nudged) >= joinNudgeEvery { |
| 107 | switch d.Stage { |
| 108 | case stageInvited: |
| 109 | if err := j.requestArm(); err != nil { |
| 110 | j.logf("agent: joining: could not ask root to arm the guard: %v", err) |
| 111 | } |
| 112 | case stageGuardInstalled, stageGuardReady: |
| 113 | if err := j.report(ctx); err != nil { |
| 114 | j.logf("agent: joining: the guard-ready report was not published: %v", err) |
| 115 | } |
| 116 | } |
| 117 | nudged = now |
| 118 | } |
| 119 | } |
| 120 | select { |
| 121 | case <-ctx.Done(): |
| 122 | return last, ctx.Err() |
| 123 | case <-time.After(j.every): |
| 124 | } |
| 125 | } |
| 126 | } |
| 127 | |
| 128 | // productionJoining is joining on a real box. args are the agent's own |
| 129 | // (edge.Run's flags): the health check listens on -http, as the edge will. |
| 130 | func productionJoining(args []string, facts modeFacts, getenv func(string) string) joining { |
| 131 | addr := joiningHTTPAddr(args) |
| 132 | return joining{ |
| 133 | facts: facts, |
| 134 | every: joinEvery, |
| 135 | serve: func(ctx context.Context) error { |
| 136 | srv := &http.Server{Addr: addr, ReadHeaderTimeout: 10 * time.Second, Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 137 | if !lbhealth.Serve(w, r, lbhealth.Default, time.Now()) { |
| 138 | http.NotFound(w, r) |
| 139 | } |
| 140 | })} |
| 141 | go func() { <-ctx.Done(); _ = srv.Close() }() |
| 142 | return srv.ListenAndServe() |
| 143 | }, |
| 144 | record: func(context.Context) { boxRecordLoop() }, |
| 145 | report: func(ctx context.Context) error { |
| 146 | store, _, err := openStatusStore(ctx, getenv) |
| 147 | if err != nil { |
| 148 | return err |
| 149 | } |
| 150 | p := status.Probes{MachineID: readMachineID(), InstanceID: cachedInstanceID(), Region: strings.TrimSpace(getenv("CARLOS_REGION")), |
| 151 | Release: version.Build, Platform: runtime.GOOS + "-" + runtime.GOARCH, SystemChannel: boxSystemChannel(getenv)} |
| 152 | claimReport(&p) |
| 153 | box := status.CollectBox(nil, p, time.Now()) |
| 154 | if groupReferenced(getenv) { |
| 155 | // Under the group reference the console derives (census |
| 156 | // ruling region-ha-2b), as statusLoop publishes it. |
| 157 | box.Label = facts.box |
| 158 | } |
| 159 | return status.WriteBox(ctx, store, box) |
| 160 | }, |
| 161 | requestArm: func() error { |
| 162 | return os.WriteFile(filepath.Join(dataDir(), joinRequestMarker), []byte(time.Now().UTC().Format(time.RFC3339Nano)+"\n"), 0o644) |
| 163 | }, |
| 164 | logf: log.Printf, |
| 165 | } |
| 166 | } |
| 167 | |
| 168 | // joiningHTTPAddr is the agent's -http flag, ":80" when it is not given. |
| 169 | func joiningHTTPAddr(args []string) string { |
| 170 | fs := flag.NewFlagSet("agent", flag.ContinueOnError) |
| 171 | fs.SetOutput(io.Discard) |
| 172 | addr := fs.String("http", ":80", "") |
| 173 | fs.String("registry", "", "") |
| 174 | fs.String("acme-cache", "", "") |
| 175 | fs.String("https", "", "") |
| 176 | fs.String("fallback", "", "") |
| 177 | fs.Bool("dev", false, "") |
| 178 | _ = fs.Parse(args) |
| 179 | return *addr |
| 180 | } |
| 181 | |
| 182 | // decideAgentMode is RunAgent's gate. With claim mode or region HA off |
| 183 | // there is no joining mode, and the agent hosts as on main. Otherwise it |
| 184 | // selects the mode, waiting out an unknown one, and on joining runs join |
| 185 | // until the box can host or ctx ends; it reports whether this process |
| 186 | // should go on to start the claim agent. |
| 187 | func decideAgentMode(ctx context.Context, getenv func(string) string, open func(context.Context, func(string) string) (modeFacts, error), join func(context.Context, modeFacts, modeDecision) error, wait time.Duration, logf func(string, ...any)) bool { |
| 188 | if !hostwrite.Enabled(getenv) || !regionHAEnabled(getenv) { |
| 189 | return true |
| 190 | } |
| 191 | var last string |
| 192 | note := func(err error) bool { |
| 193 | if err.Error() != last { |
| 194 | logf("agent: mode: %v; checking again", err) |
| 195 | last = err.Error() |
| 196 | } |
| 197 | select { |
| 198 | case <-ctx.Done(): |
| 199 | return false |
| 200 | case <-time.After(wait): |
| 201 | return true |
| 202 | } |
| 203 | } |
| 204 | for { |
| 205 | facts, err := open(ctx, getenv) |
| 206 | if err != nil { |
| 207 | if !note(modeUnknown("the deployment's stores and this box's guard", err)) { |
| 208 | return false |
| 209 | } |
| 210 | continue |
| 211 | } |
| 212 | d, err := selectMode(ctx, facts) |
| 213 | if err != nil { |
| 214 | if !note(err) { |
| 215 | return false |
| 216 | } |
| 217 | continue |
| 218 | } |
| 219 | if d.Mode == modeHosting { |
| 220 | return true |
| 221 | } |
| 222 | logf("agent: %s is joining (%s): it hosts nothing until it is enrolled", d.Box, d.Stage) |
| 223 | if err := join(ctx, facts, d); err != nil && !errors.Is(err, context.Canceled) { |
| 224 | logf("agent: joining: %v", err) |
| 225 | } |
| 226 | return false |
| 227 | } |
| 228 | } |
| 229 | |
| 230 | // hostingGate decides, for RunAgent, whether this process starts the claim |
| 231 | // agent, and records the answer in agentHosting. SIGTERM or SIGINT while |
| 232 | // joining ends the process cleanly. |
| 233 | func hostingGate(args []string) bool { |
| 234 | ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, os.Interrupt) |
| 235 | defer stop() |
| 236 | hosting := decideAgentMode(ctx, os.Getenv, productionModeFacts, func(ctx context.Context, facts modeFacts, _ modeDecision) error { |
| 237 | _, err := productionJoining(args, facts, os.Getenv).run(ctx) |
| 238 | return err |
| 239 | }, joinEvery, log.Printf) |
| 240 | agentHosting.Store(hosting) |
| 241 | return hosting |
| 242 | } |
| 243 | |
| 244 | // groupReferenceBackoff is how long agentEntry waits after the n-th |
| 245 | // failed read of a group box's reference (n from 1): a second, doubling, |
| 246 | // at most 30 seconds. |
| 247 | func groupReferenceBackoff(n int) time.Duration { |
| 248 | d := time.Second |
| 249 | for ; n > 1 && d < 30*time.Second; n-- { |
| 250 | d *= 2 |
| 251 | } |
| 252 | return min(d, 30*time.Second) |
| 253 | } |
| 254 | |
| 255 | // agentEntry is RunAgent's order. A box its region's group launched (the |
| 256 | // CARLOS_BOX_REFERENCE=group marker) first resolves its group reference, |
| 257 | // whatever the claim and region HA switches say: metricsBoxLabel answers |
| 258 | // "" until it resolves, and the loops runAgent starts (the edges file, |
| 259 | // the registry snapshot, metrics, usage, the layout box, the moves pass) |
| 260 | // read the label once and keep it for the process. It asks IMDS again |
| 261 | // with backoff, logging each failure's cause, until it answers or the |
| 262 | // process is told to stop. Any other box resolves nothing. Then the gate |
| 263 | // (hostingGate) decides whether this process starts the agent (run). |
| 264 | func agentEntry(getenv func(string) string, resolve func() (string, error), backoff func(int) time.Duration, logf func(string, ...any), gate func() bool, run func()) { |
| 265 | if groupReferenced(getenv) { |
| 266 | ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, os.Interrupt) |
| 267 | ok := awaitGroupReference(ctx, resolve, backoff, logf) |
| 268 | stop() |
| 269 | if !ok { |
| 270 | return |
| 271 | } |
| 272 | } |
| 273 | if !gate() { |
| 274 | return |
| 275 | } |
| 276 | run() |
| 277 | } |
| 278 | |
| 279 | // awaitGroupReference asks resolve until it answers, waiting backoff(n) |
| 280 | // after the n-th failure; false when ctx ends first. |
| 281 | func awaitGroupReference(ctx context.Context, resolve func() (string, error), backoff func(int) time.Duration, logf func(string, ...any)) bool { |
| 282 | for n := 1; ; n++ { |
| 283 | ref, err := resolve() |
| 284 | if err == nil { |
| 285 | logf("agent: this box's group reference is %s", ref) |
| 286 | return true |
| 287 | } |
| 288 | wait := backoff(n) |
| 289 | logf("agent: this box's group reference could not be read from its identity document (attempt %d): %v; the agent starts once it is, asking again in %s", n, err, wait) |
| 290 | select { |
| 291 | case <-ctx.Done(): |
| 292 | return false |
| 293 | case <-time.After(wait): |
| 294 | } |
| 295 | } |
| 296 | } |
| 297 | |
amadan