123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242 |
- // Copyright 2021 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 mirror
-
- import (
- "context"
- "errors"
- "io"
- "net/url"
- "regexp"
- "time"
-
- "code.gitea.io/gitea/models"
- "code.gitea.io/gitea/modules/git"
- "code.gitea.io/gitea/modules/lfs"
- "code.gitea.io/gitea/modules/log"
- "code.gitea.io/gitea/modules/repository"
- "code.gitea.io/gitea/modules/setting"
- "code.gitea.io/gitea/modules/timeutil"
- "code.gitea.io/gitea/modules/util"
- )
-
- var stripExitStatus = regexp.MustCompile(`exit status \d+ - `)
-
- // AddPushMirrorRemote registers the push mirror remote.
- func AddPushMirrorRemote(m *models.PushMirror, addr string) error {
- addRemoteAndConfig := func(addr, path string) error {
- if _, err := git.NewCommand("remote", "add", "--mirror=push", m.RemoteName, addr).RunInDir(path); err != nil {
- return err
- }
- if _, err := git.NewCommand("config", "--add", "remote."+m.RemoteName+".push", "+refs/heads/*:refs/heads/*").RunInDir(path); err != nil {
- return err
- }
- if _, err := git.NewCommand("config", "--add", "remote."+m.RemoteName+".push", "+refs/tags/*:refs/tags/*").RunInDir(path); err != nil {
- return err
- }
- return nil
- }
-
- if err := addRemoteAndConfig(addr, m.Repo.RepoPath()); err != nil {
- return err
- }
-
- if m.Repo.HasWiki() {
- wikiRemoteURL := repository.WikiRemoteURL(addr)
- if len(wikiRemoteURL) > 0 {
- if err := addRemoteAndConfig(wikiRemoteURL, m.Repo.WikiPath()); err != nil {
- return err
- }
- }
- }
-
- return nil
- }
-
- // RemovePushMirrorRemote removes the push mirror remote.
- func RemovePushMirrorRemote(m *models.PushMirror) error {
- cmd := git.NewCommand("remote", "rm", m.RemoteName)
-
- if _, err := cmd.RunInDir(m.Repo.RepoPath()); err != nil {
- return err
- }
-
- if m.Repo.HasWiki() {
- if _, err := cmd.RunInDir(m.Repo.WikiPath()); err != nil {
- // The wiki remote may not exist
- log.Warn("Wiki Remote[%d] could not be removed: %v", m.ID, err)
- }
- }
-
- return nil
- }
-
- // SyncPushMirror starts the sync of the push mirror and schedules the next run.
- func SyncPushMirror(ctx context.Context, mirrorID int64) bool {
- log.Trace("SyncPushMirror [mirror: %d]", mirrorID)
- defer func() {
- err := recover()
- if err == nil {
- return
- }
- // There was a panic whilst syncPushMirror...
- log.Error("PANIC whilst syncPushMirror[%d] Panic: %v\nStacktrace: %s", mirrorID, err, log.Stack(2))
- }()
-
- m, err := models.GetPushMirrorByID(mirrorID)
- if err != nil {
- log.Error("GetPushMirrorByID [%d]: %v", mirrorID, err)
- return false
- }
-
- m.LastError = ""
-
- log.Trace("SyncPushMirror [mirror: %d][repo: %-v]: Running Sync", m.ID, m.Repo)
- err = runPushSync(ctx, m)
- if err != nil {
- log.Error("SyncPushMirror [mirror: %d][repo: %-v]: %v", m.ID, m.Repo, err)
- m.LastError = stripExitStatus.ReplaceAllLiteralString(err.Error(), "")
- }
-
- m.LastUpdateUnix = timeutil.TimeStampNow()
-
- if err := models.UpdatePushMirror(m); err != nil {
- log.Error("UpdatePushMirror [%d]: %v", m.ID, err)
-
- return false
- }
-
- log.Trace("SyncPushMirror [mirror: %d][repo: %-v]: Finished", m.ID, m.Repo)
-
- return err == nil
- }
-
- func runPushSync(ctx context.Context, m *models.PushMirror) error {
- timeout := time.Duration(setting.Git.Timeout.Mirror) * time.Second
-
- performPush := func(path string) error {
- remoteAddr, err := git.GetRemoteAddress(path, m.RemoteName)
- if err != nil {
- log.Error("GetRemoteAddress(%s) Error %v", path, err)
- return errors.New("Unexpected error")
- }
-
- if setting.LFS.StartServer {
- log.Trace("SyncMirrors [repo: %-v]: syncing LFS objects...", m.Repo)
-
- gitRepo, err := git.OpenRepository(path)
- if err != nil {
- log.Error("OpenRepository: %v", err)
- return errors.New("Unexpected error")
- }
- defer gitRepo.Close()
-
- ep := lfs.DetermineEndpoint(remoteAddr.String(), "")
- if err := pushAllLFSObjects(ctx, gitRepo, ep, false); err != nil {
- return util.NewURLSanitizedError(err, remoteAddr, true)
- }
- }
-
- log.Trace("Pushing %s mirror[%d] remote %s", path, m.ID, m.RemoteName)
-
- if err := git.Push(path, git.PushOptions{
- Remote: m.RemoteName,
- Force: true,
- Mirror: true,
- Timeout: timeout,
- }); err != nil {
- log.Error("Error pushing %s mirror[%d] remote %s: %v", path, m.ID, m.RemoteName, err)
-
- return util.NewURLSanitizedError(err, remoteAddr, true)
- }
-
- return nil
- }
-
- err := performPush(m.Repo.RepoPath())
- if err != nil {
- return err
- }
-
- if m.Repo.HasWiki() {
- wikiPath := m.Repo.WikiPath()
- _, err := git.GetRemoteAddress(wikiPath, m.RemoteName)
- if err == nil {
- err := performPush(wikiPath)
- if err != nil {
- return err
- }
- } else {
- log.Trace("Skipping wiki: No remote configured")
- }
- }
-
- return nil
- }
-
- func pushAllLFSObjects(ctx context.Context, gitRepo *git.Repository, endpoint *url.URL, skipTLSVerify bool) error {
- client := lfs.NewClient(endpoint, skipTLSVerify)
- contentStore := lfs.NewContentStore()
-
- pointerChan := make(chan lfs.PointerBlob)
- errChan := make(chan error, 1)
- go lfs.SearchPointerBlobs(ctx, gitRepo, pointerChan, errChan)
-
- uploadObjects := func(pointers []lfs.Pointer) error {
- err := client.Upload(ctx, pointers, func(p lfs.Pointer, objectError error) (io.ReadCloser, error) {
- if objectError != nil {
- return nil, objectError
- }
-
- content, err := contentStore.Get(p)
- if err != nil {
- log.Error("Error reading LFS object %v: %v", p, err)
- }
- return content, err
- })
- if err != nil {
- select {
- case <-ctx.Done():
- return nil
- default:
- }
- }
- return err
- }
-
- var batch []lfs.Pointer
- for pointerBlob := range pointerChan {
- exists, err := contentStore.Exists(pointerBlob.Pointer)
- if err != nil {
- log.Error("Error checking if LFS object %v exists: %v", pointerBlob.Pointer, err)
- return err
- }
- if !exists {
- log.Trace("Skipping missing LFS object %v", pointerBlob.Pointer)
- continue
- }
-
- batch = append(batch, pointerBlob.Pointer)
- if len(batch) >= client.BatchSize() {
- if err := uploadObjects(batch); err != nil {
- return err
- }
- batch = nil
- }
- }
- if len(batch) > 0 {
- if err := uploadObjects(batch); err != nil {
- return err
- }
- }
-
- err, has := <-errChan
- if has {
- log.Error("Error enumerating LFS objects for repository: %v", err)
- return err
- }
-
- return nil
- }
|