| 1 | //go:build linux && rootrunner |
| 2 | |
| 3 | package host |
| 4 | |
| 5 | import ( |
| 6 | "bytes" |
| 7 | "context" |
| 8 | "encoding/json" |
| 9 | "io" |
| 10 | "net/http" |
| 11 | "net/http/httptest" |
| 12 | "os" |
| 13 | "os/exec" |
| 14 | "os/signal" |
| 15 | "runtime" |
| 16 | "strings" |
| 17 | "sync" |
| 18 | "syscall" |
| 19 | "testing" |
| 20 | "time" |
| 21 | |
| 22 | "github.com/carlosframework/platform/internal/hostbox" |
| 23 | "github.com/carlosframework/platform/internal/hostclaim" |
| 24 | execlease "github.com/carlosframework/platform/internal/hostexec" |
| 25 | "github.com/carlosframework/platform/internal/hostguard" |
| 26 | "github.com/carlosframework/platform/internal/instancectl" |
| 27 | "github.com/carlosframework/platform/internal/park" |
| 28 | ) |
| 29 | |
| 30 | type lifecycleControlTrace struct { |
| 31 | ClaimRootWithLaunch |
| 32 | t *testing.T |
| 33 | } |
| 34 | |
| 35 | type lifecycleStopTrace struct { |
| 36 | claimUnitStopKernel |
| 37 | t *testing.T |
| 38 | once sync.Once |
| 39 | } |
| 40 | |
| 41 | func (k *lifecycleStopTrace) VerifyStopped(ctx context.Context, target hostguard.FenceTarget, inv StopInventory, ack DrainAcknowledgement) (StopKernelEvidence, error) { |
| 42 | evidence, err := k.claimUnitStopKernel.VerifyStopped(ctx, target, inv, ack) |
| 43 | if err != nil { |
| 44 | k.once.Do(func() { |
| 45 | if _, descriptor, ok := strings.Cut(err.Error(), "descriptor remains at "); ok { |
| 46 | name, linkErr := os.Readlink(descriptor) |
| 47 | info, infoErr := os.ReadFile(strings.Replace(descriptor, "/fd/", "/fdinfo/", 1)) |
| 48 | k.t.Logf("retained descriptor %s target=%s link=%v info=%s read=%v", descriptor, name, linkErr, info, infoErr) |
| 49 | } |
| 50 | journal, readErr := k.Journal.Read(ctx, target) |
| 51 | k.t.Logf("stop kernel refusal: %v; journal=%+v read=%v acknowledgement=%+v inventory=%+v", err, journal, readErr, ack, inv) |
| 52 | }) |
| 53 | } |
| 54 | return evidence, err |
| 55 | } |
| 56 | |
| 57 | type lifecycleResponse struct { |
| 58 | http.ResponseWriter |
| 59 | status int |
| 60 | body bytes.Buffer |
| 61 | } |
| 62 | |
| 63 | func (w *lifecycleResponse) Unwrap() http.ResponseWriter { return w.ResponseWriter } |
| 64 | func (w *lifecycleResponse) WriteHeader(status int) { |
| 65 | w.status = status |
| 66 | w.ResponseWriter.WriteHeader(status) |
| 67 | } |
| 68 | func (w *lifecycleResponse) Write(body []byte) (int, error) { |
| 69 | if w.body.Len() < 64<<10 { |
| 70 | w.body.Write(body) |
| 71 | } |
| 72 | return w.ResponseWriter.Write(body) |
| 73 | } |
| 74 | func (h lifecycleControlTrace) ServeHTTP(w http.ResponseWriter, r *http.Request) { |
| 75 | var q hostbox.RoleLaunchRequest |
| 76 | var execution execlease.ExecutionRequest |
| 77 | if r.URL.Path == "/claim/launch" || r.URL.Path == "/claim/execution" { |
| 78 | raw, _ := io.ReadAll(r.Body) |
| 79 | r.Body = io.NopCloser(bytes.NewReader(raw)) |
| 80 | if r.URL.Path == "/claim/launch" { |
| 81 | json.Unmarshal(raw, &q) |
| 82 | } else { |
| 83 | json.Unmarshal(raw, &execution) |
| 84 | } |
| 85 | } |
| 86 | started := time.Now() |
| 87 | response := &lifecycleResponse{ResponseWriter: w} |
| 88 | h.ClaimRootWithLaunch.ServeHTTP(response, r) |
| 89 | if response.status >= 400 || bytes.Contains(response.body.Bytes(), []byte(`"Failed":true`)) { |
| 90 | if execution.Host != "" { |
| 91 | raw, err := h.Root.executionRuntime.state.Read(execution.Host + ".json") |
| 92 | diagnostic, cancel := context.WithTimeout(context.WithoutCancel(r.Context()), time.Second) |
| 93 | a, authorityErr := h.Root.executionAuthorityWithMigration(diagnostic, h.Root.executionRuntime.state, execution, execlease.ControllerProcess{}) |
| 94 | cancel() |
| 95 | h.t.Logf("execution current authority=%+v err=%v", a, authorityErr) |
| 96 | h.t.Logf("execution failure request=%+v retained=%s read=%v", execution, raw, err) |
| 97 | } |
| 98 | h.t.Logf("root RPC %s status=%d reply=%s", r.URL.Path, response.status, response.body.String()) |
| 99 | h.t.Log("root RPC elapsed", time.Since(started)) |
| 100 | } |
| 101 | if q.Op == hostbox.RoleLaunchStop && q.Reservation != nil { |
| 102 | in, err := h.Launch.Journal.Read(r.Context(), q.Reservation.Target) |
| 103 | h.t.Logf("stop record err=%v request=%+v record=%+v containment=%v", err, *q.Reservation, in, validatePersistedUnitContainment(r.Context(), in, h.Launch.Units)) |
| 104 | } |
| 105 | } |
| 106 | |
| 107 | type lifecycleKernelTrace struct { |
| 108 | ClaimLaunchKernel |
| 109 | t *testing.T |
| 110 | } |
| 111 | |
| 112 | func (k lifecycleKernelTrace) Observe(s instancectl.ClaimUnitStarted) (ClaimLaunchObserved, error) { |
| 113 | got, err := k.ClaimLaunchKernel.Observe(s) |
| 114 | k.t.Logf("launch observation: started=%+v observed=%+v error=%v", s, got, err) |
| 115 | return got, err |
| 116 | } |
| 117 | func (k lifecycleKernelTrace) VerifyDigest(ctx context.Context, path string, id hostbox.LaunchFile, digest string) error { |
| 118 | err := k.ClaimLaunchKernel.VerifyDigest(ctx, path, id, digest) |
| 119 | k.t.Log("launch digest", err) |
| 120 | return err |
| 121 | } |
| 122 | |
| 123 | type lifecycleUnitsTrace struct { |
| 124 | ClaimLaunchUnits |
| 125 | t *testing.T |
| 126 | } |
| 127 | |
| 128 | func (u lifecycleUnitsTrace) Placement(host string) (string, string, error) { |
| 129 | return claimLaunchPlacement(u.ClaimLaunchUnits, host) |
| 130 | } |
| 131 | func (u lifecycleUnitsTrace) StopInvocation(ctx context.Context, host, inv string) error { |
| 132 | err := claimStopInvocation(ctx, u.ClaimLaunchUnits, host, inv) |
| 133 | u.t.Log("stop invocation", inv, err) |
| 134 | return err |
| 135 | } |
| 136 | func (u lifecycleUnitsTrace) Gone(ctx context.Context, host, inv string) error { |
| 137 | err := u.ClaimLaunchUnits.Gone(ctx, host, inv) |
| 138 | u.t.Log("gone", inv, err) |
| 139 | return err |
| 140 | } |
| 141 | func (u lifecycleUnitsTrace) StartReserved(ctx context.Context, in instancectl.ClaimUnit, l instancectl.Launch, p instancectl.ClaimExecPermit, check func() error) (instancectl.ClaimUnitStarted, error) { |
| 142 | got, err := u.ClaimLaunchUnits.StartReserved(ctx, in, l, p, func() error { e := check(); u.t.Log("unit start revalidation", e); return e }) |
| 143 | u.t.Log("start reserved", got, err) |
| 144 | if err != nil { |
| 145 | raw, _ := exec.Command("journalctl", "-u", instancectl.UnitName(instancectl.RoleRun, in.Host), "--no-pager", "-n", "60").CombinedOutput() |
| 146 | u.t.Log(string(raw)) |
| 147 | } |
| 148 | return got, err |
| 149 | } |
| 150 | |
| 151 | type lifecycleRegionalTrace struct { |
| 152 | hostbox.RegionalEvidence |
| 153 | t *testing.T |
| 154 | once sync.Once |
| 155 | } |
| 156 | |
| 157 | func (r *lifecycleRegionalTrace) VerifyObligation(ctx context.Context, o hostclaim.Observed, ob hostclaim.Obligation) (hostclaim.ParkScope, error) { |
| 158 | scope, err := r.RegionalEvidence.(hostbox.RegionalObligationEvidence).VerifyObligation(ctx, o, ob) |
| 159 | if err != nil { |
| 160 | r.once.Do(func() { r.t.Logf("regional evidence: %v; obligation=%+v holder=%+v", err, ob, o.Claim.Holder) }) |
| 161 | } |
| 162 | return scope, err |
| 163 | } |
| 164 | |
| 165 | func lifecycleTraceRegional(t *testing.T, p *ClaimProviders) { |
| 166 | worker := p.Regional.(hostbox.RegionalWorker) |
| 167 | worker.Regional.Engine.Now = func() time.Time { return time.Now().Add(park.SignatureAge + hostbox.RequestLimit + time.Second) } |
| 168 | worker.Evidence = &lifecycleRegionalTrace{RegionalEvidence: worker.Evidence, t: t} |
| 169 | view, authorize := worker.Regional.Engine.View, worker.Regional.Engine.Authorize |
| 170 | var viewOnce, authorizeOnce sync.Once |
| 171 | worker.Regional.Engine.View = func(ctx context.Context, o hostclaim.Observed) (park.View, error) { |
| 172 | v, err := view(ctx, o) |
| 173 | if err != nil { |
| 174 | viewOnce.Do(func() { t.Log("regional view", err) }) |
| 175 | } |
| 176 | return v, err |
| 177 | } |
| 178 | worker.Regional.Engine.Authorize = func(ctx context.Context, o hostclaim.Observed, ob hostclaim.Obligation) error { |
| 179 | err := authorize(ctx, o, ob) |
| 180 | if err != nil { |
| 181 | authorizeOnce.Do(func() { t.Log("regional authorization", err) }) |
| 182 | } |
| 183 | return err |
| 184 | } |
| 185 | p.Regional = worker |
| 186 | } |
| 187 | |
| 188 | func lifecycleTraceParkState(t *testing.T, p ClaimProviders, o hostclaim.Observed) { |
| 189 | engine := p.Regional.(hostbox.RegionalWorker).Regional.Engine |
| 190 | v, err := engine.View(t.Context(), o) |
| 191 | t.Logf("park view=%+v err=%v holder=%+v", v, err, o.Claim.Holder) |
| 192 | if err != nil || o.Claim.Holder == nil { |
| 193 | return |
| 194 | } |
| 195 | live, err := park.LiveGenerations(o.Claim, v, o.Claim.Holder.Location) |
| 196 | t.Log("park live", live, err) |
| 197 | versions, err := engine.Store.ListVersions(t.Context(), engine.Bucket, o.Claim.Holder.Location+"/") |
| 198 | t.Log("park listing", err) |
| 199 | for _, version := range versions { |
| 200 | if strings.HasSuffix(version.Key, "/lock.json") { |
| 201 | continue |
| 202 | } |
| 203 | generation, known := v.GenerationOf(park.Ref{Key: version.Key, VersionID: version.VersionID}) |
| 204 | t.Logf("park version=%+v generation=%q known=%v", version, generation, known) |
| 205 | } |
| 206 | } |
| 207 | |
| 208 | func TestPlan14LifecycleTraceStreams(t *testing.T) { |
| 209 | sink := httptest.NewRecorder() |
| 210 | w := &lifecycleResponse{ResponseWriter: sink} |
| 211 | if _, err := w.Write([]byte("fence")); err != nil { |
| 212 | t.Fatal(err) |
| 213 | } |
| 214 | if err := http.NewResponseController(w).Flush(); err != nil { |
| 215 | t.Fatal(err) |
| 216 | } |
| 217 | if !sink.Flushed || sink.Body.String() != "fence" { |
| 218 | t.Fatal("drain frame buffered") |
| 219 | } |
| 220 | } |
| 221 | |
| 222 | func lifecycleStackTrace(t *testing.T) { |
| 223 | signals := make(chan os.Signal, 1) |
| 224 | signal.Notify(signals, syscall.SIGUSR1) |
| 225 | done := make(chan struct{}) |
| 226 | t.Cleanup(func() { signal.Stop(signals); close(done) }) |
| 227 | go func() { |
| 228 | for { |
| 229 | select { |
| 230 | case <-done: |
| 231 | return |
| 232 | case <-signals: |
| 233 | buf := make([]byte, 2<<20) |
| 234 | n := runtime.Stack(buf, true) |
| 235 | t.Logf("lifecycle goroutines:\n%s", buf[:n]) |
| 236 | } |
| 237 | } |
| 238 | }() |
| 239 | } |
| 240 | |
| 241 | func lifecycleCrashDiagnostic(t *testing.T) { |
| 242 | for _, unit := range []string{"carlos-edge.service", "carlos-instance-control.service"} { |
| 243 | out, err := exec.Command("systemctl", "kill", "--kill-who=main", "--signal=SIGUSR1", unit).CombinedOutput() |
| 244 | t.Logf("stack signal %s: %s %v", unit, out, err) |
| 245 | } |
| 246 | time.Sleep(time.Second) |
| 247 | out, err := exec.Command("journalctl", "-u", "carlos-edge.service", "--no-pager", "-o", "cat").CombinedOutput() |
| 248 | t.Logf("full edge log: %s %v", out, err) |
| 249 | } |
| 250 | |
amadan