jobs

Documentation

Index

• Constants
• func GetStateString(state clusterState) string
• func ParseStateString(state string) clusterState
• type Clusters
  • func NewClusters(deployerClient *clients.DeployerClient, config *viper.Viper) (*Clusters, error)
  • func (clusters *Clusters) DeleteCluster(runId string)
  • func (clusters *Clusters) GetState(runId string) clusterState
  • func (clusters *Clusters) ReloadClusterState() error
  • func (clusters *Clusters) ReserveDeployment(config *viper.Viper, applicationConfig *models.ApplicationConfig, ...) <-chan ReserveResult
  • func (clusters *Clusters) SetState(runId string, state clusterState)
  • func (clusters *Clusters) UnreserveDeployment(runId string, deleteCluster bool, log *logging.Logger) <-chan UnreserveResult
• type DeploymentFiles
  • func NewDeploymentFiles(config *viper.Viper) (DeploymentFiles, error)
• type FailedJobs
  • func NewFailedJobs() *FailedJobs
  • func (jobs *FailedJobs) AddJob(job Job)
• type Job
• type JobDeploymentConfig
  • func (config JobDeploymentConfig) GetNodes() []deployer.ClusterNode
• type JobManager
  • func NewJobManager(config *viper.Viper) (*JobManager, error)
  • func (manager *JobManager) AddJob(job Job)
  • func (manager *JobManager) FindJob(id string) (Job, error)
  • func (manager *JobManager) GetFailedJobs() []Job
  • func (manager *JobManager) GetJobs() []Job
• type JobResults
• type JobSummary
• type ReserveResult
• type S3DeploymentFiles
  • func (files *S3DeploymentFiles) DownloadDeployment(fileUrl string) (*deployer.Deployment, error)
• type UnreserveResult
• type Worker
  • func (worker *Worker) Run()
  • func (Worker *Worker) RunDirectJob(job Job) error
  • func (worker *Worker) RunJob(job Job) error

Constants

const (
	DEPLOYING   = 0
	AVAILABLE   = 1
	RESERVED    = 2
	UNRESERVING = 3
	FAILED      = 4

	ErrMaxClusters = "Max clusters reached"
)

Possible deployment states

const (
	JOB_QUEUED    = "QUEUED"
	JOB_RESERVING = "RESERVING"
	JOB_RUNNING   = "RUNNING"
	JOB_FINISHED  = "FINISHED"
	JOB_FAILED    = "FAILED"
)

Functions

func GetStateString

func GetStateString(state clusterState) string

func ParseStateString

func ParseStateString(state string) clusterState

Types

type Clusters

type Clusters struct {
	ClusterStore   blobstore.BlobStore
	Config         *viper.Viper
	DeployerClient *clients.DeployerClient

	MaxClusters int
	Deployments []*cluster
	// contains filtered or unexported fields
}

func NewClusters

func NewClusters(deployerClient *clients.DeployerClient, config *viper.Viper) (*Clusters, error)

func (*Clusters) DeleteCluster

func (clusters *Clusters) DeleteCluster(runId string)

func (*Clusters) GetState

func (clusters *Clusters) GetState(runId string) clusterState

func (*Clusters) ReloadClusterState

func (clusters *Clusters) ReloadClusterState() error

func (*Clusters) ReserveDeployment

func (clusters *Clusters) ReserveDeployment(
	config *viper.Viper,
	applicationConfig *models.ApplicationConfig,
	jobDeploymentConfig JobDeploymentConfig,
	runId string,
	log *logging.Logger) <-chan ReserveResult

func (*Clusters) SetState

func (clusters *Clusters) SetState(runId string, state clusterState)

func (*Clusters) UnreserveDeployment

func (clusters *Clusters) UnreserveDeployment(runId string, deleteCluster bool, log *logging.Logger) <-chan UnreserveResult

type DeploymentFiles

type DeploymentFiles interface {
	DownloadDeployment(fileName string) (*deployer.Deployment, error)
}

func NewDeploymentFiles

func NewDeploymentFiles(config *viper.Viper) (DeploymentFiles, error)

type FailedJobs

type FailedJobs struct {
	Jobs []Job
	// contains filtered or unexported fields
}

func NewFailedJobs

func NewFailedJobs() *FailedJobs

func (*FailedJobs) AddJob

func (jobs *FailedJobs) AddJob(job Job)

type Job

type Job interface {
	GetId() string
	GetApplicationConfig() *models.ApplicationConfig
	GetJobDeploymentConfig() JobDeploymentConfig
	GetLog() *log.FileLog
	Run(deploymentId string) error
	GetState() string
	SetState(state string)
	GetSummary() JobSummary
	SetFailed(error string)
	GetResults() <-chan *JobResults
	IsSkipUnreserveOnFailure() bool
	IsDirectJob() bool
}

type JobDeploymentConfig

type JobDeploymentConfig struct {
	Nodes []deployer.ClusterNode
}

func (JobDeploymentConfig) GetNodes

func (config JobDeploymentConfig) GetNodes() []deployer.ClusterNode

type JobManager

type JobManager struct {
	Queue      chan Job
	Jobs       map[string]Job
	Workers    []*Worker
	FailedJobs *FailedJobs
	// contains filtered or unexported fields
}

func NewJobManager

func NewJobManager(config *viper.Viper) (*JobManager, error)

func (*JobManager) AddJob

func (manager *JobManager) AddJob(job Job)

func (*JobManager) FindJob

func (manager *JobManager) FindJob(id string) (Job, error)

func (*JobManager) GetFailedJobs

func (manager *JobManager) GetFailedJobs() []Job

func (*JobManager) GetJobs

func (manager *JobManager) GetJobs() []Job

type JobResults

type JobResults struct {
	Error string
	Data  interface{}
}

type JobSummary

type JobSummary struct {
	DeploymentId string    `json:"deploymentId"`
	RunId        string    `json:"runId"`
	Status       string    `json:"status"`
	Create       time.Time `json:"create"`
}

type ReserveResult

type ReserveResult struct {
	DeploymentId string
	Err          string
}

type S3DeploymentFiles

type S3DeploymentFiles struct {
	// contains filtered or unexported fields
}

func (*S3DeploymentFiles) DownloadDeployment

func (files *S3DeploymentFiles) DownloadDeployment(fileUrl string) (*deployer.Deployment, error)

type UnreserveResult

type UnreserveResult struct {
	RunId string
	Err   string
}

type Worker

type Worker struct {
	Id               int
	Jobs             <-chan Job
	FailedJobs       *FailedJobs
	RetryReservation bool
	Config           *viper.Viper
	Clusters         *Clusters
}

func (*Worker) Run

func (worker *Worker) Run()

func (*Worker) RunDirectJob

func (Worker *Worker) RunDirectJob(job Job) error

func (*Worker) RunJob

func (worker *Worker) RunJob(job Job) error