storage

Documentation

Index

• Variables
• func WithZkConnection(c *zk.Conn) zkStorageOpt
• func WithZkPrefix(prefix string) zkStorageOpt
• func WithZkServers(servers []string) zkStorageOpt
• type Pipeline
  • func (p *Pipeline) GenerateTasks() []Task
  • func (p *Pipeline) RequiredCPU() int
  • func (p *Pipeline) RequiredMem() datasize.ByteSize
  • func (p *Pipeline) SatisfiesOffer(taskID TaskID, offer msg.Offer) bool
  • func (p *Pipeline) TaskInfo(taskID TaskID, offer msg.Offer) []msg.TaskInfo
• type Storage
  • func NewStorageBackend(uri string) (Storage, error)
  • func NewZkStorage(connectOpt zkStorageOpt, options ...zkStorageOpt) (Storage, error)
• type Task
  • func (t *Task) IsRunnable() bool
• type TaskID
  • func NewTaskID(pipelineID string, taskID string, procID string) TaskID
  • func (t TaskID) GroupID() string
  • func (t TaskID) PipelineID() string
  • func (t TaskID) ProcessorID() string
  • func (t TaskID) String() string
  • func (t TaskID) WithProcessID(id string) TaskID
• type TaskProcess

Variables

var ErrConflict = errors.New("Storage backend conflict")

var ErrPipelineExists = errors.New("A pipeline already exists with this ID.")

var ErrPipelineNotExists = errors.New("This pipeline does not exist.")

Functions

func WithZkConnection

func WithZkConnection(c *zk.Conn) zkStorageOpt

func WithZkPrefix

func WithZkPrefix(prefix string) zkStorageOpt

func WithZkServers

func WithZkServers(servers []string) zkStorageOpt

Types

type Pipeline

type Pipeline struct {
	ID        string `json:"id"`
	Key       string `json:"key"`
	Container struct {
		Image      string `json:"image"`
		Network    string `json:"network"`
		Parameters []struct {
			Key   string `value:"key"`
			Value string `json:"value"`
		} `json:"parameters"`
	} `json:"container"`
	Processors map[string]struct {
		ID          string            `json:"id"`
		Shell       bool              `json:"shell"`
		Command     string            `json:"command"`
		Arguments   []string          `json:"arguments"`
		Environment map[string]string `json:"environment"`
		Resources   struct {
			CPU  float64 `json:"cpu"`
			Mem  float64 `json:"mem"`
			Disk float64 `json:"disk"`
		} `json:"resources"`
		PortMapping []struct {
			HostPort      int    `json:"host_port"`
			ContainerPort int    `json:"container_port"`
			Protocol      string `json:"protocol"`
		} `json:"port_mappings"`
		KillGracePeriod int `json:"kill_grace_period"`
	} `json:"processes"`
	Instances         int               `json:"instances"`
	Roles             []string          `json:"roles"`
	Labels            map[string]string `json:"labels"`
	Environment       map[string]string `json:"environment"`
	Created           time.Time         `json:"created"`
	Updated           time.Time         `json:"updated"`
	ExecutorResources struct {
		CPU  float64 `json:"cpu"`
		Mem  float64 `json:"mem"`
		Disk float64 `json:"disk"`
	} `json:"executor_resources"`
	Metadata map[string]string `json:"metadata,omitempty"`
	Tasks    []Task            `json:"tasks,omitempty"`
}

func (*Pipeline) GenerateTasks

func (p *Pipeline) GenerateTasks() []Task

func (*Pipeline) RequiredCPU

func (p *Pipeline) RequiredCPU() int

func (*Pipeline) RequiredMem

func (p *Pipeline) RequiredMem() datasize.ByteSize

func (*Pipeline) SatisfiesOffer

func (p *Pipeline) SatisfiesOffer(taskID TaskID, offer msg.Offer) bool

func (*Pipeline) TaskInfo

func (p *Pipeline) TaskInfo(taskID TaskID, offer msg.Offer) []msg.TaskInfo

type Storage

type Storage interface {
	FrameworkID() (string, error)
	SetFrameworkID(string) error

	Tasks(bool) ([]string, <-chan zk.Event, error)
	SaveTasks(map[string]*Task) error
	TasksInfo() (map[string]*Task, error)

	Pipelines(bool) (map[string]Pipeline, <-chan zk.Event, error)
	SavePipeline(Pipeline, bool) error
	ResizePipeline(string, int) error
	DeletePipeline(string) error

	LastAutoscale() (time.Time, error)
	SetLastAutoscale(time.Time) error
}

func NewStorageBackend

func NewStorageBackend(uri string) (Storage, error)

func NewZkStorage

func NewZkStorage(connectOpt zkStorageOpt, options ...zkStorageOpt) (Storage, error)

type Task

type Task struct {
	ID          TaskID                  `json:"id"`
	PipelineID  string                  `json:"pipeline_id"`
	ProcessorID string                  `json:"processor_id"`
	Message     string                  `json:"message,omitempty"`
	Launching   bool                    `json:"-"`
	Launched    time.Time               `json:"launched"`
	Failures    int                     `json:"failures"`
	Updated     time.Time               `json:"updated"`
	Ok          bool                    `json:"ok"`
	Failing     bool                    `json:"failing"`
	Killing     bool                    `json:"killing"`
	Processes   map[string]*TaskProcess `json:"processes"`
}

func (*Task) IsRunnable

func (t *Task) IsRunnable() bool

type TaskID

type TaskID string

func NewTaskID

func NewTaskID(pipelineID string, taskID string, procID string) TaskID

func (TaskID) GroupID

func (t TaskID) GroupID() string

func (TaskID) PipelineID

func (t TaskID) PipelineID() string

func (TaskID) ProcessorID

func (t TaskID) ProcessorID() string

func (TaskID) String

func (t TaskID) String() string

func (TaskID) WithProcessID

func (t TaskID) WithProcessID(id string) TaskID

type TaskProcess

type TaskProcess struct {
	ID      string    `json:"id"`
	Status  string    `json:"status,omitempty"`
	AgentID string    `json:"agent_id,omitempty"`
	Updated time.Time `json:"updated"`
}