2018-08-07 01:46:17 +02:00
|
|
|
// Copyright 2016-2018, Pulumi Corporation.
|
|
|
|
//
|
|
|
|
// 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 deploy
|
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
|
|
|
|
|
|
|
"github.com/pkg/errors"
|
2018-08-13 22:41:01 +02:00
|
|
|
"github.com/pulumi/pulumi/pkg/diag"
|
2018-08-21 23:05:00 +02:00
|
|
|
"github.com/pulumi/pulumi/pkg/resource"
|
2018-08-23 02:52:46 +02:00
|
|
|
"github.com/pulumi/pulumi/pkg/resource/deploy/providers"
|
|
|
|
"github.com/pulumi/pulumi/pkg/resource/graph"
|
2018-08-21 23:05:00 +02:00
|
|
|
"github.com/pulumi/pulumi/pkg/util/contract"
|
2018-08-07 01:46:17 +02:00
|
|
|
"github.com/pulumi/pulumi/pkg/util/logging"
|
2018-09-06 00:08:09 +02:00
|
|
|
"github.com/pulumi/pulumi/pkg/util/result"
|
2018-08-07 01:46:17 +02:00
|
|
|
)
|
|
|
|
|
2018-08-21 23:05:00 +02:00
|
|
|
// planExecutor is responsible for taking a plan and driving it to completion.
|
2018-08-07 01:46:17 +02:00
|
|
|
// Its primary responsibility is to own a `stepGenerator` and `stepExecutor`, serving
|
|
|
|
// as the glue that links the two subsystems together.
|
2018-08-21 23:05:00 +02:00
|
|
|
type planExecutor struct {
|
|
|
|
plan *Plan // The plan that we are executing
|
2018-08-07 01:46:17 +02:00
|
|
|
|
|
|
|
stepGen *stepGenerator // step generator owned by this plan
|
|
|
|
stepExec *stepExecutor // step executor owned by this plan
|
2018-08-21 23:05:00 +02:00
|
|
|
}
|
2018-08-07 01:46:17 +02:00
|
|
|
|
2018-08-21 23:05:00 +02:00
|
|
|
// execError creates an error appropriate for returning from planExecutor.Execute.
|
|
|
|
func execError(message string, preview bool) error {
|
|
|
|
kind := "update"
|
|
|
|
if preview {
|
|
|
|
kind = "preview"
|
|
|
|
}
|
|
|
|
return errors.New(kind + " " + message)
|
|
|
|
}
|
|
|
|
|
|
|
|
// reportError reports a single error to the executor's diag stream with the indicated URN for context.
|
|
|
|
func (pe *planExecutor) reportError(urn resource.URN, err error) {
|
2018-08-31 21:33:01 +02:00
|
|
|
pe.plan.Diag().Errorf(diag.RawMessage(urn, err.Error()))
|
2018-08-07 01:46:17 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// Execute executes a plan to completion, using the given cancellation context and running a preview
|
|
|
|
// or update.
|
2018-08-23 02:52:46 +02:00
|
|
|
func (pe *planExecutor) Execute(callerCtx context.Context, opts Options, preview bool) error {
|
|
|
|
// Set up a goroutine that will signal cancellation to the plan's plugins if the caller context is cancelled. We do
|
2018-08-21 23:05:00 +02:00
|
|
|
// not hang this off of the context we create below because we do not want the failure of a single step to cause
|
|
|
|
// other steps to fail.
|
|
|
|
done := make(chan bool)
|
2018-09-11 00:18:25 +02:00
|
|
|
defer close(done)
|
2018-08-07 01:46:17 +02:00
|
|
|
go func() {
|
|
|
|
select {
|
2018-08-23 02:52:46 +02:00
|
|
|
case <-callerCtx.Done():
|
2018-09-11 00:18:25 +02:00
|
|
|
logging.V(4).Infof("planExecutor.Execute(...): signalling cancellation to providers...")
|
2018-08-21 23:05:00 +02:00
|
|
|
cancelErr := pe.plan.ctx.Host.SignalCancellation()
|
|
|
|
if cancelErr != nil {
|
2018-09-11 00:18:25 +02:00
|
|
|
logging.V(4).Infof("planExecutor.Execute(...): failed to signal cancellation to providers: %v", cancelErr)
|
2018-08-21 23:05:00 +02:00
|
|
|
}
|
|
|
|
case <-done:
|
2018-09-11 00:18:25 +02:00
|
|
|
logging.V(4).Infof("planExecutor.Execute(...): exiting provider canceller")
|
2018-08-07 01:46:17 +02:00
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
2018-08-23 02:52:46 +02:00
|
|
|
// Before doing anything else, optionally refresh each resource in the base checkpoint.
|
|
|
|
if opts.Refresh {
|
|
|
|
if err := pe.refresh(callerCtx, opts, preview); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
if opts.RefreshOnly {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Begin iterating the source.
|
|
|
|
src, err := pe.plan.source.Iterate(callerCtx, opts, pe.plan)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2018-09-11 01:48:14 +02:00
|
|
|
// Set up a step generator for this plan.
|
|
|
|
pe.stepGen = newStepGenerator(pe.plan, opts)
|
|
|
|
|
|
|
|
// Retire any pending deletes that are currently present in this plan.
|
|
|
|
if err = pe.retirePendingDeletes(callerCtx, opts, preview); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2018-08-21 23:05:00 +02:00
|
|
|
// Derive a cancellable context for this plan. We will only cancel this context if some piece of the plan's
|
|
|
|
// execution fails.
|
2018-08-23 02:52:46 +02:00
|
|
|
ctx, cancel := context.WithCancel(callerCtx)
|
2018-08-21 23:05:00 +02:00
|
|
|
|
|
|
|
// Set up a step generator and executor for this plan.
|
2018-09-05 23:00:28 +02:00
|
|
|
pe.stepExec = newStepExecutor(ctx, cancel, pe.plan, opts, preview, false)
|
2018-08-21 23:05:00 +02:00
|
|
|
|
|
|
|
// We iterate the source in its own goroutine because iteration is blocking and we want the main loop to be able to
|
|
|
|
// respond to cancellation requests promptly.
|
2018-08-07 01:46:17 +02:00
|
|
|
type nextEvent struct {
|
|
|
|
Event SourceEvent
|
|
|
|
Error error
|
|
|
|
}
|
|
|
|
incomingEvents := make(chan nextEvent)
|
|
|
|
go func() {
|
|
|
|
for {
|
2018-08-21 23:05:00 +02:00
|
|
|
event, sourceErr := src.Next()
|
2018-08-07 01:46:17 +02:00
|
|
|
select {
|
|
|
|
case incomingEvents <- nextEvent{event, sourceErr}:
|
2018-08-21 23:05:00 +02:00
|
|
|
if event == nil {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
case <-done:
|
2018-09-11 00:18:25 +02:00
|
|
|
logging.V(4).Infof("planExecutor.Execute(...): incoming events goroutine exiting")
|
2018-08-07 01:46:17 +02:00
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
|
|
|
// The main loop. We'll continuously select for incoming events and the cancellation signal. There are
|
|
|
|
// a three ways we can exit this loop:
|
|
|
|
// 1. The SourceIterator sends us a `nil` event. This means that we're done processing source events and
|
|
|
|
// we should begin processing deletes.
|
|
|
|
// 2. The SourceIterator sends us an error. This means some error occurred in the source program and we
|
|
|
|
// should bail.
|
|
|
|
// 3. The stepExecCancel cancel context gets canceled. This means some error occurred in the step executor
|
|
|
|
// and we need to bail. This can also happen if the user hits Ctrl-C.
|
2018-08-21 23:05:00 +02:00
|
|
|
canceled, err := func() (bool, error) {
|
2018-09-11 00:18:25 +02:00
|
|
|
logging.V(4).Infof("planExecutor.Execute(...): waiting for incoming events")
|
2018-08-21 23:05:00 +02:00
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case event := <-incomingEvents:
|
2018-09-11 00:18:25 +02:00
|
|
|
logging.V(4).Infof("planExecutor.Execute(...): incoming event (nil? %v, %v)", event.Event == nil, event.Error)
|
2018-08-21 23:05:00 +02:00
|
|
|
|
|
|
|
if event.Error != nil {
|
|
|
|
pe.reportError("", event.Error)
|
|
|
|
cancel()
|
|
|
|
return false, event.Error
|
|
|
|
}
|
|
|
|
|
|
|
|
if event.Event == nil {
|
2018-09-28 00:49:08 +02:00
|
|
|
deleteSteps := pe.stepGen.GenerateDeletes()
|
|
|
|
deletes := pe.stepGen.ScheduleDeletes(deleteSteps)
|
|
|
|
|
|
|
|
// ScheduleDeletes gives us a list of lists of steps. Each list of steps can safely be executed in
|
|
|
|
// parallel, but each list must execute completes before the next list can safely begin executing.
|
|
|
|
//
|
|
|
|
// This is not "true" delete parallelism, since there may be resources that could safely begin
|
|
|
|
// deleting but we won't until the previous set of deletes fully completes. This approximation is
|
|
|
|
// conservative, but correct.
|
|
|
|
for _, antichain := range deletes {
|
|
|
|
logging.V(4).Infof("planExecutor.Execute(...): beginning delete antichain")
|
|
|
|
tok := pe.stepExec.ExecuteParallel(antichain)
|
|
|
|
tok.Wait(ctx)
|
|
|
|
logging.V(4).Infof("planExecutor.Execute(...): antichain complete")
|
|
|
|
}
|
2018-08-21 23:05:00 +02:00
|
|
|
|
2018-09-28 00:49:08 +02:00
|
|
|
// We're done here - signal completion so that the step executor knows to terminate.
|
|
|
|
pe.stepExec.SignalCompletion()
|
2018-08-21 23:05:00 +02:00
|
|
|
return false, nil
|
|
|
|
}
|
|
|
|
|
2018-09-06 00:08:09 +02:00
|
|
|
if res := pe.handleSingleEvent(event.Event); res != nil {
|
|
|
|
if resErr := res.Error(); resErr != nil {
|
2018-09-11 00:18:25 +02:00
|
|
|
logging.V(4).Infof("planExecutor.Execute(...): error handling event: %v", resErr)
|
2018-09-06 00:08:09 +02:00
|
|
|
pe.reportError(pe.plan.generateEventURN(event.Event), resErr)
|
|
|
|
}
|
2018-08-21 23:05:00 +02:00
|
|
|
cancel()
|
2018-09-06 00:08:09 +02:00
|
|
|
return false, result.TODO()
|
2018-08-21 23:05:00 +02:00
|
|
|
}
|
|
|
|
case <-ctx.Done():
|
2018-09-11 00:18:25 +02:00
|
|
|
logging.V(4).Infof("planExecutor.Execute(...): context finished: %v", ctx.Err())
|
2018-08-21 23:05:00 +02:00
|
|
|
|
2018-08-23 02:52:46 +02:00
|
|
|
// NOTE: we use the presence of an error in the caller context in order to distinguish caller-initiated
|
2018-08-21 23:05:00 +02:00
|
|
|
// cancellation from internally-initiated cancellation.
|
2018-08-23 02:52:46 +02:00
|
|
|
return callerCtx.Err() != nil, nil
|
2018-08-07 01:46:17 +02:00
|
|
|
}
|
|
|
|
}
|
2018-08-21 23:05:00 +02:00
|
|
|
}()
|
2018-08-07 01:46:17 +02:00
|
|
|
|
|
|
|
pe.stepExec.WaitForCompletion()
|
2018-09-11 00:18:25 +02:00
|
|
|
logging.V(4).Infof("planExecutor.Execute(...): step executor has completed")
|
2018-08-07 01:46:17 +02:00
|
|
|
|
2018-08-21 23:05:00 +02:00
|
|
|
// Figure out if execution failed and why. Step generation and execution errors trump cancellation.
|
|
|
|
if err != nil || pe.stepExec.Errored() {
|
|
|
|
err = execError("failed", preview)
|
|
|
|
} else if canceled {
|
|
|
|
err = execError("canceled", preview)
|
2018-08-07 01:46:17 +02:00
|
|
|
}
|
2018-08-22 04:33:59 +02:00
|
|
|
return err
|
2018-08-07 01:46:17 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// handleSingleEvent handles a single source event. For all incoming events, it produces a chain that needs
|
|
|
|
// to be executed and schedules the chain for execution.
|
2018-09-06 00:08:09 +02:00
|
|
|
func (pe *planExecutor) handleSingleEvent(event SourceEvent) *result.Result {
|
2018-08-21 23:05:00 +02:00
|
|
|
contract.Require(event != nil, "event != nil")
|
2018-08-07 01:46:17 +02:00
|
|
|
|
2018-08-21 23:05:00 +02:00
|
|
|
var steps []Step
|
2018-09-06 00:08:09 +02:00
|
|
|
var res *result.Result
|
2018-08-07 01:46:17 +02:00
|
|
|
switch e := event.(type) {
|
|
|
|
case RegisterResourceEvent:
|
2018-09-11 00:18:25 +02:00
|
|
|
logging.V(4).Infof("planExecutor.handleSingleEvent(...): received RegisterResourceEvent")
|
2018-09-06 00:08:09 +02:00
|
|
|
steps, res = pe.stepGen.GenerateSteps(e)
|
2018-08-07 01:46:17 +02:00
|
|
|
case ReadResourceEvent:
|
2018-09-11 00:18:25 +02:00
|
|
|
logging.V(4).Infof("planExecutor.handleSingleEvent(...): received ReadResourceEvent")
|
2018-09-06 00:08:09 +02:00
|
|
|
steps, res = pe.stepGen.GenerateReadSteps(e)
|
2018-08-07 01:46:17 +02:00
|
|
|
case RegisterResourceOutputsEvent:
|
2018-09-11 00:18:25 +02:00
|
|
|
logging.V(4).Infof("planExecutor.handleSingleEvent(...): received register resource outputs")
|
2018-08-07 01:46:17 +02:00
|
|
|
pe.stepExec.ExecuteRegisterResourceOutputs(e)
|
2018-08-21 23:05:00 +02:00
|
|
|
return nil
|
2018-08-07 01:46:17 +02:00
|
|
|
}
|
|
|
|
|
2018-09-06 00:08:09 +02:00
|
|
|
if res != nil {
|
|
|
|
return res
|
2018-08-07 01:46:17 +02:00
|
|
|
}
|
2018-09-06 00:08:09 +02:00
|
|
|
|
2018-09-28 00:49:08 +02:00
|
|
|
pe.stepExec.ExecuteSerial(steps)
|
2018-08-21 23:05:00 +02:00
|
|
|
return nil
|
2018-08-07 01:46:17 +02:00
|
|
|
}
|
2018-08-23 02:52:46 +02:00
|
|
|
|
2018-09-11 01:48:14 +02:00
|
|
|
// retirePendingDeletes deletes all resources that are pending deletion. Run before the start of a plan, this pass
|
|
|
|
// ensures that the engine never sees any resources that are pending deletion from a previous plan.
|
|
|
|
//
|
|
|
|
// retirePendingDeletes re-uses the plan executor's step generator but uses its own step executor.
|
|
|
|
func (pe *planExecutor) retirePendingDeletes(callerCtx context.Context, opts Options, preview bool) error {
|
|
|
|
contract.Require(pe.stepGen != nil, "pe.stepGen != nil")
|
|
|
|
steps := pe.stepGen.GeneratePendingDeletes()
|
|
|
|
if len(steps) == 0 {
|
|
|
|
logging.V(4).Infoln("planExecutor.retirePendingDeletes(...): no pending deletions")
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
logging.V(4).Infof("planExecutor.retirePendingDeletes(...): executing %d steps", len(steps))
|
|
|
|
ctx, cancel := context.WithCancel(callerCtx)
|
|
|
|
|
|
|
|
stepExec := newStepExecutor(ctx, cancel, pe.plan, opts, preview, false)
|
2018-09-28 00:49:08 +02:00
|
|
|
antichains := pe.stepGen.ScheduleDeletes(steps)
|
2018-09-11 01:48:14 +02:00
|
|
|
// Submit the deletes for execution and wait for them all to retire.
|
2018-09-28 00:49:08 +02:00
|
|
|
for _, antichain := range antichains {
|
|
|
|
for _, step := range antichain {
|
|
|
|
pe.plan.Ctx().StatusDiag.Infof(diag.RawMessage(step.URN(), "completing deletion from previous update"))
|
|
|
|
}
|
2018-09-11 01:48:14 +02:00
|
|
|
|
2018-09-28 00:49:08 +02:00
|
|
|
tok := stepExec.ExecuteParallel(antichain)
|
|
|
|
tok.Wait(ctx)
|
2018-09-11 01:48:14 +02:00
|
|
|
}
|
|
|
|
|
2018-09-28 00:49:08 +02:00
|
|
|
stepExec.SignalCompletion()
|
2018-09-11 01:48:14 +02:00
|
|
|
stepExec.WaitForCompletion()
|
|
|
|
|
|
|
|
// Like Refresh, we use the presence of an error in the caller's context to detect whether or not we have been
|
|
|
|
// cancelled.
|
|
|
|
canceled := callerCtx.Err() != nil
|
|
|
|
if stepExec.Errored() {
|
|
|
|
return execError("failed", preview)
|
|
|
|
} else if canceled {
|
|
|
|
return execError("canceled", preview)
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2018-08-23 02:52:46 +02:00
|
|
|
// refresh refreshes the state of the base checkpoint file for the current plan in memory.
|
|
|
|
func (pe *planExecutor) refresh(callerCtx context.Context, opts Options, preview bool) error {
|
|
|
|
prev := pe.plan.prev
|
|
|
|
if prev == nil || len(prev.Resources) == 0 {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2018-08-30 06:00:05 +02:00
|
|
|
// Create a refresh step for each resource in the old snapshot.
|
2018-08-23 02:52:46 +02:00
|
|
|
steps := make([]Step, len(prev.Resources))
|
|
|
|
for i := range prev.Resources {
|
2018-08-30 06:00:05 +02:00
|
|
|
steps[i] = NewRefreshStep(pe.plan, prev.Resources[i], nil)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Fire up a worker pool and issue each refresh in turn.
|
|
|
|
ctx, cancel := context.WithCancel(callerCtx)
|
2018-09-05 23:00:28 +02:00
|
|
|
stepExec := newStepExecutor(ctx, cancel, pe.plan, opts, preview, true)
|
2018-09-28 00:49:08 +02:00
|
|
|
stepExec.ExecuteParallel(steps)
|
2018-08-23 02:52:46 +02:00
|
|
|
stepExec.SignalCompletion()
|
|
|
|
stepExec.WaitForCompletion()
|
|
|
|
|
|
|
|
// Rebuild this plan's map of old resources and dependency graph, stripping out any deleted resources and repairing
|
|
|
|
// dependency lists as necessary. Note that this updates the base snapshot _in memory_, so it is critical that any
|
|
|
|
// components that use the snapshot refer to the same instance and avoid reading it concurrently with this rebuild.
|
|
|
|
//
|
|
|
|
// The process of repairing dependency lists is a bit subtle. Because multiple physical resources may share a URN,
|
|
|
|
// the ability of a particular URN to be referenced in a dependency list can change based on the dependent
|
|
|
|
// resource's position in the resource list. For example, consider the following list of resources, where each
|
|
|
|
// resource is a (URN, ID, Dependencies) tuple:
|
|
|
|
//
|
|
|
|
// [ (A, 0, []), (B, 0, [A]), (A, 1, []), (A, 2, []), (C, 0, [A]) ]
|
|
|
|
//
|
|
|
|
// Let `(A, 0, [])` and `(A, 2, [])` be deleted by the refresh. This produces the following intermediate list
|
|
|
|
// before dependency lists are repaired:
|
|
|
|
//
|
|
|
|
// [ (B, 0, [A]), (A, 1, []), (C, 0, [A]) ]
|
|
|
|
//
|
|
|
|
// In order to repair the dependency lists, we iterate over the intermediate resource list, keeping track of which
|
|
|
|
// URNs refer to at least one physical resource at each point in the list, and remove any dependencies that refer
|
|
|
|
// to URNs that do not refer to any physical resources. This process produces the following final list:
|
|
|
|
//
|
|
|
|
// [ (B, 0, []), (A, 1, []), (C, 0, [A]) ]
|
|
|
|
//
|
|
|
|
// Note that the correctness of this process depends on the fact that the list of resources is a topological sort
|
|
|
|
// of its corresponding dependency graph, so a resource always appears in the list after any resources on which it
|
|
|
|
// may depend.
|
|
|
|
resources := make([]*resource.State, 0, len(prev.Resources))
|
|
|
|
referenceable := make(map[resource.URN]bool)
|
|
|
|
olds := make(map[resource.URN]*resource.State)
|
|
|
|
for _, s := range steps {
|
|
|
|
new := s.New()
|
|
|
|
if new == nil {
|
|
|
|
contract.Assert(s.Old().Custom)
|
|
|
|
contract.Assert(!providers.IsProviderType(s.Old().Type))
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
// Remove any deleted resources from this resource's dependency list.
|
|
|
|
if len(new.Dependencies) != 0 {
|
|
|
|
deps := make([]resource.URN, 0, len(new.Dependencies))
|
|
|
|
for _, d := range new.Dependencies {
|
|
|
|
if referenceable[d] {
|
|
|
|
deps = append(deps, d)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
new.Dependencies = deps
|
|
|
|
}
|
|
|
|
|
|
|
|
// Add this resource to the resource list and mark it as referenceable.
|
|
|
|
resources = append(resources, new)
|
|
|
|
referenceable[new.URN] = true
|
|
|
|
|
|
|
|
// Do not record resources that are pending deletion in the "olds" lookup table.
|
|
|
|
if !new.Delete {
|
|
|
|
olds[new.URN] = new
|
|
|
|
}
|
|
|
|
}
|
|
|
|
pe.plan.prev.Resources = resources
|
|
|
|
pe.plan.olds, pe.plan.depGraph = olds, graph.NewDependencyGraph(resources)
|
|
|
|
|
2018-08-30 06:00:05 +02:00
|
|
|
// NOTE: we use the presence of an error in the caller context in order to distinguish caller-initiated
|
|
|
|
// cancellation from internally-initiated cancellation.
|
|
|
|
canceled := callerCtx.Err() != nil
|
|
|
|
|
2018-08-23 02:52:46 +02:00
|
|
|
if stepExec.Errored() {
|
|
|
|
return execError("failed", preview)
|
|
|
|
} else if canceled {
|
|
|
|
return execError("canceled", preview)
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|