// LintWorkflowFile lints a json file, or multiple workflow manifest in a single yaml file. Ignores
// non-workflow manifests
175806
func LintWorkflowFile(filePath string, strict bool) error {
	body, err := ioutil.ReadFile(filePath)
	if err != nil {
		return errors.Errorf(errors.CodeBadRequest, "Can't read from file: %s, err: %v", filePath, err)
	}
	var workflows []wfv1.Workflow
	if json.IsJSON(body) {
		var wf wfv1.Workflow
		if strict {
			err = json.UnmarshalStrict(body, &wf)
		} else {
			err = json.Unmarshal(body, &wf)
		}
		if err == nil {
			workflows = []wfv1.Workflow{wf}
		} else {
			if wf.Kind != "" && wf.Kind != workflow.Kind {
				// If we get here, it was a k8s manifest which was not of type 'Workflow'
				// We ignore these since we only care about validating Workflow manifests.
				return nil
			}
		}
	} else {
		workflows, err = common.SplitYAMLFile(body, strict)
	}
	if err != nil {
		return errors.Errorf(errors.CodeBadRequest, "%s failed to parse: %v", filePath, err)
	}
	for _, wf := range workflows {
		err = ValidateWorkflow(&wf, ValidateOpts{Lint: true})
		if err != nil {
			return errors.Errorf(errors.CodeBadRequest, "%s: %s", filePath, err.Error())
		}
	}
	return nil
}
829058
func SubmitWorkflow(wfIf v1alpha1.WorkflowInterface, wf *wfv1.Workflow, opts *SubmitOpts) (*wfv1.Workflow, error) {
	if opts == nil {
		opts = &SubmitOpts{}
	}
	if opts.Entrypoint != "" {
		wf.Spec.Entrypoint = opts.Entrypoint
	}
	if opts.ServiceAccount != "" {
		wf.Spec.ServiceAccountName = opts.ServiceAccount
	}
	if opts.InstanceID != "" {
		labels := wf.GetLabels()
		if labels == nil {
			labels = make(map[string]string)
		}
		labels[common.LabelKeyControllerInstanceID] = opts.InstanceID
		wf.SetLabels(labels)
	}
	if len(opts.Parameters) > 0 || opts.ParameterFile != "" {
		newParams := make([]wfv1.Parameter, 0)
		passedParams := make(map[string]bool)
		for _, paramStr := range opts.Parameters {
			parts := strings.SplitN(paramStr, "=", 2)
			if len(parts) == 1 {
				return nil, fmt.Errorf("Expected parameter of the form: NAME=VALUE. Received: %s", paramStr)
			}
			param := wfv1.Parameter{
				Name:  parts[0],
				Value: &parts[1],
			}
			newParams = append(newParams, param)
			passedParams[param.Name] = true
		}

		// Add parameters from a parameter-file, if one was provided
		if opts.ParameterFile != "" {
			var body []byte
			var err error
			if cmdutil.IsURL(opts.ParameterFile) {
				response, err := http.Get(opts.ParameterFile)
				if err != nil {
					return nil, errors.InternalWrapError(err)
				}
				body, err = ioutil.ReadAll(response.Body)
				_ = response.Body.Close()
				if err != nil {
					return nil, errors.InternalWrapError(err)
				}
			} else {
				body, err = ioutil.ReadFile(opts.ParameterFile)
				if err != nil {
					return nil, errors.InternalWrapError(err)
				}
			}

			yamlParams := map[string]interface{}{}
			err = yaml.Unmarshal(body, &yamlParams)
			if err != nil {
				return nil, errors.InternalWrapError(err)
			}

			for k, v := range yamlParams {
				value := fmt.Sprintf("%v", v)
				param := wfv1.Parameter{
					Name:  k,
					Value: &value,
				}
				if _, ok := passedParams[param.Name]; ok {
					// this parameter was overridden via command line
					continue
				}
				newParams = append(newParams, param)
				passedParams[param.Name] = true
			}
		}

		for _, param := range wf.Spec.Arguments.Parameters {
			if _, ok := passedParams[param.Name]; ok {
				// this parameter was overridden via command line
				continue
			}
			newParams = append(newParams, param)
		}
		wf.Spec.Arguments.Parameters = newParams
	}
	if opts.GenerateName != "" {
		wf.ObjectMeta.GenerateName = opts.GenerateName
	}
	if opts.Name != "" {
		wf.ObjectMeta.Name = opts.Name
	}
	if opts.OwnerReference != nil {
		wf.SetOwnerReferences(append(wf.GetOwnerReferences(), *opts.OwnerReference))
	}

	err := validate.ValidateWorkflow(wf, validate.ValidateOpts{})
	if err != nil {
		return nil, err
	}
	return wfIf.Create(wf)
}
369880
func (l *Linter) LintFiles(files map[string][]byte) ([]Problem, error) {
	pkg := &pkg{
		fset:  token.NewFileSet(),
		files: make(map[string]*file),
	}
	var pkgName string
	for filename, src := range files {
		if isGenerated(src) {
			continue // See issue #239
		}
		f, err := parser.ParseFile(pkg.fset, filename, src, parser.ParseComments)
		if err != nil {
			return nil, err
		}
		if pkgName == "" {
			pkgName = f.Name.Name
		} else if f.Name.Name != pkgName {
			return nil, fmt.Errorf("%s is in package %s, not %s", filename, f.Name.Name, pkgName)
		}
		pkg.files[filename] = &file{
			pkg:      pkg,
			f:        f,
			fset:     pkg.fset,
			src:      src,
			filename: filename,
		}
	}
	if len(pkg.files) == 0 {
		return nil, nil
	}
	return pkg.lint(), nil
}
948435
func NewYamlSourceFromFile(file string) (InputSourceContext, error) {
	ysc := &yamlSourceContext{FilePath: file}
	var results map[interface{}]interface{}
	err := readCommandYaml(ysc.FilePath, &results)
	if err != nil {
		return nil, fmt.Errorf("Unable to load Yaml file '%s': inner error: \n'%v'", ysc.FilePath, err.Error())
	}

	return &MapInputSource{valueMap: results}, nil
}
723941
func (p *logPrinter) printLiveWorkflowLogs(workflowName string, wfClient workflowv1.WorkflowInterface, timeByPod map[string]*time.Time) {
	logs := make(chan logEntry)
	streamedPods := make(map[string]bool)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	processPods := func(wf *v1alpha1.Workflow) {
		err := util.DecompressWorkflow(wf)
		if err != nil {
			log.Warn(err)
			return
		}
		for id := range wf.Status.Nodes {
			node := wf.Status.Nodes[id]
			if node.Type == v1alpha1.NodeTypePod && node.Phase != v1alpha1.NodeError && streamedPods[node.ID] == false {
				streamedPods[node.ID] = true
				go func() {
					var sinceTimePtr *metav1.Time
					podTime := timeByPod[node.ID]
					if podTime != nil {
						sinceTime := metav1.NewTime(podTime.Add(time.Second))
						sinceTimePtr = &sinceTime
					}
					err := p.getPodLogs(ctx, getDisplayName(node), node.ID, wf.Namespace, true, nil, nil, sinceTimePtr, func(entry logEntry) {
						logs <- entry
					})
					if err != nil {
						log.Warn(err)
					}
				}()
			}
		}
	}

	go func() {
		defer close(logs)
		fieldSelector := fields.ParseSelectorOrDie(fmt.Sprintf("metadata.name=%s", workflowName))
		listOpts := metav1.ListOptions{FieldSelector: fieldSelector.String()}
		lw := &cache.ListWatch{
			ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
				return wfClient.List(listOpts)
			},
			WatchFunc: func(options metav1.ListOptions) (pkgwatch.Interface, error) {
				return wfClient.Watch(listOpts)
			},
		}
		_, err := watch.UntilWithSync(ctx, lw, &v1alpha1.Workflow{}, nil, func(event pkgwatch.Event) (b bool, e error) {
			if wf, ok := event.Object.(*v1alpha1.Workflow); ok {
				if !wf.Status.Completed() {
					processPods(wf)
				}
				return wf.Status.Completed(), nil
			}
			return true, nil
		})
		if err != nil {
			log.Fatal(err)
		}
	}()

	for entry := range logs {
		p.printLogEntry(entry)
	}
}
487256
func (we *WorkflowExecutor) saveLogToFile(mainCtrID, path string) error {
	outFile, err := os.Create(path)
	if err != nil {
		return errors.InternalWrapError(err)
	}
	defer func() { _ = outFile.Close() }()
	reader, err := we.RuntimeExecutor.GetOutputStream(mainCtrID, true)
	if err != nil {
		return err
	}
	defer func() { _ = reader.Close() }()
	_, err = io.Copy(outFile, reader)
	if err != nil {
		return errors.InternalWrapError(err)
	}
	return nil
}
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
}
92769
func isValidWorkflowFieldName(name string) []string {
	var errs []string
	if len(name) > workflowFieldMaxLength {
		errs = append(errs, apivalidation.MaxLenError(workflowFieldMaxLength))
	}
	if !workflowFieldNameRegexp.MatchString(name) {
		msg := workflowFieldNameErrMsg + " (e.g. My-name1-2, 123-NAME)"
		errs = append(errs, msg)
	}
	return errs
}
47984
func (a *Client) PutWorkflows(params *PutWorkflowsParams, authInfo runtime.ClientAuthInfoWriter) (*PutWorkflowsOK, error) {
	// TODO: Validate the params before sending
	if params == nil {
		params = NewPutWorkflowsParams()
	}

	result, err := a.transport.Submit(&runtime.ClientOperation{
		ID:                 "PutWorkflows",
		Method:             "PUT",
		PathPattern:        "/workflows",
		ProducesMediaTypes: []string{"application/json", "application/x-gzip"},
		ConsumesMediaTypes: []string{"application/json"},
		Schemes:            []string{"http", "https"},
		Params:             params,
		Reader:             &PutWorkflowsReader{formats: a.formats},
		AuthInfo:           authInfo,
	})
	if err != nil {
		return nil, err
	}
	return result.(*PutWorkflowsOK), nil
}
1192292
func (c *FakeWorkflows) Create(workflow *v1alpha1.Workflow) (result *v1alpha1.Workflow, err error) {
	obj, err := c.Fake.
		Invokes(testing.NewCreateAction(workflowsResource, c.ns, workflow), &v1alpha1.Workflow{})

	if obj == nil {
		return nil, err
	}
	return obj.(*v1alpha1.Workflow), err
}
140730
func DecompressWorkflow(wf *wfv1.Workflow) error {
	if wf.Status.CompressedNodes != "" {
		nodeContent, err := file.DecodeDecompressString(wf.Status.CompressedNodes)
		if err != nil {
			return errors.InternalWrapError(err)
		}
		err = json.Unmarshal([]byte(nodeContent), &wf.Status.Nodes)
		if err != nil {
			return err
		}
		wf.Status.CompressedNodes = ""
	}
	return nil
}
171049
func (a *Client) UploadTaskFile(params *UploadTaskFileParams, authInfo runtime.ClientAuthInfoWriter) (*UploadTaskFileCreated, error) {
	// TODO: Validate the params before sending
	if params == nil {
		params = NewUploadTaskFileParams()
	}

	result, err := a.transport.Submit(&runtime.ClientOperation{
		ID:                 "uploadTaskFile",
		Method:             "POST",
		PathPattern:        "/tasks/{id}/files",
		ProducesMediaTypes: []string{"application/vnd.goswagger.examples.task-tracker.v1+json"},
		ConsumesMediaTypes: []string{"multipart/form-data"},
		Schemes:            []string{"http", "https"},
		Params:             params,
		Reader:             &UploadTaskFileReader{formats: a.formats},
		AuthInfo:           authInfo,
		Context:            params.Context,
		Client:             params.HTTPClient,
	})
	if err != nil {
		return nil, err
	}
	return result.(*UploadTaskFileCreated), nil

}
104966
func (e *fileExtractor) procfile(ctx context.Context, id string) (string, error) {
	p := ""

	c, err := e.client.InspectContainer(id)
	if err != nil {
		return "", err
	}

	if c.Config != nil {
		p = c.Config.WorkingDir
	}

	return path.Join(p, empire.Procfile), nil
}
30938
func CloseFile(err error, f io.Closer) error {
	if lerr := f.Close(); lerr != nil && err == nil {
		err = lerr
	}
	return err
}
1176147
func LoadFile(config interface{}, file string, opts Options) error {
	return LoadFiles(config, []string{file}, opts)
}
851571
func (env *Glisp) ParseFile(file string) ([]Sexp, error) {
	in, err := os.Open(file)
	if err != nil {
		return nil, err
	}

	var exp []Sexp

	exp, err = env.ParseStream(in)

	in.Close()

	return exp, err
}
109289
func (h *LangHandler) lint(ctx context.Context, bctx *build.Context, conn jsonrpc2.JSONRPC2, args []string, files []string) error {
	if h.linter == nil {
		return nil
	}

	diags, err := h.linter.Lint(ctx, bctx, args...)
	if err != nil {
		return err
	}

	return h.publishDiagnostics(ctx, conn, diags, h.config.LintTool, files)
}