service

Documentation

Index

• Constants
• type AssetCompiler
• type FilesCompiler
• type InputCompiler
  • func NewJobInputCompiler(tenantService TenantService, compiler TemplateCompiler, ...) *InputCompiler
  • func (i InputCompiler) Compile(ctx context.Context, job *scheduler.Job, config scheduler.RunConfig, ...) (*scheduler.ExecutorInput, error)
• type JobInputCompiler
• type JobRepository
• type JobRunAssetsCompiler
  • func NewJobAssetsCompiler(engine FilesCompiler, pluginRepo PluginRepo) *JobRunAssetsCompiler
  • func (c *JobRunAssetsCompiler) CompileJobRunAssets(ctx context.Context, job *scheduler.Job, systemEnvVars map[string]string, ...) (map[string]string, error)
• type JobRunRepository
• type JobRunService
  • func NewJobRunService(logger log.Logger, jobRepo JobRepository, jobRunRepo JobRunRepository, ...) *JobRunService
  • func (s JobRunService) GetJobRuns(ctx context.Context, projectName tenant.ProjectName, jobName scheduler.JobName, ...) ([]*scheduler.JobRunStatus, error)
  • func (s JobRunService) JobRunInput(ctx context.Context, projectName tenant.ProjectName, jobName scheduler.JobName, ...) (*scheduler.ExecutorInput, error)
  • func (s JobRunService) UpdateJobState(ctx context.Context, event scheduler.Event) error
  • func (s JobRunService) UploadToScheduler(ctx context.Context, projectName tenant.ProjectName) error
• type Notifier
• type NotifyService
  • func NewNotifyService(l log.Logger, jobRepo JobRepository, tenantService TenantService, ...) *NotifyService
  • func (n *NotifyService) Close() error
  • func (n NotifyService) Push(ctx context.Context, event scheduler.Event) error
• type OperatorRunRepository
• type PluginRepo
• type PriorityResolver
• type Scheduler
• type TemplateCompiler
• type TenantService

Constants

const (
	SecretsStringToMatch = ".secret."

	TimeISOFormat = time.RFC3339
)

const (
	NotificationSchemeSlack     = "slack"
	NotificationSchemePagerDuty = "pagerduty"
)

Types

type AssetCompiler

type AssetCompiler interface {
	CompileJobRunAssets(ctx context.Context, job *scheduler.Job, systemEnvVars map[string]string, scheduledAt time.Time, contextForTask map[string]interface{}) (map[string]string, error)
}

type FilesCompiler

type FilesCompiler interface {
	Compile(fileMap map[string]string, context map[string]any) (map[string]string, error)
}

type InputCompiler

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

func NewJobInputCompiler

func NewJobInputCompiler(tenantService TenantService, compiler TemplateCompiler, assetCompiler AssetCompiler) *InputCompiler

func (InputCompiler) Compile

func (i InputCompiler) Compile(ctx context.Context, job *scheduler.Job, config scheduler.RunConfig, executedAt time.Time) (*scheduler.ExecutorInput, error)

type JobInputCompiler

type JobInputCompiler interface {
	Compile(ctx context.Context, job *scheduler.Job, config scheduler.RunConfig, executedAt time.Time) (*scheduler.ExecutorInput, error)
}

type JobRepository

type JobRepository interface {
	GetJob(ctx context.Context, name tenant.ProjectName, jobName scheduler.JobName) (*scheduler.Job, error)
	GetJobDetails(ctx context.Context, projectName tenant.ProjectName, jobName scheduler.JobName) (*scheduler.JobWithDetails, error)
	GetAll(ctx context.Context, projectName tenant.ProjectName) ([]*scheduler.JobWithDetails, error)
}

type JobRunAssetsCompiler

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

func NewJobAssetsCompiler

func NewJobAssetsCompiler(engine FilesCompiler, pluginRepo PluginRepo) *JobRunAssetsCompiler

func (*JobRunAssetsCompiler) CompileJobRunAssets

func (c *JobRunAssetsCompiler) CompileJobRunAssets(ctx context.Context, job *scheduler.Job, systemEnvVars map[string]string, scheduledAt time.Time, contextForTask map[string]interface{}) (map[string]string, error)

