schedulerapi

Documentation

Index

• Constants
• Variables
• func Initialize() error
• func SetupRoutes(router *mux.Router)
• type InputSplitter
• type InputSplitterByLine
  • func NewInputSplitterByLine(linesPerFile int) InputSplitterByLine
  • func (splitter InputSplitterByLine) Split(ctx context.Context, reader io.Reader, containerURL azblob.ContainerURL, ...) (int, error)
• type Job
  • func NewJob(job, finalizerJob *batchv1.Job) Job
• type JobCreateRequest
• type Route

Constants

const BufferSizeWriteLimit int = 1024 * 1024 * 3.5 // 3.5 MB

BufferSizeWriteLimit defines the approximated size to write content to a blob (the actual limit is 4MB, we are stopping before)

const JobBlobPrefixEnvVarName = "BLOBPREFIX"

JobBlobPrefixEnvVarName defines the environment variable containing the storage blob prefix name for the job

const JobCorrelationIDLabelName = "job_correlation_id"

JobCorrelationIDLabelName defines the label name to correlate jobs with additional ones

const JobEventGridSasKeyEnvVarName = "EVENTGRIDSASKEY"

JobEventGridSasKeyEnvVarName defines the environment variable containing the event grid sas key for the job

const JobEventGridTopicEndpointEnvVarName = "EVENTGRIDTOPICENDPOINT"

JobEventGridTopicEndpointEnvVarName defines the environment variable containing the event grid endpoint for the job

const JobHasFinalizerAnnotationName = "job_has_finalizer"

JobHasFinalizerAnnotationName defines if job has a finalizer

const JobHasWatcherAnnotationName = "job_has_watcher"

JobHasWatcherAnnotationName defines if job has a companion watcher job

const JobIDEnvVarName = "JOBID"

JobIDEnvVarName defines the environment variable containing the job identifier

const JobItemsPerJobEnvVarName = "ITEMSPERJOB"

JobItemsPerJobEnvVarName defines the environment variable containing the number of items to be processed per job

const JobNameEnvVarName = "JOBNAME"

JobNameEnvVarName defines the environment variable containing the job name

const JobStorageAccountKeyEnvVarName = "STORAGEKEY"

JobStorageAccountKeyEnvVarName defines the environment variable containing the storage account key

const JobStorageAccountNameEnvVarName = "STORAGEACCOUNT"

JobStorageAccountNameEnvVarName defines the environment variable containing the storage account name

const JobStorageConnectionStringEnvVarName = "STORAGECONNECTIONSTRING"

JobStorageConnectionStringEnvVarName defines the environment variable containing the storage connection string for the job

const JobStorageContainerEnvVarName = "CONTAINER"

JobStorageContainerEnvVarName defines the environment variable containing the storage container for the job

const MainJobAnnotationName = "job_is_main"

MainJobAnnotationName defines annotation name for main job flag

const StorageBlobPrefixAnnotationName = "storage_blob_prefix"

StorageBlobPrefixAnnotationName defines the label name containing the storage blob input prefix

const StorageContainerAnnotationName = "storage_container"

StorageContainerAnnotationName defines the label name containing the storage container name

Variables

var (

	// JobMemoryLimit defines the memory limit for the job pod when running on local cluster
	JobMemoryLimit = flag.String("jobMemoryLimit", getEnvString("JOBMEMORYLIMIT", "256Mi"), "Job Memory limit for local cluster (256Mi by default)")
)

Functions

func Initialize

func Initialize() error

Initialize configures the api

func SetupRoutes

func SetupRoutes(router *mux.Router)

SetupRoutes register the api routes

Types

type InputSplitter

type InputSplitter interface {
	Split(ctx context.Context, reader io.Reader, containerURL azblob.ContainerURL, jobNamePrefix string) (int, error)
}

InputSplitter defines a splitting input file type

type InputSplitterByLine

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

InputSplitterByLine splits a big file in lines

func NewInputSplitterByLine

func NewInputSplitterByLine(linesPerFile int) InputSplitterByLine

NewInputSplitterByLine creates a new instance of InputSplitterByLine

func (InputSplitterByLine) Split

func (splitter InputSplitterByLine) Split(ctx context.Context, reader io.Reader, containerURL azblob.ContainerURL, jobNamePrefix string) (int, error)

Split create smaller files from a big calculation input

type Job

type Job struct {
	ID                string                         `json:"id"`
	Status            string                         `json:"status"`
	StartTime         *time.Time                     `json:"startTime,omitempty"`
	CompletionTime    *time.Time                     `json:"completionTime,omitempty"`
	Active            int                            `json:"active,omitempty"`
	Succeeded         int                            `json:"succeeded,omitempty"`
	Failed            int                            `json:"failed,omitempty"`
	Parallelism       int                            `json:"parallelism,omitempty"`
	Completions       int                            `json:"completions,omitempty"`
	StorageContainer  string                         `json:"storageContainer,omitempty"`
	StorageBlobPrefix string                         `json:"storageBlobPrefix,omitempty"`
	ExecutionLocation scheduler.JobExecutionLocation `json:"executionLocation"`
}

Job defines a working item

func NewJob

func NewJob(job, finalizerJob *batchv1.Job) Job

NewJob creates a job from a existing K8s job

type JobCreateRequest

type JobCreateRequest struct {
	ID                string `json:"id"`
	StorageContainer  string `json:"storageContainer,omitempty"`
	StorageBlobPrefix string `json:"storageBlobPrefix,omitempty"`
}

JobCreateRequest defines a request to create a job where the files already exist

type Route

type Route struct {
	Name        string
	Method      string
	Pattern     string
	HandlerFunc http.HandlerFunc
}

Route - object representing a route handler