task

Documentation

Index

• Constants
• func Run(ctx context.Context, rediscli *redis.Client, db *database.Database, ...) error
• type ApplicationTasker
  • func MustNewApplicationTasker(db *database.Database, gitp *git.SimpleLocalProvider, argo *argo.Client, ...) *ApplicationTasker
  • func (t *ApplicationTasker) ProvideFuntions() map[string]interface{}
• type ClusterSyncTasker
  • func (s *ClusterSyncTasker) Crontasks() map[string]Task
  • func (t *ClusterSyncTasker) ProvideFuntions() map[string]interface{}
  • func (t *ClusterSyncTasker) Sync(ctx context.Context) error
• type CronTask
• type CronTasker
• type HelmSyncTasker
  • func (s *HelmSyncTasker) Crontasks() map[string]Task
  • func (t *HelmSyncTasker) ProvideFuntions() map[string]interface{}
  • func (t *HelmSyncTasker) SyncCharts(ctx context.Context) error
• type ProcessorContext
  • func (p *ProcessorContext) RegisterTasker(taskers ...Tasker) error
  • func (p *ProcessorContext) Run(ctx context.Context) error
  • func (p *ProcessorContext) RunCronTasksWithLock(ctx context.Context) error
• type SampleTasker
  • func (s *SampleTasker) Crontasks() map[string]Task
  • func (s *SampleTasker) ProvideFuntions() map[string]interface{}
• type Task
• type TaskArchiverTasker
  • func NewTaskArchiverTasker(databse *database.Database, redis *redis.Client) *TaskArchiverTasker
  • func (t *TaskArchiverTasker) ArchiveOutdated(ctx context.Context) error
  • func (t *TaskArchiverTasker) Crontasks() map[string]Task
  • func (t *TaskArchiverTasker) ProvideFuntions() map[string]interface{}
  • func (t *TaskArchiverTasker) RemoveOffline(ctx context.Context) error
• type Tasker

Constants

const RemoveConsumerDuration = 5 * time.Minute

const TaskFunction_ArchiveTasks = "task-archive"

const TaskFunction_ClusterSync = "cluster-sync"

const TaskFunction_HelmSyncCharts = "helm-sync-charts"

Functions

func Run

func Run(ctx context.Context, rediscli *redis.Client,
	db *database.Database,
	gitp *git.SimpleLocalProvider,
	argocd *argo.Client,
	helmOptions *helm.Options,
	agents *agents.ClientSet,
) error

Types

type ApplicationTasker

type ApplicationTasker struct {
	*application.ApplicationProcessor
}

func MustNewApplicationTasker

func MustNewApplicationTasker(db *database.Database, gitp *git.SimpleLocalProvider, argo *argo.Client, redis *redis.Client, agents *agents.ClientSet) *ApplicationTasker

func (*ApplicationTasker) ProvideFuntions

func (t *ApplicationTasker) ProvideFuntions() map[string]interface{}

type ClusterSyncTasker

type ClusterSyncTasker struct {
	DB *database.Database
	// contains filtered or unexported fields
}

func (*ClusterSyncTasker) Crontasks

func (s *ClusterSyncTasker) Crontasks() map[string]Task

func (*ClusterSyncTasker) ProvideFuntions

func (t *ClusterSyncTasker) ProvideFuntions() map[string]interface{}

func (*ClusterSyncTasker) Sync

func (t *ClusterSyncTasker) Sync(ctx context.Context) error

type CronTask

type CronTask struct {
	CronExp string
	Task    Task
}

type CronTasker

type CronTasker interface {
	Crontasks() map[string]Task // cron表达式 -> 任务
}

某些任务处理本身会有定时任务，可以实现该接口

type HelmSyncTasker

type HelmSyncTasker struct {
	DB           *database.Database
	ChartRepoUrl string
}

func (*HelmSyncTasker) Crontasks

func (s *HelmSyncTasker) Crontasks() map[string]Task

func (*HelmSyncTasker) ProvideFuntions

func (t *HelmSyncTasker) ProvideFuntions() map[string]interface{}

func (*HelmSyncTasker) SyncCharts

func (t *HelmSyncTasker) SyncCharts(ctx context.Context) error

type ProcessorContext

type ProcessorContext struct {
	Logger logr.Logger
	// contains filtered or unexported fields
}

func (*ProcessorContext) RegisterTasker

func (p *ProcessorContext) RegisterTasker(taskers ...Tasker) error

func (*ProcessorContext) Run

func (p *ProcessorContext) Run(ctx context.Context) error

func (*ProcessorContext) RunCronTasksWithLock

func (p *ProcessorContext) RunCronTasksWithLock(ctx context.Context) error

由于worker是多副本的，且crontask 只能在 worker上运行cron。 为了避免多个worker都执行，使用redis 锁选择一个worker来触发这些crontask

type SampleTasker

type SampleTasker struct{}

func (*SampleTasker) Crontasks

func (s *SampleTasker) Crontasks() map[string]Task

func (*SampleTasker) ProvideFuntions

func (s *SampleTasker) ProvideFuntions() map[string]interface{}

type Task

type Task = workflow.Task

type TaskArchiverTasker

type TaskArchiverTasker struct {
	Databse *database.Database

	Redis *redis.Client
	// contains filtered or unexported fields
}

用于转移超过时间的任务记录至database

func NewTaskArchiverTasker

func NewTaskArchiverTasker(databse *database.Database, redis *redis.Client) *TaskArchiverTasker

func (*TaskArchiverTasker) ArchiveOutdated

func (t *TaskArchiverTasker) ArchiveOutdated(ctx context.Context) error

func (*TaskArchiverTasker) Crontasks

func (t *TaskArchiverTasker) Crontasks() map[string]Task

func (*TaskArchiverTasker) ProvideFuntions

func (t *TaskArchiverTasker) ProvideFuntions() map[string]interface{}

func (*TaskArchiverTasker) RemoveOffline

func (t *TaskArchiverTasker) RemoveOffline(ctx context.Context) error

RemoveOffline 删除长时间不活跃的worker正在处理的任务

type Tasker

type Tasker interface {
	ProvideFuntions() map[string]interface{}
}