diff options
Diffstat (limited to 'services/cron')
| -rw-r--r-- | services/cron/cron.go | 94 | ||||
| -rw-r--r-- | services/cron/setting.go | 87 | ||||
| -rw-r--r-- | services/cron/tasks.go | 169 | ||||
| -rw-r--r-- | services/cron/tasks_basic.go | 141 | ||||
| -rw-r--r-- | services/cron/tasks_extended.go | 166 |
5 files changed, 657 insertions, 0 deletions
diff --git a/services/cron/cron.go b/services/cron/cron.go new file mode 100644 index 0000000000..ae309bd866 --- /dev/null +++ b/services/cron/cron.go @@ -0,0 +1,94 @@ +// Copyright 2014 The Gogs Authors. All rights reserved. +// Copyright 2019 The Gitea Authors. All rights reserved. +// Use of this source code is governed by a MIT-style +// license that can be found in the LICENSE file. + +package cron + +import ( + "context" + "time" + + "code.gitea.io/gitea/modules/graceful" + "code.gitea.io/gitea/modules/sync" + + "github.com/gogs/cron" +) + +var c = cron.New() + +// Prevent duplicate running tasks. +var taskStatusTable = sync.NewStatusTable() + +// NewContext begins cron tasks +// Each cron task is run within the shutdown context as a running server +// AtShutdown the cron server is stopped +func NewContext() { + initBasicTasks() + initExtendedTasks() + + lock.Lock() + for _, task := range tasks { + if task.IsEnabled() && task.DoRunAtStart() { + go task.Run() + } + } + + c.Start() + started = true + lock.Unlock() + graceful.GetManager().RunAtShutdown(context.Background(), func() { + c.Stop() + lock.Lock() + started = false + lock.Unlock() + }) + +} + +// TaskTableRow represents a task row in the tasks table +type TaskTableRow struct { + Name string + Spec string + Next time.Time + Prev time.Time + ExecTimes int64 +} + +// TaskTable represents a table of tasks +type TaskTable []*TaskTableRow + +// ListTasks returns all running cron tasks. +func ListTasks() TaskTable { + entries := c.Entries() + eMap := map[string]*cron.Entry{} + for _, e := range entries { + eMap[e.Description] = e + } + lock.Lock() + defer lock.Unlock() + tTable := make([]*TaskTableRow, 0, len(tasks)) + for _, task := range tasks { + spec := "-" + var ( + next time.Time + prev time.Time + ) + if e, ok := eMap[task.Name]; ok { + spec = e.Spec + next = e.Next + prev = e.Prev + } + task.lock.Lock() + tTable = append(tTable, &TaskTableRow{ + Name: task.Name, + Spec: spec, + Next: next, + Prev: prev, + ExecTimes: task.ExecTimes, + }) + task.lock.Unlock() + } + + return tTable +} diff --git a/services/cron/setting.go b/services/cron/setting.go new file mode 100644 index 0000000000..d55e5b60ad --- /dev/null +++ b/services/cron/setting.go @@ -0,0 +1,87 @@ +// Copyright 2020 The Gitea Authors. All rights reserved. +// Use of this source code is governed by a MIT-style +// license that can be found in the LICENSE file. + +package cron + +import ( + "time" + + "code.gitea.io/gitea/models" + "github.com/unknwon/i18n" +) + +// Config represents a basic configuration interface that cron task +type Config interface { + IsEnabled() bool + DoRunAtStart() bool + GetSchedule() string + FormatMessage(name, status string, doer *models.User, args ...interface{}) string + DoNoticeOnSuccess() bool +} + +// BaseConfig represents the basic config for a Cron task +type BaseConfig struct { + Enabled bool + RunAtStart bool + Schedule string + NoSuccessNotice bool +} + +// OlderThanConfig represents a cron task with OlderThan setting +type OlderThanConfig struct { + BaseConfig + OlderThan time.Duration +} + +// UpdateExistingConfig represents a cron task with UpdateExisting setting +type UpdateExistingConfig struct { + BaseConfig + UpdateExisting bool +} + +// CleanupHookTaskConfig represents a cron task with settings to cleanup hook_task +type CleanupHookTaskConfig struct { + BaseConfig + CleanupType string + OlderThan time.Duration + NumberToKeep int +} + +// GetSchedule returns the schedule for the base config +func (b *BaseConfig) GetSchedule() string { + return b.Schedule +} + +// IsEnabled returns the enabled status for the config +func (b *BaseConfig) IsEnabled() bool { + return b.Enabled +} + +// DoRunAtStart returns whether the task should be run at the start +func (b *BaseConfig) DoRunAtStart() bool { + return b.RunAtStart +} + +// DoNoticeOnSuccess returns whether a success notice should be posted +func (b *BaseConfig) DoNoticeOnSuccess() bool { + return !b.NoSuccessNotice +} + +// FormatMessage returns a message for the task +func (b *BaseConfig) FormatMessage(name, status string, doer *models.User, args ...interface{}) string { + realArgs := make([]interface{}, 0, len(args)+2) + realArgs = append(realArgs, i18n.Tr("en-US", "admin.dashboard."+name)) + if doer == nil { + realArgs = append(realArgs, "(Cron)") + } else { + realArgs = append(realArgs, doer.Name) + } + if len(args) > 0 { + realArgs = append(realArgs, args...) + } + if doer == nil || (doer.ID == -1 && doer.Name == "(Cron)") { + return i18n.Tr("en-US", "admin.dashboard.cron."+status, realArgs...) + } + return i18n.Tr("en-US", "admin.dashboard.task."+status, realArgs...) +} diff --git a/services/cron/tasks.go b/services/cron/tasks.go new file mode 100644 index 0000000000..56c363e0b8 --- /dev/null +++ b/services/cron/tasks.go @@ -0,0 +1,169 @@ +// Copyright 2020 The Gitea Authors. All rights reserved. +// Use of this source code is governed by a MIT-style +// license that can be found in the LICENSE file. + +package cron + +import ( + "context" + "fmt" + "reflect" + "sync" + + "code.gitea.io/gitea/models" + "code.gitea.io/gitea/models/db" + "code.gitea.io/gitea/modules/graceful" + "code.gitea.io/gitea/modules/log" + "code.gitea.io/gitea/modules/process" + "code.gitea.io/gitea/modules/setting" +) + +var lock = sync.Mutex{} +var started = false +var tasks = []*Task{} +var tasksMap = map[string]*Task{} + +// Task represents a Cron task +type Task struct { + lock sync.Mutex + Name string + config Config + fun func(context.Context, *models.User, Config) error + ExecTimes int64 +} + +// DoRunAtStart returns if this task should run at the start +func (t *Task) DoRunAtStart() bool { + return t.config.DoRunAtStart() +} + +// IsEnabled returns if this task is enabled as cron task +func (t *Task) IsEnabled() bool { + return t.config.IsEnabled() +} + +// GetConfig will return a copy of the task's config +func (t *Task) GetConfig() Config { + if reflect.TypeOf(t.config).Kind() == reflect.Ptr { + // Pointer: + return reflect.New(reflect.ValueOf(t.config).Elem().Type()).Interface().(Config) + } + // Not pointer: + return reflect.New(reflect.TypeOf(t.config)).Elem().Interface().(Config) +} + +// Run will run the task incrementing the cron counter with no user defined +func (t *Task) Run() { + t.RunWithUser(&models.User{ + ID: -1, + Name: "(Cron)", + LowerName: "(cron)", + }, t.config) +} + +// RunWithUser will run the task incrementing the cron counter at the time with User +func (t *Task) RunWithUser(doer *models.User, config Config) { + if !taskStatusTable.StartIfNotRunning(t.Name) { + return + } + t.lock.Lock() + if config == nil { + config = t.config + } + t.ExecTimes++ + t.lock.Unlock() + defer func() { + taskStatusTable.Stop(t.Name) + if err := recover(); err != nil { + // Recover a panic within the + combinedErr := fmt.Errorf("%s\n%s", err, log.Stack(2)) + log.Error("PANIC whilst running task: %s Value: %v", t.Name, combinedErr) + } + }() + graceful.GetManager().RunWithShutdownContext(func(baseCtx context.Context) { + ctx, cancel := context.WithCancel(baseCtx) + defer cancel() + pm := process.GetManager() + pid := pm.Add(config.FormatMessage(t.Name, "process", doer), cancel) + defer pm.Remove(pid) + if err := t.fun(ctx, doer, config); err != nil { + if db.IsErrCancelled(err) { + message := err.(db.ErrCancelled).Message + if err := models.CreateNotice(models.NoticeTask, config.FormatMessage(t.Name, "aborted", doer, message)); err != nil { + log.Error("CreateNotice: %v", err) + } + return + } + if err := models.CreateNotice(models.NoticeTask, config.FormatMessage(t.Name, "error", doer, err)); err != nil { + log.Error("CreateNotice: %v", err) + } + return + } + if config.DoNoticeOnSuccess() { + if err := models.CreateNotice(models.NoticeTask, config.FormatMessage(t.Name, "finished", doer)); err != nil { + log.Error("CreateNotice: %v", err) + } + } + }) +} + +// GetTask gets the named task +func GetTask(name string) *Task { + lock.Lock() + defer lock.Unlock() + log.Info("Getting %s in %v", name, tasksMap[name]) + + return tasksMap[name] +} + +// RegisterTask allows a task to be registered with the cron service +func RegisterTask(name string, config Config, fun func(context.Context, *models.User, Config) error) error { + log.Debug("Registering task: %s", name) + _, err := setting.GetCronSettings(name, config) + if err != nil { + log.Error("Unable to register cron task with name: %s Error: %v", name, err) + return err + } + + task := &Task{ + Name: name, + config: config, + fun: fun, + } + lock.Lock() + locked := true + defer func() { + if locked { + lock.Unlock() + } + }() + if _, has := tasksMap[task.Name]; has { + log.Error("A task with this name: %s has already been registered", name) + return fmt.Errorf("duplicate task with name: %s", task.Name) + } + + if config.IsEnabled() { + // We cannot use the entry return as there is no way to lock it + if _, err = c.AddJob(name, config.GetSchedule(), task); err != nil { + log.Error("Unable to register cron task with name: %s Error: %v", name, err) + return err + } + } + + tasks = append(tasks, task) + tasksMap[task.Name] = task + if started && config.IsEnabled() && config.DoRunAtStart() { + lock.Unlock() + locked = false + task.Run() + } + + return nil +} + +// RegisterTaskFatal will register a task but if there is an error log.Fatal +func RegisterTaskFatal(name string, config Config, fun func(context.Context, *models.User, Config) error) { + if err := RegisterTask(name, config, fun); err != nil { + log.Fatal("Unable to register cron task %s Error: %v", name, err) + } +} diff --git a/services/cron/tasks_basic.go b/services/cron/tasks_basic.go new file mode 100644 index 0000000000..a42c031b36 --- /dev/null +++ b/services/cron/tasks_basic.go @@ -0,0 +1,141 @@ +// Copyright 2020 The Gitea Authors. All rights reserved. +// Use of this source code is governed by a MIT-style +// license that can be found in the LICENSE file. + +package cron + +import ( + "context" + "time" + + "code.gitea.io/gitea/models" + "code.gitea.io/gitea/models/webhook" + "code.gitea.io/gitea/modules/migrations" + repository_service "code.gitea.io/gitea/modules/repository" + "code.gitea.io/gitea/modules/setting" + "code.gitea.io/gitea/services/auth" + mirror_service "code.gitea.io/gitea/services/mirror" +) + +func registerUpdateMirrorTask() { + RegisterTaskFatal("update_mirrors", &BaseConfig{ + Enabled: true, + RunAtStart: false, + Schedule: "@every 10m", + NoSuccessNotice: true, + }, func(ctx context.Context, _ *models.User, _ Config) error { + return mirror_service.Update(ctx) + }) +} + +func registerRepoHealthCheck() { + type RepoHealthCheckConfig struct { + BaseConfig + Timeout time.Duration + Args []string `delim:" "` + } + RegisterTaskFatal("repo_health_check", &RepoHealthCheckConfig{ + BaseConfig: BaseConfig{ + Enabled: true, + RunAtStart: false, + Schedule: "@midnight", + }, + Timeout: 60 * time.Second, + Args: []string{}, + }, func(ctx context.Context, _ *models.User, config Config) error { + rhcConfig := config.(*RepoHealthCheckConfig) + return repository_service.GitFsck(ctx, rhcConfig.Timeout, rhcConfig.Args) + }) +} + +func registerCheckRepoStats() { + RegisterTaskFatal("check_repo_stats", &BaseConfig{ + Enabled: true, + RunAtStart: true, + Schedule: "@midnight", + }, func(ctx context.Context, _ *models.User, _ Config) error { + return models.CheckRepoStats(ctx) + }) +} + +func registerArchiveCleanup() { + RegisterTaskFatal("archive_cleanup", &OlderThanConfig{ + BaseConfig: BaseConfig{ + Enabled: true, + RunAtStart: true, + Schedule: "@midnight", + }, + OlderThan: 24 * time.Hour, + }, func(ctx context.Context, _ *models.User, config Config) error { + acConfig := config.(*OlderThanConfig) + return models.DeleteOldRepositoryArchives(ctx, acConfig.OlderThan) + }) +} + +func registerSyncExternalUsers() { + RegisterTaskFatal("sync_external_users", &UpdateExistingConfig{ + BaseConfig: BaseConfig{ + Enabled: true, + RunAtStart: false, + Schedule: "@midnight", + }, + UpdateExisting: true, + }, func(ctx context.Context, _ *models.User, config Config) error { + realConfig := config.(*UpdateExistingConfig) + return auth.SyncExternalUsers(ctx, realConfig.UpdateExisting) + }) +} + +func registerDeletedBranchesCleanup() { + RegisterTaskFatal("deleted_branches_cleanup", &OlderThanConfig{ + BaseConfig: BaseConfig{ + Enabled: true, + RunAtStart: true, + Schedule: "@midnight", + }, + OlderThan: 24 * time.Hour, + }, func(ctx context.Context, _ *models.User, config Config) error { + realConfig := config.(*OlderThanConfig) + models.RemoveOldDeletedBranches(ctx, realConfig.OlderThan) + return nil + }) +} + +func registerUpdateMigrationPosterID() { + RegisterTaskFatal("update_migration_poster_id", &BaseConfig{ + Enabled: true, + RunAtStart: true, + Schedule: "@midnight", + }, func(ctx context.Context, _ *models.User, _ Config) error { + return migrations.UpdateMigrationPosterID(ctx) + }) +} + +func registerCleanupHookTaskTable() { + RegisterTaskFatal("cleanup_hook_task_table", &CleanupHookTaskConfig{ + BaseConfig: BaseConfig{ + Enabled: true, + RunAtStart: false, + Schedule: "@midnight", + }, + CleanupType: "OlderThan", + OlderThan: 168 * time.Hour, + NumberToKeep: 10, + }, func(ctx context.Context, _ *models.User, config Config) error { + realConfig := config.(*CleanupHookTaskConfig) + return webhook.CleanupHookTaskTable(ctx, webhook.ToHookTaskCleanupType(realConfig.CleanupType), realConfig.OlderThan, realConfig.NumberToKeep) + }) +} + +func initBasicTasks() { + registerUpdateMirrorTask() + registerRepoHealthCheck() + registerCheckRepoStats() + registerArchiveCleanup() + registerSyncExternalUsers() + registerDeletedBranchesCleanup() + if !setting.Repository.DisableMigrations { + registerUpdateMigrationPosterID() + } + registerCleanupHookTaskTable() +} diff --git a/services/cron/tasks_extended.go b/services/cron/tasks_extended.go new file mode 100644 index 0000000000..4ddcd44537 --- /dev/null +++ b/services/cron/tasks_extended.go @@ -0,0 +1,166 @@ +// Copyright 2020 The Gitea Authors. All rights reserved. +// Use of this source code is governed by a MIT-style +// license that can be found in the LICENSE file. + +package cron + +import ( + "context" + "time" + + "code.gitea.io/gitea/models" + repo_module "code.gitea.io/gitea/modules/repository" + "code.gitea.io/gitea/modules/setting" + "code.gitea.io/gitea/modules/updatechecker" + repo_service "code.gitea.io/gitea/services/repository" +) + +func registerDeleteInactiveUsers() { + RegisterTaskFatal("delete_inactive_accounts", &OlderThanConfig{ + BaseConfig: BaseConfig{ + Enabled: false, + RunAtStart: false, + Schedule: "@annually", + }, + OlderThan: 0 * time.Second, + }, func(ctx context.Context, _ *models.User, config Config) error { + olderThanConfig := config.(*OlderThanConfig) + return models.DeleteInactiveUsers(ctx, olderThanConfig.OlderThan) + }) +} + +func registerDeleteRepositoryArchives() { + RegisterTaskFatal("delete_repo_archives", &BaseConfig{ + Enabled: false, + RunAtStart: false, + Schedule: "@annually", + }, func(ctx context.Context, _ *models.User, _ Config) error { + return repo_service.DeleteRepositoryArchives(ctx) + }) +} + +func registerGarbageCollectRepositories() { + type RepoHealthCheckConfig struct { + BaseConfig + Timeout time.Duration + Args []string `delim:" "` + } + RegisterTaskFatal("git_gc_repos", &RepoHealthCheckConfig{ + BaseConfig: BaseConfig{ + Enabled: false, + RunAtStart: false, + Schedule: "@every 72h", + }, + Timeout: time.Duration(setting.Git.Timeout.GC) * time.Second, + Args: setting.Git.GCArgs, + }, func(ctx context.Context, _ *models.User, config Config) error { + rhcConfig := config.(*RepoHealthCheckConfig) + return repo_module.GitGcRepos(ctx, rhcConfig.Timeout, rhcConfig.Args...) + }) +} + +func registerRewriteAllPublicKeys() { + RegisterTaskFatal("resync_all_sshkeys", &BaseConfig{ + Enabled: false, + RunAtStart: false, + Schedule: "@every 72h", + }, func(_ context.Context, _ *models.User, _ Config) error { + return models.RewriteAllPublicKeys() + }) +} + +func registerRewriteAllPrincipalKeys() { + RegisterTaskFatal("resync_all_sshprincipals", &BaseConfig{ + Enabled: false, + RunAtStart: false, + Schedule: "@every 72h", + }, func(_ context.Context, _ *models.User, _ Config) error { + return models.RewriteAllPrincipalKeys() + }) +} + +func registerRepositoryUpdateHook() { + RegisterTaskFatal("resync_all_hooks", &BaseConfig{ + Enabled: false, + RunAtStart: false, + Schedule: "@every 72h", + }, func(ctx context.Context, _ *models.User, _ Config) error { + return repo_module.SyncRepositoryHooks(ctx) + }) +} + +func registerReinitMissingRepositories() { + RegisterTaskFatal("reinit_missing_repos", &BaseConfig{ + Enabled: false, + RunAtStart: false, + Schedule: "@every 72h", + }, func(ctx context.Context, _ *models.User, _ Config) error { + return repo_module.ReinitMissingRepositories(ctx) + }) +} + +func registerDeleteMissingRepositories() { + RegisterTaskFatal("delete_missing_repos", &BaseConfig{ + Enabled: false, + RunAtStart: false, + Schedule: "@every 72h", + }, func(ctx context.Context, user *models.User, _ Config) error { + return repo_module.DeleteMissingRepositories(ctx, user) + }) +} + +func registerRemoveRandomAvatars() { + RegisterTaskFatal("delete_generated_repository_avatars", &BaseConfig{ + Enabled: false, + RunAtStart: false, + Schedule: "@every 72h", + }, func(ctx context.Context, _ *models.User, _ Config) error { + return models.RemoveRandomAvatars(ctx) + }) +} + +func registerDeleteOldActions() { + RegisterTaskFatal("delete_old_actions", &OlderThanConfig{ + BaseConfig: BaseConfig{ + Enabled: false, + RunAtStart: false, + Schedule: "@every 168h", + }, + OlderThan: 365 * 24 * time.Hour, + }, func(ctx context.Context, _ *models.User, config Config) error { + olderThanConfig := config.(*OlderThanConfig) + return models.DeleteOldActions(olderThanConfig.OlderThan) + }) +} + +func registerUpdateGiteaChecker() { + type UpdateCheckerConfig struct { + BaseConfig + HTTPEndpoint string + } + RegisterTaskFatal("update_checker", &UpdateCheckerConfig{ + BaseConfig: BaseConfig{ + Enabled: true, + RunAtStart: false, + Schedule: "@every 168h", + }, + HTTPEndpoint: "https://dl.gitea.io/gitea/version.json", + }, func(ctx context.Context, _ *models.User, config Config) error { + updateCheckerConfig := config.(*UpdateCheckerConfig) + return updatechecker.GiteaUpdateChecker(updateCheckerConfig.HTTPEndpoint) + }) +} + +func initExtendedTasks() { + registerDeleteInactiveUsers() + registerDeleteRepositoryArchives() + registerGarbageCollectRepositories() + registerRewriteAllPublicKeys() + registerRewriteAllPrincipalKeys() + registerRepositoryUpdateHook() + registerReinitMissingRepositories() + registerDeleteMissingRepositories() + registerRemoveRandomAvatars() + registerDeleteOldActions() + registerUpdateGiteaChecker() +} |