Skip to content

Move resource checks to deployment reconciliation #9653

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
kaspersjo wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Move resource checks to deployment reconciliation #9653

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 0 additions & 3 deletions admin/admin.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,6 @@ type Options struct {
ExternalURL string
FrontendURL string
ProvisionerSetJSON string
ProvisionerMaxConcurrency int
DefaultProvisioner string
Version version.Version
MetricsProjectOrg string
Expand All @@ -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
Expand Down Expand Up @@ -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,
Expand Down
107 changes: 94 additions & 13 deletions admin/deployments.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"reflect"
"regexp"
"strings"
"time"
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Comment on lines +543 to +546

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 6, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Reprovision when annotations-only updates occur

When a project update only changes annotations (admin/projects.go marks oldProj.Annotations as deployment-impacting), the desired/current RuntimeArgs are equal, so this new branch calls CheckDeploymentInner instead of provisionRuntime. The Kubernetes check only triggers reprovision for version/template/org plan/custom domain drift (admin/provisioner/kubernetes/kubernetes.go), so templates that consume project annotations via DeploymentAnnotations keep the old rendered values even though UpdateDeploymentsForProject scheduled reconciliation; before this change provisionRuntime always reapplied them.

Useful? React with 👍 / 👎.

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
Expand All @@ -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]
Expand Down
12 changes: 5 additions & 7 deletions admin/jobs/river/reconcile_deployment.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
83 changes: 22 additions & 61 deletions admin/jobs/river/validate_deployments.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@ package river

import (
"context"
"sync"
"time"

"github.com/rilldata/rill/admin"
Expand All @@ -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
}
Expand Down Expand Up @@ -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
Expand Down
2 changes: 0 additions & 2 deletions cli/cmd/admin/start.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"`
Expand Down Expand Up @@ -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,
Expand Down
Loading