mirror.go 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454
  1. // Copyright 2016 The Gogs Authors. All rights reserved.
  2. // Use of this source code is governed by a MIT-style
  3. // license that can be found in the LICENSE file.
  4. package db
  5. import (
  6. "fmt"
  7. "net/url"
  8. "strings"
  9. "time"
  10. "github.com/unknwon/com"
  11. "gopkg.in/ini.v1"
  12. log "unknwon.dev/clog/v2"
  13. "xorm.io/xorm"
  14. "github.com/gogs/git-module"
  15. "gogs.io/gogs/internal/conf"
  16. "gogs.io/gogs/internal/db/errors"
  17. "gogs.io/gogs/internal/process"
  18. "gogs.io/gogs/internal/sync"
  19. )
  20. var MirrorQueue = sync.NewUniqueQueue(1000)
  21. // Mirror represents mirror information of a repository.
  22. type Mirror struct {
  23. ID int64
  24. RepoID int64
  25. Repo *Repository `xorm:"-" json:"-"`
  26. Interval int // Hour.
  27. EnablePrune bool `xorm:"NOT NULL DEFAULT true"`
  28. // Last and next sync time of Git data from upstream
  29. LastSync time.Time `xorm:"-" json:"-"`
  30. LastSyncUnix int64 `xorm:"updated_unix"`
  31. NextSync time.Time `xorm:"-" json:"-"`
  32. NextSyncUnix int64 `xorm:"next_update_unix"`
  33. address string `xorm:"-" json:"-"`
  34. }
  35. func (m *Mirror) BeforeInsert() {
  36. m.NextSyncUnix = m.NextSync.Unix()
  37. }
  38. func (m *Mirror) BeforeUpdate() {
  39. m.LastSyncUnix = m.LastSync.Unix()
  40. m.NextSyncUnix = m.NextSync.Unix()
  41. }
  42. func (m *Mirror) AfterSet(colName string, _ xorm.Cell) {
  43. var err error
  44. switch colName {
  45. case "repo_id":
  46. m.Repo, err = GetRepositoryByID(m.RepoID)
  47. if err != nil {
  48. log.Error("GetRepositoryByID [%d]: %v", m.ID, err)
  49. }
  50. case "updated_unix":
  51. m.LastSync = time.Unix(m.LastSyncUnix, 0).Local()
  52. case "next_update_unix":
  53. m.NextSync = time.Unix(m.NextSyncUnix, 0).Local()
  54. }
  55. }
  56. // ScheduleNextSync calculates and sets next sync time based on repostiroy mirror setting.
  57. func (m *Mirror) ScheduleNextSync() {
  58. m.NextSync = time.Now().Add(time.Duration(m.Interval) * time.Hour)
  59. }
  60. func (m *Mirror) readAddress() {
  61. if len(m.address) > 0 {
  62. return
  63. }
  64. cfg, err := ini.LoadSources(
  65. ini.LoadOptions{IgnoreInlineComment: true},
  66. m.Repo.GitConfigPath(),
  67. )
  68. if err != nil {
  69. log.Error("load config: %v", err)
  70. return
  71. }
  72. m.address = cfg.Section("remote \"origin\"").Key("url").Value()
  73. }
  74. // HandleMirrorCredentials replaces user credentials from HTTP/HTTPS URL
  75. // with placeholder <credentials>.
  76. // It returns original string if protocol is not HTTP/HTTPS.
  77. // TODO(unknwon): Use url.Parse.
  78. func HandleMirrorCredentials(url string, mosaics bool) string {
  79. i := strings.Index(url, "@")
  80. if i == -1 {
  81. return url
  82. }
  83. start := strings.Index(url, "://")
  84. if start == -1 {
  85. return url
  86. }
  87. if mosaics {
  88. return url[:start+3] + "<credentials>" + url[i:]
  89. }
  90. return url[:start+3] + url[i+1:]
  91. }
  92. // Address returns mirror address from Git repository config without credentials.
  93. func (m *Mirror) Address() string {
  94. m.readAddress()
  95. return HandleMirrorCredentials(m.address, false)
  96. }
  97. // MosaicsAddress returns mirror address from Git repository config with credentials under mosaics.
  98. func (m *Mirror) MosaicsAddress() string {
  99. m.readAddress()
  100. return HandleMirrorCredentials(m.address, true)
  101. }
  102. // RawAddress returns raw mirror address directly from Git repository config.
  103. func (m *Mirror) RawAddress() string {
  104. m.readAddress()
  105. return m.address
  106. }
  107. // SaveAddress writes new address to Git repository config.
  108. func (m *Mirror) SaveAddress(addr string) error {
  109. repoPath := m.Repo.RepoPath()
  110. err := git.RepoRemoveRemote(repoPath, "origin")
  111. if err != nil {
  112. return fmt.Errorf("remove remote 'origin': %v", err)
  113. }
  114. addrURL, err := url.Parse(addr)
  115. if err != nil {
  116. return err
  117. }
  118. err = git.RepoAddRemote(repoPath, "origin", addrURL.String(), git.AddRemoteOptions{MirrorFetch: true})
  119. if err != nil {
  120. return fmt.Errorf("add remote 'origin': %v", err)
  121. }
  122. return nil
  123. }
  124. const gitShortEmptyID = "0000000"
  125. // mirrorSyncResult contains information of a updated reference.
  126. // If the oldCommitID is "0000000", it means a new reference, the value of newCommitID is empty.
  127. // If the newCommitID is "0000000", it means the reference is deleted, the value of oldCommitID is empty.
  128. type mirrorSyncResult struct {
  129. refName string
  130. oldCommitID string
  131. newCommitID string
  132. }
  133. // parseRemoteUpdateOutput detects create, update and delete operations of references from upstream.
  134. func parseRemoteUpdateOutput(output string) []*mirrorSyncResult {
  135. results := make([]*mirrorSyncResult, 0, 3)
  136. lines := strings.Split(output, "\n")
  137. for i := range lines {
  138. // Make sure reference name is presented before continue
  139. idx := strings.Index(lines[i], "-> ")
  140. if idx == -1 {
  141. continue
  142. }
  143. refName := lines[i][idx+3:]
  144. switch {
  145. case strings.HasPrefix(lines[i], " * "): // New reference
  146. results = append(results, &mirrorSyncResult{
  147. refName: refName,
  148. oldCommitID: gitShortEmptyID,
  149. })
  150. case strings.HasPrefix(lines[i], " - "): // Delete reference
  151. results = append(results, &mirrorSyncResult{
  152. refName: refName,
  153. newCommitID: gitShortEmptyID,
  154. })
  155. case strings.HasPrefix(lines[i], " "): // New commits of a reference
  156. delimIdx := strings.Index(lines[i][3:], " ")
  157. if delimIdx == -1 {
  158. log.Error("SHA delimiter not found: %q", lines[i])
  159. continue
  160. }
  161. shas := strings.Split(lines[i][3:delimIdx+3], "..")
  162. if len(shas) != 2 {
  163. log.Error("Expect two SHAs but not what found: %q", lines[i])
  164. continue
  165. }
  166. results = append(results, &mirrorSyncResult{
  167. refName: refName,
  168. oldCommitID: shas[0],
  169. newCommitID: shas[1],
  170. })
  171. default:
  172. log.Warn("parseRemoteUpdateOutput: unexpected update line %q", lines[i])
  173. }
  174. }
  175. return results
  176. }
  177. // runSync returns true if sync finished without error.
  178. func (m *Mirror) runSync() ([]*mirrorSyncResult, bool) {
  179. repoPath := m.Repo.RepoPath()
  180. wikiPath := m.Repo.WikiPath()
  181. timeout := time.Duration(conf.Git.Timeout.Mirror) * time.Second
  182. // Do a fast-fail testing against on repository URL to ensure it is accessible under
  183. // good condition to prevent long blocking on URL resolution without syncing anything.
  184. if !git.IsURLAccessible(time.Minute, m.RawAddress()) {
  185. desc := fmt.Sprintf("Source URL of mirror repository '%s' is not accessible: %s", m.Repo.FullName(), m.MosaicsAddress())
  186. if err := CreateRepositoryNotice(desc); err != nil {
  187. log.Error("CreateRepositoryNotice: %v", err)
  188. }
  189. return nil, false
  190. }
  191. gitArgs := []string{"remote", "update"}
  192. if m.EnablePrune {
  193. gitArgs = append(gitArgs, "--prune")
  194. }
  195. _, stderr, err := process.ExecDir(
  196. timeout, repoPath, fmt.Sprintf("Mirror.runSync: %s", repoPath),
  197. "git", gitArgs...)
  198. if err != nil {
  199. desc := fmt.Sprintf("Failed to update mirror repository '%s': %s", repoPath, stderr)
  200. log.Error(desc)
  201. if err = CreateRepositoryNotice(desc); err != nil {
  202. log.Error("CreateRepositoryNotice: %v", err)
  203. }
  204. return nil, false
  205. }
  206. output := stderr
  207. if err := m.Repo.UpdateSize(); err != nil {
  208. log.Error("UpdateSize [repo_id: %d]: %v", m.Repo.ID, err)
  209. }
  210. if m.Repo.HasWiki() {
  211. // Even if wiki sync failed, we still want results from the main repository
  212. if _, stderr, err := process.ExecDir(
  213. timeout, wikiPath, fmt.Sprintf("Mirror.runSync: %s", wikiPath),
  214. "git", "remote", "update", "--prune"); err != nil {
  215. desc := fmt.Sprintf("Failed to update mirror wiki repository '%s': %s", wikiPath, stderr)
  216. log.Error(desc)
  217. if err = CreateRepositoryNotice(desc); err != nil {
  218. log.Error("CreateRepositoryNotice: %v", err)
  219. }
  220. }
  221. }
  222. return parseRemoteUpdateOutput(output), true
  223. }
  224. func getMirrorByRepoID(e Engine, repoID int64) (*Mirror, error) {
  225. m := &Mirror{RepoID: repoID}
  226. has, err := e.Get(m)
  227. if err != nil {
  228. return nil, err
  229. } else if !has {
  230. return nil, errors.MirrorNotExist{RepoID: repoID}
  231. }
  232. return m, nil
  233. }
  234. // GetMirrorByRepoID returns mirror information of a repository.
  235. func GetMirrorByRepoID(repoID int64) (*Mirror, error) {
  236. return getMirrorByRepoID(x, repoID)
  237. }
  238. func updateMirror(e Engine, m *Mirror) error {
  239. _, err := e.ID(m.ID).AllCols().Update(m)
  240. return err
  241. }
  242. func UpdateMirror(m *Mirror) error {
  243. return updateMirror(x, m)
  244. }
  245. func DeleteMirrorByRepoID(repoID int64) error {
  246. _, err := x.Delete(&Mirror{RepoID: repoID})
  247. return err
  248. }
  249. // MirrorUpdate checks and updates mirror repositories.
  250. func MirrorUpdate() {
  251. if taskStatusTable.IsRunning(_MIRROR_UPDATE) {
  252. return
  253. }
  254. taskStatusTable.Start(_MIRROR_UPDATE)
  255. defer taskStatusTable.Stop(_MIRROR_UPDATE)
  256. log.Trace("Doing: MirrorUpdate")
  257. if err := x.Where("next_update_unix<=?", time.Now().Unix()).Iterate(new(Mirror), func(idx int, bean interface{}) error {
  258. m := bean.(*Mirror)
  259. if m.Repo == nil {
  260. log.Error("Disconnected mirror repository found: %d", m.ID)
  261. return nil
  262. }
  263. MirrorQueue.Add(m.RepoID)
  264. return nil
  265. }); err != nil {
  266. log.Error("MirrorUpdate: %v", err)
  267. }
  268. }
  269. // SyncMirrors checks and syncs mirrors.
  270. // TODO: sync more mirrors at same time.
  271. func SyncMirrors() {
  272. // Start listening on new sync requests.
  273. for repoID := range MirrorQueue.Queue() {
  274. log.Trace("SyncMirrors [repo_id: %s]", repoID)
  275. MirrorQueue.Remove(repoID)
  276. m, err := GetMirrorByRepoID(com.StrTo(repoID).MustInt64())
  277. if err != nil {
  278. log.Error("GetMirrorByRepoID [%d]: %v", m.RepoID, err)
  279. continue
  280. }
  281. results, ok := m.runSync()
  282. if !ok {
  283. continue
  284. }
  285. m.ScheduleNextSync()
  286. if err = UpdateMirror(m); err != nil {
  287. log.Error("UpdateMirror [%d]: %v", m.RepoID, err)
  288. continue
  289. }
  290. // TODO:
  291. // - Create "Mirror Sync" webhook event
  292. // - Create mirror sync (create, push and delete) events and trigger the "mirror sync" webhooks
  293. if len(results) == 0 {
  294. log.Trace("SyncMirrors [repo_id: %d]: no commits fetched", m.RepoID)
  295. }
  296. gitRepo, err := git.Open(m.Repo.RepoPath())
  297. if err != nil {
  298. log.Error("Failed to open repository [repo_id: %d]: %v", m.RepoID, err)
  299. continue
  300. }
  301. for _, result := range results {
  302. // Discard GitHub pull requests, i.e. refs/pull/*
  303. if strings.HasPrefix(result.refName, "refs/pull/") {
  304. continue
  305. }
  306. // Delete reference
  307. if result.newCommitID == gitShortEmptyID {
  308. if err = MirrorSyncDeleteAction(m.Repo, result.refName); err != nil {
  309. log.Error("MirrorSyncDeleteAction [repo_id: %d]: %v", m.RepoID, err)
  310. }
  311. continue
  312. }
  313. // New reference
  314. isNewRef := false
  315. if result.oldCommitID == gitShortEmptyID {
  316. if err = MirrorSyncCreateAction(m.Repo, result.refName); err != nil {
  317. log.Error("MirrorSyncCreateAction [repo_id: %d]: %v", m.RepoID, err)
  318. continue
  319. }
  320. isNewRef = true
  321. }
  322. // Push commits
  323. var commits []*git.Commit
  324. var oldCommitID string
  325. var newCommitID string
  326. if !isNewRef {
  327. oldCommitID, err = gitRepo.RevParse(result.oldCommitID)
  328. if err != nil {
  329. log.Error("Failed to parse revision [repo_id: %d, old_commit_id: %s]: %v", m.RepoID, result.oldCommitID, err)
  330. continue
  331. }
  332. newCommitID, err = gitRepo.RevParse(result.newCommitID)
  333. if err != nil {
  334. log.Error("Failed to parse revision [repo_id: %d, new_commit_id: %s]: %v", m.RepoID, result.newCommitID, err)
  335. continue
  336. }
  337. commits, err = gitRepo.RevList([]string{oldCommitID + "..." + newCommitID})
  338. if err != nil {
  339. log.Error("Failed to list commits [repo_id: %d, old_commit_id: %s, new_commit_id: %s]: %v", m.RepoID, oldCommitID, newCommitID, err)
  340. continue
  341. }
  342. } else if gitRepo.HasBranch(result.refName) {
  343. refNewCommit, err := gitRepo.BranchCommit(result.refName)
  344. if err != nil {
  345. log.Error("Failed to get branch commit [repo_id: %d, branch: %s]: %v", m.RepoID, result.refName, err)
  346. continue
  347. }
  348. // TODO(unknwon): Get the commits for the new ref until the closest ancestor branch like GitHub does.
  349. commits, err = refNewCommit.Ancestors(git.LogOptions{MaxCount: 9})
  350. if err != nil {
  351. log.Error("Failed to get ancestors [repo_id: %d, commit_id: %s]: %v", m.RepoID, refNewCommit.ID, err)
  352. continue
  353. }
  354. // Put the latest commit in front of ancestors
  355. commits = append([]*git.Commit{refNewCommit}, commits...)
  356. oldCommitID = git.EmptyID
  357. newCommitID = refNewCommit.ID.String()
  358. }
  359. if err = MirrorSyncPushAction(m.Repo, MirrorSyncPushActionOptions{
  360. RefName: result.refName,
  361. OldCommitID: oldCommitID,
  362. NewCommitID: newCommitID,
  363. Commits: CommitsToPushCommits(commits),
  364. }); err != nil {
  365. log.Error("MirrorSyncPushAction [repo_id: %d]: %v", m.RepoID, err)
  366. continue
  367. }
  368. }
  369. if _, err = x.Exec("UPDATE mirror SET updated_unix = ? WHERE repo_id = ?", time.Now().Unix(), m.RepoID); err != nil {
  370. log.Error("Update 'mirror.updated_unix' [%d]: %v", m.RepoID, err)
  371. continue
  372. }
  373. // Get latest commit date and compare to current repository updated time,
  374. // update if latest commit date is newer.
  375. latestCommitTime, err := gitRepo.LatestCommitTime()
  376. if err != nil {
  377. log.Error("GetLatestCommitDate [%d]: %v", m.RepoID, err)
  378. continue
  379. } else if !latestCommitTime.After(m.Repo.Updated) {
  380. continue
  381. }
  382. if _, err = x.Exec("UPDATE repository SET updated_unix = ? WHERE id = ?", latestCommitTime.Unix(), m.RepoID); err != nil {
  383. log.Error("Update 'repository.updated_unix' [%d]: %v", m.RepoID, err)
  384. continue
  385. }
  386. }
  387. }
  388. func InitSyncMirrors() {
  389. go SyncMirrors()
  390. }