// RunWorkflowWithConfiguration ...
767036
func runWorkflowWithConfiguration(
	startTime time.Time,
	workflowToRunID string,
	bitriseConfig models.BitriseDataModel,
	secretEnvironments []envmanModels.EnvironmentItemModel) (models.BuildRunResultsModel, error) {

	workflowToRun, exist := bitriseConfig.Workflows[workflowToRunID]
	if !exist {
		return models.BuildRunResultsModel{}, fmt.Errorf("Specified Workflow (%s) does not exist", workflowToRunID)
	}

	if workflowToRun.Title == "" {
		workflowToRun.Title = workflowToRunID
	}

	// Envman setup
	if err := os.Setenv(configs.EnvstorePathEnvKey, configs.OutputEnvstorePath); err != nil {
		return models.BuildRunResultsModel{}, fmt.Errorf("Failed to add env, err: %s", err)
	}

	if err := os.Setenv(configs.FormattedOutputPathEnvKey, configs.FormattedOutputPath); err != nil {
		return models.BuildRunResultsModel{}, fmt.Errorf("Failed to add env, err: %s", err)
	}

	if err := tools.EnvmanInit(); err != nil {
		return models.BuildRunResultsModel{}, errors.New("Failed to run envman init")
	}

	// App level environment
	environments := append(secretEnvironments, bitriseConfig.App.Environments...)

	if err := os.Setenv("BITRISE_TRIGGERED_WORKFLOW_ID", workflowToRunID); err != nil {
		return models.BuildRunResultsModel{}, fmt.Errorf("Failed to set BITRISE_TRIGGERED_WORKFLOW_ID env: %s", err)
	}
	if err := os.Setenv("BITRISE_TRIGGERED_WORKFLOW_TITLE", workflowToRun.Title); err != nil {
		return models.BuildRunResultsModel{}, fmt.Errorf("Failed to set BITRISE_TRIGGERED_WORKFLOW_TITLE env: %s", err)
	}

	environments = append(environments, workflowToRun.Environments...)

	lastWorkflowID, err := lastWorkflowIDInConfig(workflowToRunID, bitriseConfig)
	if err != nil {
		return models.BuildRunResultsModel{}, fmt.Errorf("Failed to get last workflow id: %s", err)
	}

	// Bootstrap Toolkits
	for _, aToolkit := range toolkits.AllSupportedToolkits() {
		toolkitName := aToolkit.ToolkitName()
		if !aToolkit.IsToolAvailableInPATH() {
			// don't bootstrap if any preinstalled version is available,
			// the toolkit's `PrepareForStepRun` can bootstrap for itself later if required
			// or if the system installed version is not sufficient
			if err := aToolkit.Bootstrap(); err != nil {
				return models.BuildRunResultsModel{}, fmt.Errorf("Failed to bootstrap the required toolkit for the step (%s), error: %s",
					toolkitName, err)
			}
		}
	}

	//
	buildRunResults := models.BuildRunResultsModel{
		StartTime:      startTime,
		StepmanUpdates: map[string]int{},
		ProjectType:    bitriseConfig.ProjectType,
	}

	buildRunResults, err = activateAndRunWorkflow(
		workflowToRunID, workflowToRun, bitriseConfig,
		buildRunResults,
		&environments, secretEnvironments,
		lastWorkflowID)
	if err != nil {
		return buildRunResults, errors.New("[BITRISE_CLI] - Failed to activate and run workflow " + workflowToRunID)
	}

	// Build finished
	bitrise.PrintSummary(buildRunResults)

	// Trigger WorkflowRunDidFinish
	if err := plugins.TriggerEvent(plugins.DidFinishRun, buildRunResults); err != nil {
		log.Warnf("Failed to trigger WorkflowRunDidFinish, error: %s", err)
	}

	return buildRunResults, nil
}
714612
func (s *ContinueAsNewWorkflowExecutionDecisionAttributes) SetWorkflowTypeVersion(v string) *ContinueAsNewWorkflowExecutionDecisionAttributes {
	s.WorkflowTypeVersion = &v
	return s
}
970978
func (s *WorkflowExecutionInfo) SetWorkflowType(v *WorkflowType) *WorkflowExecutionInfo {
	s.WorkflowType = v
	return s
}
1083081
func (fcd *FlowDriver) RunFlowContext(parentCtx context.Context, fc *flow.Flow) {

	// task fusion to minimize disk IO
	fcd.stepGroups, fcd.taskGroups = plan.GroupTasks(fc)
	fcd.logExecutionPlan(fc)

	// create the scheduler
	sched := scheduler.New(
		fcd.Option.Master,
		&scheduler.Option{
			DataCenter:   fcd.Option.DataCenter,
			Rack:         fcd.Option.Rack,
			TaskMemoryMB: fcd.Option.TaskMemoryMB,
			Module:       fcd.Option.Module,
			FlowHashcode: fc.HashCode,
			IsProfiling:  fcd.Option.IsProfiling,
		},
	)

	// best effort to clean data on agent disk
	// this may need more improvements
	defer fcd.cleanup(sched, fc)

	ctx, cancel := context.WithCancel(parentCtx)

	on_interrupt.OnInterrupt(func() {
		println("interrupted ...")
		fcd.printDistributedStatus(os.Stderr)
		cancel()
		fcd.cleanup(sched, fc)
	}, nil)

	// schedule to run the steps
	var wg, reportWg sync.WaitGroup
	for _, taskGroup := range fcd.taskGroups {
		wg.Add(1)
		go func(taskGroup *plan.TaskGroup) {
			sched.ExecuteTaskGroup(ctx, fc, fcd.GetTaskGroupStatus(taskGroup), &wg, taskGroup,
				fcd.Option.FlowBid/float64(len(fcd.taskGroups)), fcd.Option.RequiredFiles)
		}(taskGroup)
	}
	go sched.Market.FetcherLoop()

	stopChan := make(chan bool)
	reportWg.Add(1)
	go fcd.reportStatus(ctx, &reportWg, fcd.Option.Master, stopChan)

	log.Printf("Start Job Status URL http://%s/job/%d", fcd.Option.Master, fcd.status.GetId())

	wg.Wait()

	stopChan <- true
	reportWg.Wait()

}
485102
func (schemaSwap *Swap) Run(ctx context.Context, manager *workflow.Manager, workflowInfo *topo.WorkflowInfo) error {
	schemaSwap.ctx = ctx
	schemaSwap.workflowManager = manager
	schemaSwap.topoServer = manager.TopoServer()
	schemaSwap.tabletClient = tmclient.NewTabletManagerClient()

	log.Infof("Starting schema swap on keyspace %v with the following SQL: %v", schemaSwap.keyspace, schemaSwap.sql)

	for {
		err := schemaSwap.executeSwap()
		if err == nil {
			return nil
		}
		// If context is cancelled then return right away, otherwise move on to allow
		// user to retry.
		select {
		case <-schemaSwap.ctx.Done():
			schemaSwap.setUIMessage(fmt.Sprintf("Error: %v", err))
			return err
		default:
		}

		schemaSwap.retryMutex.Lock()
		retryAction := &workflow.Action{
			Name:  "Retry",
			State: workflow.ActionStateEnabled,
			Style: workflow.ActionStyleWaiting,
		}
		schemaSwap.rootUINode.Actions = []*workflow.Action{retryAction}
		schemaSwap.rootUINode.Listener = schemaSwap
		schemaSwap.retryChannel = make(chan struct{})
		schemaSwap.retryMutex.Unlock()
		// setUIMessage broadcasts changes to rootUINode.
		schemaSwap.setUIMessage(fmt.Sprintf("Error: %v", err))

		select {
		case <-schemaSwap.retryChannel:
			schemaSwap.setUIMessage("Retrying schema swap after an error")
			continue
		case <-schemaSwap.ctx.Done():
			schemaSwap.closeRetryChannel()
			return schemaSwap.ctx.Err()
		}
	}
}
1037580
func SuspendWorkflow(wfIf v1alpha1.WorkflowInterface, workflowName string) error {
	err := wait.ExponentialBackoff(retry.DefaultRetry, func() (bool, error) {
		wf, err := wfIf.Get(workflowName, metav1.GetOptions{})
		if err != nil {
			return false, err
		}
		if IsWorkflowCompleted(wf) {
			return false, errSuspendedCompletedWorkflow
		}
		if wf.Spec.Suspend == nil || *wf.Spec.Suspend != true {
			wf.Spec.Suspend = pointer.BoolPtr(true)
			wf, err = wfIf.Update(wf)
			if err != nil {
				if apierr.IsConflict(err) {
					return false, nil
				}
				return false, err
			}
		}
		return true, nil
	})
	if err != nil {
		return err
	}
	return nil
}
1352140
func (c *Client) DescribeWorkflowExecution(request DescribeWorkflowExecutionRequest) (*DescribeWorkflowExecutionResponse, error) {
	resp := &DescribeWorkflowExecutionResponse{}
	err := c.swfReqWithResponse("DescribeWorkflowExecution", request, resp)
	return resp, err
}
1262164
func (s *ChildWorkflowExecutionFailedEventAttributes) SetWorkflowExecution(v *WorkflowExecution) *ChildWorkflowExecutionFailedEventAttributes {
	s.WorkflowExecution = v
	return s
}
1315338
func (s *AwsSecurityFindingFilters) SetWorkflowState(v []*StringFilter) *AwsSecurityFindingFilters {
	s.WorkflowState = v
	return s
}
778106
func (v *version) Workflows() WorkflowInformer {
	return &workflowInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
}
488496
func CompleteWorkflow() Decider {
	return func(ctx *FSMContext, h *swf.HistoryEvent, data interface{}) Outcome {
		Log.Printf("at=complete-workflow workflowId=%s", LS(ctx.WorkflowId))
		return ctx.CompleteWorkflow(data)
	}
}
1007069
func (f *FSMContext) FailWorkflow(data interface{}, details *string) Outcome {
	d := &swf.Decision{
		DecisionType: S(swf.DecisionTypeFailWorkflowExecution),
		FailWorkflowExecutionDecisionAttributes: &swf.FailWorkflowExecutionDecisionAttributes{
			Details: details,
		},
	}
	return Outcome{
		State:     FailedState,
		Data:      data,
		Decisions: f.Decision(d),
	}
}
1316521
func (a *Client) PostWorkflows(params *PostWorkflowsParams, authInfo runtime.ClientAuthInfoWriter) (*PostWorkflowsOK, error) {
	// TODO: Validate the params before sending
	if params == nil {
		params = NewPostWorkflowsParams()
	}

	result, err := a.transport.Submit(&runtime.ClientOperation{
		ID:                 "PostWorkflows",
		Method:             "POST",
		PathPattern:        "/workflows",
		ProducesMediaTypes: []string{"application/json", "application/x-gzip"},
		ConsumesMediaTypes: []string{"application/json"},
		Schemes:            []string{"http", "https"},
		Params:             params,
		Reader:             &PostWorkflowsReader{formats: a.formats},
		AuthInfo:           authInfo,
	})
	if err != nil {
		return nil, err
	}
	return result.(*PostWorkflowsOK), nil
}
933251
func (f *FSMContext) ContinueWorkflow(data interface{}, decisions ...*swf.Decision) Outcome {
	if len(decisions) == 0 || *decisions[len(decisions)-1].DecisionType != swf.DecisionTypeContinueAsNewWorkflowExecution {
		decisions = append(decisions, f.ContinueWorkflowDecision(f.State, data))
	}
	return Outcome{
		State:     CompleteState,
		Data:      data,
		Decisions: decisions,
	}
}
140598
func InitWorkflowClient(ns ...string) v1alpha1.WorkflowInterface {
	if wfClient != nil {
		return wfClient
	}
	initKubeClient()
	var namespace string
	var err error
	if len(ns) > 0 {
		namespace = ns[0]
	} else {
		namespace, _, err = clientConfig.Namespace()
		if err != nil {
			log.Fatal(err)
		}
	}
	wfcs := wfclientset.NewForConfigOrDie(restConfig)
	wfClient = wfcs.ArgoprojV1alpha1().Workflows(namespace)
	return wfClient
}
512310
func (s *ChildWorkflowExecutionTimedOutEventAttributes) SetWorkflowType(v *WorkflowType) *ChildWorkflowExecutionTimedOutEventAttributes {
	s.WorkflowType = v
	return s
}
1084401
func (cmd *Command) RunWith(args []string, writer io.Writer, resp http.ResponseWriter, httpMethods ...string) error {
	fds := []uintptr{}
	if writer == nil {
		writer = colorable.NewColorableStdout()
		fds = append(fds, os.Stdout.Fd())
	}
	clr := color.Color{}
	colorSwitch(&clr, writer, fds...)

	var ctx *Context
	var suggestion string
	ctx, suggestion, err := cmd.prepare(clr, args, writer, resp, httpMethods...)
	if err == ExitError {
		return nil
	}

	if err != nil {
		if cmd.OnRootPrepareError != nil {
			err = cmd.OnRootPrepareError(err)
		}
		if err != nil {
			return wrapErr(err, suggestion, clr)
		}
		return nil
	}

	if ctx.command.NoHook {
		return ctx.command.Fn(ctx)
	}

	funcs := []func(*Context) error{
		ctx.command.OnBefore,
		cmd.OnRootBefore,
		ctx.command.Fn,
		cmd.OnRootAfter,
		ctx.command.OnAfter,
	}
	for _, f := range funcs {
		if f != nil {
			if err := f(ctx); err != nil {
				if err == ExitError {
					return nil
				}
				return err
			}
		}
	}
	return nil
}