func (w *Workflow) ID() string {
	return w.id
}
ID is the unique identifyier for this Workflow.
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
}