service

Documentation

Index

• Constants
• Variables
• type Engine
• type JobPluginService
  • func NewJobPluginService(pluginRepo PluginRepo, engine Engine, logger log.Logger) *JobPluginService
  • func (p JobPluginService) GenerateDestination(ctx context.Context, tnnt *tenant.WithDetails, task job.Task) (job.ResourceURN, error)
  • func (p JobPluginService) GenerateUpstreams(ctx context.Context, jobTenant *tenant.WithDetails, spec *job.Spec, ...) ([]job.ResourceURN, error)
  • func (p JobPluginService) Info(_ context.Context, taskName job.TaskName) (*plugin.Info, error)
• type JobRepository
• type JobService
  • func NewJobService(repo JobRepository, pluginService PluginService, ...) *JobService
  • func (j JobService) Add(ctx context.Context, jobTenant tenant.Tenant, specs []*job.Spec) error
  • func (j JobService) Delete(ctx context.Context, jobTenant tenant.Tenant, jobName job.Name, cleanFlag bool, ...) (affectedDownstream []job.FullName, err error)
  • func (j JobService) Get(ctx context.Context, jobTenant tenant.Tenant, jobName job.Name) (*job.Job, error)
  • func (j JobService) GetByFilter(ctx context.Context, filters ...filter.FilterOpt) ([]*job.Job, error)
  • func (j JobService) GetDownstream(ctx context.Context, subjectJob *job.Job, localJob bool) ([]*job.Downstream, error)
  • func (j JobService) GetJobBasicInfo(ctx context.Context, jobTenant tenant.Tenant, jobName job.Name, spec *job.Spec) (*job.Job, writer.BufferedLogger)
  • func (j JobService) GetTaskInfo(ctx context.Context, task job.Task) (*plugin.Info, error)
  • func (j JobService) GetUpstreamsToInspect(ctx context.Context, subjectJob *job.Job, localJob bool) ([]*job.Upstream, error)
  • func (j JobService) Refresh(ctx context.Context, projectName tenant.ProjectName, namespaceNames []string, ...) (err error)
  • func (j JobService) ReplaceAll(ctx context.Context, jobTenant tenant.Tenant, specs []*job.Spec, ...) error
  • func (j JobService) Update(ctx context.Context, jobTenant tenant.Tenant, specs []*job.Spec) error
  • func (j JobService) Validate(ctx context.Context, jobTenant tenant.Tenant, jobSpecs []*job.Spec, ...) error
• type PluginRepo
• type PluginService
• type TenantDetailsGetter
• type UpstreamResolver

Constants

const (
	ConcurrentTicketPerSec = 40
	ConcurrentLimit        = 600
)

const (
	TimeISOFormat = time.RFC3339
)

Variables

var (
	ErrUpstreamModNotFound = errors.New("upstream mod not found for plugin")
	ErrYamlModNotExist     = errors.New("yaml mod not found for plugin")
)

Types

type Engine

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

type JobPluginService

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

func NewJobPluginService

func NewJobPluginService(pluginRepo PluginRepo, engine Engine, logger log.Logger) *JobPluginService

func (JobPluginService) GenerateDestination

func (p JobPluginService) GenerateDestination(ctx context.Context, tnnt *tenant.WithDetails, task job.Task) (job.ResourceURN, error)

func (JobPluginService) GenerateUpstreams

func (p JobPluginService) GenerateUpstreams(ctx context.Context, jobTenant *tenant.WithDetails, spec *job.Spec, dryRun bool) ([]job.ResourceURN, error)

func (JobPluginService) Info

func (p JobPluginService) Info(_ context.Context, taskName job.TaskName) (*plugin.Info, error)

type JobRepository

