| 1 | //go:build linux |
| 2 | |
| 3 | package host |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "net" |
| 10 | "os" |
| 11 | "path/filepath" |
| 12 | "strings" |
| 13 | "testing" |
| 14 | "time" |
| 15 | |
| 16 | "github.com/carlosframework/platform/internal/hostbox" |
| 17 | "github.com/carlosframework/platform/internal/hostclaim" |
| 18 | execlease "github.com/carlosframework/platform/internal/hostexec" |
| 19 | "github.com/carlosframework/platform/internal/hostwrite" |
| 20 | "github.com/carlosframework/platform/internal/instancectl" |
| 21 | "github.com/carlosframework/platform/internal/provision" |
| 22 | ) |
| 23 | |
| 24 | type rev2Local struct{ err error } |
| 25 | |
| 26 | func (l *rev2Local) Check(context.Context, hostbox.Actor, *hostclaim.Claim, hostbox.Selection) error { |
| 27 | return l.err |
| 28 | } |
| 29 | |
| 30 | type rev2Clock struct { |
| 31 | wall time.Time |
| 32 | boot int64 |
| 33 | } |
| 34 | |
| 35 | func (c *rev2Clock) Now() (time.Time, int64, error) { return c.wall, c.boot, nil } |
| 36 | func (c *rev2Clock) After(d time.Duration) <-chan time.Time { return time.After(d) } |
| 37 | |
| 38 | type rev2Attempts struct { |
| 39 | attempt hostclaim.WakeRetryAttempt |
| 40 | op hostclaim.Op |
| 41 | stamp, session uint64 |
| 42 | } |
| 43 | |
| 44 | func (a *rev2Attempts) ReadAttempt(_ context.Context, o hostclaim.Observed) (hostclaim.WakeRetryAttempt, error) { |
| 45 | if o.Claim.Op.ID != a.op.ID || o.Claim.Op.Kind != a.op.Kind || o.Claim.Op.Phase != a.op.Phase || o.Claim.Holder.Stamp != a.stamp || o.Claim.Holder.Session != a.session { |
| 46 | return hostclaim.WakeRetryAttempt{}, hostbox.ErrChanged |
| 47 | } |
| 48 | return a.attempt, nil |
| 49 | } |
| 50 | |
| 51 | func configureRev2Launch(t *testing.T, f *launchFixture) { |
| 52 | t.Helper() |
| 53 | f.root.DataRoot = f.root.Guard.DataDir |
| 54 | f.root.Local = &rev2Local{} |
| 55 | f.root.Clock = &rev2Clock{wall: time.Now(), boot: 100} |
| 56 | f.root.ReadyBudget, f.root.ReadyPoll = 40*time.Millisecond, time.Millisecond |
| 57 | f.root.ExecLockDir = t.TempDir() |
| 58 | f.kernel.digest = strings.Repeat("a", 64) |
| 59 | f.kernel.observed.Running = f.kernel.observed.Executable |
| 60 | f.req.Spec = &hostbox.LaunchSpec{Identity: hostwrite.Identity{Config: "config", Release: "release"}, Executable: f.kernel.path, Digest: f.kernel.digest, Args: []string{"--socket", filepath.Join(provision.InstanceRunDir(f.target.Host), "app.sock"), "--db", provision.NestedDBPath(f.root.DataRoot, f.target.Host)}, Env: []string{"CARLOS_ADMIN_TOKEN=test"}} |
| 61 | digest, err := f.req.Spec.Hash() |
| 62 | if err != nil { |
| 63 | t.Fatal(err) |
| 64 | } |
| 65 | f.req.Reservation.Inputs = digest |
| 66 | a := hostclaim.WakeRetryAttempt{ID: "attempt-1", InvocationID: "inv-1", LaunchInputs: digest, Box: f.target.Box, Enrollment: f.target.Enrollment, BootID: f.peer.BootID, ControllerID: f.peer.ControllerID, Session: f.target.Session} |
| 67 | if f.claims.obs.Claim.Op.Kind == hostclaim.KindServing { |
| 68 | f.claims.set(func(c *hostclaim.Claim) { |
| 69 | c.Holder.WakeRetry = &hostclaim.WakeRetry{EpisodeID: "episode", State: "attempting", AttemptsReserved: 1, Attempt: &a} |
| 70 | }) |
| 71 | } else { |
| 72 | f.root.Attempts = map[string]claimLaunchAttemptReader{f.claims.obs.Claim.Op.Kind: &rev2Attempts{attempt: a, op: f.claims.obs.Claim.Op, stamp: f.target.Stamp, session: f.target.Session}} |
| 73 | } |
| 74 | configureLaunchParticipation(t, f) |
| 75 | } |
| 76 | func writeRev2Local(t *testing.T, f *launchFixture, status string, deadline int64) { |
| 77 | t.Helper() |
| 78 | q := execlease.ExecutionRequest{Schema: 1, ID: "begin", Operation: "executionBegin", Host: f.target.Host, Incarnation: f.target.Incarnation, Stamp: f.target.Stamp, Session: f.target.Session, Op: f.target.Op} |
| 79 | r := execlease.ExecutionRecord{Schema: 1, Request: q, Peer: execlease.ControllerProcess{KernelBootID: f.peer.BootID, PID: f.peer.PID, Start: f.peer.ProcessStart}, Authority: execlease.ExecutionAuthority{Box: f.target.Box, Enrollment: f.target.Enrollment, Migration: 1, Installation: "root-installation"}, Broker: "root", Token: "root-token", Status: status, SafeUntil: deadline, LeaseETag: "root-etag", Expires: time.Now().Add(time.Minute), Replay: map[string]execlease.ExecutionReplay{}} |
| 80 | raw, _ := json.Marshal(r) |
| 81 | if err := f.root.executionState.Write(f.target.Host+".json", raw); err != nil { |
| 82 | t.Fatal(err) |
| 83 | } |
| 84 | } |
| 85 | |
| 86 | func rev2Operation() hostclaim.Op { |
| 87 | return hostclaim.Op{Kind: hostclaim.KindRelocating, ID: "operation", Phase: "probing", Exec: "box-1", Journal: map[string]string{"install_stamp": "11"}} |
| 88 | } |
| 89 | |
| 90 | func TestPlan11Rev2OperationVerifierAndLocalState(t *testing.T) { |
| 91 | for _, name := range []string{"valid", "no verifier", "wrong attempt", "wrong input", "expired", "pending", "lost", "admission closed", "released live lock", "missing state"} { |
| 92 | t.Run(name, func(t *testing.T) { |
| 93 | f := newLaunchFixture(t, rev2Operation()) |
| 94 | switch name { |
| 95 | case "no verifier": |
| 96 | f.root.Attempts = nil |
| 97 | case "wrong attempt": |
| 98 | f.root.Attempts[hostclaim.KindRelocating].(*rev2Attempts).attempt.InvocationID = "other" |
| 99 | case "wrong input": |
| 100 | f.req.Spec.Env = []string{"CARLOS_ADMIN_TOKEN=other"} |
| 101 | case "expired": |
| 102 | writeRev2Local(t, f, "active", 100) |
| 103 | case "pending": |
| 104 | writeRev2Local(t, f, "pending", 1000) |
| 105 | case "lost": |
| 106 | writeRev2Local(t, f, "lost", 1000) |
| 107 | case "admission closed": |
| 108 | f.root.Local.(*rev2Local).err = execlease.ErrLost |
| 109 | case "released live lock": |
| 110 | if err := os.Remove(filepath.Join(f.root.executionState.(*launchParticipationDisk).dir, f.target.Host+".live")); err != nil { |
| 111 | t.Fatal(err) |
| 112 | } |
| 113 | case "missing state": |
| 114 | if err := os.Remove(filepath.Join(f.root.executionState.(*launchParticipationDisk).dir, f.target.Host+".json")); err != nil { |
| 115 | t.Fatal(err) |
| 116 | } |
| 117 | } |
| 118 | _, err := f.root.Serve(t.Context(), f.peer, f.req) |
| 119 | if name == "valid" { |
| 120 | if err != nil { |
| 121 | t.Fatal(err) |
| 122 | } |
| 123 | return |
| 124 | } |
| 125 | if err == nil { |
| 126 | t.Fatal("unsafe operation launch") |
| 127 | } |
| 128 | f.noReservation(t, name) |
| 129 | }) |
| 130 | } |
| 131 | } |
| 132 | |
| 133 | func TestPlan11Rev2RechecksAfterUnitPreparation(t *testing.T) { |
| 134 | for _, name := range []string{"deadline", "quiet", "closed", "reservation", "roster"} { |
| 135 | t.Run(name, func(t *testing.T) { |
| 136 | f := newLaunchFixture(t, rev2Operation()) |
| 137 | f.units.beforeCheck = func() { |
| 138 | switch name { |
| 139 | case "deadline": |
| 140 | f.root.Clock.(*rev2Clock).boot = 1000 |
| 141 | case "quiet": |
| 142 | f.claims.set(func(c *hostclaim.Claim) { c.QuietAfter = map[string]time.Time{"eu": time.Now().Add(time.Hour)} }) |
| 143 | case "closed": |
| 144 | f.root.Local.(*rev2Local).err = execlease.ErrLost |
| 145 | case "reservation": |
| 146 | f.root.Attempts[hostclaim.KindRelocating].(*rev2Attempts).attempt.ID = "replaced" |
| 147 | case "roster": |
| 148 | f.actor.Eligible = false |
| 149 | } |
| 150 | } |
| 151 | if _, err := f.root.Serve(t.Context(), f.peer, f.req); err == nil { |
| 152 | t.Fatal("stale final check") |
| 153 | } |
| 154 | if starts, _ := f.units.counts(); starts != 0 { |
| 155 | t.Fatal("tenant code started") |
| 156 | } |
| 157 | got, err := f.journal.UnitReservation(t.Context(), f.target, f.req.Reservation.ID) |
| 158 | if err != nil || got.State != hostbox.UnitSettled { |
| 159 | t.Fatalf("unstarted intention not settled: %+v %v", got, err) |
| 160 | } |
| 161 | }) |
| 162 | } |
| 163 | } |
| 164 | |
| 165 | func TestPlan11Rev2WaitsForSameInvocation(t *testing.T) { |
| 166 | for _, name := range []string{"slow valid", "timeout", "replacement invocation", "reused pid", "wrong executable", "cancel"} { |
| 167 | t.Run(name, func(t *testing.T) { |
| 168 | f := newLaunchFixture(t, servingOp()) |
| 169 | calls := 0 |
| 170 | ctx, cancel := context.WithCancel(t.Context()) |
| 171 | defer cancel() |
| 172 | f.kernel.observe = func(instancectl.ClaimUnitStarted) (ClaimLaunchObserved, error) { |
| 173 | calls++ |
| 174 | got := f.kernel.observed |
| 175 | if name == "replacement invocation" && calls == 1 { |
| 176 | f.units.sameErr = instancectl.ErrClaimUnit |
| 177 | } |
| 178 | if name == "reused pid" && calls > 1 { |
| 179 | got.ProcessStart = "reused" |
| 180 | } |
| 181 | if name == "cancel" { |
| 182 | cancel() |
| 183 | } |
| 184 | if calls < 4 || name == "timeout" { |
| 185 | return ClaimLaunchObserved{ProcessStart: got.ProcessStart, Group: got.Group}, errClaimLaunchPending |
| 186 | } |
| 187 | if name == "wrong executable" { |
| 188 | got.Running.Inode++ |
| 189 | } |
| 190 | return got, nil |
| 191 | } |
| 192 | _, err := f.root.Serve(ctx, f.peer, f.req) |
| 193 | starts, stops := f.units.counts() |
| 194 | if starts != 1 || f.claims.obs.Claim.Holder.WakeRetry.AttemptsReserved != 1 { |
| 195 | t.Fatalf("readiness consumed an attempt: %d", starts) |
| 196 | } |
| 197 | if name == "slow valid" { |
| 198 | if err != nil || stops != 0 || calls != 4 { |
| 199 | t.Fatalf("valid start killed: %v calls=%d stops=%d", err, calls, stops) |
| 200 | } |
| 201 | return |
| 202 | } |
| 203 | wantStops := 1 |
| 204 | if name == "replacement invocation" { |
| 205 | wantStops = 0 |
| 206 | } |
| 207 | if err == nil || stops != wantStops { |
| 208 | t.Fatalf("unverified start granted or replacement stopped: %v stops=%d", err, stops) |
| 209 | } |
| 210 | if name == "timeout" && !errors.Is(err, context.DeadlineExceeded) { |
| 211 | t.Fatal(err) |
| 212 | } |
| 213 | }) |
| 214 | } |
| 215 | } |
| 216 | |
| 217 | func TestPlan11Rev2DigestAndExecutable(t *testing.T) { |
| 218 | f := newLaunchFixture(t, servingOp()) |
| 219 | s := f.req.Spec |
| 220 | want, err := ClaimLaunchDigest(ClaimLaunchInputs{Identity: s.Identity, Executable: s.Executable, ExecutableSHA256: s.Digest, Args: s.Args, Env: s.Env}) |
| 221 | got, e := s.Hash() |
| 222 | if err != nil || e != nil || got != want { |
| 223 | t.Fatalf("digest contract drift: %q %q %v %v", got, want, err, e) |
| 224 | } |
| 225 | for _, name := range []string{"config", "release", "executable", "sha", "args", "env", "inode"} { |
| 226 | t.Run(name, func(t *testing.T) { |
| 227 | f := newLaunchFixture(t, servingOp()) |
| 228 | s := *f.req.Spec |
| 229 | f.req.Spec = &s |
| 230 | switch name { |
| 231 | case "config": |
| 232 | s.Identity.Config = "other" |
| 233 | case "release": |
| 234 | s.Identity.Release = "other" |
| 235 | case "executable": |
| 236 | s.Executable = "/opt/other" |
| 237 | case "sha": |
| 238 | s.Digest = strings.Repeat("b", 64) |
| 239 | case "args": |
| 240 | s.Args = append([]string(nil), s.Args...) |
| 241 | s.Args[1] += "x" |
| 242 | case "env": |
| 243 | s.Env = []string{"CARLOS_ADMIN_TOKEN=other"} |
| 244 | case "inode": |
| 245 | f.req.Reservation.Executable.Inode++ |
| 246 | } |
| 247 | if _, err := f.root.Serve(t.Context(), f.peer, f.req); err == nil { |
| 248 | t.Fatal("input substitution launched") |
| 249 | } |
| 250 | f.noReservation(t, name) |
| 251 | }) |
| 252 | } |
| 253 | } |
| 254 | |
| 255 | func TestPlan11Rev2KindPreflightRefusesUnverifiedOperation(t *testing.T) { |
| 256 | f := newLaunchFixture(t, servingOp()) |
| 257 | f.root.Authenticate = func(context.Context, net.Conn) (StopPeer, error) { return f.peer, nil } |
| 258 | client := NewClaimRoleRootClient(serveClaimRoot(t, ClaimRootWithLaunch{Launch: f.root})) |
| 259 | roles := &ClaimRoleAssembly{Root: client} |
| 260 | for _, kind := range []string{hostclaim.KindLanding, hostclaim.KindRelocating, "unknown"} { |
| 261 | if err := roles.PreflightKind(t.Context(), kind); err == nil { |
| 262 | t.Fatalf("unverified %s passed prerequisites", kind) |
| 263 | } |
| 264 | } |
| 265 | f.noReservation(t, "unsupported kinds") |
| 266 | if err := roles.PreflightKind(t.Context(), hostclaim.KindServing); err != nil { |
| 267 | t.Fatal(err) |
| 268 | } |
| 269 | } |
| 270 | |
amadan