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/task | |
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/task')
-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 |
5 files changed, 378 insertions, 0 deletions
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) +} |