api

Documentation

Index

• func RegisterRunningPlugin(p *Plugin)
• func StopAllRunningPlugins()
• func TryCollect(result map[string]interface{})
• func TryCount(key string, val time.Duration)
• func TryGetCount(key string) int
• type CacheConfigInjector
• type CacheSystem
• type CacheSystemType
• type Counter
  • func (j *Counter) Count(key string, val time.Duration)
  • func (j *Counter) GetAndReset(key string) int
• type DeployableCache
• type FileInfo
• type FileSystem
• type FileSystemType
• type InputSplit
  • func (i InputSplit) Size() int64
• type JobInfo
• type Metrics
  • func CollectMetrics(fields map[string]string) (*Metrics, error)
  • func (j *Metrics) AddField(key string, description string) error
  • func (j *Metrics) Collect(result map[string]interface{})
  • func (j *Metrics) Info() string
  • func (j *Metrics) Reset()
  • func (j *Metrics) Start()
• type Phase
• type Plugin
  • func (p *Plugin) GetConnection() grpc.ClientConnInterface
  • func (p *Plugin) Init() error
  • func (p *Plugin) Interact(in io.Reader) error
  • func (p *Plugin) IsConnected() bool
  • func (p *Plugin) IsReady() bool
  • func (p *Plugin) Start(args ...string) error
  • func (p *Plugin) Stop()
• type PollingStrategy
• type Task
• type TaskInfo
• type TaskResult

Functions

func RegisterRunningPlugin

func RegisterRunningPlugin(p *Plugin)

func StopAllRunningPlugins

func StopAllRunningPlugins()

func TryCollect

func TryCollect(result map[string]interface{})

func TryCount

func TryCount(key string, val time.Duration)

func TryGetCount

func TryGetCount(key string) int

Types

type CacheConfigInjector

type CacheConfigInjector interface {
	CacheSystem() DeployableCache
}

CacheConfigInjector interface to be used by a function to inject the cache system config into a deployment package

type CacheSystem

type CacheSystem interface {
	FileSystem
	DeployableCache

	Flush(system FileSystem) error
	Clear() error
}

CacheSystem represent a ephemeral file system used for intermediate state between map/reduce phases

type CacheSystemType

type CacheSystemType int

CacheSystemType is an identifier for supported FileSystems

const (
	NoCache CacheSystemType = iota
	InMemory
	Redis
	Olric
	EFS
)

Identifiers for supported FileSystemTypes

type Counter

type Counter struct {
	Counter map[string]time.Duration
	sync.Mutex
}

func (*Counter) Count

func (j *Counter) Count(key string, val time.Duration)

func (*Counter) GetAndReset

func (j *Counter) GetAndReset(key string) int

type DeployableCache

type DeployableCache interface {
	//Deploy will deploy a cache based on the config and viper values - can use plugins
	Deploy() error
	//Undeploy will remove a prior deployment
	Undeploy() error
	//Check checks if the cache is deployable, e.g. if the plugin is running, all configs are set. Should not interact with the cloud provider, just locally check if everything is ready
	Check() error
	//FunctionInjector can be used by function deployment code to modify function deployments to use the underling cache system, warning needs to be implemented for each platform individually
	FunctionInjector() CacheConfigInjector
}

type FileInfo

type FileInfo struct {
	Name string // file path
	Size int64  // file size in bytes
}

FileInfo provides information about a file

type FileSystem

type FileSystem interface {
	ListFiles(pathGlob string) ([]FileInfo, error)
	Stat(filePath string) (FileInfo, error)
	OpenReader(filePath string, startAt int64) (io.ReadCloser, error)
	OpenWriter(filePath string) (io.WriteCloser, error)
	Delete(filePath string) error
	Join(elem ...string) string
	Split(path string) []string
	Init() error
}

FileSystem provides the file backend for MapReduce jobs. Input data is read from a file system. Intermediate and output data is written to a file system. This is abstracted to allow remote filesystems like S3 to be supported.

type FileSystemType

type FileSystemType int

FileSystemType is an identifier for supported FileSystems

const (
	Local FileSystemType = iota
	S3
	MINIO
)

Identifiers for supported FileSystemTypes

type InputSplit

type InputSplit struct {
	Filename    string // The file that the input split operates on
	StartOffset int64  // The starting byte index of the split in the file
	EndOffset   int64  // The ending byte index (inclusive) of the split in the file
}

InputSplit contains the information about a contiguous chunk of an input file. startOffset and endOffset are inclusive. For example, if the startOffset was 10 and the endOffset was 14, then the InputSplit would describe a 5 byte chunk of the file.

func (InputSplit) Size

func (i InputSplit) Size() int64

Size returns the number of bytes that the InputSplit spans

type JobInfo

