mirror of
https://codeberg.org/davrot/forgejo.git
synced 2025-07-10 16:00:04 +02:00
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
This commit is contained in:
parent
0a96e59884
commit
f2a3abc683
37 changed files with 1192 additions and 222 deletions
|
@ -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 {
|
||||
|
|
|
@ -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
|
||||
|
|
|
@ -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
|
||||
|
|
|
@ -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,
|
||||
|
|
|
@ -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
|
||||
}
|
||||
|
|
|
@ -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
|
||||
}
|
||||
|
||||
|
|
|
@ -1043,4 +1043,5 @@ func NewServices() {
|
|||
newNotifyMailService()
|
||||
newWebhookService()
|
||||
newIndexerService()
|
||||
newTaskService()
|
||||
}
|
||||
|
|
25
modules/setting/task.go
Normal file
25
modules/setting/task.go
Normal file
|
@ -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")
|
||||
}
|
|
@ -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
|
||||
}
|
||||
|
|
34
modules/structs/task.go
Normal file
34
modules/structs/task.go
Normal file
|
@ -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
|
||||
)
|
120
modules/task/migrate.go
Normal file
120
modules/task/migrate.go
Normal file
|
@ -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")
|
||||
}
|
14
modules/task/queue.go
Normal file
14
modules/task/queue.go
Normal file
|
@ -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()
|
||||
}
|
48
modules/task/queue_channel.go
Normal file
48
modules/task/queue_channel.go
Normal file
|
@ -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)
|
||||
}
|
130
modules/task/queue_redis.go
Normal file
130
modules/task/queue_redis.go
Normal file
|
@ -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
|
||||
}
|
66
modules/task/task.go
Normal file
66
modules/task/task.go
Normal file
|
@ -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)
|
||||
}
|
Loading…
Add table
Add a link
Reference in a new issue