scheduler

package
v0.23.4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 20, 2025 License: Apache-2.0 Imports: 21 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func FromChangelogRow added in v0.22.12 

func FromChangelogRow(row pgx.Row) (*changelog, error)

func GetStore added in v0.22.14 

func GetStore(urn string) (string, error)

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
	Webhook json.RawMessage

	StaticUpstreams pq.StringArray
	HTTPUpstreams   json.RawMessage

	TaskName   string
	TaskConfig json.RawMessage

	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) GetAllResolvedUpstreams added in v0.22.4 

func (j *JobRepository) GetAllResolvedUpstreams(ctx context.Context) (map[scheduler.JobName][]scheduler.JobName, error)

func (*JobRepository) GetChangelogs added in v0.22.12 

func (j *JobRepository) GetChangelogs(ctx context.Context, filter scheduler.ChangelogFilter) ([]*scheduler.Changelog, 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)

func (*JobRepository) GetJobs added in v0.8.0 

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

func (*JobRepository) GetJobsByLabels added in v0.22.4 

func (j *JobRepository) GetJobsByLabels(ctx context.Context, projectName tenant.ProjectName, labels map[string]string) ([]*scheduler.JobWithDetails, error)

func (*JobRepository) GetSummaryByNames added in v0.22.4 

func (j *JobRepository) GetSummaryByNames(ctx context.Context, jobNames []scheduler.JobName) (map[scheduler.JobName]*scheduler.JobSummary, 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, window interval.Interval, 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) GetByScheduledTimes added in v0.9.7 

func (j *JobRunRepository) GetByScheduledTimes(ctx context.Context, t tenant.Tenant, jobName scheduler.JobName, scheduleTimes []time.Time) ([]*scheduler.JobRun, error)

func (*JobRunRepository) GetLatestRun added in v0.17.0 

func (j *JobRunRepository) GetLatestRun(ctx context.Context, project tenant.ProjectName, jobName scheduler.JobName, runState *scheduler.State) (*scheduler.JobRun, error)

func (*JobRunRepository) GetPercentileDurationByJobNames added in v0.22.4 

func (j *JobRunRepository) GetPercentileDurationByJobNames(ctx context.Context, jobNames []scheduler.JobName, operators map[string][]string, referenceTime time.Time, lastNRuns, percentile int) (map[scheduler.JobName]*time.Duration, error)

func (*JobRunRepository) GetRunSummaryByIdentifiers added in v0.22.4 

func (j *JobRunRepository) GetRunSummaryByIdentifiers(ctx context.Context, identifiers []scheduler.JobRunIdentifier) ([]*scheduler.JobRunSummary, error)

func (*JobRunRepository) GetRunsByInterval added in v0.19.0 

func (j *JobRunRepository) GetRunsByInterval(ctx context.Context, project tenant.ProjectName, jobName scheduler.JobName, interval interval.Interval) ([]*scheduler.JobRun, error)

func (*JobRunRepository) GetRunsByTimeRange added in v0.17.0 

func (j *JobRunRepository) GetRunsByTimeRange(ctx context.Context, project tenant.ProjectName, jobName scheduler.JobName, runState *scheduler.State, since, until 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, jobName scheduler.JobName, projectName tenant.ProjectName, scheduleTimes []time.Time) error

func (*JobRunRepository) UpdateState added in v0.7.0 

func (j *JobRunRepository) UpdateState(ctx context.Context, jobRunID uuid.UUID, status scheduler.State) 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 Metadata added in v0.8.3 

type Metadata struct {
	Resource  *MetadataResource
	Scheduler map[string]string
}

type MetadataResource added in v0.8.3 

type MetadataResource struct {
	Request *MetadataResourceConfig
	Limit   *MetadataResourceConfig
}

type MetadataResourceConfig added in v0.8.3 

type MetadataResourceConfig struct {
	CPU    string
	Memory string
}

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 OperatorsSLA added in v0.22.0 

type OperatorsSLA struct {
	ID                uuid.UUID
	JobName           string
	ProjectName       string
	OperatorName      string
	RunID             string
	OperatorType      string
	SLATime           time.Time
	AlertTag          string
	ScheduledAt       time.Time
	OperatorStartTime time.Time

	WorkerSignature string
	WorkerLockUntil time.Time

	CreatedAt time.Time
	UpdatedAt time.Time
}

func SLAFromRow added in v0.22.0 

func SLAFromRow(row pgx.Row) (*OperatorsSLA, error)

type ReplayRepository added in v0.7.0 

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

func NewReplayRepository added in v0.7.0 

func NewReplayRepository(db *pgxpool.Pool) *ReplayRepository

func (ReplayRepository) AcquireReplayRequest added in v0.17.0 

func (r ReplayRepository) AcquireReplayRequest(ctx context.Context, replayID uuid.UUID, unhandledClassifierDuration time.Duration) error

func (ReplayRepository) CancelReplayRequest added in v0.17.0 

func (r ReplayRepository) CancelReplayRequest(ctx context.Context, id uuid.UUID, message string) error

func (ReplayRepository) GetReplayByFilters added in v0.16.1 

func (r ReplayRepository) GetReplayByFilters(ctx context.Context, projectName tenant.ProjectName, filters ...filter.FilterOpt) ([]*scheduler.ReplayWithRun, error)

func (ReplayRepository) GetReplayByID added in v0.7.0 