type JobInfo struct {
	JobId int

	//Total number of Inputs for this job
	Splits int
	//SplitSize of this job in byte
	SplitSize int64
	//the maximum number of bytes per pin in the map phase
	MapBinSize int64
	//Maximum input size for reduce function
	ReduceBinSize int64
	//Maximum number of allowed concurrent function calls
	MaxConcurrency int
	//Used Backend Type, e.g., whisk, local or lambda ...
	Backend string
	//Used function Memmory in Megabyte
	FunctionMemory int
	//CacheType reference
	CacheType int

	//estimated lines of code for the user defined map function
	MapLOC int
	//estimated lines of code for the user defined reduce function
	ReduceLoc int
}

type Metrics

type Metrics struct {
	Fields map[string]string
	// contains filtered or unexported fields
}

func CollectMetrics

func CollectMetrics(fields map[string]string) (*Metrics, error)

CollectMetrics creates or gets the Metrics Singleton, and starts the activation log writer. Provided fields will be added to the log.

func (*Metrics) AddField

func (j *Metrics) AddField(key string, description string) error

func (*Metrics) Collect

func (j *Metrics) Collect(result map[string]interface{})

func (*Metrics) Info

func (j *Metrics) Info() string

func (*Metrics) Reset

func (j *Metrics) Reset()

func (*Metrics) Start

func (j *Metrics) Start()

type Phase

type Phase int

Phase is a descriptor of the phase (i.e. Map or Reduce) of a Job

const (
	MapPhase Phase = iota
	ReducePhase
	SortMapPhase
	SortReducePhase
)

Descriptors of the Job phase

type Plugin

type Plugin struct {
	FullName       string `json:"name"`       // FullName of the plugin, must be a go `get`-able package
	ExecutableName string `json:"executable"` // Name of the executable installed by go install
	// contains filtered or unexported fields
}

func (*Plugin) GetConnection

func (p *Plugin) GetConnection() grpc.ClientConnInterface

func (*Plugin) Init

func (p *Plugin) Init() error

func (*Plugin) Interact

func (p *Plugin) Interact(in io.Reader) error

func (*Plugin) IsConnected

func (p *Plugin) IsConnected() bool

func (*Plugin) IsReady

func (p *Plugin) IsReady() bool

func (*Plugin) Start

func (p *Plugin) Start(args ...string) error

func (*Plugin) Stop

func (p *Plugin) Stop()

type PollingStrategy

type PollingStrategy interface {
	/*StartJob initializes a Job, all subsequent TaskUpdates are treated
	  as related to this Job. Calling StartJob again indicates the start of a
	  new job and the end of the last job.
	*/
	StartJob(JobInfo) error

	/*
		TaskUpdate updates metadata related to a task. Usually called after a Polling
		attempt.
	*/
	TaskUpdate(TaskInfo) error

	/*
		Poll blocks until the given task should be polled.
		Poll returns a channel that returns once a poll should be performed.
		This channel should only fire once. To cancel a poll use the context.
	*/
	Poll(context context.Context, RId string) (<-chan interface{}, error)
}

type Task

type Task struct {
	JobNumber        int
	Phase            Phase
	BinID            uint
	IntermediateBins uint
	Splits           []InputSplit
	FileSystemType   FileSystemType
	CacheSystemType  CacheSystemType
	WorkingLocation  string
	Cleanup          bool
}

Task defines a serialized description of a single unit of work in a MapReduce job, as well as the necessary information for a remote executor to initialize itself and begin working.

type TaskInfo

type TaskInfo struct {
	RId string
	//unique job id
	JobId int
	//unique task id
	TaskId int
	//indecates map/reduce phase
	Phase int
	//time the task was sent to the backend
	RequestStart time.Time
	//time the task was successfully polled by the backend
	RequestReceived time.Time

	//Duration of the task Execution
	ExecutionDuration time.Duration
	//RuntimeId - semi unique identifier of the used execution runtime
	RuntimeId string

	//Number of Inputs for this Task
	NumberOfInputs int
	//Number of Polls for this Task
	NumberOfPolls int
	//Indicates if this task is completed, e.g., executed successfully
	Completed bool
	//Indicates if this task failed
	Failed bool
}

type TaskResult

type TaskResult struct {
	BytesRead    int
	BytesWritten int

	Log string

	HId    string `json:"HId"`    //host identifier
	CId    string `json:"CId"`    //runtime identifier
	JId    string `json:"JId"`    //job identifier
	RId    string `json:"RId"`    //request identifier (by platform)
	CStart int64  `json:"cStart"` //start of runtime
	EStart int64  `json:"eStart"` //start of request
	EEnd   int64  `json:"eEnd"`   //end of request

	CWT int64 `json:"CWT"` //Cache WriteTime
	CRT int64 `json:"CRT"` //Cache ReadTime
	SWT int64 `json:"SWT"` //S3 WriteTime
	SRT int64 `json:"SRT"` //S3 ReadTime
}