#2018 able to sync now for mirrors

- Refactor code to use sync.UniqueQueue
- Closes #3509

1 files changed, 243 insertions, 0 deletions

diff --git a/models/repo_mirror.go b/models/repo_mirror.go
new file mode 100644
index 0000000000..53014e9d6d
--- /dev/null
+++ b/models/repo_mirror.go
@@ -0,0 +1,243 @@
+// Copyright 2016 The Gogs 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 models
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/Unknwon/com"
+	"github.com/go-xorm/xorm"
+	"gopkg.in/ini.v1"
+
+	"github.com/gogits/gogs/modules/log"
+	"github.com/gogits/gogs/modules/process"
+	"github.com/gogits/gogs/modules/setting"
+	"github.com/gogits/gogs/modules/sync"
+)
+
+var MirrorQueue = sync.NewUniqueQueue(setting.Repository.MirrorQueueLength)
+
+// Mirror represents mirror information of a repository.
+type Mirror struct {
+	ID          int64 `xorm:"pk autoincr"`
+	RepoID      int64
+	Repo        *Repository `xorm:"-"`
+	Interval    int         // Hour.
+	EnablePrune bool        `xorm:"NOT NULL DEFAULT true"`
+
+	Updated        time.Time `xorm:"-"`
+	UpdatedUnix    int64
+	NextUpdate     time.Time `xorm:"-"`
+	NextUpdateUnix int64
+
+	address string `xorm:"-"`
+}
+
+func (m *Mirror) BeforeInsert() {
+	m.NextUpdateUnix = m.NextUpdate.Unix()
+}
+
+func (m *Mirror) BeforeUpdate() {
+	m.UpdatedUnix = time.Now().Unix()
+	m.NextUpdateUnix = m.NextUpdate.Unix()
+}
+
+func (m *Mirror) AfterSet(colName string, _ xorm.Cell) {
+	var err error
+	switch colName {
+	case "repo_id":
+		m.Repo, err = GetRepositoryByID(m.RepoID)
+		if err != nil {
+			log.Error(3, "GetRepositoryByID[%d]: %v", m.ID, err)
+		}
+	case "updated_unix":
+		m.Updated = time.Unix(m.UpdatedUnix, 0).Local()
+	case "next_updated_unix":
+		m.NextUpdate = time.Unix(m.NextUpdateUnix, 0).Local()
+	}
+}
+
+// ScheduleNextUpdate calculates and sets next update time.
+func (m *Mirror) ScheduleNextUpdate() {
+	m.NextUpdate = time.Now().Add(time.Duration(m.Interval) * time.Hour)
+}
+
+func (m *Mirror) readAddress() {
+	if len(m.address) > 0 {
+		return
+	}
+
+	cfg, err := ini.Load(m.Repo.GitConfigPath())
+	if err != nil {
+		log.Error(4, "Load: %v", err)
+		return
+	}
+	m.address = cfg.Section("remote \"origin\"").Key("url").Value()
+}
+
+// HandleCloneUserCredentials replaces user credentials from HTTP/HTTPS URL
+// with placeholder <credentials>.
+// It will fail for any other forms of clone addresses.
+func HandleCloneUserCredentials(url string, mosaics bool) string {
+	i := strings.Index(url, "@")
+	if i == -1 {
+		return url
+	}
+	start := strings.Index(url, "://")
+	if start == -1 {
+		return url
+	}
+	if mosaics {
+		return url[:start+3] + "<credentials>" + url[i:]
+	}
+	return url[:start+3] + url[i+1:]
+}
+
+// Address returns mirror address from Git repository config without credentials.
+func (m *Mirror) Address() string {
+	m.readAddress()
+	return HandleCloneUserCredentials(m.address, false)
+}
+
+// FullAddress returns mirror address from Git repository config.
+func (m *Mirror) FullAddress() string {
+	m.readAddress()
+	return m.address
+}
+
+// SaveAddress writes new address to Git repository config.
+func (m *Mirror) SaveAddress(addr string) error {
+	configPath := m.Repo.GitConfigPath()
+	cfg, err := ini.Load(configPath)
+	if err != nil {
+		return fmt.Errorf("Load: %v", err)
+	}
+
+	cfg.Section("remote \"origin\"").Key("url").SetValue(addr)
+	return cfg.SaveToIndent(configPath, "\t")
+}
+
+// runSync returns true if sync finished without error.
+func (m *Mirror) runSync() bool {
+	repoPath := m.Repo.RepoPath()
+	wikiPath := m.Repo.WikiPath()
+	timeout := time.Duration(setting.Git.Timeout.Mirror) * time.Second
+
+	gitArgs := []string{"remote", "update"}
+	if m.EnablePrune {
+		gitArgs = append(gitArgs, "--prune")
+	}
+
+	if _, stderr, err := process.ExecDir(
+		timeout, repoPath, fmt.Sprintf("runSync: %s", repoPath),
+		"git", gitArgs...); err != nil {
+		desc := fmt.Sprintf("Fail to update mirror repository '%s': %s", repoPath, stderr)
+		log.Error(4, desc)
+		if err = CreateRepositoryNotice(desc); err != nil {
+			log.Error(4, "CreateRepositoryNotice: %v", err)
+		}
+		return false
+	}
+	if m.Repo.HasWiki() {
+		if _, stderr, err := process.ExecDir(
+			timeout, wikiPath, fmt.Sprintf("runSync: %s", wikiPath),
+			"git", "remote", "update", "--prune"); err != nil {
+			desc := fmt.Sprintf("Fail to update mirror wiki repository '%s': %s", wikiPath, stderr)
+			log.Error(4, desc)
+			if err = CreateRepositoryNotice(desc); err != nil {
+				log.Error(4, "CreateRepositoryNotice: %v", err)
+			}
+			return false
+		}
+	}
+
+	return true
+}
+
+func getMirrorByRepoID(e Engine, repoID int64) (*Mirror, error) {
+	m := &Mirror{RepoID: repoID}
+	has, err := e.Get(m)
+	if err != nil {
+		return nil, err
+	} else if !has {
+		return nil, ErrMirrorNotExist
+	}
+	return m, nil
+}
+
+// GetMirrorByRepoID returns mirror information of a repository.
+func GetMirrorByRepoID(repoID int64) (*Mirror, error) {
+	return getMirrorByRepoID(x, repoID)
+}
+
+func updateMirror(e Engine, m *Mirror) error {
+	_, err := e.Id(m.ID).AllCols().Update(m)
+	return err
+}
+
+func UpdateMirror(m *Mirror) error {
+	return updateMirror(x, m)
+}
+
+func DeleteMirrorByRepoID(repoID int64) error {
+	_, err := x.Delete(&Mirror{RepoID: repoID})
+	return err
+}
+
+// MirrorUpdate checks and updates mirror repositories.
+func MirrorUpdate() {
+	if taskStatusTable.IsRunning(_MIRROR_UPDATE) {
+		return
+	}
+	taskStatusTable.Start(_MIRROR_UPDATE)
+	defer taskStatusTable.Stop(_MIRROR_UPDATE)
+
+	log.Trace("Doing: MirrorUpdate")
+
+	if err := x.Where("next_update_unix<=?", time.Now().Unix()).Iterate(new(Mirror), func(idx int, bean interface{}) error {
+		m := bean.(*Mirror)
+		if m.Repo == nil {
+			log.Error(4, "Disconnected mirror repository found: %d", m.ID)
+			return nil
+		}
+
+		MirrorQueue.Add(m.RepoID)
+		return nil
+	}); err != nil {
+		log.Error(4, "MirrorUpdate: %v", err)
+	}
+}
+
+// SyncMirrors checks and syncs mirrors.
+// TODO: sync more mirrors at same time.
+func SyncMirrors() {
+	// Start listening on new sync requests.
+	for repoID := range MirrorQueue.Queue() {
+		log.Trace("SyncMirrors [repo_id: %v]", repoID)
+		MirrorQueue.Remove(repoID)
+
+		m, err := GetMirrorByRepoID(com.StrTo(repoID).MustInt64())
+		if err != nil {
+			log.Error(4, "GetMirrorByRepoID [%d]: %v", repoID, err)
+			continue
+		}
+
+		if !m.runSync() {
+			continue
+		}
+
+		m.ScheduleNextUpdate()
+		if err = UpdateMirror(m); err != nil {
+			log.Error(4, "UpdateMirror [%d]: %v", repoID, err)
+			continue
+		}
+	}
+}
+
+func InitSyncMirrors() {
+	go SyncMirrors()
+}