diff options
author | Lunny Xiao <xiaolunwen@gmail.com> | 2019-10-13 21:23:14 +0800 |
---|---|---|
committer | GitHub <noreply@github.com> | 2019-10-13 21:23:14 +0800 |
commit | f2a3abc683ad4b2177b7c7c6160a2c0b4316120a (patch) | |
tree | 3b92f34b9bb9a015072f511dc5cf6340af18eda5 /modules | |
parent | 0a96e59884ca5c4fedc8c3d166d97f35b245ad6e (diff) | |
download | gitea-f2a3abc683ad4b2177b7c7c6160a2c0b4316120a.tar.gz gitea-f2a3abc683ad4b2177b7c7c6160a2c0b4316120a.zip |
Move migrating repository from frontend to backend (#6200)
* move migrating to backend
* add loading image when migrating and fix tests
* fix format
* fix lint
* add redis task queue support and improve docs
* add redis vendor
* fix vet
* add database migrations and fix app.ini sample
* add comments for task section on app.ini.sample
* Update models/migrations/v84.go
Co-Authored-By: lunny <xiaolunwen@gmail.com>
* Update models/repo.go
Co-Authored-By: lunny <xiaolunwen@gmail.com>
* move migrating to backend
* add loading image when migrating and fix tests
* fix fmt
* add redis task queue support and improve docs
* fix fixtures
* fix fixtures
* fix duplicate function on index.js
* fix tests
* rename repository statuses
* check if repository is being create when SSH request
* fix lint
* fix template
* some improvements
* fix template
* unified migrate options
* fix lint
* fix loading page
* refactor
* When gitea restart, don't restart the running tasks because we may have servel gitea instances, that may break the migration
* fix js
* Update models/repo.go
Co-Authored-By: guillep2k <18600385+guillep2k@users.noreply.github.com>
* Update docs/content/doc/advanced/config-cheat-sheet.en-us.md
Co-Authored-By: guillep2k <18600385+guillep2k@users.noreply.github.com>
* fix tests
* rename ErrTaskIsNotExist to ErrTaskDoesNotExist
* delete release after add one on tests to make it run happy
* fix tests
* fix tests
* improve codes
* fix lint
* fix lint
* fix migrations
Diffstat (limited to 'modules')
-rw-r--r-- | modules/context/repo.go | 35 | ||||
-rw-r--r-- | modules/migrations/base/options.go | 21 | ||||
-rw-r--r-- | modules/migrations/gitea.go | 36 | ||||
-rw-r--r-- | modules/migrations/gitea_test.go | 7 | ||||
-rw-r--r-- | modules/migrations/github.go | 4 | ||||
-rw-r--r-- | modules/migrations/migrate.go | 12 | ||||
-rw-r--r-- | modules/setting/setting.go | 1 | ||||
-rw-r--r-- | modules/setting/task.go | 25 | ||||
-rw-r--r-- | modules/structs/repo.go | 16 | ||||
-rw-r--r-- | modules/structs/task.go | 34 | ||||
-rw-r--r-- | modules/task/migrate.go | 120 | ||||
-rw-r--r-- | modules/task/queue.go | 14 | ||||
-rw-r--r-- | modules/task/queue_channel.go | 48 | ||||
-rw-r--r-- | modules/task/queue_redis.go | 130 | ||||
-rw-r--r-- | modules/task/task.go | 66 |
15 files changed, 515 insertions, 54 deletions
diff --git a/modules/context/repo.go b/modules/context/repo.go index 3caf583f83..f4af19a0e8 100644 --- a/modules/context/repo.go +++ b/modules/context/repo.go @@ -146,6 +146,9 @@ func (r *Repository) FileExists(path string, branch string) (bool, error) { // GetEditorconfig returns the .editorconfig definition if found in the // HEAD of the default repo branch. func (r *Repository) GetEditorconfig() (*editorconfig.Editorconfig, error) { + if r.GitRepo == nil { + return nil, nil + } commit, err := r.GitRepo.GetBranchCommit(r.Repository.DefaultBranch) if err != nil { return nil, err @@ -358,12 +361,6 @@ func RepoAssignment() macaron.Handler { return } - gitRepo, err := git.OpenRepository(models.RepoPath(userName, repoName)) - if err != nil { - ctx.ServerError("RepoAssignment Invalid repo "+models.RepoPath(userName, repoName), err) - return - } - ctx.Repo.GitRepo = gitRepo ctx.Repo.RepoLink = repo.Link() ctx.Data["RepoLink"] = ctx.Repo.RepoLink ctx.Data["RepoRelPath"] = ctx.Repo.Owner.Name + "/" + ctx.Repo.Repository.Name @@ -373,13 +370,6 @@ func RepoAssignment() macaron.Handler { ctx.Data["RepoExternalIssuesLink"] = unit.ExternalTrackerConfig().ExternalTrackerURL } - tags, err := ctx.Repo.GitRepo.GetTags() - if err != nil { - ctx.ServerError("GetTags", err) - return - } - ctx.Data["Tags"] = tags - count, err := models.GetReleaseCountByRepoID(ctx.Repo.Repository.ID, models.FindReleasesOptions{ IncludeDrafts: false, IncludeTags: true, @@ -425,12 +415,25 @@ func RepoAssignment() macaron.Handler { } // repo is empty and display enable - if ctx.Repo.Repository.IsEmpty { + if ctx.Repo.Repository.IsEmpty || ctx.Repo.Repository.IsBeingCreated() { ctx.Data["BranchName"] = ctx.Repo.Repository.DefaultBranch return } - ctx.Data["TagName"] = ctx.Repo.TagName + gitRepo, err := git.OpenRepository(models.RepoPath(userName, repoName)) + if err != nil { + ctx.ServerError("RepoAssignment Invalid repo "+models.RepoPath(userName, repoName), err) + return + } + ctx.Repo.GitRepo = gitRepo + + tags, err := ctx.Repo.GitRepo.GetTags() + if err != nil { + ctx.ServerError("GetTags", err) + return + } + ctx.Data["Tags"] = tags + brs, err := ctx.Repo.GitRepo.GetBranches() if err != nil { ctx.ServerError("GetBranches", err) @@ -439,6 +442,8 @@ func RepoAssignment() macaron.Handler { ctx.Data["Branches"] = brs ctx.Data["BranchesCount"] = len(brs) + ctx.Data["TagName"] = ctx.Repo.TagName + // If not branch selected, try default one. // If default branch doesn't exists, fall back to some other branch. if len(ctx.Repo.BranchName) == 0 { diff --git a/modules/migrations/base/options.go b/modules/migrations/base/options.go index ba7fdc6815..2d180b61d9 100644 --- a/modules/migrations/base/options.go +++ b/modules/migrations/base/options.go @@ -5,22 +5,7 @@ package base -// MigrateOptions defines the way a repository gets migrated -type MigrateOptions struct { - RemoteURL string - AuthUsername string - AuthPassword string - Name string - Description string - OriginalURL string +import "code.gitea.io/gitea/modules/structs" - Wiki bool - Issues bool - Milestones bool - Labels bool - Releases bool - Comments bool - PullRequests bool - Private bool - Mirror bool -} +// MigrateOptions defines the way a repository gets migrated +type MigrateOptions = structs.MigrateRepoOption diff --git a/modules/migrations/gitea.go b/modules/migrations/gitea.go index 1edac47a6e..ab3b0b9f69 100644 --- a/modules/migrations/gitea.go +++ b/modules/migrations/gitea.go @@ -22,6 +22,7 @@ import ( "code.gitea.io/gitea/modules/log" "code.gitea.io/gitea/modules/migrations/base" "code.gitea.io/gitea/modules/setting" + "code.gitea.io/gitea/modules/structs" "code.gitea.io/gitea/modules/timeutil" gouuid "github.com/satori/go.uuid" @@ -90,16 +91,33 @@ func (g *GiteaLocalUploader) CreateRepo(repo *base.Repository, opts base.Migrate remoteAddr = u.String() } - r, err := models.MigrateRepository(g.doer, owner, models.MigrateRepoOptions{ - Name: g.repoName, - Description: repo.Description, - OriginalURL: repo.OriginalURL, - IsMirror: repo.IsMirror, - RemoteAddr: remoteAddr, - IsPrivate: repo.IsPrivate, - Wiki: opts.Wiki, - SyncReleasesWithTags: !opts.Releases, // if didn't get releases, then sync them from tags + var r *models.Repository + if opts.MigrateToRepoID <= 0 { + r, err = models.CreateRepository(g.doer, owner, models.CreateRepoOptions{ + Name: g.repoName, + Description: repo.Description, + OriginalURL: repo.OriginalURL, + IsPrivate: opts.Private, + IsMirror: opts.Mirror, + Status: models.RepositoryBeingMigrated, + }) + } else { + r, err = models.GetRepositoryByID(opts.MigrateToRepoID) + } + if err != nil { + return err + } + + r, err = models.MigrateRepositoryGitData(g.doer, owner, r, structs.MigrateRepoOption{ + RepoName: g.repoName, + Description: repo.Description, + Mirror: repo.IsMirror, + CloneAddr: remoteAddr, + Private: repo.IsPrivate, + Wiki: opts.Wiki, + Releases: opts.Releases, // if didn't get releases, then sync them from tags }) + g.repo = r if err != nil { return err diff --git a/modules/migrations/gitea_test.go b/modules/migrations/gitea_test.go index 88a3a6d218..73c119a15d 100644 --- a/modules/migrations/gitea_test.go +++ b/modules/migrations/gitea_test.go @@ -10,6 +10,7 @@ import ( "time" "code.gitea.io/gitea/models" + "code.gitea.io/gitea/modules/structs" "code.gitea.io/gitea/modules/util" "github.com/stretchr/testify/assert" @@ -29,9 +30,9 @@ func TestGiteaUploadRepo(t *testing.T) { uploader = NewGiteaLocalUploader(user, user.Name, repoName) ) - err := migrateRepository(downloader, uploader, MigrateOptions{ - RemoteURL: "https://github.com/go-xorm/builder", - Name: repoName, + err := migrateRepository(downloader, uploader, structs.MigrateRepoOption{ + CloneAddr: "https://github.com/go-xorm/builder", + RepoName: repoName, AuthUsername: "", Wiki: true, diff --git a/modules/migrations/github.go b/modules/migrations/github.go index 754f98941c..1c5d96c03d 100644 --- a/modules/migrations/github.go +++ b/modules/migrations/github.go @@ -34,7 +34,7 @@ type GithubDownloaderV3Factory struct { // Match returns ture if the migration remote URL matched this downloader factory func (f *GithubDownloaderV3Factory) Match(opts base.MigrateOptions) (bool, error) { - u, err := url.Parse(opts.RemoteURL) + u, err := url.Parse(opts.CloneAddr) if err != nil { return false, err } @@ -44,7 +44,7 @@ func (f *GithubDownloaderV3Factory) Match(opts base.MigrateOptions) (bool, error // New returns a Downloader related to this factory according MigrateOptions func (f *GithubDownloaderV3Factory) New(opts base.MigrateOptions) (base.Downloader, error) { - u, err := url.Parse(opts.RemoteURL) + u, err := url.Parse(opts.CloneAddr) if err != nil { return nil, err } diff --git a/modules/migrations/migrate.go b/modules/migrations/migrate.go index 27782cb940..3f5c0d1118 100644 --- a/modules/migrations/migrate.go +++ b/modules/migrations/migrate.go @@ -6,6 +6,8 @@ package migrations import ( + "fmt" + "code.gitea.io/gitea/models" "code.gitea.io/gitea/modules/log" "code.gitea.io/gitea/modules/migrations/base" @@ -27,7 +29,7 @@ func RegisterDownloaderFactory(factory base.DownloaderFactory) { func MigrateRepository(doer *models.User, ownerName string, opts base.MigrateOptions) (*models.Repository, error) { var ( downloader base.Downloader - uploader = NewGiteaLocalUploader(doer, ownerName, opts.Name) + uploader = NewGiteaLocalUploader(doer, ownerName, opts.RepoName) ) for _, factory := range factories { @@ -50,14 +52,18 @@ func MigrateRepository(doer *models.User, ownerName string, opts base.MigrateOpt opts.Comments = false opts.Issues = false opts.PullRequests = false - downloader = NewPlainGitDownloader(ownerName, opts.Name, opts.RemoteURL) - log.Trace("Will migrate from git: %s", opts.RemoteURL) + downloader = NewPlainGitDownloader(ownerName, opts.RepoName, opts.CloneAddr) + log.Trace("Will migrate from git: %s", opts.CloneAddr) } if err := migrateRepository(downloader, uploader, opts); err != nil { if err1 := uploader.Rollback(); err1 != nil { log.Error("rollback failed: %v", err1) } + + if err2 := models.CreateRepositoryNotice(fmt.Sprintf("Migrate repository from %s failed: %v", opts.CloneAddr, err)); err2 != nil { + log.Error("create respotiry notice failed: ", err2) + } return nil, err } diff --git a/modules/setting/setting.go b/modules/setting/setting.go index 5e476854b2..8c61bdbb77 100644 --- a/modules/setting/setting.go +++ b/modules/setting/setting.go @@ -1043,4 +1043,5 @@ func NewServices() { newNotifyMailService() newWebhookService() newIndexerService() + newTaskService() } diff --git a/modules/setting/task.go b/modules/setting/task.go new file mode 100644 index 0000000000..97704d4a4d --- /dev/null +++ b/modules/setting/task.go @@ -0,0 +1,25 @@ +// 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 setting + +var ( + // Task settings + Task = struct { + QueueType string + QueueLength int + QueueConnStr string + }{ + QueueType: ChannelQueueType, + QueueLength: 1000, + QueueConnStr: "addrs=127.0.0.1:6379 db=0", + } +) + +func newTaskService() { + sec := Cfg.Section("task") + Task.QueueType = sec.Key("QUEUE_TYPE").MustString(ChannelQueueType) + Task.QueueLength = sec.Key("QUEUE_LENGTH").MustInt(1000) + Task.QueueConnStr = sec.Key("QUEUE_CONN_STR").MustString("addrs=127.0.0.1:6379 db=0") +} diff --git a/modules/structs/repo.go b/modules/structs/repo.go index 87396d6ce9..57f1768a0b 100644 --- a/modules/structs/repo.go +++ b/modules/structs/repo.go @@ -162,8 +162,16 @@ type MigrateRepoOption struct { // required: true UID int `json:"uid" binding:"Required"` // required: true - RepoName string `json:"repo_name" binding:"Required"` - Mirror bool `json:"mirror"` - Private bool `json:"private"` - Description string `json:"description"` + RepoName string `json:"repo_name" binding:"Required"` + Mirror bool `json:"mirror"` + Private bool `json:"private"` + Description string `json:"description"` + Wiki bool + Issues bool + Milestones bool + Labels bool + Releases bool + Comments bool + PullRequests bool + MigrateToRepoID int64 } diff --git a/modules/structs/task.go b/modules/structs/task.go new file mode 100644 index 0000000000..e83d0437ce --- /dev/null +++ b/modules/structs/task.go @@ -0,0 +1,34 @@ +// Copyright 2019 Gitea. All rights reserved. +// Use of this source code is governed by a MIT-style +// license that can be found in the LICENSE file. + +package structs + +// TaskType defines task type +type TaskType int + +// all kinds of task types +const ( + TaskTypeMigrateRepo TaskType = iota // migrate repository from external or local disk +) + +// Name returns the task type name +func (taskType TaskType) Name() string { + switch taskType { + case TaskTypeMigrateRepo: + return "Migrate Repository" + } + return "" +} + +// TaskStatus defines task status +type TaskStatus int + +// enumerate all the kinds of task status +const ( + TaskStatusQueue TaskStatus = iota // 0 task is queue + TaskStatusRunning // 1 task is running + TaskStatusStopped // 2 task is stopped + TaskStatusFailed // 3 task is failed + TaskStatusFinished // 4 task is finished +) diff --git a/modules/task/migrate.go b/modules/task/migrate.go new file mode 100644 index 0000000000..5d15a506d7 --- /dev/null +++ b/modules/task/migrate.go @@ -0,0 +1,120 @@ +// Copyright 2019 Gitea. All rights reserved. +// Use of this source code is governed by a MIT-style +// license that can be found in the LICENSE file. + +package task + +import ( + "bytes" + "errors" + "fmt" + "strings" + + "code.gitea.io/gitea/models" + "code.gitea.io/gitea/modules/log" + "code.gitea.io/gitea/modules/migrations" + "code.gitea.io/gitea/modules/notification" + "code.gitea.io/gitea/modules/structs" + "code.gitea.io/gitea/modules/timeutil" + "code.gitea.io/gitea/modules/util" +) + +func handleCreateError(owner *models.User, err error, name string) error { + switch { + case models.IsErrReachLimitOfRepo(err): + return fmt.Errorf("You have already reached your limit of %d repositories", owner.MaxCreationLimit()) + case models.IsErrRepoAlreadyExist(err): + return errors.New("The repository name is already used") + case models.IsErrNameReserved(err): + return fmt.Errorf("The repository name '%s' is reserved", err.(models.ErrNameReserved).Name) + case models.IsErrNamePatternNotAllowed(err): + return fmt.Errorf("The pattern '%s' is not allowed in a repository name", err.(models.ErrNamePatternNotAllowed).Pattern) + default: + return err + } +} + +func runMigrateTask(t *models.Task) (err error) { + defer func() { + if e := recover(); e != nil { + var buf bytes.Buffer + fmt.Fprintf(&buf, "Handler crashed with error: %v", log.Stack(2)) + + err = errors.New(buf.String()) + } + + if err == nil { + err = models.FinishMigrateTask(t) + if err == nil { + notification.NotifyMigrateRepository(t.Doer, t.Owner, t.Repo) + return + } + + log.Error("FinishMigrateTask failed: %s", err.Error()) + } + + t.EndTime = timeutil.TimeStampNow() + t.Status = structs.TaskStatusFailed + t.Errors = err.Error() + if err := t.UpdateCols("status", "errors", "end_time"); err != nil { + log.Error("Task UpdateCols failed: %s", err.Error()) + } + + if t.Repo != nil { + if errDelete := models.DeleteRepository(t.Doer, t.OwnerID, t.Repo.ID); errDelete != nil { + log.Error("DeleteRepository: %v", errDelete) + } + } + }() + + if err := t.LoadRepo(); err != nil { + return err + } + + // if repository is ready, then just finsih the task + if t.Repo.Status == models.RepositoryReady { + return nil + } + + if err := t.LoadDoer(); err != nil { + return err + } + if err := t.LoadOwner(); err != nil { + return err + } + t.StartTime = timeutil.TimeStampNow() + t.Status = structs.TaskStatusRunning + if err := t.UpdateCols("start_time", "status"); err != nil { + return err + } + + var opts *structs.MigrateRepoOption + opts, err = t.MigrateConfig() + if err != nil { + return err + } + + opts.MigrateToRepoID = t.RepoID + repo, err := migrations.MigrateRepository(t.Doer, t.Owner.Name, *opts) + if err == nil { + notification.NotifyMigrateRepository(t.Doer, t.Owner, repo) + + log.Trace("Repository migrated [%d]: %s/%s", repo.ID, t.Owner.Name, repo.Name) + return nil + } + + if models.IsErrRepoAlreadyExist(err) { + return errors.New("The repository name is already used") + } + + // remoteAddr may contain credentials, so we sanitize it + err = util.URLSanitizedError(err, opts.CloneAddr) + if strings.Contains(err.Error(), "Authentication failed") || + strings.Contains(err.Error(), "could not read Username") { + return fmt.Errorf("Authentication failed: %v", err.Error()) + } else if strings.Contains(err.Error(), "fatal:") { + return fmt.Errorf("Migration failed: %v", err.Error()) + } + + return handleCreateError(t.Owner, err, "MigratePost") +} diff --git a/modules/task/queue.go b/modules/task/queue.go new file mode 100644 index 0000000000..ddee0b3d46 --- /dev/null +++ b/modules/task/queue.go @@ -0,0 +1,14 @@ +// Copyright 2019 Gitea. All rights reserved. +// Use of this source code is governed by a MIT-style +// license that can be found in the LICENSE file. + +package task + +import "code.gitea.io/gitea/models" + +// Queue defines an interface to run task queue +type Queue interface { + Run() error + Push(*models.Task) error + Stop() +} diff --git a/modules/task/queue_channel.go b/modules/task/queue_channel.go new file mode 100644 index 0000000000..da541f4755 --- /dev/null +++ b/modules/task/queue_channel.go @@ -0,0 +1,48 @@ +// 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 task + +import ( + "code.gitea.io/gitea/models" + "code.gitea.io/gitea/modules/log" +) + +var ( + _ Queue = &ChannelQueue{} +) + +// ChannelQueue implements +type ChannelQueue struct { + queue chan *models.Task +} + +// NewChannelQueue create a memory channel queue +func NewChannelQueue(queueLen int) *ChannelQueue { + return &ChannelQueue{ + queue: make(chan *models.Task, queueLen), + } +} + +// Run starts to run the queue +func (c *ChannelQueue) Run() error { + for task := range c.queue { + err := Run(task) + if err != nil { + log.Error("Run task failed: %s", err.Error()) + } + } + return nil +} + +// Push will push the task ID to queue +func (c *ChannelQueue) Push(task *models.Task) error { + c.queue <- task + return nil +} + +// Stop stop the queue +func (c *ChannelQueue) Stop() { + close(c.queue) +} diff --git a/modules/task/queue_redis.go b/modules/task/queue_redis.go new file mode 100644 index 0000000000..127de0cdbf --- /dev/null +++ b/modules/task/queue_redis.go @@ -0,0 +1,130 @@ +// 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 task + +import ( + "encoding/json" + "errors" + "strconv" + "strings" + "time" + + "code.gitea.io/gitea/models" + "code.gitea.io/gitea/modules/log" + + "github.com/go-redis/redis" +) + +var ( + _ Queue = &RedisQueue{} +) + +type redisClient interface { + RPush(key string, args ...interface{}) *redis.IntCmd + LPop(key string) *redis.StringCmd + Ping() *redis.StatusCmd +} + +// RedisQueue redis queue +type RedisQueue struct { + client redisClient + queueName string + closeChan chan bool +} + +func parseConnStr(connStr string) (addrs, password string, dbIdx int, err error) { + fields := strings.Fields(connStr) + for _, f := range fields { + items := strings.SplitN(f, "=", 2) + if len(items) < 2 { + continue + } + switch strings.ToLower(items[0]) { + case "addrs": + addrs = items[1] + case "password": + password = items[1] + case "db": + dbIdx, err = strconv.Atoi(items[1]) + if err != nil { + return + } + } + } + return +} + +// NewRedisQueue creates single redis or cluster redis queue +func NewRedisQueue(addrs string, password string, dbIdx int) (*RedisQueue, error) { + dbs := strings.Split(addrs, ",") + var queue = RedisQueue{ + queueName: "task_queue", + closeChan: make(chan bool), + } + if len(dbs) == 0 { + return nil, errors.New("no redis host found") + } else if len(dbs) == 1 { + queue.client = redis.NewClient(&redis.Options{ + Addr: strings.TrimSpace(dbs[0]), // use default Addr + Password: password, // no password set + DB: dbIdx, // use default DB + }) + } else { + // cluster will ignore db + queue.client = redis.NewClusterClient(&redis.ClusterOptions{ + Addrs: dbs, + Password: password, + }) + } + if err := queue.client.Ping().Err(); err != nil { + return nil, err + } + return &queue, nil +} + +// Run starts to run the queue +func (r *RedisQueue) Run() error { + for { + select { + case <-r.closeChan: + return nil + case <-time.After(time.Millisecond * 100): + } + + bs, err := r.client.LPop(r.queueName).Bytes() + if err != nil { + if err != redis.Nil { + log.Error("LPop failed: %v", err) + } + time.Sleep(time.Millisecond * 100) + continue + } + + var task models.Task + err = json.Unmarshal(bs, &task) + if err != nil { + log.Error("Unmarshal task failed: %s", err.Error()) + } else { + err = Run(&task) + if err != nil { + log.Error("Run task failed: %s", err.Error()) + } + } + } +} + +// Push implements Queue +func (r *RedisQueue) Push(task *models.Task) error { + bs, err := json.Marshal(task) + if err != nil { + return err + } + return r.client.RPush(r.queueName, bs).Err() +} + +// Stop stop the queue +func (r *RedisQueue) Stop() { + r.closeChan <- true +} diff --git a/modules/task/task.go b/modules/task/task.go new file mode 100644 index 0000000000..64744afe7a --- /dev/null +++ b/modules/task/task.go @@ -0,0 +1,66 @@ +// Copyright 2019 Gitea. All rights reserved. +// Use of this source code is governed by a MIT-style +// license that can be found in the LICENSE file. + +package task + +import ( + "fmt" + + "code.gitea.io/gitea/models" + "code.gitea.io/gitea/modules/log" + "code.gitea.io/gitea/modules/migrations/base" + "code.gitea.io/gitea/modules/setting" + "code.gitea.io/gitea/modules/structs" +) + +// taskQueue is a global queue of tasks +var taskQueue Queue + +// Run a task +func Run(t *models.Task) error { + switch t.Type { + case structs.TaskTypeMigrateRepo: + return runMigrateTask(t) + default: + return fmt.Errorf("Unknow task type: %d", t.Type) + } +} + +// Init will start the service to get all unfinished tasks and run them +func Init() error { + switch setting.Task.QueueType { + case setting.ChannelQueueType: + taskQueue = NewChannelQueue(setting.Task.QueueLength) + case setting.RedisQueueType: + var err error + addrs, pass, idx, err := parseConnStr(setting.Task.QueueConnStr) + if err != nil { + return err + } + taskQueue, err = NewRedisQueue(addrs, pass, idx) + if err != nil { + return err + } + default: + return fmt.Errorf("Unsupported task queue type: %v", setting.Task.QueueType) + } + + go func() { + if err := taskQueue.Run(); err != nil { + log.Error("taskQueue.Run end failed: %v", err) + } + }() + + return nil +} + +// MigrateRepository add migration repository to task +func MigrateRepository(doer, u *models.User, opts base.MigrateOptions) error { + task, err := models.CreateMigrateTask(doer, u, opts) + if err != nil { + return err + } + + return taskQueue.Push(task) +} |