request

Documentation

Overview

Package request provides an interface for managing requests.

Index

• Constants
• Variables
• type Manager
  • func NewManager(config ManagerConfig) Manager
• type ManagerConfig
• type Resumer
  • func NewResumer(cfg ResumerConfig) Resumer
• type ResumerConfig

Constants

const (
	DB_TRIES      = 3
	DB_RETRY_WAIT = time.Duration(500 * time.Millisecond)
	JR_TRIES      = 5
	JR_RETRY_WAIT = time.Duration(5 * time.Second)
)

Variables

var (
	ErrNotUpdated      = errors.New("no row updated")
	ErrMultipleUpdated = errors.New("multiple rows updated/deleted, expected single-row update/delete")
)

Types

type Manager

type Manager interface {
	// Create creates a request and saves it to the db. The request is not
	// started; its state is pending until Start is called.
	Create(proto.CreateRequest) (proto.Request, error)

	// Get retrieves the request corresponding to the provided id,
	// without its job chain or parameters set.
	Get(requestId string) (proto.Request, error)

	// Get retrieves the request corresponding to the provided id,
	// with its job chain and parameters.
	GetWithJC(requestId string) (proto.Request, error)

	// Start starts a request (sends it to the JR).
	Start(requestId string) error

	// Stop stops a request (sends a stop signal to the JR).
	Stop(requestId string) error

	// Finish marks a request as being finished. It gets the request's final
	// state from the proto.FinishRequest argument.
	Finish(requestId string, finishParams proto.FinishRequest) error

	// Fail a pending request (if it can't be started for some reason).
	FailPending(requestId string) error

	// Specs returns a list of all the request specs the the RM knows about.
	Specs() []proto.RequestSpec

	// JobChain returns the job chain for the given request id.
	JobChain(requestId string) (proto.JobChain, error)

	// Find returns a list of requests that match the given filter criteria,
	// in descending order by create time (i.e. most recent first) and ascending
	// by request id where create time is not unique. Returned requests do
	// not have job chain or args set.
	Find(filter proto.RequestFilter) ([]proto.Request, error)
}

A Manager creates and manages the life cycle of requests.

func NewManager

func NewManager(config ManagerConfig) Manager

type ManagerConfig

type ManagerConfig struct {
	GrapherFactory grapher.GrapherFactory
	DBConnector    *sql.DB
	JRClient       jr.Client
	DefaultJRURL   string
	ShutdownChan   chan struct{}
}

type Resumer

type Resumer interface {
	// Suspend marks a running request as suspended and saves the corresponding
	// suspended job chain.
	Suspend(sjc proto.SuspendedJobChain) error

	// ResumeAll tries to resume all the SJCs currently stored in the database.
	ResumeAll()

	// Resume tries to resume a single SJC given its id and a connection to the
	// database. The SJC must be claimed (`rm_host` field for the SJC must be set
	// to the hostname given when creating the Resumer) before calling Resume, or
	// it will fail.
	Resume(id string) error

	// Cleanup cleans up abandoned and old SJCs. Abandoned SJCs are those that have
	// been claimed by an RM (`rm_host` field set) but have not been updated in a
	// while, meaning the RM resuming them probably crashed. These SJCs are
	// unclaimed (set `rm_host` to null) so they can be resumed in the future. Old
	// SJCs are those which haven't been resumed within the TTL provided when
	// creating the Resumer (rounded to the nearest second). They're deleted and
	// their requests' states set to FAILED.
	Cleanup()
}

func NewResumer

func NewResumer(cfg ResumerConfig) Resumer

type ResumerConfig

type ResumerConfig struct {
	RequestManager       Manager
	DBConnector          *sql.DB
	JRClient             jr.Client
	DefaultJRURL         string
	RMHost               string
	ShutdownChan         chan struct{}
	SuspendedJobChainTTL time.Duration
}