Feb 2, 2022 · Feb 2, 2022 · Feb 2, 2022 · Feb 2, 2022 · Feb 2, 2022 · Feb 2, 2022
diff --git a/provisioner/terraform/provision.go b/provisioner/terraform/provision.go
 "github.com/hashicorp/terraform-exec/tfexec"
 "golang.org/x/xerrors"

 "cdr.dev/slog"

 "github.com/coder/coder/provisionersdk/proto"
 )

 resources := make([]*proto.Resource, 0)
 if state.Values != nil {
 for _, resource := range state.Values.RootModule.Resources {
 t.logger.Debug(ctx, "appending state file", slog.F("name", resource.Name), slog.F("type", resource.Type))
 resources = append(resources, &proto.Resource{
 Name: resource.Name,
 Type: resource.Type,
 })
 }
 }

 t.logger.Debug(ctx, "sending completion response")
 return stream.Send(&proto.Provision_Response{
 Type: &proto.Provision_Response_Complete{
 Complete: &proto.Provision_Complete{
diff --git a/provisionerd/provisionerd.go b/provisionerd/provisionerd.go
 daemon := &provisionerDaemon{
 clientDialer: clientDialer,
 opts:         opts,

 closeContext: ctx,
 closeCancel:  ctxCancel,
 closed:       make(chan struct{}),
 acquiredJobCancel    context.CancelFunc
 acquiredJobCancelled atomic.Bool
 acquiredJobRunning   atomic.Bool
 acquiredJobDone chan struct{}
 acquiredJobGroupsync.WaitGroup
 }

 // Connect establishes a connection to coderd.

 // Locks a job in the database, and runs it!
 func (p *provisionerDaemon) acquireJob(ctx context.Context) {
 if p.isClosed() {
 return
 }
 p.acquiredJobMutex.Lock()
 defer p.acquiredJobMutex.Unlock()
 if p.isRunningJob() {
 p.opts.Logger.Warn(context.Background(), "acquire job", slog.Error(err))
 return
 }
 if p.isClosed() {
 return
 }
 if p.acquiredJob.JobId == "" {
 p.opts.Logger.Debug(context.Background(), "no jobs available")
 return
 }
 ctx, p.acquiredJobCancel = context.WithCancel(ctx)
 p.acquiredJobGroup.Add(1)
 p.acquiredJobCancelled.Store(false)
 p.acquiredJobRunning.Store(true)
 p.acquiredJobDone = make(chan struct{})

 p.opts.Logger.Info(context.Background(), "acquired job",
 slog.F("job_id", p.acquiredJob.JobId),
 slog.F("organization_name", p.acquiredJob.OrganizationName),
 slog.F("project_name", p.acquiredJob.ProjectName),
 slog.F("username", p.acquiredJob.UserName),
 slog.F("provisioner", p.acquiredJob.Provisioner),
 slog.F("job_type", p.acquiredJob.Type),
 )

 go p.runJob(ctx)
 return
 }
 p.opts.Logger.Debug(ctx, "cleaned up work directory")
 p.acquiredJobMutex.Lock()
 defer p.acquiredJobMutex.Unlock()
 p.acquiredJobRunning.Store(false)
 close(p.acquiredJobDone)
 p.acquiredJobGroup.Done()
 }()
 // It's safe to cast this ProvisionerType. This data is coming directly from coderd.
 provisioner, hasProvisioner := p.opts.Provisioners[p.acquiredJob.Provisioner]
 }

 p.acquiredJobCancel()
 p.opts.Logger.Info(context.Background(), "completed job")
 p.opts.Logger.Info(context.Background(), "completed job", slog.F("job_id", p.acquiredJob.JobId))
 }

 func (p *provisionerDaemon) runProjectImport(ctx context.Context, provisioner sdkproto.DRPCProvisionerClient, job *proto.AcquiredJob_ProjectImport_) {
 func (p *provisionerDaemon) closeWithError(err error) error {
 p.closeMutex.Lock()
 defer p.closeMutex.Unlock()

 if p.isClosed() {
 return p.closeError
 }
 if !p.acquiredJobCancelled.Load() {
 p.cancelActiveJob(errMsg)
 }
 <-p.acquiredJobDone
 p.acquiredJobGroup.Wait()
 }

 p.opts.Logger.Debug(context.Background(), "closing server with error", slog.Error(err))
Original file line number	Diff line number	Diff line change
Expand Up		@@ -12,6 +12,8 @@ import (
		"github.com/hashicorp/terraform-exec/tfexec"
		"golang.org/x/xerrors"

		"cdr.dev/slog"

		"github.com/coder/coder/provisionersdk/proto"
		)

Expand DownExpand Up		@@ -149,13 +151,15 @@ func (t terraform) Provision(request proto.Provision_Request, stream proto.DRP
		resources := make([]*proto.Resource, 0)
		if state.Values != nil {
		for _, resource := range state.Values.RootModule.Resources {
		t.logger.Debug(ctx, "appending state file", slog.F("name", resource.Name), slog.F("type", resource.Type))
		resources = append(resources, &proto.Resource{
		Name: resource.Name,
		Type: resource.Type,
		})
		}
		}

		t.logger.Debug(ctx, "sending completion response")
		return stream.Send(&proto.Provision_Response{
		Type: &proto.Provision_Response_Complete{
		Complete: &proto.Provision_Complete{
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -52,7 +52,6 @@ func New(clientDialer Dialer, opts *Options) io.Closer {
		daemon := &provisionerDaemon{
		clientDialer: clientDialer,
		opts: opts,

		closeContext: ctx,
		closeCancel: ctxCancel,
		closed: make(chan struct{}),
Expand DownExpand Up		@@ -90,7 +89,7 @@ type provisionerDaemon struct {
		acquiredJobCancel context.CancelFunc
		acquiredJobCancelled atomic.Bool
		acquiredJobRunning atomic.Bool
		acquiredJobDone chan struct{}
		acquiredJobGroupsync.WaitGroup
		}

		// Connect establishes a connection to coderd.
Expand DownExpand Up		@@ -156,6 +155,9 @@ func (p *provisionerDaemon) connect(ctx context.Context) {

		// Locks a job in the database, and runs it!
		func (p *provisionerDaemon) acquireJob(ctx context.Context) {
		if p.isClosed() {
		return
		}
		p.acquiredJobMutex.Lock()
		defer p.acquiredJobMutex.Unlock()
		if p.isRunningJob() {
Expand All		@@ -174,23 +176,22 @@ func (p *provisionerDaemon) acquireJob(ctx context.Context) {
		p.opts.Logger.Warn(context.Background(), "acquire job", slog.Error(err))
		return
		}
		if p.isClosed() {
		return
		}
		if p.acquiredJob.JobId == "" {
		p.opts.Logger.Debug(context.Background(), "no jobs available")
		return
		}
		ctx, p.acquiredJobCancel = context.WithCancel(ctx)
		p.acquiredJobGroup.Add(1)
		p.acquiredJobCancelled.Store(false)
		p.acquiredJobRunning.Store(true)
		p.acquiredJobDone = make(chan struct{})

		p.opts.Logger.Info(context.Background(), "acquired job",
		slog.F("job_id", p.acquiredJob.JobId),
		slog.F("organization_name", p.acquiredJob.OrganizationName),
		slog.F("project_name", p.acquiredJob.ProjectName),
		slog.F("username", p.acquiredJob.UserName),
		slog.F("provisioner", p.acquiredJob.Provisioner),
		slog.F("job_type", p.acquiredJob.Type),
		)

		go p.runJob(ctx)
Expand DownExpand Up		@@ -232,10 +233,8 @@ func (p *provisionerDaemon) runJob(ctx context.Context) {
		return
		}
		p.opts.Logger.Debug(ctx, "cleaned up work directory")
		p.acquiredJobMutex.Lock()
		defer p.acquiredJobMutex.Unlock()
		p.acquiredJobRunning.Store(false)
		close(p.acquiredJobDone)
		p.acquiredJobGroup.Done()
		}()
		// It's safe to cast this ProvisionerType. This data is coming directly from coderd.
		provisioner, hasProvisioner := p.opts.Provisioners[p.acquiredJob.Provisioner]
Expand DownExpand Up		@@ -328,7 +327,7 @@ func (p *provisionerDaemon) runJob(ctx context.Context) {
		}

		p.acquiredJobCancel()
		p.opts.Logger.Info(context.Background(), "completed job")
		p.opts.Logger.Info(context.Background(), "completed job", slog.F("job_id", p.acquiredJob.JobId))
		}

		func (p provisionerDaemon) runProjectImport(ctx context.Context, provisioner sdkproto.DRPCProvisionerClient, job proto.AcquiredJob_ProjectImport_) {
Expand DownExpand Up		@@ -508,6 +507,7 @@ func (p *provisionerDaemon) Close() error {
		func (p *provisionerDaemon) closeWithError(err error) error {
		p.closeMutex.Lock()
		defer p.closeMutex.Unlock()

		if p.isClosed() {
		return p.closeError
		}
Expand All		@@ -520,7 +520,7 @@ func (p *provisionerDaemon) closeWithError(err error) error {
		if !p.acquiredJobCancelled.Load() {
		p.cancelActiveJob(errMsg)
		}
		<-p.acquiredJobDone
		p.acquiredJobGroup.Wait()
		}

		p.opts.Logger.Debug(context.Background(), "closing server with error", slog.Error(err))
Expand Down