diff --git a/api/compute/compute_v1alpha/schema.gen.go b/api/compute/compute_v1alpha/schema.gen.go index 280700d4e..7757e6055 100644 --- a/api/compute/compute_v1alpha/schema.gen.go +++ b/api/compute/compute_v1alpha/schema.gen.go @@ -1075,44 +1075,58 @@ func (o *Node) InitSchema(sb *schema.SchemaBuilder) { } const ( - SandboxBoundPortId = entity.Id("dev.miren.compute/sandbox.bound_port") - SandboxContainerId = entity.Id("dev.miren.compute/sandbox.container") - SandboxExitId = entity.Id("dev.miren.compute/sandbox.exit") - SandboxHostNetworkId = entity.Id("dev.miren.compute/sandbox.hostNetwork") - SandboxLabelsId = entity.Id("dev.miren.compute/sandbox.labels") - SandboxLastActivityId = entity.Id("dev.miren.compute/sandbox.last_activity") - SandboxLogAttributeId = entity.Id("dev.miren.compute/sandbox.logAttribute") - SandboxLogEntityId = entity.Id("dev.miren.compute/sandbox.logEntity") - SandboxNetworkId = entity.Id("dev.miren.compute/sandbox.network") - SandboxRouteId = entity.Id("dev.miren.compute/sandbox.route") - SandboxSpecId = entity.Id("dev.miren.compute/sandbox.spec") - SandboxStaticHostId = entity.Id("dev.miren.compute/sandbox.static_host") - SandboxStatusId = entity.Id("dev.miren.compute/sandbox.status") - SandboxStatusPendingId = entity.Id("dev.miren.compute/status.pending") - SandboxStatusNotReadyId = entity.Id("dev.miren.compute/status.not_ready") - SandboxStatusRunningId = entity.Id("dev.miren.compute/status.running") - SandboxStatusStoppedId = entity.Id("dev.miren.compute/status.stopped") - SandboxStatusDeadId = entity.Id("dev.miren.compute/status.dead") - SandboxVolumeId = entity.Id("dev.miren.compute/sandbox.volume") + SandboxBoundPortId = entity.Id("dev.miren.compute/sandbox.bound_port") + SandboxContainerId = entity.Id("dev.miren.compute/sandbox.container") + SandboxExitId = entity.Id("dev.miren.compute/sandbox.exit") + SandboxHostNetworkId = entity.Id("dev.miren.compute/sandbox.hostNetwork") + SandboxLabelsId = entity.Id("dev.miren.compute/sandbox.labels") + SandboxLastActivityId = entity.Id("dev.miren.compute/sandbox.last_activity") + SandboxLogAttributeId = entity.Id("dev.miren.compute/sandbox.logAttribute") + SandboxLogEntityId = entity.Id("dev.miren.compute/sandbox.logEntity") + SandboxNetworkId = entity.Id("dev.miren.compute/sandbox.network") + SandboxRouteId = entity.Id("dev.miren.compute/sandbox.route") + SandboxSpecId = entity.Id("dev.miren.compute/sandbox.spec") + SandboxStartupOutcomeId = entity.Id("dev.miren.compute/sandbox.startup_outcome") + SandboxStartupOutcomeStartupRunningId = entity.Id("dev.miren.compute/startup_outcome.startup_running") + SandboxStartupOutcomeStartupFailedId = entity.Id("dev.miren.compute/startup_outcome.startup_failed") + SandboxStaticHostId = entity.Id("dev.miren.compute/sandbox.static_host") + SandboxStatusId = entity.Id("dev.miren.compute/sandbox.status") + SandboxStatusPendingId = entity.Id("dev.miren.compute/status.pending") + SandboxStatusNotReadyId = entity.Id("dev.miren.compute/status.not_ready") + SandboxStatusRunningId = entity.Id("dev.miren.compute/status.running") + SandboxStatusStoppedId = entity.Id("dev.miren.compute/status.stopped") + SandboxStatusDeadId = entity.Id("dev.miren.compute/status.dead") + SandboxVolumeId = entity.Id("dev.miren.compute/sandbox.volume") ) type Sandbox struct { - ID entity.Id `json:"id"` - BoundPort []BoundPort `cbor:"bound_port,omitempty" json:"bound_port,omitempty"` - Container []Container `cbor:"container" json:"container"` - Exit Exit `cbor:"exit,omitempty" json:"exit"` - HostNetwork bool `cbor:"hostNetwork,omitempty" json:"hostNetwork,omitempty"` - Labels []string `cbor:"labels,omitempty" json:"labels,omitempty"` - LastActivity time.Time `cbor:"last_activity,omitempty" json:"last_activity"` - LogAttribute types.Labels `cbor:"logAttribute,omitempty" json:"logAttribute,omitempty"` - LogEntity string `cbor:"logEntity,omitempty" json:"logEntity,omitempty"` - Network []Network `cbor:"network,omitempty" json:"network,omitempty"` - Route []Route `cbor:"route,omitempty" json:"route,omitempty"` - Spec SandboxSpec `cbor:"spec,omitempty" json:"spec"` - StaticHost []StaticHost `cbor:"static_host,omitempty" json:"static_host,omitempty"` - Status SandboxStatus `cbor:"status,omitempty" json:"status,omitempty"` - Volume []Volume `cbor:"volume,omitempty" json:"volume,omitempty"` -} + ID entity.Id `json:"id"` + BoundPort []BoundPort `cbor:"bound_port,omitempty" json:"bound_port,omitempty"` + Container []Container `cbor:"container" json:"container"` + Exit Exit `cbor:"exit,omitempty" json:"exit"` + HostNetwork bool `cbor:"hostNetwork,omitempty" json:"hostNetwork,omitempty"` + Labels []string `cbor:"labels,omitempty" json:"labels,omitempty"` + LastActivity time.Time `cbor:"last_activity,omitempty" json:"last_activity"` + LogAttribute types.Labels `cbor:"logAttribute,omitempty" json:"logAttribute,omitempty"` + LogEntity string `cbor:"logEntity,omitempty" json:"logEntity,omitempty"` + Network []Network `cbor:"network,omitempty" json:"network,omitempty"` + Route []Route `cbor:"route,omitempty" json:"route,omitempty"` + Spec SandboxSpec `cbor:"spec,omitempty" json:"spec"` + StartupOutcome SandboxStartupOutcome `cbor:"startup_outcome,omitempty" json:"startup_outcome,omitempty"` + StaticHost []StaticHost `cbor:"static_host,omitempty" json:"static_host,omitempty"` + Status SandboxStatus `cbor:"status,omitempty" json:"status,omitempty"` + Volume []Volume `cbor:"volume,omitempty" json:"volume,omitempty"` +} + +type SandboxStartupOutcome string + +const ( + STARTUP_RUNNING SandboxStartupOutcome = "startup_outcome.startup_running" + STARTUP_FAILED SandboxStartupOutcome = "startup_outcome.startup_failed" +) + +var sandboxstartup_outcomeFromId = map[entity.Id]SandboxStartupOutcome{SandboxStartupOutcomeStartupRunningId: STARTUP_RUNNING, SandboxStartupOutcomeStartupFailedId: STARTUP_FAILED} +var sandboxstartup_outcomeToId = map[SandboxStartupOutcome]entity.Id{STARTUP_RUNNING: SandboxStartupOutcomeStartupRunningId, STARTUP_FAILED: SandboxStartupOutcomeStartupFailedId} type SandboxStatus string @@ -1182,6 +1196,9 @@ func (o *Sandbox) Decode(e entity.AttrGetter) { if a, ok := e.Get(SandboxSpecId); ok && a.Value.Kind() == entity.KindComponent { o.Spec.Decode(a.Value.Component()) } + if a, ok := e.Get(SandboxStartupOutcomeId); ok && a.Value.Kind() == entity.KindId { + o.StartupOutcome = sandboxstartup_outcomeFromId[a.Value.Id()] + } for _, a := range e.GetAll(SandboxStaticHostId) { if a.Value.Kind() == entity.KindComponent { var v StaticHost @@ -1249,6 +1266,9 @@ func (o *Sandbox) Encode() (attrs []entity.Attr) { if !o.Spec.Empty() { attrs = append(attrs, entity.Component(SandboxSpecId, o.Spec.Encode())) } + if a, ok := sandboxstartup_outcomeToId[o.StartupOutcome]; ok { + attrs = append(attrs, entity.Ref(SandboxStartupOutcomeId, a)) + } for _, v := range o.StaticHost { attrs = append(attrs, entity.Component(SandboxStaticHostId, v.Encode())) } @@ -1296,6 +1316,9 @@ func (o *Sandbox) Empty() bool { if !o.Spec.Empty() { return false } + if o.StartupOutcome != "" { + return false + } if len(o.StaticHost) != 0 { return false } @@ -1325,6 +1348,9 @@ func (o *Sandbox) InitSchema(sb *schema.SchemaBuilder) { sb.Component("route", "dev.miren.compute/sandbox.route", schema.Doc("A network route the container uses"), schema.Many) (&Route{}).InitSchema(sb.Builder("sandbox.route")) sb.Component("spec", "dev.miren.compute/sandbox.spec", schema.Doc("Immutable sandbox configuration")) + sb.Singleton("dev.miren.compute/startup_outcome.startup_running") + sb.Singleton("dev.miren.compute/startup_outcome.startup_failed") + sb.Ref("startup_outcome", "dev.miren.compute/sandbox.startup_outcome", schema.Doc("Durable distinction between a sandbox that reached RUNNING and one that failed before then"), schema.Choices(SandboxStartupOutcomeStartupRunningId, SandboxStartupOutcomeStartupFailedId)) sb.Component("static_host", "dev.miren.compute/sandbox.static_host", schema.Doc("A name to ip mapping configured staticly for the sandbox"), schema.Many) (&StaticHost{}).InitSchema(sb.Builder("sandbox.static_host")) sb.Singleton("dev.miren.compute/status.pending") @@ -2187,12 +2213,12 @@ func (o *SandboxPool) Empty() bool { func (o *SandboxPool) InitSchema(sb *schema.SchemaBuilder) { sb.Ref("app", "dev.miren.compute/sandbox_pool.app", schema.Doc("Reference to the app this pool belongs to"), schema.Indexed, schema.Tags("dev.miren.app_ref")) - sb.Int64("consecutive_crash_count", "dev.miren.compute/sandbox_pool.consecutive_crash_count", schema.Doc("Number of consecutive quick crashes (sandboxes that died within 60s of creation)")) + sb.Int64("consecutive_crash_count", "dev.miren.compute/sandbox_pool.consecutive_crash_count", schema.Doc("Number of consecutive sandbox failures, including quick crashes after RUNNING")) sb.Time("cooldown_until", "dev.miren.compute/sandbox_pool.cooldown_until", schema.Doc("Timestamp until which new sandbox creation is paused due to crash loop")) sb.Int64("current_instances", "dev.miren.compute/sandbox_pool.current_instances", schema.Doc("Current number of sandbox instances (non-STOPPED)")) sb.Int64("desired_instances", "dev.miren.compute/sandbox_pool.desired_instances", schema.Doc("Target number of sandbox instances")) sb.Bool("ephemeral", "dev.miren.compute/sandbox_pool.ephemeral", schema.Doc("True when this pool backs an ephemeral AppVersion. Ephemeral pools never scale beyond 1 instance.")) - sb.Time("last_crash_time", "dev.miren.compute/sandbox_pool.last_crash_time", schema.Doc("Timestamp of the most recent quick crash")) + sb.Time("last_crash_time", "dev.miren.compute/sandbox_pool.last_crash_time", schema.Doc("Timestamp of the most recently counted sandbox failure")) sb.Int64("ready_instances", "dev.miren.compute/sandbox_pool.ready_instances", schema.Doc("Number of RUNNING sandboxes")) sb.Ref("referenced_by_versions", "dev.miren.compute/sandbox_pool.referenced_by_versions", schema.Doc("AppVersions that reference this pool (enables reuse when specs match)"), schema.Many, schema.Indexed) sb.Label("sandbox_labels", "dev.miren.compute/sandbox_pool.sandbox_labels", schema.Doc("Labels that will be added to the metadata of sandboxes created from this pool"), schema.Many) @@ -2312,5 +2338,5 @@ func init() { (&SandboxPool{}).InitSchema(sb) (&Schedule{}).InitSchema(sb) }) - schema.RegisterEncodedSchema("dev.miren.compute", "v1alpha", []byte("\x1f\x8b\b\x00\x00\x00\x00\x00\x00\xff\xec\\َ\xec<\x11~\r\x0e\xcba\xdf\x04d\x00\xb1\x8b\xed\b\xb8\xe5\x15\"w\\\x9d\xf6tbgl\xa7\x17\xee\x00!!\x04<\x04g\x867\x84k\xe4%\x89\x93؉\xe3F\xfaor3\xb2ݮ\xcfv\xb9\\U.\xd7\xe4\x15ST\xc3\v\x86KV\x13\x0e4+Xݴ\x12\xe0L(\x16o\xb7\xcf\xcc~yR\xbfd\x94a\xf8\xb7\xa6\xbd\xcc{\xa8\x1f\r\xc0\x7f\x8f\x98Ո\xd0\xf9\x00\xc7#\x81\n\x8b\xbf~<\x10|\xfb\x92\x1f#C\r\xc9\x11\xc6\x1c\x84\xd0c\x9d\xdd\x06yo\xe0($'\xb4|]\x02)\x18\x15\x92#B\xa5\xc05\xa2\xf7\xff\x18(\xb7YAA\x85\x0eP\xe9\xe9x\x16\xad\x91\x14\x1d\x1e\xfe8\x13\xd0d_\t\x90q(\x89\x90\xc0\x01\xe7Hj\xd2zܤ\x80\xb0$5h\x98/\x84`ZJ\x81\xe7\x04k\b2T\xa7\x8c\xf8b\x00@\x14'\xc0mEh\xa9\x11\x9e\x9d\xba\x9e\x01ж>\xab?\xf9\x05U-\x88\x7f\x9d\n\xc61\xa3\x80o_\x9dC\x0e\xd4Y\xd7\xedl\xdbС\x82\xdb7\x16I\x9c\x9ezΟ\v\xcdY\"ٚ\xcd?ڲw\xae\xaf\xc0\x01\xe1\xfb\xed\xbdgTM\x96\xe9\xdf˖\x9e)\xbbR\x1f\x93m?\xdbㄉP\xb3\xf3\xb2\xd3v\xed\xba\x90\x96\x9e\x00U\xf2t\xf7\ta\x8fk\xfb\xe8M\xf6\xccS\xaf\xf7\x02\\\x10F\xf5\x82ˮ\xe2lp\xd7V^\xbe\x87\xaa愪\x86\x93\x1a\xf1{\xae\x8e\x1bV\x10\xb7\xcf\x06\x8el\x05H\xd83{\x9dwѿF\x1e\xda?\xeb%|-\x00\x92UH\xc8\xfc\x04\x88\xcb\x03X\x81\xa7\x93\xb6\xb1\xc4\x7f>\x84\xd4p\xf6\f\x85\x81(\xbb\x8a\xa2=\x10\xbcL)\x10\xc5\av3\x94]\xc5R.\xf2\x104\xbdo\x7f4\x13-\x94a\xe3\xed\x9dg\xb7M\x87HN\xfe\xf3-\xa08,Lv`-\xc5yøtt׳Ӫ\xd6D\x14\x15\xa3@\xe5P\xb23\x9cCgs\xe8\xc8\xc9\xfe)\xa4\xe5\x06\xa4\xcc\xd5֥GS\x7f\f((\aB/K+\xd8~\x81\x05\xa1rq\u05c8\xa6W\xfd5C\xbf\x1cfh\xc1\xa8D\x84\x02w\xf8I\x86\xc6\x15v\u0381\xb3\x19p$7\xff\xf110\xd3\x1eH\xb5Ԉ\x1a}_v\x15\x87\x99z\xad__F\xa0GR\xe6GR\xc1\xc4\xf6\xf5\xcd++~\x8aX\xb1;\xcc&\x15\xe2і\x0eT\x86\x91DF\x12ti*F+\xd45\xc3\xd6P\xeb\xd2F\xea\x06ɓ\x95BU\x8aU\xc1\xcf\x06CA\xe8Q<\x86s`\x1b&\x1c\n\xc9\xf8\xddH\xe1P\x9dZt\x8f\x9a\x1bP\x80^\x9c\xbd-T5\xe2\xc0\r\xf4\xa4F\xa5a\x14\x98\xe2T\xc2\x16\xa9k\xd6RW7\x81iX\x91\xaao\xc6H\x95Fڤ\x9b<\xa7I\x83d\x18\x84$\x14\xc9δ\x9e݆)\xb7<\xbaߠ\b\xd6\xf2\x02\xac3bʱra\xd8\x12\xf2\xef\x865\xaf\xb8\x98\x8b\xe2\xc4X\x9d\x8b\x82qCK\x86j\xa7B\xdfV\x87\x9f\x18\x1a\x1ccb掞g/7\x98\x99\xbf\x84|pm\x1dV\x18\xe4Y\x9d!c\x18\xf2\u07b2\x90\xa1\xda\xf1fqЀI\n\x9dMCÙd\x05\xab4ݩ\xaf\xf9=\xedB\x16\x8dO\xee:\xb2L\x16M\xd1\xe2\xe5>-n\x16W\xa1\x87\xee\xb9\x16\xefU*\xe2\x90\xc7\xe7\xec0'\x17RA\t\xc6^=;u=܁1s\xbb\xfa\xf6\x12\x8a8\xb5\x12\xb3+͕s\xc8Z\xc3\xf5fֺI\xc1\t\x89\x899\xf6`\x8a\xe3\xf9,*W)\x8dr.T\xa1\xa7[\xf6Ez\xea׀6\xe9\x8e\a܈\x15+]Z9gs\xa0\xcc\x05\xdadx=\x8e\xab\xc2Ȭ\xbb~\x98\xb9\xe8\x1e\x99\xd2\x04Eod\x8b\xce\xc8\xf6\xa7\xc9s\x16-M\xe7my\x9c\xaf(\x89T0!}\xd8\xf1\xe4Ą\xfc=\xc8+\xe3g\xa3\xf2݆~'_\x03\x13\xedPtP\xc0\x8d\x1b\x1cm\xcbT\x02=n\u0600!d\x8e\nI.\xc4JS=n\xeaY\xfd\x1a8e=\x12+?H\xc9ɡ\x95\xae?W\x8d\xda'\xc1\x8c\x05_\xb8b\xe5\xef\xa8\xec&E\x86j\x84\a\xd0aP\xcb\xd1a6%u\x98\xbc \xcfs\xd0l\x02\xba\xc9\xf6\xfb\xa2\x1e\x06&\xeeR\xe2\v\xbbXz\xd1\x1e(Hk\xf7M9VT;f\xbc\x054M\xb7b\xce\xc6[\n\xa6a\x85\x85s\xc0l\x04\xf8\xa8\xf3\xa4A\xb69O\x9e5\x1a\x94\x12I\xb8\"#jeW\x89v\x9f4ƚ>\x15\r\x14F\x19\xe9\xd2f\xbf\xe5\xa9\xefұ1W@\x91\\\xfc\x9b\xde\xe3\xefǢ>x\x13\x9d\x8f\x93\xad\x8d\x13\x1b\x93\xd0l\xfe\xf1\xf6ud\x88\x97\xae\xaaĺ>\x15\x90\x9f%\x00G]\x84\x7f\x99\x04\xfc\xe8\xfdx>\xea\xea>$_\x97\x7f\xf3\xd8\n\xd7\xeeӏ¯\\\xb8\x1f\x85O\xbc\x91\xdf\xde[l\x05\xdd#O\xae\xe9?O\x98[\xf4\xed\xfdG\t\xe0\x11\x97\xfa\x9f$\xc0\xae\xde\xf5S@\xd3B\x00\xf3\x91\xd6\x0f\xce\xf6\x88\xc0oS׳\xcd\xea\xfd*y\x98\ab\n\xb7w>\xc9\x1e\x02\r)*|\xe5z\x9drN\xe2\xc2\x12)\x93M\x89V\xcc\xc7Y\x17\xbb\xcd\xc1\x8b\x146\xc5D7>$\xe3F\x85?\x92\xa7\xbd\x14\x1f\xf9u2\xe8\xe6\x00J\xcaa\x1f\rՇY\x1eG\xea\x821\xc9\xe5S\n}\b\xe7\x17)Ӊ\x8d\xec\xa4X\xf8ĀO\x8a\x9dZ\x89\x03\xa5\x98\xe9\x84\xf0\x90\xf4폞\xc0\x0f\xa2'\xb0!\xb6\xf1\xc3hД\xe0B\xfcm'&ְ\xf1~\xa0\x84:\xbf\"\"G\xb2\xf32o\x9e:d\U00076183\x90\x88˼a\x15)\xcc\xd4\xe9\xa4ͯ\x8f\x80\xc2\x05\xf8\x86\xe36F\xcd4\xf9\x11UWt\x17\x1bn6\x13\x14C\xaf\xedk\x16\x8f\x91\x14\x8a\x98\xe3\x87\xec\xea\xf6\xc8D\xbcH$\x04,\xe2\x8f\xfd\xff!\x8e\xd18\x87_ým;\xf8B\"I\x8a\\\x1dw\xf7\xbe\xea6\xaf\xec\xd3|\xac\xd0>9\xa0\x9bv\xeb\xa7)\xab\xd1*͘\xbb~\x15\xee.\xc5\x1fY\x17\x944&\x9cM\x9a\xe8\x1dj\x9d\x1d2P\nI\xcf\xe1\xbb\xd1s\bf\xf1\x1c\b\xd6;>\x7fU\x0fB\xb1\xaa\xad\xdd\xe3x\xb4-\xdb\xdf\xed\x17G\x88\xdc\xe2\xbfo\xdc\r\x03\x9e\xe1\x83\t\xa6h\x86t\x95\xe9\x1e\xc7\vN\x87J\xc49\xef\xbde2T\xd3m\x8aEV\xf3\x13w!\xa16.\x8fSO\xbf\xdd[\xec\xc5'\x05Ƕ\xc6ێ\x0e\x18\x90\x80\x91-\xac\xc7M\x0f\xb3E\xdf-\xf3>\x10\xf3\xecԧ\xd8\xf1:\xcdb\xaf\\}\xe2\xfd\x18\x8bǮ\xdd\v\x13\x98b\xba\vi\x11\x1b\xce.\x04[\xd0S_{X\x8c9 \x9c3ZY\xd7h\xa8\x8e\xfd\xd3͇N\x90?@^\x1el\x0e\x9c\xad\x8c.|\x9b\xa7*^*\"\xa1\xcfA\x1d\xaa\xb1\xea\xf5\xc5Q\xaf\x06s\xb1{\xe5N\xe2m\xe5\xc5\xefq\xe38\a\xcf<\xe0\x9b\xaca !4\xd2\xec\x05\xd2T\x1f\xb0oσQ[{\xfb\x8cͽ}\xc3\x18\x907\x85\xb7˓\x05\x84\xcb\x06(&\xb4\\ȼ\xb5=J\xdeR\xba\xdc\xd3\xf6(\x85dM\x03A6\xb5\"\xb3=\be27\x19\xc2\xe1\fݾO(O\xa6cL\xaaM\x9eCfc\xc8\xd8׀P\xe2\xc9F\v\xe3ٱ8e\xecyk\x8d֑\x8b\xd2y\x8c\xd0\t]:\xaf\x8f\x01&G\xd7\xe4\x96\xdb\\\xe7ۧ=\x1bi{D\xf2\xfb\x8f\xc1\x87G\x8b\x93\x9d\xc1\xc6\x01T!!\x8f\xc3\xc1٤]\x1b9\xb2\r a\xcf\xec˼\x8b\xfe5\xf1\xd0\xfeI/\xe1k\x11\x90\xa2AB\x96'@\\\x1e\xc0\n<\x9d\xb4\x8d%\xfe\xf31\xa4\x96\xb3'\xa8\fD\xed*\x8a\xf6@\xf02\xa5@\x14\x1f\xd8\xcdP\xba\x8a\xa5\\\xe4!h\xfa\xd0\xfeh&Z(\xc3\xc6\xdb\a\x81\xdd6\x1d\x129\xf9Ϸ\x88\xe2\xb00Łu\x14\x97-\xe3\xd2\xd3]O^\xabZ\x13QT\x8c\x02\x95C\xc9\xcep\x0e]̡\x13'\xfbǘ\x96\x1b\x90\n_[\xd7\x01M\xfd>\xa2\xa0<\b\xbd,\xad`\xfb\x05V\x84\xca\xc5]#\x9a^\xf5\xd7\f\xfdr\x9c\xa1\x15\xa3\x12\x11\n\xdc\xe3'\x19\x1aW\xd89\a.f\xc0\x89\xdc\xfc\xfb\xfb\xc8L{ \xd5rA\xd4\xe8\xfb\xdaUo\x879\xb9\x92\x06j0\xf6\xeaɫ\xeb\xe1\x0e\x8c\x99\xdbշ\x97Pĩ\x93\x98\xbd\xd0R9\x87\xac3\\og\xad\x9b\x14\x9c\x90\x98\x98c\x0f\xa68\x9eϢr\x95\xd2(\xe7J\x15z\xbae_\xa4\xa7~\x8dh\x13w<\xe0F\xacX\xe9\xd2\xca9\x9b\x03\x15>\xd0&\xc3\x1bp\\\x15Fa\xdd\xf5\xc3\xccE\x0fȔ&\xa8z#[9#۟\xa6\xc0Y\xb44\xce\xdb\n8_I\x12\xa9`b\xfa\xd0\xf1\xe4Ą\xfc\x1d\xc8\x17\xc6\xcfF\xe5\xfb\r\xfdN\xbeF&\xeaPtP\xc0\x8f\x1b\x1cm\xcbT\x02\x03n\u0600!d\x89*I\xae\xc4J\xd3e\xdcԳ\xfa5r\xcaz$V\x7f,%'\x87N\xfa\xfe\\3j\x9f\x043\x16|\xe1\x86տ\xa5\xd2M\x8a\f\xd5\x04\x0f\xc0aP\xcb\xd1a65\xf5\x98\xbc \xcfs\xd0b\x02\xba\xc9\xf6\x87\xa2\x1e\x06&\xedR\x12\n\xbbXz\xd1\x1d(Hk\xf7M9UT\x1d3\xde\"\x9aƭ\x98\xb3\xf1\x96\x82iXa\xe1\x1c\xb0\x18\x01>\xea\x84\xf3\xf3\x9c\xe9\xa4Fvr,|f\xc0'\xc7N\xadār\xcctFxH\x86\xf6GO\xe0\xfb\xc9\x13\xd8\x10\xdb\xf8A2hNp!\xfd\xb6\x93\x12k\xd8x?PB]\xbe \"G\xb2\xf3(iM8\x9b\xb4\xc9;\xd4y;d\xa0\x14\x92\x9e\xc3w\x93\xe7\x10\xcd\xe29\x10\xacw|\xfe\xaa\x1e\x85bMw\xf1\x8f\xe3Ѷl\x7f\xb7_\x1c!q\x8b\xff\xb6q7\fx\x81\x0f&\x98\xa2\x19\xe2*\xd3=N\x17\x1c\x87JĹ\xec\xbde2T\xf3m\x8aEV\xf3\x13w!\xe1b\\\x1e\xaf\x9e\x7f\xbb\xb7؋O\n\x9emM\xb7\x1d\x0e\x18\x90\x80\x91-\xbc\x8c\x9b\x1ef\x8b\xbe[\x96} \xe6ɫO\xb1\xd3u\x9a\xc5^\xb9\xfa\xa4\xfb1\x16\x8f\xbd\xb8\x17&0\xc5|\x17\xd2\"\xb6\x9c]\t\xb6\xa0\xa7\xbe\xf6\xb0\x18s@\xb8d\xb4\xb1\xae\xd1P\x1d\xfb\xa7\x9b\x0f\x9d \xbf\x87\xb2>\xd8\x1c8[\x19]\xf86OU<7DB\x9f\x83:TS\xd5볧^\r\xe6b\xf7Ɵ\x84>n\xf3̑!\xfc\xaf\\\x9f\xae-Y'+f\xe5\x88M\x1b\xc3\xee\x1auݎHi\xa6\x90\xaa\x9f\x00\x15c\x8a~\x1c\xdeQJh}\xfb(\x1d\xc1\x92\xbc\xadn\xfb\xe7\xe0E\x00|\x93\xb1\x8f\xe4\xbb&Z\xf5H\x16\xee\x03\xe6\xfbi\xb0\xd9kO\xbb\xa9\xa9\xc5o\x18\x03\nf(\xbb4`@\xb8n\x81b\xb5\xef\xf1\xc4bۣv\x12\x12\xefi{\xd4B\xb2\xb6\x85(\x9b:Q\xd8\x1e\x842Y\x9a\x04\xe8x\x02r\xdf'\x96\x06\xe4\x18\x93\xebr\xcc!\x8b1d\xeacG,\xaff\xa3\x01\r\xecX\x9a\xad\t<%'\x9b\x80E\xe9<&\xa8<\x97\xad\x1cb\x80IA6\xa9\xf36\x95\xfb\xf6\xe9\xc0F\xda\x1e\x89\xfc\xfeC\xf4]\xd5\xe2\x14g\xb0a\x0eU\xc8HS\xf1p6i\x97\xc0\xda\xcep/L\xae\xbb\xda9]\xf2\x13\xc2#\x14\xb4\x7f\x9a\xa2\xeeij-\x11\\\xadu\xb1\xc3\xc9-+\xf4r\xee'\x8b\x97-cM\x94;\xef\xfc^\x9br\x87CG\xdd\xc3*Pk\xf4g\xa5\n>\x93\x02.눰bT@\xd5Ir\x85\xb2\xe2H\x9c\xcaJ\xbf\xe8(\xb0\x97؏#\xc7\xe2;\xab#\xb0F\xc7\xf7:*\x89\x89\xf0\xd2I\xdb8\xff(d\x90G\x80\x1d\xe7@eI\x94%\xa7\x15\x18\xbd\xfe auto-routable. alt, ok := singleAlternativePort([]int{3000}, 8080) diff --git a/controllers/sandbox/saga_controller.go b/controllers/sandbox/saga_controller.go index d38b47ae2..575b7ded5 100644 --- a/controllers/sandbox/saga_controller.go +++ b/controllers/sandbox/saga_controller.go @@ -89,11 +89,25 @@ func (c *SandboxController) createSandboxViaSaga(ctx context.Context, co *comput // NOTE: this runs at the call site, so a crash between saga completion // and this patch leaves the entity PENDING (retried by reconciler). // Durable saga outcome declaration is future work. + current, meta, getErr := c.ops.GetSandbox(ctx, co.ID.String()) + failure := &compute.Sandbox{Status: compute.DEAD} + var revision int64 + if getErr != nil { + c.Log.Warn("failed to fetch sandbox after saga failure; leaving startup outcome unchanged", "id", co.ID, "error", getErr) + } else { + if current.Status == compute.DEAD { + return fmt.Errorf("saga sandbox creation failed: %w", err) + } + revision = meta.GetRevision() + if current.StartupOutcome != compute.STARTUP_RUNNING && current.Status != compute.RUNNING { + failure.StartupOutcome = compute.STARTUP_FAILED + } + } patchAttrs := entity.New( entity.Ref(entity.DBId, co.ID), - (&compute.Sandbox{Status: compute.DEAD}).Encode, + failure.Encode, ) - if _, patchErr := c.ops.PatchSandbox(ctx, patchAttrs.Attrs(), 0); patchErr != nil { + if _, patchErr := c.ops.PatchSandbox(ctx, patchAttrs.Attrs(), revision); patchErr != nil { c.Log.Error("failed to mark sandbox DEAD after saga failure", "id", co.ID, "error", patchErr) } diff --git a/controllers/sandbox/saga_controller_test.go b/controllers/sandbox/saga_controller_test.go index 183e4b5d5..895510606 100644 --- a/controllers/sandbox/saga_controller_test.go +++ b/controllers/sandbox/saga_controller_test.go @@ -2,6 +2,7 @@ package sandbox import ( "context" + "errors" "log/slog" "testing" "time" @@ -14,6 +15,51 @@ import ( "miren.dev/runtime/pkg/saga" ) +type sagaFailureOps struct { + SandboxEntityStore + SandboxNetworking + SandboxContainerRuntime + SandboxObservability +} + +func TestSagaFailureMarksDeadWhenFinalFetchFails(t *testing.T) { + h := newTestHarness(t) + h.networking.allocateErr = errors.New("no IPs available") + h.entities.getSandboxFunc = func(_ context.Context, _ string) (*compute.Sandbox, *entity.Meta, error) { + if h.entities.getCalls > 1 { + return nil, nil, errors.New("transient fetch failure") + } + return h.entities.sandbox, h.entities.meta, nil + } + c := &SandboxController{ + Log: slog.Default(), ops: sagaFailureOps{SandboxEntityStore: h.entities}, + executor: h.executor, sagaStorage: h.storage, + } + err := c.createSandboxViaSaga(context.Background(), h.entities.sandbox, false) + require.ErrorContains(t, err, "no IPs available") + require.Len(t, h.entities.patchCalls, 1) + patch := entity.New(h.entities.patchCalls[0]) + status, ok := patch.Get(compute.SandboxStatusId) + require.True(t, ok) + assert.Equal(t, compute.SandboxStatusDeadId, status.Value.Id()) + _, outcomeSet := patch.Get(compute.SandboxStartupOutcomeId) + assert.False(t, outcomeSet, "a failed fetch must not derive an outcome from the stale snapshot") +} + +func TestSagaFailureDoesNotRewriteDead(t *testing.T) { + h := newTestHarness(t) + h.networking.allocateErr = errors.New("no IPs available") + h.entities.sandbox.Status = compute.DEAD + h.entities.sandbox.StartupOutcome = compute.STARTUP_FAILED + c := &SandboxController{ + Log: slog.Default(), ops: sagaFailureOps{SandboxEntityStore: h.entities}, + executor: h.executor, sagaStorage: h.storage, + } + err := c.createSandboxViaSaga(context.Background(), h.entities.sandbox, false) + require.ErrorContains(t, err, "no IPs available") + require.Empty(t, h.entities.patchCalls, "already-DEAD sandbox must retain its failure timestamp") +} + // newSagaControllerForResume wires up only what sagaResumeNeeded reads // (storage + log), so no live containerd client is needed. func newSagaControllerForResume(t *testing.T) *SandboxController { diff --git a/controllers/sandbox/sandbox.go b/controllers/sandbox/sandbox.go index c6276ecfc..55dc62c60 100644 --- a/controllers/sandbox/sandbox.go +++ b/controllers/sandbox/sandbox.go @@ -2,6 +2,7 @@ package sandbox import ( "context" + stderrors "errors" "fmt" "io" "log/slog" @@ -71,9 +72,12 @@ func cleanupAttach() cio.Attach { } type containerPorts struct { - Ports []observability.BoundPort + Ports []observability.BoundPort + exited bool } +var errProcessExited = stderrors.New("sandbox process exited") + // SandboxControllerDeps holds required dependencies for SandboxController. type SandboxControllerDeps struct { Log *slog.Logger @@ -195,7 +199,12 @@ type SandboxController struct { // or unwinds on restart rather than stranding containers, addresses, and // disk leases. ops adapts this controller to the domain interfaces the // saga's actions are written against. - ops *sandboxOps + ops interface { + SandboxEntityStore + SandboxNetworking + SandboxContainerRuntime + SandboxObservability + } executor *saga.Executor sagaRegistry *saga.Registry sagaStorage saga.Storage @@ -365,8 +374,8 @@ func (c *SandboxController) SetPortStatus(id string, port observability.BoundPor func (c *SandboxController) WaitForPort(ctx context.Context, id string, port int, timeout time.Duration) error { deadline := time.Now().Add(timeout) - // Create a channel to signal when port is ready - done := make(chan struct{}) + // Signal either readiness or process exit. + done := make(chan error, 1) cancelled := make(chan struct{}) go func() { @@ -385,10 +394,14 @@ func (c *SandboxController) WaitForPort(ctx context.Context, id string, port int ports = &containerPorts{} c.portMap[id] = ports } + if ports.exited { + done <- fmt.Errorf("%w: %s while waiting for port %d", errProcessExited, id, port) + return + } for _, p := range ports.Ports { if p.Port == port { - close(done) + done <- nil return } } @@ -398,8 +411,8 @@ func (c *SandboxController) WaitForPort(ctx context.Context, id string, port int }() select { - case <-done: - return nil + case err := <-done: + return err case <-ctx.Done(): close(cancelled) c.portCond.Broadcast() // Wake up the waiting goroutine @@ -412,6 +425,12 @@ func (c *SandboxController) WaitForPort(ctx context.Context, id string, port int // return a spurious timeout even though the port is in fact bound. c.portMu.Lock() if ports, ok := c.portMap[id]; ok { + if ports.exited { + c.portMu.Unlock() + close(cancelled) + c.portCond.Broadcast() + return fmt.Errorf("%w: %s while waiting for port %d", errProcessExited, id, port) + } for _, p := range ports.Ports { if p.Port == port { c.portMu.Unlock() @@ -426,6 +445,18 @@ func (c *SandboxController) WaitForPort(ctx context.Context, id string, port int } } +func (c *SandboxController) setProcessExited(id string) { + c.portMu.Lock() + defer c.portMu.Unlock() + ports := c.portMap[id] + if ports == nil { + ports = &containerPorts{} + c.portMap[id] = ports + } + ports.exited = true + c.portCond.Broadcast() +} + // mapLegacyProtocol converts legacy PortProtocol values to SandboxSpecContainerPortProtocol func mapLegacyProtocol(legacy compute.PortProtocol) compute.SandboxSpecContainerPortProtocol { switch legacy { @@ -1327,7 +1358,7 @@ func (c *SandboxController) Create(ctx context.Context, co *compute.Sandbox, met "id", co.ID, "createdAt", createdAt, "age", age) patchAttrs := entity.New( entity.Ref(entity.DBId, co.ID), - (&compute.Sandbox{Status: compute.RUNNING}).Encode, + (&compute.Sandbox{Status: compute.RUNNING, StartupOutcome: compute.STARTUP_RUNNING}).Encode, ) _, err := c.ops.PatchSandbox(ctx, patchAttrs.Attrs(), meta.Revision) if err != nil { @@ -1356,7 +1387,7 @@ func (c *SandboxController) Create(ctx context.Context, co *compute.Sandbox, met c.Log.Info("marking unhealthy sandbox as DEAD", "id", co.ID) patchAttrs := entity.New( entity.Ref(entity.DBId, co.ID), - (&compute.Sandbox{Status: compute.DEAD}).Encode, + (&compute.Sandbox{Status: compute.DEAD, StartupOutcome: compute.STARTUP_RUNNING}).Encode, ) _, err := c.ops.PatchSandbox(ctx, patchAttrs.Attrs(), 0) if err != nil { @@ -1416,7 +1447,7 @@ func (c *SandboxController) markDeadNoRestart(ctx context.Context, co *compute.S if co.Status != compute.DEAD { patchAttrs := entity.New( entity.Ref(entity.DBId, co.ID), - (&compute.Sandbox{Status: compute.DEAD}).Encode, + (&compute.Sandbox{Status: compute.DEAD, StartupOutcome: compute.STARTUP_RUNNING}).Encode, ) result, err := c.EAC.Patch(ctx, patchAttrs.Attrs(), 0) if err != nil { @@ -2253,6 +2284,9 @@ func (c *SandboxController) BootContainers( } c.Log.Info("container started", "id", cc.ID()) + c.portMu.Lock() + c.portMap[cc.ID()] = &containerPorts{} + c.portMu.Unlock() if hub != nil { hub.SetResizer(task) @@ -2321,6 +2355,7 @@ func (c *SandboxController) monitorTaskExit( "exit_code", exitStatus.ExitCode(), "exit_time", exitStatus.ExitTime(), ) + c.setProcessExited(containerID) // We don't delete the task here so that our destroySubContainers function // has a consistent view of the state of containers and tasks. @@ -2441,9 +2476,26 @@ func (c *SandboxController) recordExit( return nil, err } + var current compute.Sandbox + current.Decode(resp.Entity().Entity()) + if current.Status == compute.DEAD { + // Do not advance UpdatedAt on an already-counted failure. + return nil, nil + } + stopped := &compute.Sandbox{Status: compute.STOPPED, Exit: exit} + if current.StartupOutcome == "" { + switch current.Status { + case compute.RUNNING: + stopped.StartupOutcome = compute.STARTUP_RUNNING + case compute.PENDING: + stopped.StartupOutcome = compute.STARTUP_FAILED + case compute.NOT_READY, compute.STOPPED, compute.DEAD: + // No lifecycle conclusion from these states alone. + } + } patchAttrs := entity.New( entity.Ref(entity.DBId, id), - (&compute.Sandbox{Status: compute.STOPPED, Exit: exit}).Encode, + stopped.Encode, ) result, err := c.EAC.Patch(ctx, patchAttrs.Attrs(), resp.Entity().Revision()) @@ -3263,21 +3315,10 @@ func (c *SandboxController) StopSandbox(ctx context.Context, id entity.Id, sb *c tmpDir := filepath.Join(c.Tempdir, "containerd", id.PathSafe()) _ = os.RemoveAll(tmpDir) - // Mark sandbox as DEAD in entity store - result, err := c.EAC.Patch(ctx, entity.New( - entity.Ref(entity.DBId, id), - (&compute.Sandbox{ - Status: compute.DEAD, - }).Encode, - ).Attrs(), 0) - if err != nil { - // We ignore if the entity is not found as we run this code path when detecting - // the sandbox entity has already been deleted. - if !errors.Is(err, cond.ErrNotFound{}) { - c.Log.Error("failed to mark sandbox as DEAD", "id", id, "error", err) - } - } else if c.writeTracker != nil && result.HasRevision() { - c.writeTracker.RecordWrite(result.Revision()) + // Use the current entity, not the cleanup snapshot: boot or an exit may + // have changed the lifecycle while resources were being torn down. + if err := c.retireSandbox(ctx, id); err != nil && !errors.Is(err, cond.ErrNotFound{}) { + c.Log.Error("failed to mark sandbox as DEAD", "id", id, "error", err) } c.Log.Info("sandbox retired", "id", id, "status", compute.DEAD) @@ -3291,6 +3332,46 @@ func (c *SandboxController) StopSandbox(ctx context.Context, id entity.Id, sb *c return nil } +func (c *SandboxController) retireSandbox(ctx context.Context, id entity.Id) error { + for attempt := range 10 { + resp, err := c.EAC.Get(ctx, id.String()) + if err != nil { + return err + } + var current compute.Sandbox + current.Decode(resp.Entity().Entity()) + if current.Status == compute.DEAD { + return nil + } + retired := &compute.Sandbox{Status: compute.DEAD} + if current.StartupOutcome == "" { + switch current.Status { + case compute.PENDING: + retired.StartupOutcome = compute.STARTUP_FAILED + case compute.RUNNING: + retired.StartupOutcome = compute.STARTUP_RUNNING + case compute.NOT_READY, compute.STOPPED, compute.DEAD: + // No lifecycle conclusion from these states alone. + } + } + result, err := c.EAC.Patch(ctx, entity.New( + entity.Ref(entity.DBId, id), + retired.Encode, + ).Attrs(), resp.Entity().Revision()) + if errors.Is(err, cond.ErrConflict{}) && attempt < 9 { + continue + } + if err != nil { + return err + } + if c.writeTracker != nil && result.HasRevision() { + c.writeTracker.RecordWrite(result.Revision()) + } + return nil + } + return nil +} + // reregisterSqliteDisks restores replication for a sandbox that outlived the // runner process. It mirrors what configureSqliteVolume did when the sandbox // first started, minus creating anything: the directory and database are diff --git a/controllers/sandbox/sandbox_test.go b/controllers/sandbox/sandbox_test.go index 8ac8f316f..1777e9df6 100644 --- a/controllers/sandbox/sandbox_test.go +++ b/controllers/sandbox/sandbox_test.go @@ -28,6 +28,7 @@ import ( "miren.dev/runtime/api/entityserver/entityserver_v1alpha" "miren.dev/runtime/observability" "miren.dev/runtime/pkg/entity" + entitytestutils "miren.dev/runtime/pkg/entity/testutils" "miren.dev/runtime/pkg/entity/types" "miren.dev/runtime/pkg/idgen" "miren.dev/runtime/pkg/saga" @@ -1694,6 +1695,41 @@ func TestMonitorTaskExitIgnoresErrorStatus(t *testing.T) { }, 5*time.Second, 50*time.Millisecond, "sandbox should remain RUNNING when exit status has an error") } +func TestRecordExitStartupOutcome(t *testing.T) { + ctx := context.Background() + server, cleanup := entitytestutils.NewInMemEntityServer(t) + defer cleanup() + c := &SandboxController{EAC: server.EAC} + + for _, tc := range []struct { + status compute.SandboxStatus + want compute.SandboxStartupOutcome + final compute.SandboxStatus + }{ + {compute.PENDING, compute.STARTUP_FAILED, compute.STOPPED}, + {compute.RUNNING, compute.STARTUP_RUNNING, compute.STOPPED}, + {compute.DEAD, "", compute.DEAD}, + } { + t.Run(string(tc.status), func(t *testing.T) { + id, err := server.Client.Create(ctx, string(tc.status), &compute.Sandbox{Status: tc.status}) + require.NoError(t, err) + before, err := server.EAC.Get(ctx, id.String()) + require.NoError(t, err) + _, err = c.recordExit(ctx, id, compute.Exit{At: time.Now(), Container: "app"}) + require.NoError(t, err) + resp, err := server.EAC.Get(ctx, id.String()) + require.NoError(t, err) + if tc.status == compute.DEAD { + require.Equal(t, before.Entity().Revision(), resp.Entity().Revision(), "late exit must not rewrite a DEAD sandbox") + } + var sb compute.Sandbox + sb.Decode(resp.Entity().Entity()) + require.Equal(t, tc.final, sb.Status) + require.Equal(t, tc.want, sb.StartupOutcome) + }) + } +} + // TestMonitorTaskExitHandlesValidExit verifies that monitorTaskExit correctly // marks a sandbox as STOPPED when receiving a valid exit status (no error). func TestMonitorTaskExitHandlesValidExit(t *testing.T) { @@ -1941,6 +1977,52 @@ func TestDeadPatchPreservesRecordedExit(t *testing.T) { r.Equal("app", got.Exit.Container) } +func TestRetireSandboxUsesCurrentLifecycle(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + deps, cleanup := testutils.NewTestDeps() + defer cleanup() + c, err := newSandboxController(deps) + require.NoError(t, err) + defer c.Close() + require.NoError(t, c.Init(ctx)) + id := entity.Id(idgen.GenNS("sb")) + stale := &compute.Sandbox{ID: id, Status: compute.PENDING} + var rpcE entityserver_v1alpha.Entity + rpcE.SetId(id.String()) + rpcE.SetAttrs(entity.New(entity.DBId, id, stale.Encode).Attrs()) + _, err = c.EAC.Put(ctx, &rpcE) + require.NoError(t, err) + + // The cleanup caller still holds PENDING, but boot has already persisted RUNNING. + _, err = c.EAC.Patch(ctx, entity.New(entity.DBId, id, + (&compute.Sandbox{Status: compute.RUNNING, StartupOutcome: compute.STARTUP_RUNNING}).Encode).Attrs(), 0) + require.NoError(t, err) + require.NoError(t, c.StopSandbox(ctx, id, stale)) + resp, err := c.EAC.Get(ctx, id.String()) + require.NoError(t, err) + var got compute.Sandbox + got.Decode(resp.Entity().Entity()) + require.Equal(t, compute.DEAD, got.Status) + require.Equal(t, compute.STARTUP_RUNNING, got.StartupOutcome) +} + +func TestRetireSandboxDoesNotRewriteDead(t *testing.T) { + ctx := context.Background() + server, cleanup := entitytestutils.NewInMemEntityServer(t) + defer cleanup() + id, err := server.Client.Create(ctx, "dead", &compute.Sandbox{Status: compute.DEAD, StartupOutcome: compute.STARTUP_FAILED}) + require.NoError(t, err) + before, err := server.EAC.Get(ctx, id.String()) + require.NoError(t, err) + + c := &SandboxController{EAC: server.EAC} + require.NoError(t, c.retireSandbox(ctx, id)) + after, err := server.EAC.Get(ctx, id.String()) + require.NoError(t, err) + require.Equal(t, before.Entity().Revision(), after.Entity().Revision(), "already-DEAD sandbox must retain its failure timestamp") +} + // A sandbox whose command must execute at most once is finished when its // containers vanish. Rebooting it would re-run the command -- for a migration, // not a recoverable mistake. diff --git a/controllers/sandbox/waitforport_race_test.go b/controllers/sandbox/waitforport_race_test.go index 652ba9e7f..20b1e7fe1 100644 --- a/controllers/sandbox/waitforport_race_test.go +++ b/controllers/sandbox/waitforport_race_test.go @@ -2,6 +2,7 @@ package sandbox import ( "context" + "errors" "log/slog" "sync" "testing" @@ -30,6 +31,28 @@ func boundPort(port int) observability.BoundPort { return observability.BoundPort{Port: port} } +func TestWaitForPortProcessExit(t *testing.T) { + c := newPortTestController() + result := make(chan error, 1) + go func() { result <- c.WaitForPort(context.Background(), "app", 8080, time.Minute) }() + c.setProcessExited("app") + select { + case err := <-result: + require.True(t, errors.Is(err, errProcessExited), "got %v", err) + case <-time.After(time.Second): + t.Fatal("port wait did not stop on process exit") + } + + // A later wait must not forget an exit or mistake a stale bound port for readiness. + c.SetPortStatus("app", boundPort(8080), observability.PortStatusBound) + require.ErrorIs(t, c.WaitForPort(context.Background(), "app", 8080, time.Minute), errProcessExited) + + other := newPortTestController() + err := other.WaitForPort(context.Background(), "still-running", 8080, 20*time.Millisecond) + require.ErrorContains(t, err, "timeout waiting for port 8080") + require.NotErrorIs(t, err, errProcessExited) +} + // TestWaitForPortNoSpuriousTimeoutWhenPortBound is the deterministic regression // guard for the WaitForPort deadline race (controllers/sandbox/sandbox.go). // diff --git a/controllers/sandboxpool/manager.go b/controllers/sandboxpool/manager.go index b2e14e299..ce804ffdf 100644 --- a/controllers/sandboxpool/manager.go +++ b/controllers/sandboxpool/manager.go @@ -86,10 +86,10 @@ func (m *Manager) Reconcile(ctx context.Context, pool *compute_v1alpha.SandboxPo // Skip crash detection for decommissioned pools (desired=0, no references). // Sandbox deaths during intentional scale-down are expected, not crashes. if pool.DesiredInstances > 0 || len(pool.ReferencedByVersions) > 0 { - newCrashes := m.countQuickCrashes(sandboxes, pool.LastCrashTime) + newCrashes, latestCrash := m.countStartupFailures(sandboxes, pool) if newCrashes > 0 { pool.ConsecutiveCrashCount += int64(newCrashes) - pool.LastCrashTime = time.Now() + pool.LastCrashTime = latestCrash pool.CooldownUntil = m.calculateBackoff(pool.ConsecutiveCrashCount) m.log.Warn("crash detected, entering cooldown", @@ -108,11 +108,14 @@ func (m *Manager) Reconcile(ctx context.Context, pool *compute_v1alpha.SandboxPo // Unreferenced pools should be allowed to scale to 0 even during cooldown isUnreferenced := len(pool.ReferencedByVersions) == 0 - // Reset DesiredInstances to prevent activator-driven accumulation - // Allow desired: 0 for unreferenced pools (deployment cleanup) + // Allow one replacement beyond the live count (RUNNING or PENDING). + // This preserves siblings after a partial failure (including node loss), + // without letting activator requests accumulate a full crash loop. targetDesired := int64(1) if isUnreferenced { targetDesired = 0 + } else if actual > 0 { + targetDesired = max(1, min(pool.DesiredInstances, actual+1)) } if pool.DesiredInstances != targetDesired { @@ -781,60 +784,59 @@ func (m *Manager) checkForStalePendingSandboxes(ctx context.Context) error { if createdAt.After(threshold) { continue } + m.retireStalePending(ctx, ent) + } - m.log.Warn("marking stale PENDING sandbox as STOPPED", - "sandbox", sb.ID, - "created_at", createdAt, - "age", now.Sub(createdAt)) + return nil +} - if _, err := m.eac.Patch(ctx, entity.New( - entity.DBId, sb.ID, - (&compute_v1alpha.Sandbox{ - Status: compute_v1alpha.STOPPED, - }).Encode, - ).Attrs(), 0); err != nil { - if errors.Is(err, cond.ErrNotFound{}) { - m.log.Warn("sandbox already deleted during stale pending check", - "sandbox", sb.ID) - } else { - m.log.Error("failed to stop stale PENDING sandbox", - "sandbox", sb.ID, - "error", err) - } - continue +func (m *Manager) retireStalePending(ctx context.Context, ent *entityserver_v1alpha.Entity) { + var sb compute_v1alpha.Sandbox + sb.Decode(ent.Entity()) + createdAt := time.UnixMilli(ent.CreatedAt()) + m.log.Warn("marking stale PENDING sandbox as STOPPED", + "sandbox", sb.ID, + "created_at", createdAt, + "age", time.Since(createdAt)) + + if _, err := m.eac.Patch(ctx, entity.New( + entity.DBId, sb.ID, + (&compute_v1alpha.Sandbox{ + Status: compute_v1alpha.STOPPED, + StartupOutcome: compute_v1alpha.STARTUP_FAILED, + }).Encode, + ).Attrs(), ent.Revision()); err != nil { + if errors.Is(err, cond.ErrNotFound{}) { + m.log.Warn("sandbox already deleted during stale pending check", "sandbox", sb.ID) + } else if errors.Is(err, cond.ErrConflict{}) { + m.log.Debug("sandbox changed during stale pending check", "sandbox", sb.ID) + } else { + m.log.Error("failed to stop stale PENDING sandbox", "sandbox", sb.ID, "error", err) } } - - return nil } -// countQuickCrashes counts sandboxes that died within 60 seconds of creation -// and occurred after lastCrashTime -func (m *Manager) countQuickCrashes(sandboxes []*sandboxWithMeta, lastCrashTime time.Time) int64 { +// countStartupFailures uses the latest counted sandbox update as a watermark. +// DEAD sandboxes must not be rewritten after counting or they can be counted again. +func (m *Manager) countStartupFailures(sandboxes []*sandboxWithMeta, pool *compute_v1alpha.SandboxPool) (int64, time.Time) { count := int64(0) - crashThreshold := 60 * time.Second - + latest := pool.LastCrashTime for _, sbm := range sandboxes { - if sbm.sandbox.Status != compute_v1alpha.DEAD { + if sbm.sandbox.Status != compute_v1alpha.DEAD || !sbm.updatedAt.After(pool.LastCrashTime) { continue } - // Check if this is a quick crash (died within 60s of creation) - lifetime := sbm.updatedAt.Sub(sbm.createdAt) - - if lifetime >= crashThreshold { - continue // Lived long enough, not a quick crash - } - - // Check if this crash is new (after lastCrashTime) - if !lastCrashTime.IsZero() && !sbm.updatedAt.After(lastCrashTime) { - continue // Already counted this crash + if sbm.sandbox.StartupOutcome != compute_v1alpha.STARTUP_FAILED && + sbm.updatedAt.Sub(sbm.createdAt) >= 60*time.Second { + continue } count++ + if sbm.updatedAt.After(latest) { + latest = sbm.updatedAt + } } - - return count + return count, latest } // backoffDuration calculates the exponential backoff duration based on consecutive crash count diff --git a/controllers/sandboxpool/manager_test.go b/controllers/sandboxpool/manager_test.go index ce0c9f178..400e75368 100644 --- a/controllers/sandboxpool/manager_test.go +++ b/controllers/sandboxpool/manager_test.go @@ -1110,6 +1110,32 @@ func TestCheckForStalePendingSandboxes(t *testing.T) { "RUNNING sandbox should remain RUNNING") } +func TestStalePendingDoesNotOverwriteRunning(t *testing.T) { + ctx := context.Background() + server, cleanup := testutils.NewInMemEntityServer(t) + defer cleanup() + server.Store.NowFunc = func() time.Time { return time.Now().Add(-6 * time.Minute) } + id, err := server.Client.Create(ctx, "pending", &compute_v1alpha.Sandbox{Status: compute_v1alpha.PENDING}) + require.NoError(t, err) + server.Store.NowFunc = nil + + // The monitor's snapshot predates a successful setRunning patch. + stale, err := server.EAC.Get(ctx, id.String()) + require.NoError(t, err) + _, err = server.EAC.Patch(ctx, entity.New(entity.DBId, id, + (&compute_v1alpha.Sandbox{Status: compute_v1alpha.RUNNING, StartupOutcome: compute_v1alpha.STARTUP_RUNNING}).Encode).Attrs(), stale.Entity().Revision()) + require.NoError(t, err) + + manager := NewManager(testutils.TestLogger(t), server.EAC) + manager.retireStalePending(ctx, stale.Entity()) + resp, err := server.EAC.Get(ctx, id.String()) + require.NoError(t, err) + var sb compute_v1alpha.Sandbox + sb.Decode(resp.Entity().Entity()) + assert.Equal(t, compute_v1alpha.RUNNING, sb.Status) + assert.Equal(t, compute_v1alpha.STARTUP_RUNNING, sb.StartupOutcome) +} + // TestStalePendingSandboxUnblocksPoolCapacity tests that marking a stale // PENDING sandbox as STOPPED frees pool capacity, allowing the pool to // create a replacement sandbox on the next reconcile. @@ -1482,3 +1508,201 @@ func TestManagerCrashResetDoesNotRecount(t *testing.T) { assert.True(t, updatedPool.CooldownUntil.IsZero(), "pool should not re-enter cooldown from old dead sandboxes") } + +func TestCountStartupFailures(t *testing.T) { + now := time.Now() + manager := &Manager{} + sandboxes := []*sandboxWithMeta{ + {sandbox: &compute_v1alpha.Sandbox{ID: "first", Status: compute_v1alpha.DEAD, StartupOutcome: compute_v1alpha.STARTUP_FAILED}, createdAt: now.Add(-6 * time.Minute), updatedAt: now.Add(-40 * time.Second)}, + {sandbox: &compute_v1alpha.Sandbox{ID: "healthy", Status: compute_v1alpha.DEAD, StartupOutcome: compute_v1alpha.STARTUP_RUNNING}, createdAt: now.Add(-time.Hour), updatedAt: now.Add(-30 * time.Second)}, + {sandbox: &compute_v1alpha.Sandbox{ID: "fast-healthy", Status: compute_v1alpha.DEAD, StartupOutcome: compute_v1alpha.STARTUP_RUNNING}, createdAt: now.Add(-20 * time.Second), updatedAt: now.Add(-10 * time.Second)}, + {sandbox: &compute_v1alpha.Sandbox{ID: "legacy", Status: compute_v1alpha.DEAD}, createdAt: now.Add(-25 * time.Second), updatedAt: now.Add(-15 * time.Second)}, + {sandbox: &compute_v1alpha.Sandbox{ID: "old-legacy", Status: compute_v1alpha.DEAD}, createdAt: now.Add(-6 * time.Minute), updatedAt: now.Add(-time.Minute)}, + {sandbox: &compute_v1alpha.Sandbox{ID: "old-failure", Status: compute_v1alpha.DEAD, StartupOutcome: compute_v1alpha.STARTUP_FAILED}, createdAt: now.Add(-10 * time.Minute), updatedAt: now.Add(-2 * time.Minute)}, + } + pool := &compute_v1alpha.SandboxPool{LastCrashTime: now.Add(-90 * time.Second)} + count, latest := manager.countStartupFailures(sandboxes, pool) + assert.Equal(t, int64(3), count, + "long pre-running failure and quick crashes count, but long-running and previously counted failures do not") + assert.Equal(t, now.Add(-10*time.Second), latest, "use the newest counted update, not time.Now or the newest healthy exit") + pool.LastCrashTime = latest + count, _ = manager.countStartupFailures(sandboxes, pool) + assert.Zero(t, count, "reconciliation must not count the same DEAD sandboxes again") +} + +func TestPartialNodeLossKeepsPoolDesiredDuringCooldown(t *testing.T) { + ctx := context.Background() + server, cleanup := testutils.NewInMemEntityServer(t) + defer cleanup() + pool := &compute_v1alpha.SandboxPool{ + Service: "web", DesiredInstances: 5, + ReferencedByVersions: []entity.Id{"ver-1"}, + SandboxSpec: compute_v1alpha.SandboxSpec{Version: "ver-1"}, + } + id, err := server.Client.Create(ctx, "pool", pool) + require.NoError(t, err) + pool.ID = id + for i := range 4 { + _, err = server.Client.Create(ctx, fmt.Sprintf("healthy-%d", i), + &compute_v1alpha.Sandbox{Status: compute_v1alpha.RUNNING, Spec: pool.SandboxSpec}, + entityserver.WithLabels(types.LabelSet("service", "web", "pool", id.String()))) + require.NoError(t, err) + } + server.Store.NowFunc = func() time.Time { return time.Now().Add(-6 * time.Minute) } + failedID, err := server.Client.Create(ctx, "lost-node", + &compute_v1alpha.Sandbox{Status: compute_v1alpha.PENDING, Spec: pool.SandboxSpec}, + entityserver.WithLabels(types.LabelSet("service", "web", "pool", id.String()))) + require.NoError(t, err) + server.Store.NowFunc = nil + _, err = server.EAC.Patch(ctx, entity.New(entity.DBId, failedID, + (&compute_v1alpha.Sandbox{Status: compute_v1alpha.DEAD, StartupOutcome: compute_v1alpha.STARTUP_FAILED}).Encode).Attrs(), 0) + require.NoError(t, err) + + manager := NewManager(testutils.TestLogger(t), server.EAC) + reconcilePool(t, ctx, server, manager, pool) + updated := getPool(t, ctx, server, id) + assert.Equal(t, int64(1), updated.ConsecutiveCrashCount) + assert.Equal(t, int64(5), updated.DesiredInstances, "a failed boot must not downsize four healthy instances") + assert.Equal(t, int64(4), updated.ReadyInstances) + assert.WithinDuration(t, time.Now().Add(10*time.Second), updated.CooldownUntil, 2*time.Second) + assert.Len(t, listSandboxesForPool(t, ctx, server, pool), 5, "replacement waits for cooldown") +} + +func TestCrashLoopCooldownCapsDesiredWithRunningSiblings(t *testing.T) { + ctx := context.Background() + server, cleanup := testutils.NewInMemEntityServer(t) + defer cleanup() + pool := &compute_v1alpha.SandboxPool{ + Service: "web", DesiredInstances: 8, + ReferencedByVersions: []entity.Id{"ver-1"}, + SandboxSpec: compute_v1alpha.SandboxSpec{Version: "ver-1"}, + } + id, err := server.Client.Create(ctx, "pool", pool) + require.NoError(t, err) + pool.ID = id + for i := range 2 { + _, err = server.Client.Create(ctx, fmt.Sprintf("running-%d", i), + &compute_v1alpha.Sandbox{Status: compute_v1alpha.RUNNING, Spec: pool.SandboxSpec}, + entityserver.WithLabels(types.LabelSet("service", "web", "pool", id.String()))) + require.NoError(t, err) + } + server.Store.NowFunc = func() time.Time { return time.Now().Add(-20 * time.Second) } + failedID, err := server.Client.Create(ctx, "quick-crash", + &compute_v1alpha.Sandbox{Status: compute_v1alpha.RUNNING, StartupOutcome: compute_v1alpha.STARTUP_RUNNING, Spec: pool.SandboxSpec}, + entityserver.WithLabels(types.LabelSet("service", "web", "pool", id.String()))) + require.NoError(t, err) + server.Store.NowFunc = nil + _, err = server.EAC.Patch(ctx, entity.New(entity.DBId, failedID, + (&compute_v1alpha.Sandbox{Status: compute_v1alpha.DEAD}).Encode).Attrs(), 0) + require.NoError(t, err) + + reconcilePool(t, ctx, server, NewManager(testutils.TestLogger(t), server.EAC), pool) + updated := getPool(t, ctx, server, id) + assert.Equal(t, int64(1), updated.ConsecutiveCrashCount) + assert.Equal(t, int64(3), updated.DesiredInstances, "only one replacement may be queued beyond the two running siblings") + assert.Equal(t, int64(2), updated.ReadyInstances) +} + +func TestCooldownPreservesPendingSiblingsAfterTheyStart(t *testing.T) { + ctx := context.Background() + server, cleanup := testutils.NewInMemEntityServer(t) + defer cleanup() + pool := &compute_v1alpha.SandboxPool{ + Service: "web", DesiredInstances: 5, + ReferencedByVersions: []entity.Id{"ver-1"}, + SandboxSpec: compute_v1alpha.SandboxSpec{Version: "ver-1"}, + } + id, err := server.Client.Create(ctx, "pool", pool) + require.NoError(t, err) + pool.ID = id + for i := range 4 { + status := compute_v1alpha.RUNNING + if i >= 2 { + status = compute_v1alpha.PENDING + } + _, err = server.Client.Create(ctx, fmt.Sprintf("sibling-%d", i), + &compute_v1alpha.Sandbox{Status: status, Spec: pool.SandboxSpec}, + entityserver.WithLabels(types.LabelSet("service", "web", "pool", id.String()))) + require.NoError(t, err) + } + _, err = server.Client.Create(ctx, "failed", + &compute_v1alpha.Sandbox{Status: compute_v1alpha.DEAD, StartupOutcome: compute_v1alpha.STARTUP_FAILED, Spec: pool.SandboxSpec}, + entityserver.WithLabels(types.LabelSet("service", "web", "pool", id.String()))) + require.NoError(t, err) + manager := NewManager(testutils.TestLogger(t), server.EAC) + reconcilePool(t, ctx, server, manager, pool) + updated := getPool(t, ctx, server, id) + assert.Equal(t, int64(5), updated.DesiredInstances) + assert.Equal(t, int64(4), updated.CurrentInstances) + assert.Equal(t, int64(2), updated.ReadyInstances) + + // Both pending siblings finish booting before the cooldown ends. + for _, sb := range listSandboxesForPool(t, ctx, server, pool) { + if sb.Status == compute_v1alpha.PENDING { + _, err = server.EAC.Patch(ctx, entity.New(entity.DBId, sb.ID, + (&compute_v1alpha.Sandbox{Status: compute_v1alpha.RUNNING, StartupOutcome: compute_v1alpha.STARTUP_RUNNING}).Encode).Attrs(), 0) + require.NoError(t, err) + } + } + _, err = server.EAC.Patch(ctx, entity.New(entity.DBId, id, + (&compute_v1alpha.SandboxPool{CooldownUntil: time.Now().Add(-time.Second)}).Encode).Attrs(), 0) + require.NoError(t, err) + reconcilePool(t, ctx, server, manager, pool) + updated = getPool(t, ctx, server, id) + assert.Equal(t, int64(5), updated.DesiredInstances) + assert.Equal(t, int64(5), updated.CurrentInstances, "a replacement may start without retiring a healthy sibling") + assert.Equal(t, int64(4), updated.ReadyInstances) +} + +func TestManagerLongStartupFailureBackoff(t *testing.T) { + ctx := context.Background() + server, cleanup := testutils.NewInMemEntityServer(t) + defer cleanup() + pool := &compute_v1alpha.SandboxPool{ + Service: "web", DesiredInstances: 1, + SandboxSpec: compute_v1alpha.SandboxSpec{Version: entity.Id("ver-1")}, + } + id, err := server.Client.Create(ctx, "pool", pool) + require.NoError(t, err) + pool.ID = id + manager := NewManager(testutils.TestLogger(t), server.EAC) + + for streak := int64(1); streak <= 2; streak++ { + created := time.Now().Add(-6 * time.Minute) + server.Store.NowFunc = func() time.Time { return created } + sbID, err := server.Client.Create(ctx, fmt.Sprintf("failed-%d", streak), + &compute_v1alpha.Sandbox{Status: compute_v1alpha.PENDING, Spec: pool.SandboxSpec}, + entityserver.WithLabels(types.LabelSet("service", "web", "pool", id.String()))) + require.NoError(t, err) + server.Store.NowFunc = nil + diedAt := time.Now().Add(-time.Duration(3-streak) * time.Second).Truncate(time.Millisecond).UTC() + server.Store.NowFunc = func() time.Time { return diedAt } + _, err = server.EAC.Patch(ctx, entity.New(entity.DBId, sbID, + (&compute_v1alpha.Sandbox{Status: compute_v1alpha.DEAD, StartupOutcome: compute_v1alpha.STARTUP_FAILED}).Encode).Attrs(), 0) + require.NoError(t, err) + server.Store.NowFunc = nil + before := time.Now() + reconcilePool(t, ctx, server, manager, pool) + updated := getPool(t, ctx, server, id) + assert.Equal(t, streak, updated.ConsecutiveCrashCount) + assert.Equal(t, diedAt, updated.LastCrashTime, "watermark must use the counted event, not reconciliation time") + assert.WithinDuration(t, before.Add(backoffDuration(streak)), updated.CooldownUntil, 2*time.Second) + assert.Equal(t, int64(0), updated.CurrentInstances) + assert.Len(t, listSandboxesForPool(t, ctx, server, pool), int(streak), "no replacement during cooldown") + // The same terminal entity must not increase the streak on another reconcile. + reconcilePool(t, ctx, server, manager, pool) + updated = getPool(t, ctx, server, id) + assert.Equal(t, streak, updated.ConsecutiveCrashCount) + } + // A deployment clears the streak but preserves the failure watermark. + _, err = server.EAC.Patch(ctx, []entity.Attr{ + entity.Ref(entity.DBId, id), + entity.Int64(compute_v1alpha.SandboxPoolConsecutiveCrashCountId, 0), + entity.Time(compute_v1alpha.SandboxPoolCooldownUntilId, time.Time{}), + }, 0) + require.NoError(t, err) + reconcilePool(t, ctx, server, manager, pool) + updated := getPool(t, ctx, server, id) + assert.Equal(t, int64(0), updated.ConsecutiveCrashCount) + assert.False(t, updated.LastCrashTime.IsZero()) +}