func (w *Workflow) IterateWorkflowSteps(cb func(step *Step)) {
	for _, step := range w.Steps {
		if step.IncludeWorkflow != nil {
			step.IncludeWorkflow.Workflow.IterateWorkflowSteps(cb)
		}
		cb(step)
	}
}
IterateWorkflowSteps iterates over all workflow steps, including included
workflow steps, and calls cb callback function
package daisy
import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"os"
	"path"
	"path/filepath"
	"reflect"
	"strconv"
	"strings"
	"sync"
	"time"
	"cloud.google.com/go/logging"
	"cloud.google.com/go/storage"
	"github.com/GoogleCloudPlatform/compute-image-tools/daisy/compute"
	"google.golang.org/api/iterator"
	"google.golang.org/api/option"
)
const defaultTimeout = "10m"
func daisyBkt(ctx context.Context, client *storage.Client, project string) (string, DError) {
	dBkt := strings.Replace(project, ":", "-", -1) + "-daisy-bkt"
	it := client.Buckets(ctx, project)
	for bucketAttrs, err := it.Next(); err != iterator.Done; bucketAttrs, err = it.Next() {
		if err != nil {
			return "", typedErr(apiError, "failed to iterate buckets", err)
		}
		if bucketAttrs.Name == dBkt {
			return dBkt, nil
		}
	}
	if err := client.Bucket(dBkt).Create(ctx, project, nil); err != nil {
		return "", typedErr(apiError, "failed to create bucket", err)
	}
	return dBkt, nil
}
type TimeRecord struct {
	Name      string
	StartTime time.Time
	EndTime   time.Time
}
type Var struct {
	Value       string
	Required    bool   `json:",omitempty"`
	Description string `json:",omitempty"`
}
func (v *Var) UnmarshalJSON(b []byte) error {
	var s string
	if err := json.Unmarshal(b, &s); err == nil {
		v.Value = s
		return nil
	}
	type aVar Var
	return json.Unmarshal(b, &struct{ *aVar }{aVar: (*aVar)(v)})
}
type Workflow struct {
	Cancel     chan struct{} `json:"-"`
	isCanceled bool
	cancelMx   sync.Mutex
	Name string `json:",omitempty"`
	Project string `json:",omitempty"`
	Zone string `json:",omitempty"`
	GCSPath string `json:",omitempty"`
	OAuthPath string `json:",omitempty"`
	Sources map[string]string `json:",omitempty"`
	Vars  map[string]Var   `json:",omitempty"`
	Steps map[string]*Step `json:",omitempty"`
	Dependencies map[string][]string `json:",omitempty"`
	DefaultTimeout string `json:",omitempty"`
	defaultTimeout time.Duration
	autovars              map[string]string
	workflowDir           string
	parent                *Workflow
	bucket                string
	scratchPath           string
	sourcesPath           string
	logsPath              string
	outsPath              string
	username              string
	externalLogging       bool
	gcsLoggingDisabled    bool
	cloudLoggingDisabled  bool
	stdoutLoggingDisabled bool
	id                    string
	Logger                Logger `json:"-"`
	cleanupHooks          []func() DError
	cleanupHooksMx        sync.Mutex
	recordTimeMx          sync.Mutex
	stepWait              sync.WaitGroup
	logProcessHook        func(string) string
	ComputeEndpoint    string          `json:",omitempty"`
	ComputeClient      compute.Client  `json:"-"`
	StorageClient      *storage.Client `json:"-"`
	cloudLoggingClient *logging.Client
	disks           *diskRegistry
	forwardingRules *forwardingRuleRegistry
	firewallRules   *firewallRuleRegistry
	images          *imageRegistry
	machineImages   *machineImageRegistry
	instances       *instanceRegistry
	networks        *networkRegistry
	subnetworks     *subnetworkRegistry
	targetInstances *targetInstanceRegistry
	objects         *objectRegistry
	snapshots       *snapshotRegistry
	machineTypeCache    twoDResourceCache
	instanceCache       twoDResourceCache
	diskCache           twoDResourceCache
	subnetworkCache     twoDResourceCache
	targetInstanceCache twoDResourceCache
	forwardingRuleCache twoDResourceCache
	imageCache          oneDResourceCache
	imageFamilyCache    oneDResourceCache
	machineImageCache   oneDResourceCache
	networkCache        oneDResourceCache
	firewallRuleCache   oneDResourceCache
	zonesCache          oneDResourceCache
	regionsCache        oneDResourceCache
	licenseCache        oneDResourceCache
	snapshotCache       oneDResourceCache
	stepTimeRecords             []TimeRecord
	serialControlOutputValues   map[string]string
	serialControlOutputValuesMx sync.Mutex
	ForceCleanupOnError bool
	forceCleanup bool
	cancelReason string
}
func (w *Workflow) DisableCloudLogging() {
	w.cloudLoggingDisabled = true
}
func (w *Workflow) DisableGCSLogging() {
	w.gcsLoggingDisabled = true
}
func (w *Workflow) DisableStdoutLogging() {
	w.stdoutLoggingDisabled = true
}
func (w *Workflow) AddVar(k, v string) {
	if w.Vars == nil {
		w.Vars = map[string]Var{}
	}
	w.Vars[k] = Var{Value: v}
}
func (w *Workflow) AddSerialConsoleOutputValue(k, v string) {
	w.serialControlOutputValuesMx.Lock()
	if w.serialControlOutputValues == nil {
		w.serialControlOutputValues = map[string]string{}
	}
	w.serialControlOutputValues[k] = v
	w.serialControlOutputValuesMx.Unlock()
}
func (w *Workflow) GetSerialConsoleOutputValue(k string) string {
	return w.serialControlOutputValues[k]
}
func (w *Workflow) addCleanupHook(hook func() DError) {
	w.cleanupHooksMx.Lock()
	w.cleanupHooks = append(w.cleanupHooks, hook)
	w.cleanupHooksMx.Unlock()
}
func (w *Workflow) SetLogProcessHook(hook func(string) string) {
	w.logProcessHook = hook
}
func (w *Workflow) Validate(ctx context.Context) DError {
	if err := w.PopulateClients(ctx); err != nil {
		w.CancelWorkflow()
		return Errf("error populating workflow: %v", err)
	}
	if err := w.validateRequiredFields(); err != nil {
		w.CancelWorkflow()
		return Errf("error validating workflow: %v", err)
	}
	if err := w.populate(ctx); err != nil {
		w.CancelWorkflow()
		return Errf("error populating workflow: %v", err)
	}
	w.LogWorkflowInfo("Validating workflow")
	if err := w.validate(ctx); err != nil {
		w.LogWorkflowInfo("Error validating workflow: %v", err)
		w.CancelWorkflow()
		return err
	}
	w.LogWorkflowInfo("Validation Complete")
	return nil
}
type WorkflowModifier func(*Workflow)
func (w *Workflow) Run(ctx context.Context) (err DError) {
	w.externalLogging = true
	if err = w.Validate(ctx); err != nil {
		return err
	}
	defer w.cleanup()
	defer func() {
		if err != nil {
			w.forceCleanup = w.ForceCleanupOnError
		}
	}()
	if os.Getenv("BUILD_ID") != "" {
		w.LogWorkflowInfo("Cloud Build ID: %s", os.Getenv("BUILD_ID"))
	}
	w.LogWorkflowInfo("Workflow Project: %s", w.Project)
	w.LogWorkflowInfo("Workflow Zone: %s", w.Zone)
	w.LogWorkflowInfo("Workflow GCSPath: %s", w.GCSPath)
	w.LogWorkflowInfo("Daisy scratch path: https://console.cloud.google.com/storage/browser/%s", path.Join(w.bucket, w.scratchPath))
	w.LogWorkflowInfo("Uploading sources")
	if err = w.uploadSources(ctx); err != nil {
		w.LogWorkflowInfo("Error uploading sources: %v", err)
		w.CancelWorkflow()
		return err
	}
	w.LogWorkflowInfo("Running workflow")
	defer func() {
		for k, v := range w.serialControlOutputValues {
			w.LogWorkflowInfo("Serial-output value -> %v:%v", k, v)
		}
	}()
	if err = w.run(ctx); err != nil {
		w.LogWorkflowInfo("Error running workflow: %v", err)
		return err
	}
	return nil
}
func (w *Workflow) recordStepTime(stepName string, startTime time.Time, endTime time.Time) {
	if w.parent == nil {
		w.recordTimeMx.Lock()
		w.stepTimeRecords = append(w.stepTimeRecords, TimeRecord{stepName, startTime, endTime})
		w.recordTimeMx.Unlock()
	} else {
		w.parent.recordStepTime(fmt.Sprintf("%s.%s", w.Name, stepName), startTime, endTime)
	}
}
func (w *Workflow) GetStepTimeRecords() []TimeRecord {
	return w.stepTimeRecords
}
func (w *Workflow) cleanup() {
	startTime := time.Now()
	w.LogWorkflowInfo("Workflow %q cleaning up (this may take up to 2 minutes).", w.Name)
	select {
	case <-w.Cancel:
	default:
		w.CancelWorkflow()
	}
	c := make(chan struct{})
	go func() {
		w.stepWait.Wait()
		close(c)
	}()
	select {
	case <-c:
	case <-time.After(4 * time.Second):
	}
	for _, hook := range w.cleanupHooks {
		if err := hook(); err != nil {
			w.LogWorkflowInfo("Error returned from cleanup hook: %s", err)
		}
	}
	w.LogWorkflowInfo("Workflow %q finished cleanup.", w.Name)
	w.recordStepTime("workflow cleanup", startTime, time.Now())
}
func (w *Workflow) genName(n string) string {
	name := w.Name
	for parent := w.parent; parent != nil; parent = parent.parent {
		name = parent.Name + "-" + name
	}
	prefix := name
	if n != "" {
		prefix = fmt.Sprintf("%s-%s", n, name)
	}
	if len(prefix) > 57 {
		prefix = prefix[0:56]
	}
	result := fmt.Sprintf("%s-%s", prefix, w.id)
	if len(result) > 64 {
		result = result[0:63]
	}
	return strings.ToLower(result)
}
func (w *Workflow) getSourceGCSAPIPath(s string) string {
	return fmt.Sprintf("%s/%s", gcsAPIBase, path.Join(w.bucket, w.sourcesPath, s))
}
func (w *Workflow) PopulateClients(ctx context.Context, options ...option.ClientOption) error {
	var (
		err            error
		computeOptions []option.ClientOption
		storageOptions []option.ClientOption
		loggingOptions []option.ClientOption
	)
	if len(options) > 0 {
		computeOptions = options
		storageOptions = options
		loggingOptions = options
	} else {
		computeOptions = []option.ClientOption{option.WithCredentialsFile(w.OAuthPath)}
		storageOptions = []option.ClientOption{option.WithCredentialsFile(w.OAuthPath)}
		loggingOptions = []option.ClientOption{option.WithCredentialsFile(w.OAuthPath)}
	}
	if w.ComputeEndpoint != "" {
		computeOptions = append(computeOptions, option.WithEndpoint(w.ComputeEndpoint))
	}
	if w.ComputeClient == nil {
		w.ComputeClient, err = compute.NewClient(ctx, computeOptions...)
		if err != nil {
			return typedErr(apiError, "failed to create compute client", err)
		}
	}
	if w.StorageClient == nil {
		w.StorageClient, err = storage.NewClient(ctx, storageOptions...)
		if err != nil {
			return err
		}
	}
	if w.externalLogging && !w.cloudLoggingDisabled && w.cloudLoggingClient == nil {
		w.cloudLoggingClient, err = logging.NewClient(ctx, w.Project, loggingOptions...)
		if err != nil {
			return err
		}
	}
	return nil
}
func (w *Workflow) populateStep(ctx context.Context, s *Step) DError {
	if s.Timeout == "" {
		s.Timeout = w.DefaultTimeout
	}
	timeout, err := time.ParseDuration(s.Timeout)
	if err != nil {
		return newErr(fmt.Sprintf("failed to parse duration for workflow %v, step %v", w.Name, s.name), err)
	}
	s.timeout = timeout
	var derr DError
	var step stepImpl
	if step, derr = s.stepImpl(); derr != nil {
		return derr
	}
	return step.populate(ctx, s)
}
func (w *Workflow) populate(ctx context.Context) DError {
	for k, v := range w.Vars {
		if v.Required && v.Value == "" {
			return Errf("cannot populate workflow, required var %q is unset", k)
		}
	}
	cwd, _ := os.Getwd()
	now := time.Now().UTC()
	w.username = getUser()
	w.autovars = map[string]string{
		"ID":        w.id,
		"DATE":      now.Format("20060102"),
		"DATETIME":  now.Format("20060102150405"),
		"TIMESTAMP": strconv.FormatInt(now.Unix(), 10),
		"USERNAME":  w.username,
		"WFDIR":     w.workflowDir,
		"CWD":       cwd,
	}
	var replacements []string
	for k, v := range w.autovars {
		replacements = append(replacements, fmt.Sprintf("${%s}", k), v)
	}
	for k, v := range w.Vars {
		replacements = append(replacements, fmt.Sprintf("${%s}", k), v.Value)
	}
	substitute(reflect.ValueOf(w).Elem(), strings.NewReplacer(replacements...))
	timeout, err := time.ParseDuration(w.DefaultTimeout)
	if err != nil {
		return Errf("failed to parse timeout for workflow: %v", err)
	}
	w.defaultTimeout = timeout
	if w.GCSPath == "" {
		dBkt, err := daisyBkt(ctx, w.StorageClient, w.Project)
		if err != nil {
			return err
		}
		w.GCSPath = "gs://" + dBkt
	}
	bkt, p, derr := splitGCSPath(w.GCSPath)
	if derr != nil {
		return derr
	}
	w.bucket = bkt
	w.scratchPath = path.Join(p, fmt.Sprintf("daisy-%s-%s-%s", w.Name, now.Format("20060102-15:04:05"), w.id))
	w.sourcesPath = path.Join(w.scratchPath, "sources")
	w.logsPath = path.Join(w.scratchPath, "logs")
	w.outsPath = path.Join(w.scratchPath, "outs")
	w.autovars["NAME"] = w.Name
	w.autovars["FULLNAME"] = w.genName("")
	w.autovars["ZONE"] = w.Zone
	w.autovars["PROJECT"] = w.Project
	w.autovars["GCSPATH"] = w.GCSPath
	w.autovars["SCRATCHPATH"] = fmt.Sprintf("gs://%s/%s", w.bucket, w.scratchPath)
	w.autovars["SOURCESPATH"] = fmt.Sprintf("gs://%s/%s", w.bucket, w.sourcesPath)
	w.autovars["LOGSPATH"] = fmt.Sprintf("gs://%s/%s", w.bucket, w.logsPath)
	w.autovars["OUTSPATH"] = fmt.Sprintf("gs://%s/%s", w.bucket, w.outsPath)
	replacements = []string{}
	for k, v := range w.autovars {
		replacements = append(replacements, fmt.Sprintf("${%s}", k), v)
	}
	substitute(reflect.ValueOf(w).Elem(), strings.NewReplacer(replacements...))
	if w.Logger == nil {
		w.createLogger(ctx)
	}
	for name, s := range w.Steps {
		s.name = name
		s.w = w
		if err := w.populateStep(ctx, s); err != nil {
			return Errf("error populating step %q: %v", name, err)
		}
	}
	if err := w.validateVarsSubbed(); err != nil {
		return err
	}
	if err := w.substituteSourceVars(ctx, reflect.ValueOf(w).Elem()); err != nil {
		return err
	}
	return nil
}
func (w *Workflow) AddDependency(dependent *Step, dependencies ...*Step) error {
	if _, ok := w.Steps[dependent.name]; !ok {
		return fmt.Errorf("can't create dependency: step %q does not exist", dependent.name)
	}
	if w.Dependencies == nil {
		w.Dependencies = map[string][]string{}
	}
	for _, dependency := range dependencies {
		if _, ok := w.Steps[dependency.name]; !ok {
			return fmt.Errorf("can't create dependency: step %q does not exist", dependency.name)
		}
		if !strIn(dependency.name, w.Dependencies[dependent.name]) {  
			w.Dependencies[dependent.name] = append(w.Dependencies[dependent.name], dependency.name)
		}
	}
	return nil
}
func (w *Workflow) includeWorkflow(iw *Workflow) {
	iw.Cancel = w.Cancel
	iw.parent = w
	iw.disks = w.disks
	iw.forwardingRules = w.forwardingRules
	iw.firewallRules = w.firewallRules
	iw.images = w.images
	iw.machineImages = w.machineImages
	iw.instances = w.instances
	iw.networks = w.networks
	iw.subnetworks = w.subnetworks
	iw.targetInstances = w.targetInstances
	iw.snapshots = w.snapshots
	iw.objects = w.objects
}
func (w *Workflow) ID() string {
	return w.id
}
func (w *Workflow) NewIncludedWorkflowFromFile(file string) (*Workflow, DError) {
	iw := New()
	w.includeWorkflow(iw)
	if !filepath.IsAbs(file) {
		file = filepath.Join(w.workflowDir, file)
	}
	if err := readWorkflow(file, iw); err != nil {
		return nil, err
	}
	return iw, nil
}
func (w *Workflow) NewStep(name string) (*Step, error) {
	if _, ok := w.Steps[name]; ok {
		return nil, fmt.Errorf("can't create step %q: a step already exists with that name", name)
	}
	s := &Step{name: name, w: w}
	if w.Steps == nil {
		w.Steps = map[string]*Step{}
	}
	w.Steps[name] = s
	return s, nil
}
func (w *Workflow) NewSubWorkflow() *Workflow {
	sw := New()
	sw.Cancel = w.Cancel
	sw.parent = w
	return sw
}
func (w *Workflow) NewSubWorkflowFromFile(file string) (*Workflow, DError) {
	sw := w.NewSubWorkflow()
	if !filepath.IsAbs(file) {
		file = filepath.Join(w.workflowDir, file)
	}
	if err := readWorkflow(file, sw); err != nil {
		return nil, err
	}
	return sw, nil
}
func (w *Workflow) Print(ctx context.Context) {
	w.externalLogging = false
	if err := w.PopulateClients(ctx); err != nil {
		fmt.Println("Error running PopulateClients:", err)
	}
	if err := w.populate(ctx); err != nil {
		fmt.Println("Error running populate:", err)
	}
	b, err := json.MarshalIndent(w, "", "  ")
	if err != nil {
		fmt.Println("Error marshalling workflow for printing:", err)
	}
	fmt.Println(string(b))
}
func (w *Workflow) run(ctx context.Context) DError {
	return w.traverseDAG(func(s *Step) DError {
		return w.runStep(ctx, s)
	})
}
func (w *Workflow) runStep(ctx context.Context, s *Step) DError {
	timeout := make(chan struct{})
	go func() {
		time.Sleep(s.timeout)
		close(timeout)
	}()
	e := make(chan DError)
	go func() {
		e <- s.run(ctx)
	}()
	select {
	case err := <-e:
		return err
	case <-timeout:
		return s.getTimeoutError()
	}
}
func (w *Workflow) traverseDAG(f func(*Step) DError) DError {
	waiting := map[string][]string{}
	var running []string
	start := map[string]chan DError{}
	done := map[string]chan DError{}
	for name := range w.Steps {
		waiting[name] = w.Dependencies[name]
		start[name] = make(chan DError)
		done[name] = make(chan DError)
	}
	for name, s := range w.Steps {
		go func(name string, s *Step) {
			if err := <-start[name]; err != nil {
				done[name] <- err
			} else if err := f(s); err != nil {
				done[name] <- err
			}
			close(done[name])
		}(name, s)
	}
	for len(waiting) != 0 || len(running) != 0 {
		select {
		case <-w.Cancel:
			waiting = map[string][]string{}
		default:
		}
		for name, deps := range waiting {
			if len(deps) == 0 {
				delete(waiting, name)
				running = append(running, name)
				close(start[name])
			}
		}
		if len(running) == 0 {
			continue
		}
		finished, err := stepsListen(running, done)
		if err != nil {
			return err
		}
		for name, deps := range waiting {
			waiting[name] = filter(deps, finished)
		}
		running = filter(running, finished)
	}
	return nil
}
func New() *Workflow {
	w := &Workflow{Cancel: make(chan struct{})}
	w.Sources = map[string]string{}
	w.Vars = map[string]Var{}
	w.Steps = map[string]*Step{}
	w.Dependencies = map[string][]string{}
	w.DefaultTimeout = defaultTimeout
	w.autovars = map[string]string{}
	w.disks = newDiskRegistry(w)
	w.forwardingRules = newForwardingRuleRegistry(w)
	w.firewallRules = newFirewallRuleRegistry(w)
	w.images = newImageRegistry(w)
	w.machineImages = newMachineImageRegistry(w)
	w.instances = newInstanceRegistry(w)
	w.networks = newNetworkRegistry(w)
	w.subnetworks = newSubnetworkRegistry(w)
	w.objects = newObjectRegistry(w)
	w.targetInstances = newTargetInstanceRegistry(w)
	w.snapshots = newSnapshotRegistry(w)
	w.addCleanupHook(func() DError {
		w.instances.cleanup()  
		w.images.cleanup()
		w.machineImages.cleanup()
		w.disks.cleanup()
		w.forwardingRules.cleanup()
		w.targetInstances.cleanup()
		w.firewallRules.cleanup()
		w.subnetworks.cleanup()
		w.networks.cleanup()
		w.snapshots.cleanup()
		return nil
	})
	w.id = randString(5)
	return w
}
func NewFromFile(file string) (w *Workflow, err error) {
	w = New()
	if err := readWorkflow(file, w); err != nil {
		return nil, err
	}
	return w, nil
}
func JSONError(file string, data []byte, err error) error {
	sErr, ok := err.(*json.SyntaxError)
	if !ok {
		return err
	}
	start := bytes.LastIndex(data[:sErr.Offset], []byte("\n")) + 1
	end := len(data)
	if i := bytes.Index(data[start:], []byte("\n")); i >= 0 {
		end = start + i
	}
	line := bytes.Count(data[:start], []byte("\n")) + 1
	pos := int(sErr.Offset) - start
	if pos != 0 {
		pos = pos - 1
	}
	return fmt.Errorf("%s: JSON syntax error in line %d: %s \n%s\n%s^", file, line, err, data[start:end], strings.Repeat(" ", pos))
}
func readWorkflow(file string, w *Workflow) (derr DError) {
	data, err := ioutil.ReadFile(file)
	if err != nil {
		return newErr("failed to read workflow file", err)
	}
	w.workflowDir, err = filepath.Abs(filepath.Dir(file))
	if err != nil {
		return newErr("failed to get absolute path of workflow file", err)
	}
	if err := json.Unmarshal(data, &w); err != nil {
		return newErr("failed to unmarshal workflow file", JSONError(file, data, err))
	}
	if w.OAuthPath != "" && !filepath.IsAbs(w.OAuthPath) {
		w.OAuthPath = filepath.Join(w.workflowDir, w.OAuthPath)
	}
	for name, step := range w.Steps {
		step.name = name
		step.w = w
		if step.SubWorkflow != nil &&
			step.SubWorkflow.Path != "" &&
			!hasVariableDeclaration(step.SubWorkflow.Path) {
			step.SubWorkflow.Workflow, derr = w.NewSubWorkflowFromFile(step.SubWorkflow.Path)
		} else if step.IncludeWorkflow != nil &&
			step.IncludeWorkflow.Path != "" &&
			!hasVariableDeclaration(step.IncludeWorkflow.Path) {
			step.IncludeWorkflow.Workflow, derr = w.NewIncludedWorkflowFromFile(step.IncludeWorkflow.Path)
		} else {
			continue
		}
		if derr != nil {
			return derr
		}
	}
	return nil
}
func stepsListen(names []string, chans map[string]chan DError) (string, DError) {
	cases := make([]reflect.SelectCase, len(names))
	for i, name := range names {
		cases[i] = reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(chans[name])}
	}
	caseIndex, value, recvOk := reflect.Select(cases)
	name := names[caseIndex]
	if recvOk {
		return name, value.Interface().(DError)
	}
	return name, nil
}