func (r ReplayRepository) GetReplayByID(ctx context.Context, replayID uuid.UUID) (*scheduler.ReplayWithRun, error)

func (ReplayRepository) GetReplayJobConfig added in v0.7.0 

func (r ReplayRepository) GetReplayJobConfig(ctx context.Context, jobTenant tenant.Tenant, jobName scheduler.JobName, scheduledAt time.Time) (map[string]string, error)

func (ReplayRepository) GetReplayRequestByID added in v0.17.0 

func (r ReplayRepository) GetReplayRequestByID(ctx context.Context, replayID uuid.UUID) (*scheduler.Replay, error)

func (ReplayRepository) GetReplayRequestsByStatus added in v0.7.0 

func (r ReplayRepository) GetReplayRequestsByStatus(ctx context.Context, statusList []scheduler.ReplayState) ([]*scheduler.Replay, error)

func (ReplayRepository) GetReplaysByProject added in v0.7.0 

func (r ReplayRepository) GetReplaysByProject(ctx context.Context, projectName tenant.ProjectName, dayLimits int) ([]*scheduler.Replay, error)

func (ReplayRepository) RegisterReplay added in v0.7.0 

func (r ReplayRepository) RegisterReplay(ctx context.Context, replay *scheduler.Replay, runs []*scheduler.JobRunStatus) (uuid.UUID, error)

func (ReplayRepository) ScanAbandonedReplayRequests added in v0.17.0 

func (r ReplayRepository) ScanAbandonedReplayRequests(ctx context.Context, unhandledClassifierDuration time.Duration) ([]*scheduler.Replay, error)

func (ReplayRepository) UpdateReplay added in v0.7.0 

func (r ReplayRepository) UpdateReplay(ctx context.Context, id uuid.UUID, replayStatus scheduler.ReplayState, runs []*scheduler.JobRunStatus, message string) error

func (ReplayRepository) UpdateReplayHeartbeat added in v0.17.2 

func (r ReplayRepository) UpdateReplayHeartbeat(ctx context.Context, id uuid.UUID) error

func (ReplayRepository) UpdateReplayRuns added in v0.17.0 

func (r ReplayRepository) UpdateReplayRuns(ctx context.Context, id uuid.UUID, runs []*scheduler.JobRunStatus) error

func (ReplayRepository) UpdateReplayStatus added in v0.7.0 

func (r ReplayRepository) UpdateReplayStatus(ctx context.Context, id uuid.UUID, replayStatus scheduler.ReplayState, message string) error

type Retry 

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

type SLARepository added in v0.22.0 

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

func NewSLARepository added in v0.22.0 

func NewSLARepository(pool *pgxpool.Pool) *SLARepository

func (*SLARepository) FinishSLA added in v0.22.0 

func (s *SLARepository) FinishSLA(ctx context.Context, projectName tenant.ProjectName, jobName, operatorName, operatorType, runID string, operatorEndTime time.Time) error

func (*SLARepository) GetExpiredSLAsForProcessing added in v0.22.0 

func (s *SLARepository) GetExpiredSLAsForProcessing(ctx context.Context, signature string, processingDuration time.Duration) ([]*scheduler.OperatorsSLA, error)

func (*SLARepository) GetPredictedSLAJobNamesWithinTimeRange added in v0.23.2 

func (s *SLARepository) GetPredictedSLAJobNamesWithinTimeRange(ctx context.Context, from, to time.Time) ([]scheduler.JobName, error)

func (*SLARepository) RegisterSLA added in v0.22.0 

func (s *SLARepository) RegisterSLA(ctx context.Context, projectName tenant.ProjectName, jobName, operatorName, operatorType, runID string, slaTime time.Time, alertTag string, scheduledAt, operatorStartTime time.Time) error

func (*SLARepository) RemoveProcessedSLA added in v0.22.0 

func (s *SLARepository) RemoveProcessedSLA(ctx context.Context, slaID uuid.UUID) error

func (*SLARepository) StorePredictedSLABreach added in v0.22.8 

func (s *SLARepository) StorePredictedSLABreach(ctx context.Context, jobTargetName, jobCauseName scheduler.JobName, jobScheduledAt time.Time, cause string, referenceTime time.Time, config map[string]interface{}, lineages []interface{}) error

func (*SLARepository) UpdateSLA added in v0.22.0 

func (s *SLARepository) UpdateSLA(ctx context.Context, projectName tenant.ProjectName, jobName, operatorName, operatorType, runID string, slaTime, operatorStartTime time.Time) error

type Schedule 

type Schedule struct {
	StartDate     time.Time
	EndDate       *time.Time
	Interval      string
	DependsOnPast bool
	CatchUp       bool
	Retry         *Retry
}

type ThirdPartyUpstream added in v0.22.14 

type ThirdPartyUpstream struct {
	JobID                        uuid.UUID
	JobName                      string
	ProjectName                  string
	UpstreamThirdPartyType       string
	UpstreamThirdPartyIdentifier string
	UpstreamThirdPartyConfig     map[string]string
	CreatedAt                    time.Time
}

type Window 

type Window struct {
	WindowSize       string `json:",omitempty"`
	WindowShiftBy    string `json:",omitempty"`
	WindowTruncateTo string `json:",omitempty"`
	WindowLocation   string `json:",omitempty"`
	WindowOffset     string `json:",omitempty"`
	Preset           string `json:",omitempty"`
	Type             string
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.