diff --git a/apiserver/controllers/controllers.go b/apiserver/controllers/controllers.go index 98d6a68e0..a46a6c7e3 100644 --- a/apiserver/controllers/controllers.go +++ b/apiserver/controllers/controllers.go @@ -123,6 +123,9 @@ func handleError(ctx context.Context, w http.ResponseWriter, err error) { case errors.Is(err, gErrors.ErrDuplicateEntity), errors.Is(err, &gErrors.ConflictError{}): w.WriteHeader(http.StatusConflict) apiErr.Error = "Conflict" + case errors.Is(err, gErrors.ErrUnprocessable): + w.WriteHeader(http.StatusUnprocessableEntity) + apiErr.Error = "Unprocessable Content" default: w.WriteHeader(http.StatusInternalServerError) apiErr.Error = "Server error" diff --git a/config/external.go b/config/external.go index cbb8978ef..6b9d1c4ef 100644 --- a/config/external.go +++ b/config/external.go @@ -19,6 +19,7 @@ import ( "os" "path/filepath" "strings" + "time" "github.com/cloudbase/garm-provider-common/util/exec" ) @@ -48,6 +49,19 @@ type External struct { // EnvironmentVariables is a list of environment variable names that will be // passed to the external binary together with their values. EnvironmentVariables []string `toml:"environment_variables" json:"environment-variables"` + // ExecTimeoutSeconds is the maximum number of seconds a single invocation + // of the provider binary may run. When exceeded, the binary is killed and + // the operation fails; for instance creation this marks the instance as + // error and normal cleanup takes over. This bounds how long an instance + // can sit in the creating/deleting states if a provider binary hangs. + // A value of 0 (the default) disables the timeout. + ExecTimeoutSeconds uint `toml:"exec_timeout_seconds" json:"exec-timeout-seconds"` +} + +// ExecTimeout returns the configured provider binary execution timeout as a +// time.Duration. A zero duration means no timeout is enforced. +func (e *External) ExecTimeout() time.Duration { + return time.Duration(e.ExecTimeoutSeconds) * time.Second } func (e *External) GetEnvironmentVariables() []string { diff --git a/database/sql/instances.go b/database/sql/instances.go index 604379947..06d5b31d4 100644 --- a/database/sql/instances.go +++ b/database/sql/instances.go @@ -31,6 +31,7 @@ import ( runnerErrors "github.com/cloudbase/garm-provider-common/errors" commonParams "github.com/cloudbase/garm-provider-common/params" "github.com/cloudbase/garm/database/common" + garmErrors "github.com/cloudbase/garm/internal/errors" "github.com/cloudbase/garm/params" ) @@ -319,7 +320,7 @@ func (s *sqlDatabase) validateRunnerStatusTransition(current, newStatus params.R } if !slices.Contains(allowedTransitions, newStatus) { - return runnerErrors.NewBadRequestError("invalid runner status transition from %s to %s", current, newStatus) + return garmErrors.NewRunnerTransitionError(current, newStatus) } return nil } @@ -338,7 +339,7 @@ func (s *sqlDatabase) validateInstanceStatusTransition(current, newStatus common } if !slices.Contains(allowedTransitions, newStatus) { - return runnerErrors.NewBadRequestError("invalid instance status transition from %s to %s", current, newStatus) + return runnerErrors.NewInstanceTransitionError(current, newStatus) } return nil } diff --git a/doc/providers.md b/doc/providers.md index af0e7b938..dd3f71c77 100644 --- a/doc/providers.md +++ b/doc/providers.md @@ -50,6 +50,24 @@ However, some providers need access to host environment variables for authentica This passes all environment variables starting with `AWS_` to the provider. You can also specify exact variable names (e.g., `["AZURE_CLIENT_ID", "AZURE_TENANT_ID"]`). +## Bounding provider execution time + +Instance creation is always bounded by the pool or scale set's `runner_bootstrap_timeout`. If the provider binary does not return within that window, it is killed, the instance is marked as `error` and it is cleaned up through the normal lifecycle. Past that point the runner would be considered failed anyway, so there is no reason to keep waiting on a hung create. + +All other provider operations (delete, list, etc.) run without a deadline by default. You can bound every invocation of a provider binary with `exec_timeout_seconds`: + +```toml +[provider.external] + provider_executable = "/opt/garm/providers.d/garm-provider-aws" + config_file = "/etc/garm/garm-provider-aws.toml" + exec_timeout_seconds = 900 +``` + +When the timeout is exceeded, the binary is killed and the operation fails. + +> [!IMPORTANT] +> For instance creation, the effective deadline is the smaller of `exec_timeout_seconds` and the pool or scale set's `runner_bootstrap_timeout`. If you set `exec_timeout_seconds`, keep it below your bootstrap timeouts — a create that is allowed to outlive the bootstrap timeout only produces a runner that is already considered failed. A value of `0` (the default) leaves non-create operations unbounded. + ## Listing configured providers ```bash diff --git a/go.mod b/go.mod index 4941b565f..de04e8687 100644 --- a/go.mod +++ b/go.mod @@ -6,7 +6,7 @@ require ( github.com/BurntSushi/toml v1.6.0 github.com/alecthomas/chroma/v2 v2.27.0 github.com/bradleyfalzon/ghinstallation/v2 v2.19.0 - github.com/cloudbase/garm-provider-common v0.1.10-0.20260707095006-33396540b57b + github.com/cloudbase/garm-provider-common v0.1.10-0.20260707224136-4f5d9bf80949 github.com/felixge/httpsnoop v1.1.0 github.com/gdamore/tcell/v2 v2.13.10 github.com/go-gormigrate/gormigrate/v2 v2.1.6 diff --git a/go.sum b/go.sum index 4e95b6242..a742d4134 100644 --- a/go.sum +++ b/go.sum @@ -25,8 +25,8 @@ github.com/chzyer/test v1.0.0 h1:p3BQDXSxOhOG0P9z6/hGnII4LGiEPOYBhs8asl/fC04= github.com/chzyer/test v1.0.0/go.mod h1:2JlltgoNkt4TW/z9V/IzDdFaMTM2JPIi26O1pF38GC8= github.com/clipperhouse/uax29/v2 v2.7.0 h1:+gs4oBZ2gPfVrKPthwbMzWZDaAFPGYK72F0NJv2v7Vk= github.com/clipperhouse/uax29/v2 v2.7.0/go.mod h1:EFJ2TJMRUaplDxHKj1qAEhCtQPW2tJSwu5BF98AuoVM= -github.com/cloudbase/garm-provider-common v0.1.10-0.20260707095006-33396540b57b h1:70W/+KXIfnNXy8KcT17CdnDG5U7yflS3rxvIgfIKzDw= -github.com/cloudbase/garm-provider-common v0.1.10-0.20260707095006-33396540b57b/go.mod h1:9UvQ/bHlwgSAeLFV94by9M7kRg3d0H3kemkjcFzJMmc= +github.com/cloudbase/garm-provider-common v0.1.10-0.20260707224136-4f5d9bf80949 h1:Ksf13uqPZhEcGwAHT1pmAXU3+LcAKl1yqyLWiKIxLTc= +github.com/cloudbase/garm-provider-common v0.1.10-0.20260707224136-4f5d9bf80949/go.mod h1:9UvQ/bHlwgSAeLFV94by9M7kRg3d0H3kemkjcFzJMmc= github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= diff --git a/internal/errors/errors.go b/internal/errors/errors.go new file mode 100644 index 000000000..b42e945c2 --- /dev/null +++ b/internal/errors/errors.go @@ -0,0 +1,104 @@ +// Copyright 2026 Cloudbase Solutions SRL +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. You may obtain +// a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +// License for the specific language governing permissions and limitations +// under the License. +package errors + +import ( + "fmt" + + runnerErrors "github.com/cloudbase/garm-provider-common/errors" + commonParams "github.com/cloudbase/garm-provider-common/params" + "github.com/cloudbase/garm/params" +) + +// NewRunnerTransitionError returns a RunnerTransitionError describing an invalid +// runner status state machine transition. +func NewRunnerTransitionError(from, to params.RunnerStatus) error { + return &RunnerTransitionError{ + From: from, + To: to, + } +} + +// RunnerTransitionError is returned when a requested runner status transition is +// rejected by the state machine. It carries the current (From) and requested +// (To) statuses as their proper type, so callers can compare them directly (via +// errors.As) without casting through strings. It also reports as a +// BadRequestError, so it maps to an HTTP 400. +type RunnerTransitionError struct { + From params.RunnerStatus + To params.RunnerStatus +} + +func (e *RunnerTransitionError) Error() string { + return fmt.Sprintf("invalid runner status transition from %s to %s", e.From, e.To) +} + +func (e *RunnerTransitionError) Is(target error) bool { + if target == nil { + return false + } + + switch target.(type) { + case *RunnerTransitionError, *runnerErrors.BadRequestError: + return true + default: + return false + } +} + +// InstanceIsBeingDeleted reports whether an instance status is on the deletion +// lane, meaning the runner is already on its way out. Note that the lane is +// not strictly monotonic: a failed provider delete moves the instance from +// deleting to error. Tolerating a status observed here is still safe because +// reconciliation re-drives error back to pending_delete until the delete +// succeeds — the system converges by retry, and the intent behind a rejected +// transition to pending_delete (the runner should be removed) remains +// satisfied. Pair it with InstanceTransitionError.From. +func InstanceIsBeingDeleted(s commonParams.InstanceStatus) bool { + switch s { + case commonParams.InstancePendingDelete, commonParams.InstancePendingForceDelete, + commonParams.InstanceDeleting, commonParams.InstanceDeleted: + return true + default: + return false + } +} + +// InstanceIsProvisioning reports whether an instance status indicates the +// provider is still creating the instance (or is about to). The provider +// worker owns this stage of the lifecycle: deletion cannot be requested until +// the create call returns (creating only transitions to error or running). +// Pair it with InstanceTransitionError.From to decide whether a rejected +// transition to pending_delete should be deferred to a later reconciliation +// pass instead of being treated as fatal. +func InstanceIsProvisioning(s commonParams.InstanceStatus) bool { + switch s { + case commonParams.InstancePendingCreate, commonParams.InstanceCreating: + return true + default: + return false + } +} + +// RunnerIsTerminal reports whether a runner status is terminal, meaning a +// transition to active can no longer succeed and a late "started" message for +// the runner is moot. Pair it with RunnerTransitionError.From. +func RunnerIsTerminal(s params.RunnerStatus) bool { + switch s { + case params.RunnerTerminated, params.RunnerFailed: + return true + default: + return false + } +} diff --git a/params/params.go b/params/params.go index 86543e08b..204f10ed6 100644 --- a/params/params.go +++ b/params/params.go @@ -280,12 +280,23 @@ var RunnerStatusTransitions = map[RunnerStatus][]RunnerStatus{ RunnerInstalling, RunnerTerminated, RunnerPending, + // The forge assigns jobs based on its own registration state; a + // "job started" event can arrive before the runner's own idle + // callback (or the image may never send callbacks at all), so a + // runner the forge considers busy is authoritatively active. + RunnerActive, }, RunnerInstalling: { RunnerFailed, RunnerIdle, RunnerTerminated, RunnerInstalling, + // Same as RunnerPending: the forge saying a job started on this + // runner is proof it finished installing. + RunnerActive, + // When an agent manages the runner, the runner itself is offline, + // until the process is started and set online by the agent. + RunnerOffline, }, RunnerIdle: { RunnerOffline, diff --git a/runner/pool/pool.go b/runner/pool/pool.go index a2d5e441f..a3756a46d 100644 --- a/runner/pool/pool.go +++ b/runner/pool/pool.go @@ -39,6 +39,7 @@ import ( "github.com/cloudbase/garm/cache" dbCommon "github.com/cloudbase/garm/database/common" "github.com/cloudbase/garm/database/watcher" + garmErrors "github.com/cloudbase/garm/internal/errors" "github.com/cloudbase/garm/locking" "github.com/cloudbase/garm/params" "github.com/cloudbase/garm/runner/common" @@ -346,6 +347,12 @@ func (r *basePoolManager) handleCompletedJob(ctx context.Context, jobParams para if errors.Is(err, runnerErrors.ErrNotFound) { return nil } + var te *runnerErrors.InstanceTransitionError + if errors.As(err, &te) && garmErrors.InstanceIsBeingDeleted(te.From) { + // Another code path already moved the instance into the deletion + // lane; the runner is being removed, which is what we wanted. + return nil + } slog.With(slog.Any("error", err)).ErrorContext( ctx, "failed to update runner status", "runner_name", util.SanitizeLogEntry(jobParams.RunnerName)) @@ -1130,7 +1137,13 @@ func (r *basePoolManager) addInstanceToProvider(instance params.Instance) error ProviderBaseParams: r.getProviderBaseParams(pool), }, } - providerInstance, err := provider.CreateInstance(r.ctx, bootstrapArgs, createInstanceParams) + // Bound the create call by the pool's bootstrap timeout; past that point + // the runner is considered failed anyway. The provider-level exec timeout + // (if configured and smaller) still applies; the effective deadline is + // whichever is sooner. + createCtx, cancel := context.WithTimeout(r.ctx, time.Duration(pool.RunnerTimeout())*time.Minute) + defer cancel() + providerInstance, err := provider.CreateInstance(createCtx, bootstrapArgs, createInstanceParams) if err != nil { instanceIDToDelete = instance.Name return fmt.Errorf("error creating instance: %w", err) diff --git a/runner/providers/v0.1.0/external.go b/runner/providers/v0.1.0/external.go index bb96f4d7c..66c194c06 100644 --- a/runner/providers/v0.1.0/external.go +++ b/runner/providers/v0.1.0/external.go @@ -70,6 +70,25 @@ type external struct { environmentVariables []string } +// execWithTimeout invokes the provider binary, bounding the call by the +// provider's configured exec timeout (if any). On expiry the child process is +// killed and a ProviderError is returned so the operation fails and normal +// cleanup takes over, instead of the instance being stuck in a transient +// state (e.g. creating) for as long as a hung binary sits there. +func (e *external) execWithTimeout(ctx context.Context, stdinData []byte, environ []string) ([]byte, error) { + timeout := e.cfg.External.ExecTimeout() + if timeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, timeout) + defer cancel() + } + out, err := garmExec.Exec(ctx, e.execPath, stdinData, environ) + if err != nil && errors.Is(ctx.Err(), context.DeadlineExceeded) { + return nil, garmErrors.NewProviderError("provider binary %s timed out (context deadline exceeded)", e.execPath) + } + return out, err +} + // CreateInstance creates a new compute instance in the provider. func (e *external) CreateInstance(ctx context.Context, bootstrapParams commonParams.BootstrapInstance, _ common.CreateInstanceParams) (commonParams.ProviderInstance, error) { asEnv := []string{ @@ -90,7 +109,7 @@ func (e *external) CreateInstance(ctx context.Context, bootstrapParams commonPar e.cfg.Name, // label: provider ).Inc() - out, err := garmExec.Exec(ctx, e.execPath, asJs, asEnv) + out, err := e.execWithTimeout(ctx, asJs, asEnv) if err != nil { metrics.InstanceOperationFailedCount.WithLabelValues( "CreateInstance", // label: operation @@ -137,7 +156,7 @@ func (e *external) DeleteInstance(ctx context.Context, instance string, _ common "DeleteInstance", // label: operation e.cfg.Name, // label: provider ).Inc() - _, err := garmExec.Exec(ctx, e.execPath, nil, asEnv) + _, err := e.execWithTimeout(ctx, nil, asEnv) if err != nil { var exitErr *exec.ExitError if !errors.As(err, &exitErr) || exitErr.ExitCode() != commonExecution.ExitCodeNotFound { @@ -168,7 +187,7 @@ func (e *external) GetInstance(ctx context.Context, instance string, _ common.Ge "GetInstance", // label: operation e.cfg.Name, // label: provider ).Inc() - out, err := garmExec.Exec(ctx, e.execPath, nil, asEnv) + out, err := e.execWithTimeout(ctx, nil, asEnv) if err != nil { metrics.InstanceOperationFailedCount.WithLabelValues( "GetInstance", // label: operation @@ -212,7 +231,7 @@ func (e *external) ListInstances(ctx context.Context, poolID string, _ common.Li e.cfg.Name, // label: provider ).Inc() - out, err := garmExec.Exec(ctx, e.execPath, nil, asEnv) + out, err := e.execWithTimeout(ctx, nil, asEnv) if err != nil { metrics.InstanceOperationFailedCount.WithLabelValues( "ListInstances", // label: operation @@ -258,7 +277,7 @@ func (e *external) RemoveAllInstances(ctx context.Context, _ common.RemoveAllIns e.cfg.Name, // label: provider ).Inc() - _, err := garmExec.Exec(ctx, e.execPath, nil, asEnv) + _, err := e.execWithTimeout(ctx, nil, asEnv) if err != nil { metrics.InstanceOperationFailedCount.WithLabelValues( "RemoveAllInstances", // label: operation @@ -283,7 +302,7 @@ func (e *external) Stop(ctx context.Context, instance string, _ common.StopParam "Stop", // label: operation e.cfg.Name, // label: provider ).Inc() - _, err := garmExec.Exec(ctx, e.execPath, nil, asEnv) + _, err := e.execWithTimeout(ctx, nil, asEnv) if err != nil { metrics.InstanceOperationFailedCount.WithLabelValues( "Stop", // label: operation @@ -309,7 +328,7 @@ func (e *external) Start(ctx context.Context, instance string, _ common.StartPar e.cfg.Name, // label: provider ).Inc() - _, err := garmExec.Exec(ctx, e.execPath, nil, asEnv) + _, err := e.execWithTimeout(ctx, nil, asEnv) if err != nil { metrics.InstanceOperationFailedCount.WithLabelValues( "Start", // label: operation diff --git a/runner/providers/v0.1.1/external.go b/runner/providers/v0.1.1/external.go index 6e43dce7f..3ff8affa2 100644 --- a/runner/providers/v0.1.1/external.go +++ b/runner/providers/v0.1.1/external.go @@ -69,6 +69,25 @@ type external struct { environmentVariables []string } +// execWithTimeout invokes the provider binary, bounding the call by the +// provider's configured exec timeout (if any). On expiry the child process is +// killed and a ProviderError is returned so the operation fails and normal +// cleanup takes over, instead of the instance being stuck in a transient +// state (e.g. creating) for as long as a hung binary sits there. +func (e *external) execWithTimeout(ctx context.Context, stdinData []byte, environ []string) ([]byte, error) { + timeout := e.cfg.External.ExecTimeout() + if timeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, timeout) + defer cancel() + } + out, err := garmExec.Exec(ctx, e.execPath, stdinData, environ) + if err != nil && errors.Is(ctx.Err(), context.DeadlineExceeded) { + return nil, garmErrors.NewProviderError("provider binary %s timed out (context deadline exceeded)", e.execPath) + } + return out, err +} + // CreateInstance creates a new compute instance in the provider. func (e *external) CreateInstance(ctx context.Context, bootstrapParams commonParams.BootstrapInstance, _ common.CreateInstanceParams) (commonParams.ProviderInstance, error) { extraspecs := bootstrapParams.ExtraSpecs @@ -97,7 +116,7 @@ func (e *external) CreateInstance(ctx context.Context, bootstrapParams commonPar e.cfg.Name, // label: provider ).Inc() - out, err := garmExec.Exec(ctx, e.execPath, asJs, asEnv) + out, err := e.execWithTimeout(ctx, asJs, asEnv) if err != nil { metrics.InstanceOperationFailedCount.WithLabelValues( "CreateInstance", // label: operation @@ -153,7 +172,7 @@ func (e *external) DeleteInstance(ctx context.Context, instance string, deleteIn "DeleteInstance", // label: operation e.cfg.Name, // label: provider ).Inc() - _, err = garmExec.Exec(ctx, e.execPath, nil, asEnv) + _, err = e.execWithTimeout(ctx, nil, asEnv) if err != nil { var exitErr *exec.ExitError if !errors.As(err, &exitErr) || exitErr.ExitCode() != commonExecution.ExitCodeNotFound { @@ -193,7 +212,7 @@ func (e *external) GetInstance(ctx context.Context, instance string, getInstance "GetInstance", // label: operation e.cfg.Name, // label: provider ).Inc() - out, err := garmExec.Exec(ctx, e.execPath, nil, asEnv) + out, err := e.execWithTimeout(ctx, nil, asEnv) if err != nil { metrics.InstanceOperationFailedCount.WithLabelValues( "GetInstance", // label: operation @@ -245,7 +264,7 @@ func (e *external) ListInstances(ctx context.Context, poolID string, listInstanc e.cfg.Name, // label: provider ).Inc() - out, err := garmExec.Exec(ctx, e.execPath, nil, asEnv) + out, err := e.execWithTimeout(ctx, nil, asEnv) if err == nil { metrics.InstanceOperationFailedCount.WithLabelValues( "ListInstances", // label: operation @@ -300,7 +319,7 @@ func (e *external) RemoveAllInstances(ctx context.Context, removeAllInstances co e.cfg.Name, // label: provider ).Inc() - _, err = garmExec.Exec(ctx, e.execPath, nil, asEnv) + _, err = e.execWithTimeout(ctx, nil, asEnv) if err != nil { metrics.InstanceOperationFailedCount.WithLabelValues( "RemoveAllInstances", // label: operation @@ -334,7 +353,7 @@ func (e *external) Stop(ctx context.Context, instance string, stopParams common. "Stop", // label: operation e.cfg.Name, // label: provider ).Inc() - _, err = garmExec.Exec(ctx, e.execPath, nil, asEnv) + _, err = e.execWithTimeout(ctx, nil, asEnv) if err != nil { metrics.InstanceOperationFailedCount.WithLabelValues( "Stop", // label: operation @@ -369,7 +388,7 @@ func (e *external) Start(ctx context.Context, instance string, startParams commo e.cfg.Name, // label: provider ).Inc() - _, err = garmExec.Exec(ctx, e.execPath, nil, asEnv) + _, err = e.execWithTimeout(ctx, nil, asEnv) if err != nil { metrics.InstanceOperationFailedCount.WithLabelValues( "Start", // label: operation diff --git a/runner/runner.go b/runner/runner.go index 8b2efbee1..9f754a4fd 100644 --- a/runner/runner.go +++ b/runner/runner.go @@ -41,6 +41,7 @@ import ( "github.com/cloudbase/garm/auth" "github.com/cloudbase/garm/config" dbCommon "github.com/cloudbase/garm/database/common" + garmErrors "github.com/cloudbase/garm/internal/errors" "github.com/cloudbase/garm/params" "github.com/cloudbase/garm/runner/common" "github.com/cloudbase/garm/runner/pool" @@ -1111,6 +1112,12 @@ func (r *Runner) AddInstanceStatusMessage(ctx context.Context, param params.Inst } if _, err := r.store.UpdateInstance(r.ctx, instanceName, updateParams); err != nil { + var te *garmErrors.RunnerTransitionError + if errors.As(err, &te) && garmErrors.RunnerIsTerminal(te.From) { + // The runner is already terminal (e.g. reaped after going missing); + // its self-reported status update is moot. + return nil + } return fmt.Errorf("error updating runner agent ID: %w", err) } @@ -1265,6 +1272,12 @@ func (r *Runner) DeleteRunner(ctx context.Context, instanceName string, forceDel } _, err = r.store.UpdateInstance(r.ctx, instance.Name, updateParams) if err != nil { + var te *runnerErrors.InstanceTransitionError + if errors.As(err, &te) && garmErrors.InstanceIsBeingDeleted(te.From) { + // Another code path already moved the instance into the deletion + // lane; the deletion the caller asked for is already under way. + return nil + } return fmt.Errorf("error updating runner state: %w", err) } diff --git a/vendor/github.com/cloudbase/garm-provider-common/errors/errors.go b/vendor/github.com/cloudbase/garm-provider-common/errors/errors.go index 9a12c4ba7..239369bd0 100644 --- a/vendor/github.com/cloudbase/garm-provider-common/errors/errors.go +++ b/vendor/github.com/cloudbase/garm-provider-common/errors/errors.go @@ -14,7 +14,11 @@ package errors -import "fmt" +import ( + "fmt" + + "github.com/cloudbase/garm-provider-common/params" +) var ( // ErrUnauthorized is returned when a user does not have @@ -295,3 +299,42 @@ func (p *NoCapacityError) Is(target error) bool { _, ok := target.(*NoCapacityError) return ok } + +// NewInstanceTransitionError returns an InstanceTransitionError describing an +// invalid instance status state machine transition. +func NewInstanceTransitionError(from, to params.InstanceStatus) error { + return &InstanceTransitionError{ + baseError: baseError{ + msg: fmt.Sprintf("invalid instance status transition from %s to %s", from, to), + }, + From: from, + To: to, + } +} + +// InstanceTransitionError is returned when a requested instance status +// transition is rejected by the state machine. It carries the current (From) +// and requested (To) statuses as their proper type, so callers working with a +// params.Instance can compare them directly (via errors.As) without casting +// through strings, and decide whether the refusal is benign for their intent +// (for example, the instance is already further along the same terminal path) +// or a genuine error to stop on. It also reports as a BadRequestError, so it +// maps to an HTTP 400. +type InstanceTransitionError struct { + baseError + From params.InstanceStatus + To params.InstanceStatus +} + +func (e *InstanceTransitionError) Is(target error) bool { + if target == nil { + return false + } + + switch target.(type) { + case *InstanceTransitionError, *BadRequestError: + return true + default: + return false + } +} diff --git a/vendor/modules.txt b/vendor/modules.txt index 0a28082f2..0b9d6be11 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -25,7 +25,7 @@ github.com/chzyer/readline # github.com/clipperhouse/uax29/v2 v2.7.0 ## explicit; go 1.18 github.com/clipperhouse/uax29/v2/graphemes -# github.com/cloudbase/garm-provider-common v0.1.10-0.20260707095006-33396540b57b +# github.com/cloudbase/garm-provider-common v0.1.10-0.20260707224136-4f5d9bf80949 ## explicit; go 1.25.0 github.com/cloudbase/garm-provider-common/cloudconfig github.com/cloudbase/garm-provider-common/defaults diff --git a/workers/provider/instance_manager.go b/workers/provider/instance_manager.go index 78465f599..95b45f797 100644 --- a/workers/provider/instance_manager.go +++ b/workers/provider/instance_manager.go @@ -237,7 +237,15 @@ func (i *instanceManager) handleCreateInstanceInProvider(instance params.Instanc }, } - providerInstance, err := i.provider.CreateInstance(i.ctx, bootstrapArgs, createInstanceParams) + // Bound the create call by the scale set's bootstrap timeout. Past that + // point the runner is considered failed anyway, and reaping skips + // instances in creating, so allowing the provider call to outlive the + // bootstrap timeout would leave the instance stuck until the binary + // resolved on its own. The provider-level exec timeout (if configured and + // smaller) still applies; the effective deadline is whichever is sooner. + createCtx, cancel := context.WithTimeout(i.ctx, time.Duration(i.scaleSet.RunnerTimeout())*time.Minute) + defer cancel() + providerInstance, err := i.provider.CreateInstance(createCtx, bootstrapArgs, createInstanceParams) if err != nil { instanceIDToDelete = instance.Name return fmt.Errorf("creating instance in provider: %w", err) diff --git a/workers/provider/provider.go b/workers/provider/provider.go index 0711b7070..fa1f683d3 100644 --- a/workers/provider/provider.go +++ b/workers/provider/provider.go @@ -136,11 +136,13 @@ func (p *Provider) loadAllRunners() error { continue } // Ignore runners in "creating" state. If we're just starting up and - // we find a runner in "creating" it was most likely interrupted while - // creating. It is unlikely that it is still usable. We allow the scale set - // worker to clean it up. It will eventually be marked as pending delete and - // this worker will get an update to clean up any resources left behing by - // an incomplete creation event. + // we find a runner in "creating", the create was interrupted by a + // restart (the provider binary died with the previous process). It is + // unlikely that it is still usable. The scale set worker resolves + // these in its own Start(): it force-marks them pending_delete (or + // running, if the runner managed to register with the forge in the + // meantime), and this worker will get an update to clean up any + // resources left behind by the incomplete creation. if runner.Status == commonParams.InstanceCreating { continue } diff --git a/workers/scaleset/scaleset.go b/workers/scaleset/scaleset.go index 9197a378b..418756c0d 100644 --- a/workers/scaleset/scaleset.go +++ b/workers/scaleset/scaleset.go @@ -28,6 +28,7 @@ import ( "github.com/cloudbase/garm/cache" dbCommon "github.com/cloudbase/garm/database/common" "github.com/cloudbase/garm/database/watcher" + garmErrors "github.com/cloudbase/garm/internal/errors" "github.com/cloudbase/garm/locking" "github.com/cloudbase/garm/params" "github.com/cloudbase/garm/runner/common" @@ -350,9 +351,7 @@ func (w *Worker) setRunnerDBStatus(runner string, status commonParams.InstanceSt } newDbInstance, err := w.store.UpdateInstance(w.ctx, runner, updateParams) if err != nil { - if !errors.Is(err, runnerErrors.ErrNotFound) { - return params.Instance{}, fmt.Errorf("updating runner %s: %w", runner, err) - } + return params.Instance{}, fmt.Errorf("updating runner %s: %w", runner, err) } return newDbInstance, nil } @@ -369,7 +368,13 @@ func (w *Worker) removeRunnerFromGithubAndSetPendingDelete(runnerName string, ag } instance, err := w.setRunnerDBStatus(runnerName, commonParams.InstancePendingDelete) if err != nil { - return fmt.Errorf("updating runner %s: %w", instance.Name, err) + if errors.Is(err, runnerErrors.ErrNotFound) { + // The runner record is already gone from the database; there is + // nothing left to mark. The watcher delete event will evict it + // from the local cache. + return nil + } + return fmt.Errorf("updating runner %s: %w", runnerName, err) } w.runners[instance.ID] = instance return nil @@ -395,12 +400,14 @@ func (w *Worker) reapTimedOutRunners(runners map[string]params.RunnerReference) continue case commonParams.InstanceCreating, commonParams.InstancePendingCreate: // Instance is still being created in the provider, or is about to be - // created. The provider worker owns create timeouts/failures for this - // instance and will move it through Error -> PendingDelete on its own - // if creation fails or times out. Nothing for the scale-set-level - // reaper to do here; jumping straight to pending_delete from here is - // also not a valid status transition (creating only allows -> error - // or -> running). + // created. The provider worker owns create timeouts/failures: the + // create call carries a deadline derived from the bootstrap timeout + // (and optionally the provider exec timeout), so an instance cannot + // linger here past it; on failure or expiry the provider worker moves + // it through Error -> PendingDelete on its own. Nothing for the + // scale-set-level reaper to do; jumping straight to pending_delete + // from here is also not a valid status transition (creating only + // allows -> error or -> running). continue } @@ -483,6 +490,13 @@ func (w *Worker) consolidateRunnerState(runners []params.RunnerReference) error case commonParams.InstancePendingDelete, commonParams.InstancePendingForceDelete, commonParams.InstanceDeleting, commonParams.InstanceDeleted: continue + case commonParams.InstanceCreating: + // The provider worker owns provisioning; an instance in creating + // cannot transition to pending_delete while the create call is in + // flight. If its runner is missing from github, we will pick it up + // on a later pass, once the create returns and the instance moves + // to running or error. + continue } if _, ok := ghRunnersByName[name]; !ok { @@ -490,20 +504,35 @@ func (w *Worker) consolidateRunnerState(runners []params.RunnerReference) error slog.DebugContext(w.ctx, "runner is locked; skipping", "runner_name", name) continue } - // unlock the runner only after this function returns. This function also cross - // checks between the provider and the database, and removes left over runners. - // If we unlock early, the provider worker will attempt to remove runners that - // we set in pending_delete. This function holds the mutex, so we won't see those - // changes until we return. So we hold the instance lock here until we are done. - // That way, even if the provider sees the pending_delete status, it won't act on - // it until it manages to lock the instance. + // unlock the runner only after this function returns. These locks are + // worker-local: they serialize this worker's own goroutines (listener + // handlers, reaper, scale down) so none of them process this runner + // while we reconcile it, including the provider cross-check further + // down. The provider worker does NOT participate in instance locking; + // it reacts to the pending_delete status via watcher events as soon + // as the transition commits. Cross-worker coordination is done by the + // status state machine, not by these locks. defer locking.Unlock(name, false) slog.InfoContext(w.ctx, "runner does not exist in github; removing from provider", "runner_name", name) instance, err := w.setRunnerDBStatus(runner.Name, commonParams.InstancePendingDelete) if err != nil { - if !errors.Is(err, runnerErrors.ErrNotFound) { - return fmt.Errorf("updating runner %s: %w", instance.Name, err) + var te *runnerErrors.InstanceTransitionError + switch { + case errors.Is(err, runnerErrors.ErrNotFound): + // The record is already gone from the database. Drop it from + // the local cache rather than writing back the zero-value + // instance below. + delete(w.runners, runner.ID) + continue + case errors.As(err, &te) && garmErrors.InstanceIsBeingDeleted(te.From): + // Already being deleted by another code path; nothing to do. + continue + default: + // Don't let one runner's error abort consolidation for the + // entire scale set; log it and keep processing the rest. + slog.ErrorContext(w.ctx, "error updating runner", "runner_name", runner.Name, "error", err) + continue } } // We will get an update event anyway from the watcher, but updating the runner @@ -514,7 +543,14 @@ func (w *Worker) consolidateRunnerState(runners []params.RunnerReference) error } } - // Cross check what exists in the provider with the DB. + return w.consolidateProviderState() +} + +// consolidateProviderState cross checks what exists in the provider with the +// DB. Provider instances with no DB record are removed from the provider, and +// DB runners with no provider instance are marked as pending_delete. Must be +// called with w.mux held. +func (w *Worker) consolidateProviderState() error { pseudoPoolID, err := w.pseudoPoolID() if err != nil { return fmt.Errorf("getting pseudo pool ID: %w", err) @@ -545,8 +581,8 @@ func (w *Worker) consolidateRunnerState(runners []params.RunnerReference) error }, } - // refresh the map. It may have been mutated above. - dbRunnersByName = w.runnerByName() + // The runner cache may have been mutated by the github cross check. + dbRunnersByName := w.runnerByName() for _, runner := range providerRunners { if _, ok := dbRunnersByName[runner.Name]; !ok { slog.InfoContext(w.ctx, "runner does not exist in database; removing from provider", "runner_name", runner.Name) @@ -587,8 +623,12 @@ func (w *Worker) consolidateRunnerState(runners []params.RunnerReference) error // The runner is not in the provider anymore. Remove it from the DB. slog.InfoContext(w.ctx, "runner does not exist in provider; removing from database", "runner_name", runner.Name) if err := w.removeRunnerFromGithubAndSetPendingDelete(runner.Name, runner.AgentID); err != nil { + // Same reasoning as reapTimedOutRunners and the github cross + // check above: one poisoned runner must not abort the sweep + // for the rest of the scale set. Log it and keep going. + slog.ErrorContext(w.ctx, "error removing runner", "runner_name", runner.Name, "error", err) locking.Unlock(runner.Name, false) - return fmt.Errorf("removing runner %s: %w", runner.Name, err) + continue } } locking.Unlock(runner.Name, false) @@ -940,6 +980,18 @@ func (w *Worker) handleScaleDown() { locking.Unlock(runner.Name, true) continue } + var te *runnerErrors.InstanceTransitionError + if errors.As(err, &te) && garmErrors.InstanceIsBeingDeleted(te.From) { + // Another code path already moved the instance into the + // deletion lane; it is being removed, which is what we wanted. + // Keep the lock entry (remove=false): the runner record still + // exists and other goroutines of this worker may be blocked + // on this lock; removing the entry from under a blocked + // waiter would let two goroutines hold the same runner's lock. + removed++ + locking.Unlock(runner.Name, false) + continue + } // nolint:golangci-lint,godox // TODO: This should not happen, unless there is some issue with the database. // The UpdateInstance() function should add tenacity, but even in that case, if it diff --git a/workers/scaleset/scaleset_helper.go b/workers/scaleset/scaleset_helper.go index e0871d197..ea74abb74 100644 --- a/workers/scaleset/scaleset_helper.go +++ b/workers/scaleset/scaleset_helper.go @@ -22,6 +22,7 @@ import ( runnerErrors "github.com/cloudbase/garm-provider-common/errors" commonParams "github.com/cloudbase/garm-provider-common/params" "github.com/cloudbase/garm/cache" + garmErrors "github.com/cloudbase/garm/internal/errors" "github.com/cloudbase/garm/locking" "github.com/cloudbase/garm/params" "github.com/cloudbase/garm/util/github/scalesets" @@ -122,10 +123,43 @@ func (w *Worker) HandleJobsCompleted(jobs []params.ScaleSetJobMessage) (err erro locking.Lock(job.RunnerName, w.consumerID) _, err := w.store.UpdateInstance(w.ctx, job.RunnerName, runnerUpdateParams) if err != nil { - if !errors.Is(err, runnerErrors.ErrNotFound) { + if errors.Is(err, runnerErrors.ErrNotFound) { + locking.Unlock(job.RunnerName, true) + continue + } + // If the transition was refused because the instance is already on + // its way out (some other code path — reaper, consolidate, provider + // — is deleting it), our intent to remove the runner is already + // satisfied. te.From is the status seen inside the update's tx. + // Even if the in-flight delete later fails (deleting can bounce to + // error), reconciliation re-drives it to pending_delete, so skipping + // this update never strands the runner. + var te *runnerErrors.InstanceTransitionError + if errors.As(err, &te) && garmErrors.InstanceIsBeingDeleted(te.From) { + slog.InfoContext(w.ctx, "runner already being removed; ignoring completed job status update", + "runner_name", job.RunnerName, "from_status", te.From) + // The runner record still exists, so keep the lock entry; + // another goroutine of this worker (reaper, consolidate, scale + // down) may be blocked on it, and removing the entry from under + // a blocked waiter would let two goroutines hold the same + // runner's lock. The entry is removed once the record is gone. locking.Unlock(job.RunnerName, false) - return fmt.Errorf("updating runner %s: %w", job.RunnerName, err) + continue } + // The instance is still in creating. The provider create call has + // not returned or hit its deadline yet (creates are bounded by the + // bootstrap timeout). But somehow the compute instance it spun up managed + // to finish bootstrap, accept a job and finish, within the runner bootstrap + // timeout. While this scenario is wildly unlikely, unless the provider is + // badly broken, we need to account for it. + if errors.As(err, &te) && garmErrors.InstanceIsProvisioning(te.From) { + slog.InfoContext(w.ctx, "instance is still provisioning; deferring removal to consolidation", + "runner_name", job.RunnerName, "from_status", te.From) + locking.Unlock(job.RunnerName, false) + continue + } + locking.Unlock(job.RunnerName, false) + return fmt.Errorf("updating runner %s: %w", job.RunnerName, err) } locking.Unlock(job.RunnerName, false) } @@ -160,6 +194,20 @@ func (w *Worker) HandleJobsStarted(jobs []params.ScaleSetJobMessage) (err error) locking.Unlock(job.RunnerName, true) continue } + // If the runner was already terminal (e.g. reaped after going missing) + // by the time this started message was processed, it can't transition + // to active and the job won't run on it (github will requeue). te.From + // is the status seen inside the update's tx; nothing to do here. + var te *garmErrors.RunnerTransitionError + if errors.As(err, &te) && garmErrors.RunnerIsTerminal(te.From) { + slog.InfoContext(w.ctx, "runner already terminal; ignoring started job status update", + "runner_name", job.RunnerName, "from_status", te.From) + // Keep the lock entry: the runner record still exists and other + // goroutines of this worker may be blocked on this lock. See + // HandleJobsCompleted for details. + locking.Unlock(job.RunnerName, false) + continue + } locking.Unlock(job.RunnerName, false) return fmt.Errorf("updating runner %s: %w", job.RunnerName, err) }