type JobRepository interface {
	// TODO: remove `savedJobs` since the method's main purpose is to add, not to get
	Add(context.Context, []*job.Job) (addedJobs []*job.Job, err error)
	Update(context.Context, []*job.Job) (updatedJobs []*job.Job, err error)
	Delete(ctx context.Context, projectName tenant.ProjectName, jobName job.Name, cleanHistory bool) error

	GetByJobName(ctx context.Context, projectName tenant.ProjectName, jobName job.Name) (*job.Job, error)
	GetAllByResourceDestination(ctx context.Context, resourceDestination job.ResourceURN) ([]*job.Job, error)
	GetAllByTenant(ctx context.Context, jobTenant tenant.Tenant) ([]*job.Job, error)
	GetAllByProjectName(ctx context.Context, projectName tenant.ProjectName) ([]*job.Job, error)

	ResolveUpstreams(context.Context, tenant.ProjectName, []job.Name) (map[job.Name][]*job.Upstream, error)
	ReplaceUpstreams(context.Context, []*job.WithUpstream) error
	GetUpstreams(ctx context.Context, projectName tenant.ProjectName, jobName job.Name) ([]*job.Upstream, error)

	GetDownstreamByDestination(ctx context.Context, projectName tenant.ProjectName, destination job.ResourceURN) ([]*job.Downstream, error)
	GetDownstreamByJobName(ctx context.Context, projectName tenant.ProjectName, jobName job.Name) ([]*job.Downstream, error)
}

type JobService

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

func NewJobService

func NewJobService(repo JobRepository, pluginService PluginService, upstreamResolver UpstreamResolver, tenantDetailsGetter TenantDetailsGetter, logger log.Logger) *JobService

func (JobService) Add

func (j JobService) Add(ctx context.Context, jobTenant tenant.Tenant, specs []*job.Spec) error

func (JobService) Delete

func (j JobService) Delete(ctx context.Context, jobTenant tenant.Tenant, jobName job.Name, cleanFlag bool, forceFlag bool) (affectedDownstream []job.FullName, err error)

func (JobService) Get

func (j JobService) Get(ctx context.Context, jobTenant tenant.Tenant, jobName job.Name) (*job.Job, error)

func (JobService) GetByFilter

func (j JobService) GetByFilter(ctx context.Context, filters ...filter.FilterOpt) ([]*job.Job, error)

func (JobService) GetDownstream

func (j JobService) GetDownstream(ctx context.Context, subjectJob *job.Job, localJob bool) ([]*job.Downstream, error)

func (JobService) GetJobBasicInfo

func (j JobService) GetJobBasicInfo(ctx context.Context, jobTenant tenant.Tenant, jobName job.Name, spec *job.Spec) (*job.Job, writer.BufferedLogger)

func (JobService) GetTaskInfo

func (j JobService) GetTaskInfo(ctx context.Context, task job.Task) (*plugin.Info, error)

func (JobService) GetUpstreamsToInspect

func (j JobService) GetUpstreamsToInspect(ctx context.Context, subjectJob *job.Job, localJob bool) ([]*job.Upstream, error)

func (JobService) Refresh

func (j JobService) Refresh(ctx context.Context, projectName tenant.ProjectName, namespaceNames []string, jobNames []string, logWriter writer.LogWriter) (err error)

func (JobService) ReplaceAll

func (j JobService) ReplaceAll(ctx context.Context, jobTenant tenant.Tenant, specs []*job.Spec, jobNamesWithValidationError []job.Name, logWriter writer.LogWriter) error

func (JobService) Update

func (j JobService) Update(ctx context.Context, jobTenant tenant.Tenant, specs []*job.Spec) error

func (JobService) Validate

func (j JobService) Validate(ctx context.Context, jobTenant tenant.Tenant, jobSpecs []*job.Spec, logWriter writer.LogWriter) error

type PluginRepo

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

type PluginService

type PluginService interface {
	Info(context.Context, job.TaskName) (*plugin.Info, error)
	GenerateDestination(context.Context, *tenant.WithDetails, job.Task) (job.ResourceURN, error)
	GenerateUpstreams(ctx context.Context, jobTenant *tenant.WithDetails, spec *job.Spec, dryRun bool) ([]job.ResourceURN, error)
}

type TenantDetailsGetter

type TenantDetailsGetter interface {
	GetDetails(ctx context.Context, jobTenant tenant.Tenant) (*tenant.WithDetails, error)
}

type UpstreamResolver

type UpstreamResolver interface {
	BulkResolve(ctx context.Context, projectName tenant.ProjectName, jobs []*job.Job, logWriter writer.LogWriter) (jobWithUpstreams []*job.WithUpstream, err error)
	Resolve(ctx context.Context, subjectJob *job.Job, logWriter writer.LogWriter) ([]*job.Upstream, error)
}