dag

Documentation

Index

• Constants
• func DisplayName(name string) string
• func OptimusFuncMap() template.FuncMap
• func Quote(str string) string
• func Replace(old, newStr, name string) string
• func ReplaceDash(name string) string
• func SLAMissDuration(job *scheduler.JobWithDetails) (int64, error)
• func Trunc(c int, s string) string
• type AirflowConfig
  • func ToAirflowConfig(schedulerConf map[string]string) AirflowConfig
• type Compiler
  • func NewDagCompiler(hostname string, repo PluginRepo) (*Compiler, error)
  • func (c *Compiler) Compile(jobDetails *scheduler.JobWithDetails) ([]byte, error)
• type Hook
• type Hooks
  • func PrepareHooksForJob(job *scheduler.Job, pluginRepo PluginRepo) (Hooks, error)
  • func (h Hooks) List() []Hook
• type PluginRepo
• type Resource
  • func ToResource(resource *scheduler.Resource) *Resource
• type ResourceConfig
  • func ToResourceConfig(config *scheduler.ResourceConfig) *ResourceConfig
• type RuntimeConfig
  • func SetupRuntimeConfig(jobDetails *scheduler.JobWithDetails) RuntimeConfig
• type Task
  • func PrepareTask(job *scheduler.Job, pluginRepo PluginRepo) (Task, error)
• type TemplateContext
• type Upstream
• type Upstreams
  • func SetupUpstreams(upstreams scheduler.Upstreams, host string) Upstreams
  • func (u Upstreams) Empty() bool

Constants

const (
	EntitySchedulerAirflow = "schedulerAirflow"
)

Functions

func DisplayName

func DisplayName(name string) string

func OptimusFuncMap

func OptimusFuncMap() template.FuncMap

func Quote

func Quote(str string) string

func Replace

func Replace(old, newStr, name string) string

func ReplaceDash

func ReplaceDash(name string) string

func SLAMissDuration

func SLAMissDuration(job *scheduler.JobWithDetails) (int64, error)

func Trunc

func Trunc(c int, s string) string

Types

type AirflowConfig

type AirflowConfig struct {
	Pool  string
	Queue string
}

func ToAirflowConfig

func ToAirflowConfig(schedulerConf map[string]string) AirflowConfig

type Compiler

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

func NewDagCompiler

func NewDagCompiler(hostname string, repo PluginRepo) (*Compiler, error)

func (*Compiler) Compile

func (c *Compiler) Compile(jobDetails *scheduler.JobWithDetails) ([]byte, error)

type Hook

type Hook struct {
	Name       string
	Image      string
	Entrypoint plugin.Entrypoint
	IsFailHook bool
}

type Hooks

type Hooks struct {
	Pre          []Hook
	Post         []Hook
	Fail         []Hook
	Dependencies map[string]string
}

func PrepareHooksForJob

func PrepareHooksForJob(job *scheduler.Job, pluginRepo PluginRepo) (Hooks, error)

func (Hooks) List

func (h Hooks) List() []Hook

type PluginRepo

type PluginRepo interface {
	GetByName(name string) (*plugin.Plugin, error)
}

type Resource

type Resource struct {
	Request *ResourceConfig
	Limit   *ResourceConfig
}

func ToResource

func ToResource(resource *scheduler.Resource) *Resource

type ResourceConfig

type ResourceConfig struct {
	CPU    string
	Memory string
}

func ToResourceConfig

func ToResourceConfig(config *scheduler.ResourceConfig) *ResourceConfig

type RuntimeConfig

type RuntimeConfig struct {
	Resource *Resource
	Airflow  AirflowConfig
}

func SetupRuntimeConfig

func SetupRuntimeConfig(jobDetails *scheduler.JobWithDetails) RuntimeConfig

type Task

type Task struct {
	Name       string
	Image      string
	Entrypoint plugin.Entrypoint
}

func PrepareTask

func PrepareTask(job *scheduler.Job, pluginRepo PluginRepo) (Task, error)

type TemplateContext

type TemplateContext struct {
	JobDetails *scheduler.JobWithDetails

	Tenant          tenant.Tenant
	Version         string
	SLAMissDuration int64
	Hostname        string
	ExecutorTask    string
	ExecutorHook    string

	RuntimeConfig RuntimeConfig
	Task          Task
	Hooks         Hooks
	Priority      int
	Upstreams     Upstreams
}

type Upstream

type Upstream struct {
	JobName  string
	Tenant   tenant.Tenant
	Host     string
	TaskName string
}

type Upstreams

type Upstreams struct {
	HTTP      []*scheduler.HTTPUpstreams
	Upstreams []Upstream
}

func SetupUpstreams

func SetupUpstreams(upstreams scheduler.Upstreams, host string) Upstreams

func (Upstreams) Empty

func (u Upstreams) Empty() bool