type JobRunRepository

type JobRunRepository interface {
	GetByID(ctx context.Context, id scheduler.JobRunID) (*scheduler.JobRun, error)
	GetByScheduledAt(ctx context.Context, tenant tenant.Tenant, name scheduler.JobName, scheduledAt time.Time) (*scheduler.JobRun, error)
	Create(ctx context.Context, tenant tenant.Tenant, name scheduler.JobName, scheduledAt time.Time, slaDefinitionInSec int64) error
	Update(ctx context.Context, jobRunID uuid.UUID, endTime time.Time, jobRunStatus scheduler.State) error
	UpdateSLA(ctx context.Context, slaObjects []*scheduler.SLAObject) error
	UpdateMonitoring(ctx context.Context, jobRunID uuid.UUID, monitoring map[string]any) error
}

type JobRunService

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

func NewJobRunService

func NewJobRunService(logger log.Logger, jobRepo JobRepository, jobRunRepo JobRunRepository,
	operatorRunRepo OperatorRunRepository, scheduler Scheduler, resolver PriorityResolver, compiler JobInputCompiler) *JobRunService

func (JobRunService) GetJobRuns

func (s JobRunService) GetJobRuns(ctx context.Context, projectName tenant.ProjectName, jobName scheduler.JobName, criteria *scheduler.JobRunsCriteria) ([]*scheduler.JobRunStatus, error)

func (JobRunService) JobRunInput

func (s JobRunService) JobRunInput(ctx context.Context, projectName tenant.ProjectName, jobName scheduler.JobName, config scheduler.RunConfig) (*scheduler.ExecutorInput, error)

func (JobRunService) UpdateJobState

func (s JobRunService) UpdateJobState(ctx context.Context, event scheduler.Event) error

func (JobRunService) UploadToScheduler

func (s JobRunService) UploadToScheduler(ctx context.Context, projectName tenant.ProjectName) error

type Notifier

type Notifier interface {
	io.Closer
	Notify(ctx context.Context, attr scheduler.NotifyAttrs) error
}

type NotifyService

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

func NewNotifyService

func NewNotifyService(l log.Logger, jobRepo JobRepository, tenantService TenantService, notifyChan map[string]Notifier) *NotifyService

func (*NotifyService) Close

func (n *NotifyService) Close() error

func (NotifyService) Push

func (n NotifyService) Push(ctx context.Context, event scheduler.Event) error

type OperatorRunRepository

type OperatorRunRepository interface {
	GetOperatorRun(ctx context.Context, operatorName string, operator scheduler.OperatorType, jobRunID uuid.UUID) (*scheduler.OperatorRun, error)
	CreateOperatorRun(ctx context.Context, operatorName string, operator scheduler.OperatorType, jobRunID uuid.UUID, startTime time.Time) error
	UpdateOperatorRun(ctx context.Context, operator scheduler.OperatorType, jobRunID uuid.UUID, eventTime time.Time, state scheduler.State) error
}

type PluginRepo

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

type PriorityResolver

type PriorityResolver interface {
	Resolve(context.Context, []*scheduler.JobWithDetails) error
}

type Scheduler

type Scheduler interface {
	GetJobRuns(ctx context.Context, t tenant.Tenant, criteria *scheduler.JobRunsCriteria, jobCron *cron.ScheduleSpec) ([]*scheduler.JobRunStatus, error)
	DeployJobs(ctx context.Context, t tenant.Tenant, jobs []*scheduler.JobWithDetails) error
	ListJobs(ctx context.Context, t tenant.Tenant) ([]string, error)
	DeleteJobs(ctx context.Context, t tenant.Tenant, jobsToDelete []string) error
}

type TemplateCompiler

type TemplateCompiler interface {
	Compile(templateMap map[string]string, context map[string]any) (map[string]string, error)
}

type TenantService

type TenantService interface {
	GetDetails(ctx context.Context, tnnt tenant.Tenant) (*tenant.WithDetails, error)
	GetSecrets(ctx context.Context, tnnt tenant.Tenant) ([]*tenant.PlainTextSecret, error)
}