| 1 | package hostactors |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "time" |
| 8 | |
| 9 | "github.com/carlosframework/platform/internal/release" |
| 10 | ) |
| 11 | |
| 12 | // Group membership, confirmed by the console (region HA spec 2 §5 step 5, |
| 13 | // H16, H28). Before it invites a box its region's group launched, and |
| 14 | // again before it enrolls it, the console asks the cloud, read-only and |
| 15 | // with its own credentials, whether the instance is a current member of |
| 16 | // the region's group. No box writes or contributes membership evidence: |
| 17 | // a confirmation exists only inside this package, built by |
| 18 | // confirmMembership from one MembershipReader answer, and nothing |
| 19 | // exported takes one in. |
| 20 | // |
| 21 | // What the cloud says is the reader's business; the AWS adapter is |
| 22 | // internal/hostactors/awsmembership. This file names no provider (§25). |
| 23 | |
| 24 | // MembershipReader answers, read-only, from the cloud the roster's region |
| 25 | // entries name. It is the console's own client. |
| 26 | type MembershipReader interface { |
| 27 | // Member describes instance in awsRegion: its group, if any, and its |
| 28 | // state. An instance in no group is a Membership with no GroupName. An |
| 29 | // error is a read that failed, never an answer; the Membership beside |
| 30 | // it then carries only the RequestIDs of the calls made, so the log |
| 31 | // keeps whatever ids the cloud gave. |
| 32 | Member(ctx context.Context, awsRegion, instance string) (Membership, error) |
| 33 | } |
| 34 | |
| 35 | // Membership is one instance's group membership, as the cloud reported it. |
| 36 | type Membership struct { |
| 37 | GroupName string |
| 38 | AvailabilityZoneRegion string |
| 39 | LifecycleState string |
| 40 | InstanceState string |
| 41 | Owner string |
| 42 | // RequestIDs are the cloud's request ids for the reads, one per call |
| 43 | // made, in order: the id the cloud gave, in its answer or in its error |
| 44 | // response, or NoRequestID for a call that failed before the cloud |
| 45 | // answered. |
| 46 | RequestIDs []string |
| 47 | } |
| 48 | |
| 49 | // NoRequestID stands in a Membership's RequestIDs for a call that failed |
| 50 | // before the cloud answered, so it gave no id. It is never an id: a read |
| 51 | // that carries it is not an answer. |
| 52 | const NoRequestID = "none" |
| 53 | |
| 54 | // MembershipEvidence is what a confirmation read, for the console's log: |
| 55 | // the cloud's request ids, and when the read began. |
| 56 | type MembershipEvidence struct { |
| 57 | RequestIDs []string |
| 58 | ObservedAt time.Time |
| 59 | } |
| 60 | |
| 61 | // membershipFreshness is the write deadline: a read's write is issued |
| 62 | // less than membershipFreshness after the read's anchor, never at or after |
| 63 | // it. |
| 64 | const membershipFreshness = time.Minute |
| 65 | |
| 66 | var ( |
| 67 | // ErrMembershipUnknown is a membership read that failed. It says |
| 68 | // nothing about the instance: the console tries again with the same |
| 69 | // attempt. |
| 70 | ErrMembershipUnknown = errors.New("hostactors: group membership could not be read") |
| 71 | // ErrNotMember is the cloud's positive answer that the instance is not |
| 72 | // a current, running member of the region's group in the deployment's |
| 73 | // account and AWS region. |
| 74 | ErrNotMember = errors.New("hostactors: the instance is not a current member of the region's group") |
| 75 | // ErrMembershipStale is a read that was already membershipFreshness old |
| 76 | // or more when the describe returned: the cloud was slow, and the |
| 77 | // answer counts against the attempt. |
| 78 | ErrMembershipStale = errors.New("hostactors: the membership read is more than a minute old") |
| 79 | // ErrWriteDeadline is a write the deadline stopped (writeBound): no |
| 80 | // store call was issued at or after it, but one already in flight was |
| 81 | // cancelled, so whether it committed is unknown. The next pass reads |
| 82 | // back and describes again. |
| 83 | ErrWriteDeadline = errors.New("hostactors: the write would be issued more than a minute after its membership read") |
| 84 | ) |
| 85 | |
| 86 | // membershipConfirmation is one read of an attempt's instance. It is built |
| 87 | // only by confirmMembership, from the reader's answer, and never leaves |
| 88 | // this package. A read the cloud answered carries its request ids and |
| 89 | // time whether or not it found a member, so every read can be logged. |
| 90 | type membershipConfirmation struct { |
| 91 | requestIDs []string |
| 92 | observedAt time.Time |
| 93 | } |
| 94 | |
| 95 | // confirmMembership asks r about instance in region's AWS region, and |
| 96 | // returns the confirmation only when the answer is a current member of |
| 97 | // region's group: Pending or InService, pending or running, owned by the |
| 98 | // region's account, in its AWS region, with a request id for each call. |
| 99 | // The read's anchor, its time, is taken before the call: never later than |
| 100 | // the describe's completion, so its age is never understated. |
| 101 | // |
| 102 | // The call gets at most storeCallTimeout, as each bound store call does: |
| 103 | // neither describe sets a deadline of its own, and a connection that |
| 104 | // stalls would otherwise hold the serial pass, and every box behind this |
| 105 | // one, for as long as it stalls. A call cut off, by the timeout or by the |
| 106 | // pass's own context, is unknown whatever it returned: the attempt stands |
| 107 | // and its wait does not grow. |
| 108 | func confirmMembership(ctx context.Context, r MembershipReader, region Region, instance string, now func() time.Time) (membershipConfirmation, error) { |
| 109 | if r == nil || now == nil { |
| 110 | return membershipConfirmation{}, fmt.Errorf("%w: no reader", ErrMembershipUnknown) |
| 111 | } |
| 112 | at := now().UTC() |
| 113 | rctx, cancel := context.WithTimeout(ctx, storeCallTimeout) |
| 114 | m, err := r.Member(rctx, region.AWSRegion, instance) |
| 115 | cut := rctx.Err() |
| 116 | cancel() |
| 117 | // The ids the cloud gave are kept whatever came of the read, failed |
| 118 | // calls included, for the console's log. |
| 119 | c := membershipConfirmation{requestIDs: append([]string(nil), m.RequestIDs...), observedAt: at} |
| 120 | if cut != nil { |
| 121 | return c, fmt.Errorf("%w: the read did not finish in time: %v", ErrMembershipUnknown, cut) |
| 122 | } |
| 123 | if err != nil { |
| 124 | return c, fmt.Errorf("%w: %v", ErrMembershipUnknown, err) |
| 125 | } |
| 126 | // A reply without an id from each call is not an answer, whatever it |
| 127 | // claims, so it is never counted against the attempt. |
| 128 | if len(m.RequestIDs) < 2 { |
| 129 | return c, fmt.Errorf("%w: the cloud gave no request ids", ErrMembershipUnknown) |
| 130 | } |
| 131 | for _, id := range m.RequestIDs { |
| 132 | if id == "" || id == NoRequestID { |
| 133 | return c, fmt.Errorf("%w: the cloud gave a blank request id", ErrMembershipUnknown) |
| 134 | } |
| 135 | } |
| 136 | switch { |
| 137 | case m.GroupName != region.GroupName: |
| 138 | return c, fmt.Errorf("%w: it is in group %q", ErrNotMember, m.GroupName) |
| 139 | case m.LifecycleState != "Pending" && m.LifecycleState != "InService": |
| 140 | return c, fmt.Errorf("%w: its lifecycle state is %q", ErrNotMember, m.LifecycleState) |
| 141 | case m.InstanceState == "" || m.Owner == "": |
| 142 | // Auto Scaling knows it but EC2 does not list it (yet, or any |
| 143 | // more): no answer. |
| 144 | return c, fmt.Errorf("%w: EC2 does not list it", ErrMembershipUnknown) |
| 145 | case m.InstanceState != "pending" && m.InstanceState != "running": |
| 146 | return c, fmt.Errorf("%w: it is %q", ErrNotMember, m.InstanceState) |
| 147 | case m.Owner != region.Account || m.AvailabilityZoneRegion != region.AWSRegion: |
| 148 | return c, fmt.Errorf("%w: it is in account %q, region %q", ErrNotMember, m.Owner, m.AvailabilityZoneRegion) |
| 149 | } |
| 150 | return c, nil |
| 151 | } |
| 152 | |
| 153 | // fresh refuses a confirmation membershipFreshness old or more at now, or |
| 154 | // one from the future: the describe itself was too slow. |
| 155 | func (c membershipConfirmation) fresh(now time.Time) error { |
| 156 | if c.observedAt.IsZero() || now.Sub(c.observedAt) >= membershipFreshness || c.observedAt.After(now.Add(5*time.Second)) { |
| 157 | return fmt.Errorf("%w (read at %s, checked at %s)", ErrMembershipStale, c.observedAt.Format(time.RFC3339), now.UTC().Format(time.RFC3339)) |
| 158 | } |
| 159 | return nil |
| 160 | } |
| 161 | |
| 162 | // deadline is the last moment a write this read allows may be issued. |
| 163 | func (c membershipConfirmation) deadline() time.Time { return c.observedAt.Add(membershipFreshness) } |
| 164 | |
| 165 | // atWrite is the bound store's check before each call: at or after the |
| 166 | // deadline it is ErrWriteDeadline, unknown, never a refusal. |
| 167 | func (c membershipConfirmation) atWrite(now time.Time) error { |
| 168 | if c.observedAt.IsZero() || !now.Before(c.deadline()) { |
| 169 | return fmt.Errorf("%w (read at %s, write at %s)", ErrWriteDeadline, c.observedAt.Format(time.RFC3339), now.UTC().Format(time.RFC3339)) |
| 170 | } |
| 171 | return nil |
| 172 | } |
| 173 | |
| 174 | func (c membershipConfirmation) evidence() MembershipEvidence { |
| 175 | return MembershipEvidence{RequestIDs: append([]string(nil), c.requestIDs...), ObservedAt: c.observedAt} |
| 176 | } |
| 177 | |
| 178 | // writeBound is d with every store call bound to c's deadline (C2, S1), |
| 179 | // the single enforcement point of the write's freshness. Each store call |
| 180 | // (a read, the write, the read-back after a lost write) is refused before |
| 181 | // it is issued once the deadline is reached, by either clock, the |
| 182 | // context's or now; the returned context ends at the deadline; and each |
| 183 | // call it does issue gets at most storeCallTimeout (U2). The sealed |
| 184 | // writers (Invite, Replace and the checkMembership it makes) run on it |
| 185 | // unchanged, so no read, signature's Put or read-back of theirs is issued |
| 186 | // 60 seconds or more after the membership read's anchor, and any outcome |
| 187 | // not confirmed in time is ErrWriteDeadline, unknown. |
| 188 | func writeBound(ctx context.Context, d Directory, c membershipConfirmation, now func() time.Time) (Directory, context.Context, context.CancelFunc) { |
| 189 | bctx, cancel := context.WithTimeout(ctx, c.deadline().Sub(now())) |
| 190 | bd := d |
| 191 | bd.Store = deadlineStore{inner: d.Store, check: func(callCtx context.Context) error { |
| 192 | if bctx.Err() != nil || callCtx.Err() != nil { |
| 193 | return fmt.Errorf("%w: the deadline passed", ErrWriteDeadline) |
| 194 | } |
| 195 | return c.atWrite(now()) |
| 196 | }} |
| 197 | return bd, bctx, cancel |
| 198 | } |
| 199 | |
| 200 | // storeCallTimeout is the hard limit on one store call of a bound write |
| 201 | // (U2): a call the store has not answered within it is unknown, and its |
| 202 | // context is cancelled, so an S3 request is abandoned at the HTTP layer. |
| 203 | // It bounds how long the console waits, not when a write commits: a |
| 204 | // call cut off may still commit later, and is settled from the roster on |
| 205 | // a later pass. confirmMembership gives each membership read the same |
| 206 | // limit. A var so tests can shorten it. |
| 207 | var storeCallTimeout = 15 * time.Second |
| 208 | |
| 209 | // deadlineStore refuses every call once check says the deadline has been |
| 210 | // reached, before the call reaches the store underneath, and gives each |
| 211 | // call it issues at most storeCallTimeout. The bound cannot rest on the |
| 212 | // store alone: release.DirStore ignores its context, so the wrapper |
| 213 | // returns on its own when the call outlives the timeout or the deadline, |
| 214 | // whatever the store does. Every outcome it cannot confirm in time is |
| 215 | // ErrWriteDeadline, unknown: a call cut off in flight, one that never |
| 216 | // answered, and one that answered, even successfully, only after the |
| 217 | // deadline. |
| 218 | // |
| 219 | // The store underneath is a named field, not embedded, and every method of |
| 220 | // release.Store is written out here: a method added to the interface later |
| 221 | // fails to compile instead of reaching the store past the bound. |
| 222 | type deadlineStore struct { |
| 223 | inner release.Store |
| 224 | check func(context.Context) error |
| 225 | } |
| 226 | |
| 227 | var _ release.Store = deadlineStore{} |
| 228 | |
| 229 | // call runs one store call under the bound. |
| 230 | func (s deadlineStore) call(ctx context.Context, f func(context.Context) error) error { |
| 231 | if err := s.check(ctx); err != nil { |
| 232 | return err |
| 233 | } |
| 234 | cctx, cancel := context.WithTimeout(ctx, storeCallTimeout) |
| 235 | defer cancel() |
| 236 | done := make(chan error, 1) |
| 237 | go func() { done <- f(cctx) }() |
| 238 | var err error |
| 239 | select { |
| 240 | case err = <-done: |
| 241 | case <-cctx.Done(): |
| 242 | } |
| 243 | // Cut off, whether the store gave up on the context or not: by the |
| 244 | // deadline, or by the hard timeout. Whatever it answered, if anything, |
| 245 | // is not known to be the outcome. |
| 246 | if cctx.Err() != nil { |
| 247 | if ctx.Err() != nil { |
| 248 | return fmt.Errorf("%w: in flight at the deadline", ErrWriteDeadline) |
| 249 | } |
| 250 | return fmt.Errorf("%w: the store did not answer in time", ErrWriteDeadline) |
| 251 | } |
| 252 | if err == nil { |
| 253 | if late := s.check(ctx); late != nil { |
| 254 | return fmt.Errorf("%w: it answered only after the deadline", ErrWriteDeadline) |
| 255 | } |
| 256 | } |
| 257 | return err |
| 258 | } |
| 259 | |
| 260 | func (s deadlineStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 261 | var body []byte |
| 262 | var etag string |
| 263 | err := s.call(ctx, func(cctx context.Context) error { |
| 264 | var err error |
| 265 | body, etag, err = s.inner.Get(cctx, key) |
| 266 | return err |
| 267 | }) |
| 268 | if err != nil { |
| 269 | return nil, "", err |
| 270 | } |
| 271 | return body, etag, nil |
| 272 | } |
| 273 | |
| 274 | func (s deadlineStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| 275 | var etag string |
| 276 | err := s.call(ctx, func(cctx context.Context) error { |
| 277 | var err error |
| 278 | etag, err = s.inner.Put(cctx, key, body, ifMatch) |
| 279 | return err |
| 280 | }) |
| 281 | if err != nil { |
| 282 | return "", err |
| 283 | } |
| 284 | return etag, nil |
| 285 | } |
| 286 | |
| 287 | func (s deadlineStore) List(ctx context.Context, prefix string) ([]string, error) { |
| 288 | var keys []string |
| 289 | err := s.call(ctx, func(cctx context.Context) error { |
| 290 | var err error |
| 291 | keys, err = s.inner.List(cctx, prefix) |
| 292 | return err |
| 293 | }) |
| 294 | if err != nil { |
| 295 | return nil, err |
| 296 | } |
| 297 | return keys, nil |
| 298 | } |
| 299 | |
| 300 | func (s deadlineStore) ListDirs(ctx context.Context, prefix string) ([]string, error) { |
| 301 | var dirs []string |
| 302 | err := s.call(ctx, func(cctx context.Context) error { |
| 303 | var err error |
| 304 | dirs, err = s.inner.ListDirs(cctx, prefix) |
| 305 | return err |
| 306 | }) |
| 307 | if err != nil { |
| 308 | return nil, err |
| 309 | } |
| 310 | return dirs, nil |
| 311 | } |
| 312 | |
| 313 | func (s deadlineStore) Delete(ctx context.Context, key string) error { |
| 314 | return s.call(ctx, func(cctx context.Context) error { return s.inner.Delete(cctx, key) }) |
| 315 | } |
| 316 | |
amadan