diff --git a/admin/admin.go b/admin/admin.go index b9b5ecc1e7ff..00715aa41b93 100644 --- a/admin/admin.go +++ b/admin/admin.go @@ -27,7 +27,6 @@ type Options struct { ExternalURL string FrontendURL string ProvisionerSetJSON string - ProvisionerMaxConcurrency int DefaultProvisioner string Version version.Version MetricsProjectOrg string @@ -43,7 +42,6 @@ type Service struct { Jobs jobs.Client URLs *URLs ProvisionerSet map[string]provisioner.Provisioner - ProvisionerMaxConcurrency int Email *email.Client Github Github AI drivers.AIService @@ -129,7 +127,6 @@ func New(ctx context.Context, opts *Options, logger *zap.Logger, issuer *auth.Is DB: db, URLs: urls, ProvisionerSet: provSet, - ProvisionerMaxConcurrency: opts.ProvisionerMaxConcurrency, Email: emailClient, Github: github, AI: aiService, diff --git a/admin/deployments.go b/admin/deployments.go index 94646141f6c8..989ac130308d 100644 --- a/admin/deployments.go +++ b/admin/deployments.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "reflect" "regexp" "strings" "time" @@ -481,8 +482,14 @@ func (s *Service) DeleteDeploymentInner(ctx context.Context, depl *database.Depl return nil } -// UpdateDeploymentInner updates a deployment by updating its runtime instance and resources. -// The implementation is idempotent, enabling it to be called from a retryable background job. +// UpdateDeploymentInner reconciles a running deployment towards its desired configuration. +// It compares the deployment's desired provisioning args against the currently provisioned args +// and only reprovisions the runtime when they differ (e.g. slots, version or disk changed). +// Otherwise it runs a lightweight, drift-aware health check of the deployment's provisioner resources. +// In both cases it reloads the runtime config, which is itself change-aware and cheap when nothing has changed, +// so that variable/branch/annotation changes still propagate promptly. +// The implementation is idempotent, enabling it to be called from a retryable background job +// (both for explicit updates and for periodic reconciliation). func (s *Service) UpdateDeploymentInner(ctx context.Context, d *database.Deployment) error { // Find project and organization proj, err := s.DB.FindProject(ctx, d.ProjectID) @@ -512,32 +519,62 @@ func (s *Service) UpdateDeploymentInner(ctx context.Context, d *database.Deploym return fmt.Errorf("can't update deployment %q because its runtime has not been initialized yet", d.ID) } - // Prepare deployment annotations - annotations := s.NewDeploymentAnnotations(org, proj, d.Environment) - // Resolve slots based on environment slots, err := resolveSlots(proj, d.Environment) if err != nil { return err } - // Provision the runtime. This is idempotent and will (partially) update the existing provisioned runtime if the config has changed. - _, err = s.provisionRuntime(ctx, &provisionRuntimeOptions{ - DeploymentID: d.ID, - Environment: d.Environment, - Provisioner: pr.Provisioner, + // Determine the desired provisioning args and compare them against the currently provisioned args. + // s.Provision persists the provisioned args on the resource, so this is a stable, stateless way to detect + // whether the runtime actually needs to be reprovisioned. Other kinds of drift (e.g. templates, billing plan, + // custom domain) are handled by the drift-aware resource check below. + desiredArgs := &provisioner.RuntimeArgs{ Slots: slots, Version: runtimeVersion, + Environment: d.Environment, OverrideDiskGB: proj.OverrideDiskGB, - Annotations: annotations.ToMap(), - }) + } + currentArgs, err := provisioner.NewRuntimeArgs(pr.Args) if err != nil { return err } + if reflect.DeepEqual(desiredArgs, currentArgs) { + // No provisioning args changed. Run a drift-aware health check of the deployment's provisioner + // resources instead of reprovisioning. + err = s.CheckDeploymentInner(ctx, d) + if err != nil { + return err + } + } else { + // Provisioning args changed. Mark the deployment as updating while we reprovision the runtime. + d, err = s.DB.UpdateDeploymentStatus(ctx, d.ID, database.DeploymentStatusUpdating, "Updating...") + if err != nil { + return err + } + + // Reprovision the runtime. + // This is idempotent and will (partially) update the existing provisioned runtime. + annotations := s.NewDeploymentAnnotations(org, proj, d.Environment) + _, err = s.provisionRuntime(ctx, &provisionRuntimeOptions{ + DeploymentID: d.ID, + Environment: d.Environment, + Provisioner: pr.Provisioner, + Slots: slots, + Version: runtimeVersion, + OverrideDiskGB: proj.OverrideDiskGB, + Annotations: annotations.ToMap(), + }) + if err != nil { + return err + } + } + // Connect to the runtime and call ReloadConfig. // The runtime will pull the latest variables, annotations, and frontend_url from the admin service, - // and will also force a repo pull. + // and will also force a repo pull if the deployment config has changed. + // ReloadConfig is change-aware and cheap (no repo pull or controller restart) when nothing has changed. rt, err := s.OpenRuntimeClient(d) if err != nil { return err @@ -553,6 +590,50 @@ func (s *Service) UpdateDeploymentInner(ctx context.Context, d *database.Deploym return nil } +// CheckDeploymentInner health checks all provisioner resources for a deployment. +// The check is drift-aware (it only reprovisions when the underlying resource has actually drifted) and is +// therefore safe to run periodically. It must only be called from within the reconcile loop so that provisioner +// resources are never modified concurrently with other deployment operations. +// The implementation is idempotent, enabling it to be called from a retryable background job. +func (s *Service) CheckDeploymentInner(ctx context.Context, depl *database.Deployment) error { + // Find project and organization, we need this to build the deployment annotations + proj, err := s.DB.FindProject(ctx, depl.ProjectID) + if err != nil { + return err + } + + org, err := s.DB.FindOrganization(ctx, proj.OrganizationID) + if err != nil { + return err + } + + // Retrieve the deployment's provisioned resources + prs, err := s.DB.FindProvisionerResourcesForDeployment(ctx, depl.ID) + if err != nil { + return err + } + if len(prs) == 0 { + return nil + } + + // Build annotations for the deployment + annotations := s.NewDeploymentAnnotations(org, proj, depl.Environment) + + // Validate each provisioned resource + for _, pr := range prs { + s.Logger.Info("check deployment: checking resource", zap.String("organization_id", org.ID), zap.String("project_id", proj.ID), zap.String("deployment_id", depl.ID), zap.String("instance_id", depl.RuntimeInstanceID), zap.String("resource_id", pr.ID), zap.String("provisioner", pr.Provisioner), observability.ZapCtx(ctx)) + err := s.CheckProvisionerResource(ctx, pr, annotations) + if err != nil { + // We log the error, but continue to the next resource + s.Logger.Error("check deployment: failed to check resource", zap.String("organization_id", org.ID), zap.String("project_id", proj.ID), zap.String("deployment_id", depl.ID), zap.String("instance_id", depl.RuntimeInstanceID), zap.String("resource_id", pr.ID), zap.String("provisioner", pr.Provisioner), zap.Error(err), observability.ZapCtx(ctx)) + continue + } + s.Logger.Info("check deployment: checked resource", zap.String("organization_id", org.ID), zap.String("project_id", proj.ID), zap.String("deployment_id", depl.ID), zap.String("instance_id", depl.RuntimeInstanceID), zap.String("resource_id", pr.ID), zap.String("provisioner", pr.Provisioner), observability.ZapCtx(ctx)) + } + + return nil +} + func (s *Service) CheckProvisionerResource(ctx context.Context, pr *database.ProvisionerResource, annotations DeploymentAnnotations) error { // Find the provisioner p, ok := s.ProvisionerSet[pr.Provisioner] diff --git a/admin/jobs/river/reconcile_deployment.go b/admin/jobs/river/reconcile_deployment.go index 24c0a2b13ca6..0428bfcdeb80 100644 --- a/admin/jobs/river/reconcile_deployment.go +++ b/admin/jobs/river/reconcile_deployment.go @@ -55,13 +55,11 @@ func (w *ReconcileDeploymentWorker) Work(ctx context.Context, job *river.Job[Rec case database.DeploymentStatusRunning: // Check current status to either start or update the deployment if depl.Status == database.DeploymentStatusRunning { - // Update the deployment status to updating - depl, err = w.admin.DB.UpdateDeploymentStatus(ctx, depl.ID, database.DeploymentStatusUpdating, "Updating...") - if err != nil { - return err - } - - // Update the deployment by updating its runtime instance and resources. + // Reconcile the running deployment towards its desired configuration. + // UpdateDeploymentInner is change-aware: it only reprovisions when the provisioning args have + // changed, and otherwise performs a lightweight drift-aware resource check. We therefore keep the + // deployment in the Running status here rather than flipping it to Updating, which would otherwise flap + // on every periodic reconciliation. err := w.admin.UpdateDeploymentInner(ctx, depl) if err != nil { return err diff --git a/admin/jobs/river/validate_deployments.go b/admin/jobs/river/validate_deployments.go index 8c12a8648b56..ee67a43da27a 100644 --- a/admin/jobs/river/validate_deployments.go +++ b/admin/jobs/river/validate_deployments.go @@ -2,7 +2,6 @@ package river import ( "context" - "sync" "time" "github.com/rilldata/rill/admin" @@ -24,57 +23,29 @@ type ValidateDeploymentsWorker struct { const validateDeploymentsForProjectTimeout = 5 * time.Minute func (w *ValidateDeploymentsWorker) Work(ctx context.Context, job *river.Job[ValidateDeploymentsArgs]) error { - var wg sync.WaitGroup - ch := make(chan *database.Project) - - concurrency := 30 - if w.admin.ProvisionerMaxConcurrency > 0 { - concurrency = w.admin.ProvisionerMaxConcurrency - } else { - w.admin.Logger.Warn("validate deployments: provisioner max concurrency invalid, using default concurrency of 30", zap.Int("provisioner_max_concurrency", w.admin.ProvisionerMaxConcurrency), observability.ZapCtx(ctx)) - } - - // Setup concurrent workers - for range concurrency { - wg.Add(1) - go func() { - defer wg.Done() - // Read projects from shared channel - for proj := range ch { - w.admin.Logger.Info("validate deployments: validating project deployments", zap.String("project_id", proj.ID), observability.ZapCtx(ctx)) - err := w.validateDeploymentsForProject(ctx, proj) - if err != nil { - // We log the error, but continue to the next project - w.admin.Logger.Error("validate deployments: failed to validate project deployments", zap.String("project_id", proj.ID), zap.Error(err), observability.ZapCtx(ctx)) - } - } - }() - } - - // Iterate over batches of projects and add them to the shared channel + // Iterate over batches of projects and validate each project's deployments. + // The per-project work is lightweight (it only schedules reconcile jobs), so we process projects sequentially. limit := 100 afterID := "" - stop := false - for !stop { - // Get batch and update iterator variables + for { projs, err := w.admin.DB.FindProjects(ctx, afterID, limit) if err != nil { return err } - if len(projs) < limit { - stop = true - } - if len(projs) != 0 { - afterID = projs[len(projs)-1].ID - } for _, proj := range projs { - ch <- proj + w.admin.Logger.Info("validate deployments: validating project deployments", zap.String("project_id", proj.ID), observability.ZapCtx(ctx)) + if err := w.validateDeploymentsForProject(ctx, proj); err != nil { + // We log the error, but continue to the next project + w.admin.Logger.Error("validate deployments: failed to validate project deployments", zap.String("project_id", proj.ID), zap.Error(err), observability.ZapCtx(ctx)) + } } - } - close(ch) - wg.Wait() + if len(projs) < limit { + break + } + afterID = projs[len(projs)-1].ID + } return nil } @@ -121,28 +92,18 @@ func (w *ValidateDeploymentsWorker) validateDeploymentsForProject(ctx context.Co continue } - // Retrieve the deployment's provisioned resources - prs, err := w.admin.DB.FindProvisionerResourcesForDeployment(ctx, depl.ID) + // Schedule a reconcile job. We deliberately do not check the deployment's provisioner resources directly + // here: all modifications to provisioner resources must happen inside the reconcile loop to avoid racing + // with other deployment operations (see Service.CheckDeploymentInner / Service.UpdateDeploymentInner). + // Reconcile only ever drives a deployment toward its own DesiredStatus, is a no-op for already-consistent + // deployments (e.g. stopped/stopped, deleted/deleted), and is de-duplicated per deployment, so scheduling + // for every deployment is safe and self-healing. + _, err := w.admin.Jobs.ReconcileDeployment(ctx, depl.ID) if err != nil { - return err - } - if len(prs) == 0 { + w.admin.Logger.Error("validate deployments: failed to schedule reconcile", zap.String("organization_id", org.ID), zap.String("project_id", proj.ID), zap.String("deployment_id", depl.ID), zap.String("instance_id", depl.RuntimeInstanceID), zap.Error(err), observability.ZapCtx(ctx)) continue } - - // Build annotations for the deployment - annotations := w.admin.NewDeploymentAnnotations(org, proj, depl.Environment) - - // Validate each provisioned resource - for _, pr := range prs { - w.admin.Logger.Info("validate deployments: checking resource", zap.String("organization_id", org.ID), zap.String("project_id", proj.ID), zap.String("deployment_id", depl.ID), zap.String("instance_id", depl.RuntimeInstanceID), zap.String("resource_id", pr.ID), zap.String("provisioner", pr.Provisioner), observability.ZapCtx(ctx)) - err := w.admin.CheckProvisionerResource(ctx, pr, annotations) - if err != nil { - w.admin.Logger.Error("validate deployments: failed to check resource", zap.String("organization_id", org.ID), zap.String("project_id", proj.ID), zap.String("deployment_id", depl.ID), zap.String("instance_id", depl.RuntimeInstanceID), zap.String("resource_id", pr.ID), zap.String("provisioner", pr.Provisioner), zap.Error(err), observability.ZapCtx(ctx)) - continue - } - w.admin.Logger.Info("validate deployments: checked resource", zap.String("organization_id", org.ID), zap.String("project_id", proj.ID), zap.String("deployment_id", depl.ID), zap.String("instance_id", depl.RuntimeInstanceID), zap.String("resource_id", pr.ID), zap.String("provisioner", pr.Provisioner), observability.ZapCtx(ctx)) - } + w.admin.Logger.Info("validate deployments: scheduled reconcile", zap.String("organization_id", org.ID), zap.String("project_id", proj.ID), zap.String("deployment_id", depl.ID), zap.String("instance_id", depl.RuntimeInstanceID), observability.ZapCtx(ctx)) } return nil diff --git a/cli/cmd/admin/start.go b/cli/cmd/admin/start.go index 7f092646bfb4..ce750a210279 100644 --- a/cli/cmd/admin/start.go +++ b/cli/cmd/admin/start.go @@ -57,7 +57,6 @@ type Config struct { RiverDatabaseURL string `split_words:"true"` RedisURL string `default:"" split_words:"true"` ProvisionerSetJSON string `split_words:"true"` - ProvisionerMaxConcurrency int `default:"30" split_words:"true"` DefaultProvisioner string `split_words:"true"` Jobs []string `split_words:"true"` LogLevel zapcore.Level `default:"info" split_words:"true"` @@ -327,7 +326,6 @@ func StartCmd(ch *cmdutil.Helper) *cobra.Command { ExternalURL: conf.ExternalGRPCURL, // NOTE: using gRPC url FrontendURL: conf.FrontendURL, ProvisionerSetJSON: conf.ProvisionerSetJSON, - ProvisionerMaxConcurrency: conf.ProvisionerMaxConcurrency, DefaultProvisioner: conf.DefaultProvisioner, Version: ch.Version, MetricsProjectOrg: metricsProjectOrg,