opensource/coder
3
0
Fork 0
mirror of https://github.com/coder/coder.git synced 2026-09-24 15:04:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Refactor Provisioner to distinguish Plan and Apply (#5036)

Browse Source
This commit is contained in:
Ammar Bandukwala
2022-11-11 16:45:58 -06:00
committed by GitHub
parent 71601f4971
commit 95fb59696e
37 changed files with 942 additions and 648 deletions
Download Patch File Download Diff File Expand all files Collapse all files
provisionerd/provisionerd.go
+12 -10
View File
@@ -322,16 +322,18 @@ func (p *Server) acquireJob(ctx context.Context) {
p.activeJob = runner.New(
ctx,
job,
p,
p.opts.Logger,
p.opts.Filesystem,
p.opts.WorkDirectory,
provisioner,
p.opts.UpdateInterval,
p.opts.ForceCancelInterval,
p.opts.LogBufferInterval,
p.tracer,
p.opts.Metrics.Runner,
runner.Options{
Updater: p,
Logger: p.opts.Logger,
Filesystem: p.opts.Filesystem,
WorkDirectory: p.opts.WorkDirectory,
Provisioner: provisioner,
UpdateInterval: p.opts.UpdateInterval,
ForceCancelInterval: p.opts.ForceCancelInterval,
LogDebounceInterval: p.opts.LogBufferInterval,
Tracer: p.tracer,
Metrics: p.opts.Metrics.Runner,
},
)
go p.activeJob.Run()
provisionerd/provisionerd_test.go
+3 -3
View File
@@ -229,7 +229,7 @@ func TestProvisionerd(t *testing.T) {
didComplete atomic.Bool
didLog atomic.Bool
didAcquireJob atomic.Bool
didDryRun atomic.Bool
didDryRun = atomic.NewBool(true)
didReadme atomic.Bool
completeChan = make(chan struct{})
completeOnce sync.Once
@@ -299,8 +299,8 @@ func TestProvisionerd(t *testing.T) {
provision: func(stream sdkproto.DRPCProvisioner_ProvisionStream) error {
request, err := stream.Recv()
require.NoError(t, err)
if request.GetStart().DryRun {
didDryRun.Store(true)
if request.GetApply() != nil {
didDryRun.Store(false)
}
err = stream.Send(&sdkproto.Provision_Response{
Type: &sdkproto.Provision_Response_Log{
provisionerd/runner/runner.go
+101 -67
View File
@@ -86,19 +86,23 @@ type JobUpdater interface {
CompleteJob(ctx context.Context, in *proto.CompletedJob) error
}
type Options struct {
Updater JobUpdater
Logger slog.Logger
Filesystem afero.Fs
WorkDirectory string
Provisioner sdkproto.DRPCProvisionerClient
UpdateInterval time.Duration
ForceCancelInterval time.Duration
LogDebounceInterval time.Duration
Tracer trace.Tracer
Metrics Metrics
}
func New(
ctx context.Context,
job *proto.AcquiredJob,
updater JobUpdater,
logger slog.Logger,
filesystem afero.Fs,
workDirectory string,
provisioner sdkproto.DRPCProvisionerClient,
updateInterval time.Duration,
forceCancelInterval time.Duration,
logDebounceInterval time.Duration,
tracer trace.Tracer,
metrics Metrics,
opts Options,
) *Runner {
m := new(sync.Mutex)
@@ -107,17 +111,17 @@ func New(
gracefulContext, cancelFunc := context.WithCancel(forceStopContext)
return &Runner{
tracer: tracer,
metrics: metrics,
tracer: opts.Tracer,
metrics: opts.Metrics,
job: job,
sender: updater,
logger: logger.With(slog.F("job_id", job.JobId)),
filesystem: filesystem,
workDirectory: workDirectory,
provisioner: provisioner,
updateInterval: updateInterval,
forceCancelInterval: forceCancelInterval,
logBufferInterval: logDebounceInterval,
sender: opts.Updater,
logger: opts.Logger.With(slog.F("job_id", job.JobId)),
filesystem: opts.Filesystem,
workDirectory: opts.WorkDirectory,
provisioner: opts.Provisioner,
updateInterval: opts.UpdateInterval,
forceCancelInterval: opts.ForceCancelInterval,
logBufferInterval: opts.LogDebounceInterval,
queuedLogs: make([]*proto.Log, 0),
mutex: m,
cond: sync.NewCond(m),
@@ -667,12 +671,13 @@ func (r *Runner) runTemplateImportProvision(ctx context.Context, values []*sdkpr
}
}()
err = stream.Send(&sdkproto.Provision_Request{
Type: &sdkproto.Provision_Request_Start{
Start: &sdkproto.Provision_Start{
Directory: r.workDirectory,
Type: &sdkproto.Provision_Request_Plan{
Plan: &sdkproto.Provision_Plan{
Config: &sdkproto.Provision_Config{
Directory: r.workDirectory,
Metadata: metadata,
},
ParameterValues: values,
DryRun: true,
Metadata: metadata,
},
},
})
@@ -772,29 +777,11 @@ func (r *Runner) runTemplateDryRun(ctx context.Context) (*proto.CompletedJob, *p
}, nil
}
func (r *Runner) runWorkspaceBuild(ctx context.Context) (*proto.CompletedJob, *proto.FailedJob) {
ctx, span := r.startTrace(ctx, tracing.FuncName())
defer span.End()
var stage string
switch r.job.GetWorkspaceBuild().Metadata.WorkspaceTransition {
case sdkproto.WorkspaceTransition_START:
stage = "Starting workspace"
case sdkproto.WorkspaceTransition_STOP:
stage = "Stopping workspace"
case sdkproto.WorkspaceTransition_DESTROY:
stage = "Destroying workspace"
}
r.queueLog(ctx, &proto.Log{
Source: proto.LogSource_PROVISIONER_DAEMON,
Level: sdkproto.LogLevel_INFO,
Stage: stage,
CreatedAt: time.Now().UnixMilli(),
})
// use the notStopped so that if we attempt to gracefully cancel, the stream will still be available for us
// to send the cancel to the provisioner
func (r *Runner) buildWorkspace(ctx context.Context, stage string, req *sdkproto.Provision_Request) (
*sdkproto.Provision_Complete, *proto.FailedJob,
) {
// use the notStopped so that if we attempt to gracefully cancel, the stream
// will still be available for us to send the cancel to the provisioner
stream, err := r.provisioner.Provision(ctx)
if err != nil {
return nil, r.failedJobf("provision: %s", err)
@@ -812,16 +799,8 @@ func (r *Runner) runWorkspaceBuild(ctx context.Context) (*proto.CompletedJob, *p
})
}
}()
err = stream.Send(&sdkproto.Provision_Request{
Type: &sdkproto.Provision_Request_Start{
Start: &sdkproto.Provision_Start{
Directory: r.workDirectory,
ParameterValues: r.job.GetWorkspaceBuild().ParameterValues,
Metadata: r.job.GetWorkspaceBuild().Metadata,
State: r.job.GetWorkspaceBuild().State,
},
},
})
err = stream.Send(req)
if err != nil {
return nil, r.failedJobf("start provision: %s", err)
}
@@ -870,21 +849,76 @@ func (r *Runner) runWorkspaceBuild(ctx context.Context) (*proto.CompletedJob, *p
slog.F("state_length", len(msgType.Complete.State)),
)
// Stop looping!
return &proto.CompletedJob{
JobId: r.job.JobId,
Type: &proto.CompletedJob_WorkspaceBuild_{
WorkspaceBuild: &proto.CompletedJob_WorkspaceBuild{
State: msgType.Complete.State,
Resources: msgType.Complete.Resources,
},
},
}, nil
return msgType.Complete, nil
default:
return nil, r.failedJobf("invalid message type %T received from provisioner", msg.Type)
}
}
}
func (r *Runner) runWorkspaceBuild(ctx context.Context) (*proto.CompletedJob, *proto.FailedJob) {
ctx, span := r.startTrace(ctx, tracing.FuncName())
defer span.End()
var (
applyStage string
)
switch r.job.GetWorkspaceBuild().Metadata.WorkspaceTransition {
case sdkproto.WorkspaceTransition_START:
applyStage = "Starting workspace"
case sdkproto.WorkspaceTransition_STOP:
applyStage = "Stopping workspace"
case sdkproto.WorkspaceTransition_DESTROY:
applyStage = "Destroying workspace"
}
r.queueLog(ctx, &proto.Log{
Source: proto.LogSource_PROVISIONER_DAEMON,
Level: sdkproto.LogLevel_INFO,
Stage: applyStage,
CreatedAt: time.Now().UnixMilli(),
})
config := &sdkproto.Provision_Config{
Directory: r.workDirectory,
Metadata: r.job.GetWorkspaceBuild().Metadata,
State: r.job.GetWorkspaceBuild().State,
}
completed, failed := r.buildWorkspace(ctx, "Planning infrastructure", &sdkproto.Provision_Request{
Type: &sdkproto.Provision_Request_Plan{
Plan: &sdkproto.Provision_Plan{
Config: config,
ParameterValues: r.job.GetWorkspaceBuild().ParameterValues,
},
},
})
if failed != nil {
return nil, failed
}
completedApply, failed := r.buildWorkspace(ctx, applyStage, &sdkproto.Provision_Request{
Type: &sdkproto.Provision_Request_Apply{
Apply: &sdkproto.Provision_Apply{
Config: config,
Plan: completed.GetPlan(),
},
},
})
if failed != nil {
return nil, failed
}
return &proto.CompletedJob{
JobId: r.job.JobId,
Type: &proto.CompletedJob_WorkspaceBuild_{
WorkspaceBuild: &proto.CompletedJob_WorkspaceBuild{
State: completedApply.GetState(),
Resources: completedApply.GetResources(),
},
},
}, nil
}
func (r *Runner) failedJobf(format string, args ...interface{}) *proto.FailedJob {
return &proto.FailedJob{
JobId: r.job.JobId,