| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "strings" |
| 9 | "testing" |
| 10 | |
| 11 | "github.com/carlosframework/platform/internal/hostbox" |
| 12 | "github.com/carlosframework/platform/internal/hostclaim" |
| 13 | "github.com/carlosframework/platform/internal/hostguard" |
| 14 | ) |
| 15 | |
| 16 | type noProcessKernelFake struct { |
| 17 | stopKernelFunc |
| 18 | calls int |
| 19 | target hostguard.FenceTarget |
| 20 | err error |
| 21 | during func() |
| 22 | } |
| 23 | |
| 24 | func (k *noProcessKernelFake) VerifyNoProcess(_ context.Context, t hostguard.FenceTarget, _ StopInventory) (StopKernelEvidence, error) { |
| 25 | k.calls++ |
| 26 | k.target = t |
| 27 | if k.during != nil { |
| 28 | k.during() |
| 29 | } |
| 30 | if k.err != nil { |
| 31 | return StopKernelEvidence{}, k.err |
| 32 | } |
| 33 | return StopKernelEvidence{InventorySHA256: strings.Repeat("b", 64), LocalJournalSHA256: strings.Repeat("c", 64), CgroupIdentities: []string{"absent:carlos-exec@shop.example"}, UnitInvocationIDs: []string{}}, nil |
| 34 | } |
| 35 | |
| 36 | // changingClaims returns obs on the first read and next afterwards. |
| 37 | type changingClaims struct { |
| 38 | obs, next hostclaim.Observed |
| 39 | reads int |
| 40 | change bool |
| 41 | } |
| 42 | |
| 43 | func (c *changingClaims) Read(context.Context, string) (hostclaim.Observed, error) { |
| 44 | c.reads++ |
| 45 | if c.change && c.reads > 1 { |
| 46 | return c.next, nil |
| 47 | } |
| 48 | return c.obs, nil |
| 49 | } |
| 50 | |
| 51 | // noProcessFixture is a created incarnation that never admitted a writer, |
| 52 | // executed on box-1 (serving, session 0). |
| 53 | func noProcessFixture(t *testing.T) (ClaimStopCoordinator, StopRequest, *noProcessKernelFake, *changingClaims) { |
| 54 | t.Helper() |
| 55 | c, req, _ := claimStopFixture(t) |
| 56 | observed, err := c.Claims.Read(t.Context(), req.Host) |
| 57 | if err != nil { |
| 58 | t.Fatal(err) |
| 59 | } |
| 60 | claim := *observed.Claim |
| 61 | claim.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: req.Box} |
| 62 | claim.Holder = &hostclaim.Holder{Incarnation: "inc-1", Stamp: 11} |
| 63 | observed.Claim = &claim |
| 64 | claims := &changingClaims{obs: observed} |
| 65 | c.Claims = claims |
| 66 | kernel := &noProcessKernelFake{} |
| 67 | c.Kernel = kernel |
| 68 | c.Drainer = stopDrainFunc(func(context.Context, StopRequest, string) (DrainAcknowledgement, error) { |
| 69 | t.Fatal("a session-zero stop asked for a producer drain") |
| 70 | return DrainAcknowledgement{}, nil |
| 71 | }) |
| 72 | req.Region = "eu-west-1" |
| 73 | req.WriterSession = 0 |
| 74 | req.OperationID, req.Phase, req.ObligationID = "", "", "" |
| 75 | return c, req, kernel, claims |
| 76 | } |
| 77 | |
| 78 | func TestNoProcessWitnessPublishesASessionZeroReceipt(t *testing.T) { |
| 79 | c, req, kernel, _ := noProcessFixture(t) |
| 80 | result, err := c.FenceAndStop(t.Context(), req) |
| 81 | if err != nil { |
| 82 | t.Fatal(err) |
| 83 | } |
| 84 | if kernel.calls != 1 || kernel.target.Session != 0 || kernel.target.FenceID != "" { |
| 85 | t.Fatalf("kernel = %d %+v", kernel.calls, kernel.target) |
| 86 | } |
| 87 | var receipt StopReceipt |
| 88 | if err := json.Unmarshal(result.Bytes, &receipt); err != nil { |
| 89 | t.Fatal(err) |
| 90 | } |
| 91 | if receipt.Kind != StopReceiptNoProcess || receipt.WriterSession != 0 || receipt.FenceID != "" || receipt.ProducerDrainDigest != "" || receipt.ClaimETag != req.ExpectedClaimETag || receipt.ClaimGen != 19 { |
| 92 | t.Fatalf("receipt = %+v", receipt) |
| 93 | } |
| 94 | if !strings.Contains(result.Key, "/serving-evidence/box-1/") { |
| 95 | t.Fatalf("key = %s", result.Key) |
| 96 | } |
| 97 | stored, _, err := c.Evidence.Get(t.Context(), result.Key) |
| 98 | if err != nil || string(stored) == "" { |
| 99 | t.Fatalf("not published: %v", err) |
| 100 | } |
| 101 | if _, err := validateTransportStop(req, result); err != nil { |
| 102 | t.Fatalf("the agent's transport check refuses it: %v", err) |
| 103 | } |
| 104 | // No fence is installed: the claim's ETag, which any admission changes, |
| 105 | // is what binds the witness. |
| 106 | called := false |
| 107 | if err := c.Guard.With(t.Context(), req.Host, hostguard.Captured{Incarnation: req.Incarnation, Stamp: req.InstallationStamp}, func(*hostclaim.Claim) error { return nil }, func() error { called = true; return nil }); err != nil || !called { |
| 108 | t.Fatalf("a fence was installed: %v", err) |
| 109 | } |
| 110 | } |
| 111 | |
| 112 | func TestNoProcessWitnessRefuses(t *testing.T) { |
| 113 | for name, tc := range map[string]struct { |
| 114 | change func(*ClaimStopCoordinator, *StopRequest, *noProcessKernelFake, *changingClaims) |
| 115 | want error |
| 116 | }{ |
| 117 | "a writer was admitted": {func(_ *ClaimStopCoordinator, _ *StopRequest, _ *noProcessKernelFake, cl *changingClaims) { |
| 118 | cl.obs.Claim.Holder.Admissions = []hostclaim.AdmissionRecord{{Box: "box-1", Session: 1}} |
| 119 | }, ErrClaimStopIdentity}, |
| 120 | "a legacy holder without retention": {func(_ *ClaimStopCoordinator, r *StopRequest, _ *noProcessKernelFake, cl *changingClaims) { |
| 121 | cl.obs.Claim.Holder.WriterBox = r.Box |
| 122 | }, ErrClaimStopIdentity}, |
| 123 | "a foreign writer box is named": {func(_ *ClaimStopCoordinator, _ *StopRequest, _ *noProcessKernelFake, cl *changingClaims) { |
| 124 | cl.obs.Claim.Holder.WriterBox = "box-2" |
| 125 | }, ErrClaimStopIdentity}, |
| 126 | "another executor": {func(_ *ClaimStopCoordinator, _ *StopRequest, _ *noProcessKernelFake, cl *changingClaims) { |
| 127 | cl.obs.Claim.Op.Exec = "box-2" |
| 128 | }, ErrClaimStopIdentity}, |
| 129 | "a stale claim": {func(_ *ClaimStopCoordinator, r *StopRequest, _ *noProcessKernelFake, _ *changingClaims) { |
| 130 | r.ExpectedClaimETag = "etag-0" |
| 131 | }, ErrClaimStopIdentity}, |
| 132 | "an admission during the observation": {func(_ *ClaimStopCoordinator, _ *StopRequest, _ *noProcessKernelFake, cl *changingClaims) { |
| 133 | next := cl.obs |
| 134 | claim := *next.Claim |
| 135 | claim.Holder = &hostclaim.Holder{Incarnation: "inc-1", Stamp: 11, Session: 1, WriterBox: "box-1"} |
| 136 | next.Claim, next.ETag = &claim, "etag-2" |
| 137 | cl.next, cl.change = next, true |
| 138 | }, ErrClaimStopIdentity}, |
| 139 | "recovery": {func(_ *ClaimStopCoordinator, r *StopRequest, _ *noProcessKernelFake, _ *changingClaims) { |
| 140 | r.Recover = true |
| 141 | }, ErrClaimStopUnavailable}, |
| 142 | "cleanup": {func(_ *ClaimStopCoordinator, r *StopRequest, _ *noProcessKernelFake, _ *changingClaims) { |
| 143 | r.Cleanup = true |
| 144 | }, ErrClaimStopUnavailable}, |
| 145 | "another controller": {func(_ *ClaimStopCoordinator, r *StopRequest, _ *noProcessKernelFake, _ *changingClaims) { |
| 146 | r.ControllerID = "controller-2" |
| 147 | }, ErrClaimStopIdentity}, |
| 148 | "the kernel sees a process": {func(_ *ClaimStopCoordinator, _ *StopRequest, k *noProcessKernelFake, _ *changingClaims) { |
| 149 | k.err = ErrClaimStopInventory |
| 150 | }, ErrClaimStopInventory}, |
| 151 | "a kernel without the witness": {func(c *ClaimStopCoordinator, _ *StopRequest, _ *noProcessKernelFake, _ *changingClaims) { |
| 152 | c.Kernel = stopKernelFunc(nil) |
| 153 | }, ErrClaimStopUnavailable}, |
| 154 | } { |
| 155 | t.Run(name, func(t *testing.T) { |
| 156 | c, req, kernel, claims := noProcessFixture(t) |
| 157 | tc.change(&c, &req, kernel, claims) |
| 158 | if _, err := c.FenceAndStop(t.Context(), req); !errors.Is(err, tc.want) { |
| 159 | t.Fatalf("got %v want %v", err, tc.want) |
| 160 | } |
| 161 | keys, _ := c.Evidence.List(t.Context(), "control/") |
| 162 | if len(keys) != 0 { |
| 163 | t.Fatalf("a refused witness published %v", keys) |
| 164 | } |
| 165 | }) |
| 166 | } |
| 167 | } |
| 168 | |
| 169 | // The agent asks root for the witness with its own tuple and offers no |
| 170 | // drain: the session never launched anything. |
| 171 | func TestStopCapturedSessionZeroAsksRootForTheWitness(t *testing.T) { |
| 172 | f := newRecoveryFixture(t) |
| 173 | observed := f.read() |
| 174 | if observed.Claim.Holder.Session != 0 { |
| 175 | t.Fatalf("fixture session = %d", observed.Claim.Holder.Session) |
| 176 | } |
| 177 | var got StopRequest |
| 178 | var drain StopDrainer |
| 179 | a := ClaimAssembly{Claims: f.claims, Wake: &ClaimWakeAssembly{}} |
| 180 | a.stopLocal = func(context.Context) (claimStopLocal, error) { |
| 181 | return claimStopLocal{Actor: hostbox.Actor{Box: "box-1", Enrollment: "enrollment-1", Region: "ie", Revision: 7, Binding: "7"}, BootID: "boot-1", ControllerID: "controller-1", PID: 42, ProcessStart: "start-42"}, nil |
| 182 | } |
| 183 | a.collectStop = func(_ context.Context, r StopRequest, d StopDrainer) (StopResult, error) { |
| 184 | got, drain = r, d |
| 185 | return StopResult{}, errors.New("stop here") |
| 186 | } |
| 187 | if err := a.stopCapturedSession(t.Context(), observed); err == nil || err.Error() != "stop here" { |
| 188 | t.Fatalf("got %v", err) |
| 189 | } |
| 190 | if got.WriterSession != 0 || got.BootID != "boot-1" || got.ControllerID != "controller-1" || got.Recover || got.ExpectedClaimETag != observed.ETag || got.Region != "ie" { |
| 191 | t.Fatalf("request = %+v", got) |
| 192 | } |
| 193 | if _, ok := drain.(recoveryStopDrain); !ok { |
| 194 | t.Fatalf("drain = %T", drain) |
| 195 | } |
| 196 | } |
| 197 | |
| 198 | // A refused admission at session zero launched nothing: its cleanup needs |
| 199 | // no witness, and publishes no receipt. |
| 200 | func TestFailedSessionZeroAdmissionStopsNothing(t *testing.T) { |
| 201 | stops := 0 |
| 202 | w := &ClaimWakeAssembly{Stop: func(context.Context, hostclaim.Observed) error { stops++; return nil }} |
| 203 | o := hostclaim.Observed{Host: "shop.example", Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{Incarnation: "inc-1", Stamp: 11}, Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box-1"}}} |
| 204 | if err := w.stopAttempt(t.Context(), &claimWakeAttempt{}, o); err != nil || stops != 0 { |
| 205 | t.Fatalf("stopAttempt = %v, stops %d", err, stops) |
| 206 | } |
| 207 | o.Claim.Holder.Session = 1 |
| 208 | if err := w.stopAttempt(t.Context(), &claimWakeAttempt{}, o); err != nil || stops != 1 { |
| 209 | t.Fatalf("an admitted session was not stopped: %v, stops %d", err, stops) |
| 210 | } |
| 211 | } |
| 212 | |
| 213 | func TestNoProcessWitnessRetainsBeforePublication(t *testing.T) { |
| 214 | for _, reject := range []bool{false, true} { |
| 215 | t.Run(fmt.Sprint(reject), func(t *testing.T) { |
| 216 | c, req, _, claims := noProcessFixture(t) |
| 217 | claims.obs.Claim.Holder.WriterBox = req.Box |
| 218 | calls := 0 |
| 219 | c.RetainNoProcess = func(ctx context.Context, o hostclaim.Observed, result StopResult) error { |
| 220 | calls++ |
| 221 | if o.ETag != req.ExpectedClaimETag || o.Claim.Holder.Session != 0 || result.Target.FenceID != "" { |
| 222 | t.Fatal("wrong retained observation") |
| 223 | } |
| 224 | keys, err := c.Evidence.List(ctx, "control/") |
| 225 | if err != nil || len(keys) != 0 { |
| 226 | t.Fatal("published before retention") |
| 227 | } |
| 228 | if reject { |
| 229 | return hostbox.ErrChanged |
| 230 | } |
| 231 | return nil |
| 232 | } |
| 233 | _, err := c.FenceAndStop(t.Context(), req) |
| 234 | if calls != 1 || (err != nil) != reject { |
| 235 | t.Fatalf("retention %d %v", calls, err) |
| 236 | } |
| 237 | if reject { |
| 238 | keys, _ := c.Evidence.List(t.Context(), "control/") |
| 239 | if len(keys) != 0 { |
| 240 | t.Fatal("published despite failed retention") |
| 241 | } |
| 242 | } |
| 243 | }) |
| 244 | } |
| 245 | } |
| 246 | |
amadan