func (ep *endpoints) CreateRunV4(w http.ResponseWriter, r *http.Request) {
	var lr LaunchRequestV2
	err := ep.decodeRequest(r, &lr)
	if err != nil {
		ep.encodeError(w, exceptions.MalformedInput{ErrorString: err.Error()})
		return
	}
	if len(lr.RunTags.OwnerID) == 0 {
		ep.encodeError(w, exceptions.MalformedInput{
			ErrorString: fmt.Sprintf("run_tags must exist in body and contain [owner_id]")})
		return
	}
	if lr.Engine == nil || *lr.Engine == "ecs" {
		if lr.SparkExtension != nil {
			lr.Engine = &state.EKSSparkEngine
		} else {
			lr.Engine = &state.EKSEngine
		}
	}
	if lr.NodeLifecycle != nil {
		if !utils.StringSliceContains(state.NodeLifeCycles, *lr.NodeLifecycle) {
			ep.encodeError(w, exceptions.MalformedInput{
				ErrorString: fmt.Sprintf("Nodelifecyle must be [normal, spot]")})
			return
		}
	} else {
		lr.NodeLifecycle = &state.DefaultLifecycle
	}
	vars := mux.Vars(r)
	req := state.DefinitionExecutionRequest{
		ExecutionRequestCommon: &state.ExecutionRequestCommon{
			Env:                   lr.Env,
			OwnerID:               lr.RunTags.OwnerID,
			Command:               lr.Command,
			Memory:                lr.Memory,
			Cpu:                   lr.Cpu,
			Gpu:                   lr.Gpu,
			Engine:                lr.Engine,
			NodeLifecycle:         lr.NodeLifecycle,
			ActiveDeadlineSeconds: lr.ActiveDeadlineSeconds,
			SparkExtension:        lr.SparkExtension,
			Description:           lr.Description,
		},
	}
	run, err := ep.executionService.CreateDefinitionRunByDefinitionID(vars["definition_id"], &req)
	if err != nil {
		ep.logger.Log(
			"message", "problem creating V4 run",
			"operation", "CreateRunV4",
			"error", fmt.Sprintf("%+v", err))
		ep.encodeError(w, err)
	} else {
		ep.encodeResponse(w, run)
	}
}
Creates a new Run.
package flotilla
import (
	"encoding/json"
	"fmt"
	"github.com/gorilla/mux"
	"github.com/stitchfix/flotilla-os/exceptions"
	flotillaLog "github.com/stitchfix/flotilla-os/log"
	"github.com/stitchfix/flotilla-os/services"
	"github.com/stitchfix/flotilla-os/state"
	"github.com/stitchfix/flotilla-os/utils"
	"net/http"
	"net/url"
	"strconv"
	"strings"
)
type endpoints struct {
	executionService  services.ExecutionService
	definitionService services.DefinitionService
	templateService   services.TemplateService
	eksLogService     services.LogService
	workerService     services.WorkerService
	logger            flotillaLog.Logger
}
type listRequest struct {
	limit      int
	offset     int
	sortBy     string
	order      string
	filters    map[string][]string
	envFilters map[string]string
}
type LaunchRequest struct {
	ClusterName *string        `json:"cluster,omitempty"`
	Env         *state.EnvList `json:"env,omitempty"`
}
type LaunchRequestV2 struct {
	RunTags               RunTags `json:"run_tags"`
	Command               *string
	Memory                *int64
	Cpu                   *int64
	Gpu                   *int64
	Engine                *string
	NodeLifecycle         *string               `json:"node_lifecycle"`
	ActiveDeadlineSeconds *int64                `json:"active_deadline_seconds,omitempty"`
	SparkExtension        *state.SparkExtension `json:"spark_extension,omitempty"`
	ClusterName           *string               `json:"cluster,omitempty"`
	Env                   *state.EnvList        `json:"env,omitempty"`
	Description           *string               `json:"description,omitempty"`
}
type RunTags struct {
	OwnerEmail string `json:"owner_email"`
	TeamName   string `json:"team_name"`
	OwnerID    string `json:"owner_id"`
}
func (ep *endpoints) getURLParam(v url.Values, key string, defaultValue string) string {
	val, ok := v[key]
	if ok && len(val) > 0 {
		return val[0]
	}
	return defaultValue
}
func (ep *endpoints) getFilters(params url.Values, nonFilters map[string]bool) (map[string][]string, map[string]string) {
	filters := make(map[string][]string)
	envFilters := make(map[string]string)
	for k, v := range params {
		if !nonFilters[k] && len(v) > 0 {
			if k == "env" {
				for _, kv := range v {
					split := strings.Split(kv, "|")
					if len(split) == 2 {
						envFilters[split[0]] = split[1]
					}
				}
			} else {
				filters[k] = v
			}
		}
	}
	return filters, envFilters
}
func (ep *endpoints) decodeListRequest(r *http.Request) listRequest {
	var lr listRequest
	params := r.URL.Query()
	lr.limit, _ = strconv.Atoi(ep.getURLParam(params, "limit", "1024"))
	lr.offset, _ = strconv.Atoi(ep.getURLParam(params, "offset", "0"))
	lr.sortBy = ep.getURLParam(params, "sort_by", "group_name")
	lr.order = ep.getURLParam(params, "order", "asc")
	lr.filters, lr.envFilters = ep.getFilters(params, map[string]bool{
		"limit":   true,
		"offset":  true,
		"sort_by": true,
		"order":   true,
	})
	return lr
}
func (ep *endpoints) decodeOrderableListRequest(r *http.Request, orderable state.IOrderable) listRequest {
	var lr listRequest
	params := r.URL.Query()
	lr.limit, _ = strconv.Atoi(ep.getURLParam(params, "limit", "1024"))
	lr.offset, _ = strconv.Atoi(ep.getURLParam(params, "offset", "0"))
	lr.sortBy = ep.getURLParam(params, "sort_by", orderable.DefaultOrderField())
	lr.order = ep.getURLParam(params, "order", "asc")
	lr.filters, lr.envFilters = ep.getFilters(params, map[string]bool{
		"limit":   true,
		"offset":  true,
		"sort_by": true,
		"order":   true,
	})
	return lr
}
func (ep *endpoints) decodeRequest(r *http.Request, entity interface{}) error {
	return json.NewDecoder(r.Body).Decode(entity)
}
func (ep endpoints) encodeError(w http.ResponseWriter, err error) {
	w.Header().Set("Content-Type", "application/json; charset=utf-8")
	switch err.(type) {
	case exceptions.MalformedInput:
		w.WriteHeader(http.StatusBadRequest)
	case exceptions.ConflictingResource:
		w.WriteHeader(http.StatusConflict)
	case exceptions.MissingResource:
		w.WriteHeader(http.StatusNotFound)
	default:
		w.WriteHeader(http.StatusInternalServerError)
	}
	_ = json.NewEncoder(w).Encode(map[string]interface{}{
		"error": err.Error(),
	})
}
func (ep *endpoints) encodeResponse(w http.ResponseWriter, response interface{}) {
	w.Header().Set("Content-Type", "application/json; charset=utf-8")
	_ = json.NewEncoder(w).Encode(response)
}
func (ep *endpoints) ListDefinitions(w http.ResponseWriter, r *http.Request) {
	lr := ep.decodeListRequest(r)
	definitionList, err := ep.definitionService.List(
		lr.limit, lr.offset, lr.sortBy, lr.order, lr.filters, lr.envFilters)
	if definitionList.Definitions == nil {
		definitionList.Definitions = []state.Definition{}
	}
	if err != nil {
		ep.logger.Log(
			"message", "problem listing definitions",
			"operation", "ListDefinitions",
			"error", fmt.Sprintf("%+v", err))
		ep.encodeError(w, err)
	} else {
		response := make(map[string]interface{})
		response["total"] = definitionList.Total
		response["definitions"] = definitionList.Definitions
		response["limit"] = lr.limit
		response["offset"] = lr.offset
		response["sort_by"] = lr.sortBy
		response["order"] = lr.order
		response["env_filters"] = lr.envFilters
		for k, v := range lr.filters {
			response[k] = v
		}
		ep.encodeResponse(w, response)
	}
}
func (ep *endpoints) GetDefinition(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	definition, err := ep.definitionService.Get(vars["definition_id"])
	if err != nil {
		ep.logger.Log(
			"message", "problem getting definitions",
			"operation", "GetDefinition",
			"error", fmt.Sprintf("%+v", err),
			"definition_id", vars["definition_id"])
		ep.encodeError(w, err)
	} else {
		ep.encodeResponse(w, definition)
	}
}
func (ep *endpoints) GetDefinitionByAlias(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	definition, err := ep.definitionService.GetByAlias(vars["alias"])
	if err != nil {
		ep.logger.Log(
			"message", "problem getting definition by alias",
			"operation", "GetDefinitionByAlias",
			"error", fmt.Sprintf("%+v", err),
			"alias", vars["alias"])
		ep.encodeError(w, err)
	} else {
		ep.encodeResponse(w, definition)
	}
}
func (ep *endpoints) CreateDefinition(w http.ResponseWriter, r *http.Request) {
	var definition state.Definition
	err := ep.decodeRequest(r, &definition)
	if err != nil {
		ep.encodeError(w, exceptions.MalformedInput{ErrorString: err.Error()})
		return
	}
	created, err := ep.definitionService.Create(&definition)
	if err != nil {
		ep.logger.Log(
			"message", "problem creating definition",
			"operation", "CreateDefinition",
			"error", fmt.Sprintf("%+v", err))
		ep.encodeError(w, err)
	} else {
		ep.encodeResponse(w, created)
	}
}
func (ep *endpoints) UpdateDefinition(w http.ResponseWriter, r *http.Request) {
	var definition state.Definition
	err := ep.decodeRequest(r, &definition)
	if err != nil {
		ep.encodeError(w, exceptions.MalformedInput{ErrorString: err.Error()})
		return
	}
	vars := mux.Vars(r)
	updated, err := ep.definitionService.Update(vars["definition_id"], definition)
	if err != nil {
		ep.logger.Log(
			"message", "problem updating definition",
			"operation", "UpdateDefinition",
			"error", fmt.Sprintf("%+v", err),
			"definition_id", vars["definition_id"])
		ep.encodeError(w, err)
	} else {
		ep.encodeResponse(w, updated)
	}
}
func (ep *endpoints) DeleteDefinition(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	err := ep.definitionService.Delete(vars["definition_id"])
	if err != nil {
		ep.logger.Log(
			"message", "problem deleting definition",
			"operation", "DeleteDefinition",
			"error", fmt.Sprintf("%+v", err),
			"definition_id", vars["definition_id"])
		ep.encodeError(w, err)
	} else {
		ep.encodeResponse(w, map[string]bool{"deleted": true})
	}
}
func (ep *endpoints) ListRuns(w http.ResponseWriter, r *http.Request) {
	lr := ep.decodeListRequest(r)
	runList, err := ep.executionService.List(lr.limit, lr.offset, lr.order, lr.sortBy, lr.filters, lr.envFilters)
	if err != nil {
		ep.logger.Log(
			"message", "problem listing runs",
			"operation", "ListRuns",
			"error", fmt.Sprintf("%+v", err))
		ep.encodeError(w, err)
	} else {
		response := make(map[string]interface{})
		response["total"] = runList.Total
		response["history"] = runList.Runs
		response["limit"] = lr.limit
		response["offset"] = lr.offset
		response["sort_by"] = lr.sortBy
		response["order"] = lr.order
		response["env_filters"] = lr.envFilters
		for k, v := range lr.filters {
			response[k] = v
		}
		ep.encodeResponse(w, response)
	}
}
func (ep *endpoints) ListDefinitionRuns(w http.ResponseWriter, r *http.Request) {
	lr := ep.decodeListRequest(r)
	vars := mux.Vars(r)
	definitionID, ok := vars["definition_id"]
	if ok {
		lr.filters["definition_id"] = []string{definitionID}
	}
	runList, err := ep.executionService.List(lr.limit, lr.offset, lr.order, lr.sortBy, lr.filters, lr.envFilters)
	if err != nil {
		ep.logger.Log(
			"message", "problem listing definition runs",
			"operation", "ListDefinitionRuns",
			"error", fmt.Sprintf("%+v", err))
		ep.encodeError(w, err)
	} else {
		response := ep.createListRunsResponse(runList, lr)
		ep.encodeResponse(w, response)
	}
}
func (ep *endpoints) ListTemplateRuns(w http.ResponseWriter, r *http.Request) {
	lr := ep.decodeListRequest(r)
	vars := mux.Vars(r)
	tplID, ok := vars["template_id"]
	if ok {
		lr.filters["executable_id"] = []string{tplID}
	}
	runList, err := ep.executionService.List(lr.limit, lr.offset, lr.order, lr.sortBy, lr.filters, lr.envFilters)
	if err != nil {
		ep.logger.Log(
			"message", "problem listing runs for template",
			"operation", "ListTemplateRuns",
			"error", fmt.Sprintf("%+v", err))
		ep.encodeError(w, err)
	} else {
		response := ep.createListRunsResponse(runList, lr)
		ep.encodeResponse(w, response)
	}
}
func (ep *endpoints) createListRunsResponse(runList state.RunList, req listRequest) map[string]interface{} {
	response := make(map[string]interface{})
	response["total"] = runList.Total
	response["history"] = runList.Runs
	response["limit"] = req.limit
	response["offset"] = req.offset
	response["sort_by"] = req.sortBy
	response["order"] = req.order
	response["env_filters"] = req.envFilters
	for k, v := range req.filters {
		response[k] = v
	}
	return response
}
func (ep *endpoints) GetRun(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	run, err := ep.executionService.Get(vars["run_id"])
	if err != nil {
		ep.logger.Log(
			"message", "problem getting run",
			"operation", "GetRun",
			"error", fmt.Sprintf("%+v", err),
			"run_id", vars["run_id"])
		ep.encodeError(w, err)
	} else {
		ep.encodeResponse(w, run)
	}
}
func (ep *endpoints) GetPayload(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	run, err := ep.executionService.Get(vars["run_id"])
	if err != nil {
		ep.logger.Log(
			"message", "problem getting run",
			"operation", "GetRun",
			"error", fmt.Sprintf("%+v", err),
			"run_id", vars["run_id"])
		ep.encodeError(w, err)
	} else {
		if run.ExecutionRequestCustom != nil {
			ep.encodeResponse(w, run.ExecutionRequestCustom)
		} else {
			ep.encodeResponse(w, map[string]string{})
		}
	}
}
func (ep *endpoints) CreateRun(w http.ResponseWriter, r *http.Request) {
	var lr LaunchRequest
	err := ep.decodeRequest(r, &lr)
	if err != nil {
		ep.encodeError(w, exceptions.MalformedInput{ErrorString: err.Error()})
		return
	}
	vars := mux.Vars(r)
	req := state.DefinitionExecutionRequest{
		ExecutionRequestCommon: &state.ExecutionRequestCommon{
			Env:              lr.Env,
			OwnerID:          "v1-unknown",
			Command:          nil,
			Memory:           nil,
			Cpu:              nil,
			Gpu:              nil,
			Engine:           &state.DefaultEngine,
			EphemeralStorage: nil,
			NodeLifecycle:    nil,
		},
	}
	run, err := ep.executionService.CreateDefinitionRunByDefinitionID(vars["definition_id"], &req)
	if err != nil {
		ep.logger.Log(
			"message", "problem creating run",
			"operation", "CreateRun",
			"error", fmt.Sprintf("%+v", err))
		ep.encodeError(w, err)
	} else {
		ep.encodeResponse(w, run)
	}
}
func (ep *endpoints) CreateRunV2(w http.ResponseWriter, r *http.Request) {
	var lr LaunchRequestV2
	err := ep.decodeRequest(r, &lr)
	if err != nil {
		ep.encodeError(w, exceptions.MalformedInput{ErrorString: err.Error()})
		return
	}
	if len(lr.RunTags.OwnerEmail) == 0 || len(lr.RunTags.TeamName) == 0 {
		ep.encodeError(w, exceptions.MalformedInput{
			ErrorString: fmt.Sprintf("run_tags must exist in body and contain [owner_email] and [team_name]")})
		return
	}
	vars := mux.Vars(r)
	if lr.Engine == nil {
		if lr.SparkExtension != nil {
			lr.Engine = &state.EKSSparkEngine
		} else {
			lr.Engine = &state.EKSEngine
		}
	}
	req := state.DefinitionExecutionRequest{
		ExecutionRequestCommon: &state.ExecutionRequestCommon{
			Env:              lr.Env,
			OwnerID:          lr.RunTags.OwnerEmail,
			Command:          nil,
			Memory:           nil,
			Cpu:              nil,
			Gpu:              nil,
			Engine:           lr.Engine,
			EphemeralStorage: nil,
			NodeLifecycle:    nil,
			SparkExtension:   lr.SparkExtension,
			Description:      lr.Description,
		},
	}
	run, err := ep.executionService.CreateDefinitionRunByDefinitionID(vars["definition_id"], &req)
	if err != nil {
		ep.logger.Log(
			"message", "problem creating V2 run",
			"operation", "CreateRunV2",
			"error", fmt.Sprintf("%+v", err))
		ep.encodeError(w, err)
	} else {
		ep.encodeResponse(w, run)
	}
}