scheduler

Documentation

Index

• type Job
  • func FromRow(row pgx.Row) (*Job, error)
• type JobRepository
  • func NewJobProviderRepository(pool *pgxpool.Pool) *JobRepository
  • func (j *JobRepository) GetAll(ctx context.Context, projectName tenant.ProjectName) ([]*scheduler.JobWithDetails, error)
  • func (j *JobRepository) GetJob(ctx context.Context, projectName tenant.ProjectName, jobName scheduler.JobName) (*scheduler.Job, error)
  • func (j *JobRepository) GetJobDetails(ctx context.Context, projectName tenant.ProjectName, jobName scheduler.JobName) (*scheduler.JobWithDetails, error)
• type JobRunRepository
  • func NewJobRunRepository(pool *pgxpool.Pool) *JobRunRepository
  • func (j *JobRunRepository) Create(ctx context.Context, t tenant.Tenant, jobName scheduler.JobName, ...) error
  • func (j *JobRunRepository) GetByID(ctx context.Context, id scheduler.JobRunID) (*scheduler.JobRun, error)
  • func (j *JobRunRepository) GetByScheduledAt(ctx context.Context, t tenant.Tenant, jobName scheduler.JobName, ...) (*scheduler.JobRun, error)
  • func (j *JobRunRepository) Update(ctx context.Context, jobRunID uuid.UUID, endTime time.Time, ...) error
  • func (j *JobRunRepository) UpdateMonitoring(ctx context.Context, jobRunID uuid.UUID, monitoringValues map[string]any) error
  • func (j *JobRunRepository) UpdateSLA(ctx context.Context, slaObjects []*scheduler.SLAObject) error
• type JobUpstreams
• type OperatorRunRepository
  • func NewOperatorRunRepository(pool *pgxpool.Pool) *OperatorRunRepository
  • func (o *OperatorRunRepository) CreateOperatorRun(ctx context.Context, name string, operatorType scheduler.OperatorType, ...) error
  • func (o *OperatorRunRepository) GetOperatorRun(ctx context.Context, name string, operatorType scheduler.OperatorType, ...) (*scheduler.OperatorRun, error)
  • func (o *OperatorRunRepository) UpdateOperatorRun(ctx context.Context, operatorType scheduler.OperatorType, ...) error
• type Retry
• type Schedule
• type Window

Types

type Job

type Job struct {
	ID          uuid.UUID
	Name        string
	Version     int
	Owner       string
	Description string
	Labels      map[string]string

	Schedule   json.RawMessage
	WindowSpec json.RawMessage

	Alert json.RawMessage

	StaticUpstreams pq.StringArray
	HTTPUpstreams   json.RawMessage

	TaskName   string
	TaskConfig map[string]string

	Hooks json.RawMessage

	Assets map[string]string

	Metadata json.RawMessage

	Destination string
	Sources     pq.StringArray

	ProjectName   string `json:"project_name"`
	NamespaceName string `json:"namespace_name"`

	CreatedAt time.Time
	UpdatedAt time.Time
	DeletedAt sql.NullTime
}

func FromRow

func FromRow(row pgx.Row) (*Job, error)

type JobRepository

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

func NewJobProviderRepository

func NewJobProviderRepository(pool *pgxpool.Pool) *JobRepository

func (*JobRepository) GetAll

func (j *JobRepository) GetAll(ctx context.Context, projectName tenant.ProjectName) ([]*scheduler.JobWithDetails, error)

func (*JobRepository) GetJob

func (j *JobRepository) GetJob(ctx context.Context, projectName tenant.ProjectName, jobName scheduler.JobName) (*scheduler.Job, error)

func (*JobRepository) GetJobDetails

func (j *JobRepository) GetJobDetails(ctx context.Context, projectName tenant.ProjectName, jobName scheduler.JobName) (*scheduler.JobWithDetails, error)

type JobRunRepository

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

func NewJobRunRepository

func NewJobRunRepository(pool *pgxpool.Pool) *JobRunRepository

func (*JobRunRepository) Create

func (j *JobRunRepository) Create(ctx context.Context, t tenant.Tenant, jobName scheduler.JobName, scheduledAt time.Time, slaDefinitionInSec int64) error

func (*JobRunRepository) GetByID

func (j *JobRunRepository) GetByID(ctx context.Context, id scheduler.JobRunID) (*scheduler.JobRun, error)

func (*JobRunRepository) GetByScheduledAt

func (j *JobRunRepository) GetByScheduledAt(ctx context.Context, t tenant.Tenant, jobName scheduler.JobName, scheduledAt time.Time) (*scheduler.JobRun, error)

func (*JobRunRepository) Update

func (j *JobRunRepository) Update(ctx context.Context, jobRunID uuid.UUID, endTime time.Time, status scheduler.State) error

func (*JobRunRepository) UpdateMonitoring

func (j *JobRunRepository) UpdateMonitoring(ctx context.Context, jobRunID uuid.UUID, monitoringValues map[string]any) error

func (*JobRunRepository) UpdateSLA

func (j *JobRunRepository) UpdateSLA(ctx context.Context, slaObjects []*scheduler.SLAObject) error

type JobUpstreams

type JobUpstreams struct {
	JobID                 uuid.UUID
	JobName               string
	ProjectName           string
	UpstreamJobID         uuid.UUID
	UpstreamJobName       sql.NullString
	UpstreamResourceUrn   sql.NullString
	UpstreamProjectName   sql.NullString
	UpstreamNamespaceName sql.NullString
	UpstreamTaskName      sql.NullString
	UpstreamHost          sql.NullString
	UpstreamType          string
	UpstreamState         string
	UpstreamExternal      sql.NullBool

	CreatedAt time.Time
	UpdatedAt time.Time
}

type OperatorRunRepository

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

func NewOperatorRunRepository

func NewOperatorRunRepository(pool *pgxpool.Pool) *OperatorRunRepository

func (*OperatorRunRepository) CreateOperatorRun

func (o *OperatorRunRepository) CreateOperatorRun(ctx context.Context, name string, operatorType scheduler.OperatorType, jobRunID uuid.UUID, startTime time.Time) error

func (*OperatorRunRepository) GetOperatorRun

func (o *OperatorRunRepository) GetOperatorRun(ctx context.Context, name string, operatorType scheduler.OperatorType, jobRunID uuid.UUID) (*scheduler.OperatorRun, error)

func (*OperatorRunRepository) UpdateOperatorRun

func (o *OperatorRunRepository) UpdateOperatorRun(ctx context.Context, operatorType scheduler.OperatorType, operatorRunID uuid.UUID, eventTime time.Time, state scheduler.State) error

type Retry

type Retry struct {
	Count              int   `json:"count"`
	Delay              int32 `json:"delay"`
	ExponentialBackoff bool  `json:"exponential_backoff"`
}

type Schedule

type Schedule struct {
	StartDate     time.Time
	EndDate       *time.Time
	Interval      string
	DependsOnPast bool `json:"depends_on_past"`
	CatchUp       bool `json:"catch_up"`
	Retry         *Retry
}

type Window

type Window struct {
	WindowSize       string
	WindowOffset     string
	WindowTruncateTo string
}