agent

Documentation

Index

• Constants
• func NewSlotQueue(key string) *slotQueue
• func NewSlotQueueMgr() *slotQueueMgr
• type Agent
  • func New(da DataAccess) Agent
• type CachedDataAccess
  • func (da *CachedDataAccess) GetApp(ctx context.Context, appName string) (*models.App, error)
  • func (da *CachedDataAccess) GetRoute(ctx context.Context, appName string, routePath string) (*models.Route, error)
• type Call
• type CallOpt
  • func FromModel(mCall *models.Call) CallOpt
  • func FromRequest(appName, path string, req *http.Request, params Params) CallOpt
  • func WithWriter(w io.Writer) CallOpt
• type DataAccess
  • func NewCachedDataAccess(da DataAccess) DataAccess
  • func NewDirectDataAccess(ds models.Datastore, ls models.LogStore, mq models.MessageQueue) DataAccess
• type FunctionStats
• type Param
• type Params
• type ResourceToken
• type ResourceTracker
  • func NewResourceTracker() ResourceTracker
• type Slot
• type SlotQueueMetricType
• type Stats

Constants

const (
	Mem1MB = 1024 * 1024
	Mem1GB = 1024 * 1024 * 1024
)

Functions

func NewSlotQueue

func NewSlotQueue(key string) *slotQueue

func NewSlotQueueMgr

func NewSlotQueueMgr() *slotQueueMgr

Types

type Agent

type Agent interface {
	// GetCall will return a Call that is executable by the Agent, which
	// can be built via various CallOpt's provided to the method.
	GetCall(...CallOpt) (Call, error)

	// Submit will attempt to execute a call locally, a Call may store information
	// about itself in its Start and End methods, which will be called in Submit
	// immediately before and after the Call is executed, respectively. An error
	// will be returned if there is an issue executing the call or the error
	// may be from the call's execution itself (if, say, the container dies,
	// or the call times out).
	Submit(Call) error

	// Close will wait for any outstanding calls to complete and then exit.
	// Close is not safe to be called from multiple threads.
	io.Closer

	// Stats should be burned at the stake. adding so as to not ruffle feathers.
	// TODO this should be derived from our metrics
	Stats() Stats

	// Return the http.Handler used to handle Prometheus metric requests
	PromHandler() http.Handler
	AddCallListener(fnext.CallListener)

	// Enqueue is to use the agent's sweet sweet client bindings to remotely
	// queue async tasks and should be removed from Agent interface ASAP.
	Enqueue(context.Context, *models.Call) error
}

func New

func New(da DataAccess) Agent

type CachedDataAccess

type CachedDataAccess struct {
	DataAccess
	// contains filtered or unexported fields
}

CachedDataAccess wraps a DataAccess and caches the results of GetApp and GetRoute.

func (*CachedDataAccess) GetApp

func (da *CachedDataAccess) GetApp(ctx context.Context, appName string) (*models.App, error)

func (*CachedDataAccess) GetRoute

func (da *CachedDataAccess) GetRoute(ctx context.Context, appName string, routePath string) (*models.Route, error)

type Call

type Call interface {
	// Model will return the underlying models.Call configuration for this call.
	// TODO we could respond to async correctly from agent but layering, this
	// is only because the front end has different responses based on call type.
	// try to discourage use elsewhere until this gets pushed down more...
	Model() *models.Call

	// Start will be called before this call is executed, it may be used to
	// guarantee mutual exclusion, check docker permissions, update timestamps,
	// etc.
	// TODO Start and End can likely be unexported as they are only used in the agent,
	// and on a type which is constructed in a specific agent. meh.
	Start(ctx context.Context) error

	// End will be called immediately after attempting a call execution,
	// regardless of whether the execution failed or not. An error will be passed
	// to End, which if nil indicates a successful execution. Any error returned
	// from End will be returned as the error from Submit.
	End(ctx context.Context, err error) error
}

type CallOpt

type CallOpt func(a *agent, c *call) error

TODO build w/o closures... lazy

func FromModel

func FromModel(mCall *models.Call) CallOpt

TODO this currently relies on FromRequest having happened before to create the model here, to be a fully qualified model. We probably should double check but having a way to bypass will likely be what's used anyway unless forced.

func FromRequest

func FromRequest(appName, path string, req *http.Request, params Params) CallOpt

func WithWriter

func WithWriter(w io.Writer) CallOpt

TODO this should be required

type DataAccess

type DataAccess interface {
	// GetApp abstracts querying the datastore for an app.
	GetApp(ctx context.Context, appName string) (*models.App, error)

	// GetRoute abstracts querying the datastore for a route within an app.
	GetRoute(ctx context.Context, appName string, routePath string) (*models.Route, error)

	// Enqueue will add a Call to the queue (ultimately forwards to mq.Push).
	Enqueue(ctx context.Context, mCall *models.Call) error

	// Dequeue will query the queue for the next available Call that can be run
	// by this Agent, and reserve it (ultimately forwards to mq.Reserve).
	Dequeue(ctx context.Context) (*models.Call, error)

	// Start will attempt to start the provided Call within an appropriate
	// context.
	Start(ctx context.Context, mCall *models.Call) error

	// Finish will notify the system that the Call has been processed, and
	// fulfill the reservation in the queue if the call came from a queue.
	Finish(ctx context.Context, mCall *models.Call, stderr io.Reader, async bool) error
}

DataAccess abstracts the datastore and message queue operations done by the agent, so that API nodes and runner nodes can work with the same interface but actually operate on the data in different ways (by direct access or by mediation through an API node).

func NewCachedDataAccess

func NewCachedDataAccess(da DataAccess) DataAccess

func NewDirectDataAccess

func NewDirectDataAccess(ds models.Datastore, ls models.LogStore, mq models.MessageQueue) DataAccess

type FunctionStats

type FunctionStats struct {
	Queue    uint64
	Running  uint64
	Complete uint64
	Failed   uint64
}

statistics for an individual function

type Param

type Param struct {
	Key   string
	Value string
}

type Params

type Params []Param

type ResourceToken

type ResourceToken interface {
	// Close must be called by any thread that receives a token.
	io.Closer
}

type ResourceTracker

type ResourceTracker interface {
	WaitAsyncResource() chan struct{}
	// returns a closed channel if the resource can never me met.
	GetResourceToken(ctx context.Context, memory uint64, isAsync bool) <-chan ResourceToken
}

A simple resource (memory, cpu, disk, etc.) tracker for scheduling. TODO: add cpu, disk, network IO for future

func NewResourceTracker

func NewResourceTracker() ResourceTracker

type Slot

type Slot interface {
	Close() error
	Error() error
	// contains filtered or unexported methods
}

type SlotQueueMetricType

type SlotQueueMetricType int

const (
	SlotQueueRunner SlotQueueMetricType = iota
	SlotQueueStarter
	SlotQueueWaiter
	SlotQueueLast
)

type Stats

type Stats struct {
	// statistics for all functions combined
	Queue    uint64
	Running  uint64
	Complete uint64
	Failed   uint64
	// statistics for individual functions, keyed by function path
	FunctionStatsMap map[string]*FunctionStats
}