summaryrefslogtreecommitdiffstats
path: root/modules/task
diff options
context:
space:
mode:
authorLunny Xiao <xiaolunwen@gmail.com>2019-10-13 21:23:14 +0800
committerGitHub <noreply@github.com>2019-10-13 21:23:14 +0800
commitf2a3abc683ad4b2177b7c7c6160a2c0b4316120a (patch)
tree3b92f34b9bb9a015072f511dc5cf6340af18eda5 /modules/task
parent0a96e59884ca5c4fedc8c3d166d97f35b245ad6e (diff)
downloadgitea-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.go120
-rw-r--r--modules/task/queue.go14
-rw-r--r--modules/task/queue_channel.go48
-rw-r--r--modules/task/queue_redis.go130
-rw-r--r--modules/task/task.go66
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)
+}