本站源代码
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

402 lines
11KB

  1. // Copyright 2017 The Gitea 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 models
  5. import (
  6. "fmt"
  7. "strconv"
  8. "strings"
  9. "time"
  10. "code.gitea.io/gitea/modules/base"
  11. "code.gitea.io/gitea/modules/charset"
  12. "code.gitea.io/gitea/modules/git"
  13. "code.gitea.io/gitea/modules/graceful"
  14. "code.gitea.io/gitea/modules/indexer"
  15. "code.gitea.io/gitea/modules/log"
  16. "code.gitea.io/gitea/modules/setting"
  17. "github.com/ethantkoenig/rupture"
  18. )
  19. // RepoIndexerStatus status of a repo's entry in the repo indexer
  20. // For now, implicitly refers to default branch
  21. type RepoIndexerStatus struct {
  22. ID int64 `xorm:"pk autoincr"`
  23. RepoID int64 `xorm:"INDEX"`
  24. CommitSha string `xorm:"VARCHAR(40)"`
  25. }
  26. func (repo *Repository) getIndexerStatus() error {
  27. if repo.IndexerStatus != nil {
  28. return nil
  29. }
  30. status := &RepoIndexerStatus{RepoID: repo.ID}
  31. has, err := x.Get(status)
  32. if err != nil {
  33. return err
  34. } else if !has {
  35. status.CommitSha = ""
  36. }
  37. repo.IndexerStatus = status
  38. return nil
  39. }
  40. func (repo *Repository) updateIndexerStatus(sha string) error {
  41. if err := repo.getIndexerStatus(); err != nil {
  42. return err
  43. }
  44. if len(repo.IndexerStatus.CommitSha) == 0 {
  45. repo.IndexerStatus.CommitSha = sha
  46. _, err := x.Insert(repo.IndexerStatus)
  47. return err
  48. }
  49. repo.IndexerStatus.CommitSha = sha
  50. _, err := x.ID(repo.IndexerStatus.ID).Cols("commit_sha").
  51. Update(repo.IndexerStatus)
  52. return err
  53. }
  54. type repoIndexerOperation struct {
  55. repo *Repository
  56. deleted bool
  57. watchers []chan<- error
  58. }
  59. var repoIndexerOperationQueue chan repoIndexerOperation
  60. // InitRepoIndexer initialize the repo indexer
  61. func InitRepoIndexer() {
  62. if !setting.Indexer.RepoIndexerEnabled {
  63. return
  64. }
  65. waitChannel := make(chan time.Duration)
  66. repoIndexerOperationQueue = make(chan repoIndexerOperation, setting.Indexer.UpdateQueueLength)
  67. go func() {
  68. start := time.Now()
  69. log.Info("Initializing Repository Indexer")
  70. indexer.InitRepoIndexer(populateRepoIndexerAsynchronously)
  71. go processRepoIndexerOperationQueue()
  72. waitChannel <- time.Since(start)
  73. }()
  74. if setting.Indexer.StartupTimeout > 0 {
  75. go func() {
  76. timeout := setting.Indexer.StartupTimeout
  77. if graceful.IsChild && setting.GracefulHammerTime > 0 {
  78. timeout += setting.GracefulHammerTime
  79. }
  80. select {
  81. case duration := <-waitChannel:
  82. log.Info("Repository Indexer Initialization took %v", duration)
  83. case <-time.After(timeout):
  84. log.Fatal("Repository Indexer Initialization Timed-Out after: %v", timeout)
  85. }
  86. }()
  87. }
  88. }
  89. // populateRepoIndexerAsynchronously asynchronously populates the repo indexer
  90. // with pre-existing data. This should only be run when the indexer is created
  91. // for the first time.
  92. func populateRepoIndexerAsynchronously() error {
  93. exist, err := x.Table("repository").Exist()
  94. if err != nil {
  95. return err
  96. } else if !exist {
  97. return nil
  98. }
  99. // if there is any existing repo indexer metadata in the DB, delete it
  100. // since we are starting afresh. Also, xorm requires deletes to have a
  101. // condition, and we want to delete everything, thus 1=1.
  102. if _, err := x.Where("1=1").Delete(new(RepoIndexerStatus)); err != nil {
  103. return err
  104. }
  105. var maxRepoID int64
  106. if _, err = x.Select("MAX(id)").Table("repository").Get(&maxRepoID); err != nil {
  107. return err
  108. }
  109. go populateRepoIndexer(maxRepoID)
  110. return nil
  111. }
  112. // populateRepoIndexer populate the repo indexer with pre-existing data. This
  113. // should only be run when the indexer is created for the first time.
  114. func populateRepoIndexer(maxRepoID int64) {
  115. log.Info("Populating the repo indexer with existing repositories")
  116. // start with the maximum existing repo ID and work backwards, so that we
  117. // don't include repos that are created after gitea starts; such repos will
  118. // already be added to the indexer, and we don't need to add them again.
  119. for maxRepoID > 0 {
  120. repos := make([]*Repository, 0, RepositoryListDefaultPageSize)
  121. err := x.Where("id <= ?", maxRepoID).
  122. OrderBy("id DESC").
  123. Limit(RepositoryListDefaultPageSize).
  124. Find(&repos)
  125. if err != nil {
  126. log.Error("populateRepoIndexer: %v", err)
  127. return
  128. } else if len(repos) == 0 {
  129. break
  130. }
  131. for _, repo := range repos {
  132. repoIndexerOperationQueue <- repoIndexerOperation{
  133. repo: repo,
  134. deleted: false,
  135. }
  136. maxRepoID = repo.ID - 1
  137. }
  138. }
  139. log.Info("Done populating the repo indexer with existing repositories")
  140. }
  141. func updateRepoIndexer(repo *Repository) error {
  142. sha, err := getDefaultBranchSha(repo)
  143. if err != nil {
  144. return err
  145. }
  146. changes, err := getRepoChanges(repo, sha)
  147. if err != nil {
  148. return err
  149. } else if changes == nil {
  150. return nil
  151. }
  152. batch := indexer.RepoIndexerBatch()
  153. for _, update := range changes.Updates {
  154. if err := addUpdate(update, repo, batch); err != nil {
  155. return err
  156. }
  157. }
  158. for _, filename := range changes.RemovedFilenames {
  159. if err := addDelete(filename, repo, batch); err != nil {
  160. return err
  161. }
  162. }
  163. if err = batch.Flush(); err != nil {
  164. return err
  165. }
  166. return repo.updateIndexerStatus(sha)
  167. }
  168. // repoChanges changes (file additions/updates/removals) to a repo
  169. type repoChanges struct {
  170. Updates []fileUpdate
  171. RemovedFilenames []string
  172. }
  173. type fileUpdate struct {
  174. Filename string
  175. BlobSha string
  176. }
  177. func getDefaultBranchSha(repo *Repository) (string, error) {
  178. stdout, err := git.NewCommand("show-ref", "-s", repo.DefaultBranch).RunInDir(repo.RepoPath())
  179. if err != nil {
  180. return "", err
  181. }
  182. return strings.TrimSpace(stdout), nil
  183. }
  184. // getRepoChanges returns changes to repo since last indexer update
  185. func getRepoChanges(repo *Repository, revision string) (*repoChanges, error) {
  186. if err := repo.getIndexerStatus(); err != nil {
  187. return nil, err
  188. }
  189. if len(repo.IndexerStatus.CommitSha) == 0 {
  190. return genesisChanges(repo, revision)
  191. }
  192. return nonGenesisChanges(repo, revision)
  193. }
  194. func addUpdate(update fileUpdate, repo *Repository, batch rupture.FlushingBatch) error {
  195. stdout, err := git.NewCommand("cat-file", "-s", update.BlobSha).
  196. RunInDir(repo.RepoPath())
  197. if err != nil {
  198. return err
  199. }
  200. if size, err := strconv.Atoi(strings.TrimSpace(stdout)); err != nil {
  201. return fmt.Errorf("Misformatted git cat-file output: %v", err)
  202. } else if int64(size) > setting.Indexer.MaxIndexerFileSize {
  203. return addDelete(update.Filename, repo, batch)
  204. }
  205. fileContents, err := git.NewCommand("cat-file", "blob", update.BlobSha).
  206. RunInDirBytes(repo.RepoPath())
  207. if err != nil {
  208. return err
  209. } else if !base.IsTextFile(fileContents) {
  210. // FIXME: UTF-16 files will probably fail here
  211. return nil
  212. }
  213. indexerUpdate := indexer.RepoIndexerUpdate{
  214. Filepath: update.Filename,
  215. Op: indexer.RepoIndexerOpUpdate,
  216. Data: &indexer.RepoIndexerData{
  217. RepoID: repo.ID,
  218. Content: string(charset.ToUTF8DropErrors(fileContents)),
  219. },
  220. }
  221. return indexerUpdate.AddToFlushingBatch(batch)
  222. }
  223. func addDelete(filename string, repo *Repository, batch rupture.FlushingBatch) error {
  224. indexerUpdate := indexer.RepoIndexerUpdate{
  225. Filepath: filename,
  226. Op: indexer.RepoIndexerOpDelete,
  227. Data: &indexer.RepoIndexerData{
  228. RepoID: repo.ID,
  229. },
  230. }
  231. return indexerUpdate.AddToFlushingBatch(batch)
  232. }
  233. func isIndexable(entry *git.TreeEntry) bool {
  234. if !entry.IsRegular() && !entry.IsExecutable() {
  235. return false
  236. }
  237. name := strings.ToLower(entry.Name())
  238. for _, g := range setting.Indexer.ExcludePatterns {
  239. if g.Match(name) {
  240. return false
  241. }
  242. }
  243. for _, g := range setting.Indexer.IncludePatterns {
  244. if g.Match(name) {
  245. return true
  246. }
  247. }
  248. return len(setting.Indexer.IncludePatterns) == 0
  249. }
  250. // parseGitLsTreeOutput parses the output of a `git ls-tree -r --full-name` command
  251. func parseGitLsTreeOutput(stdout []byte) ([]fileUpdate, error) {
  252. entries, err := git.ParseTreeEntries(stdout)
  253. if err != nil {
  254. return nil, err
  255. }
  256. var idxCount = 0
  257. updates := make([]fileUpdate, len(entries))
  258. for _, entry := range entries {
  259. if isIndexable(entry) {
  260. updates[idxCount] = fileUpdate{
  261. Filename: entry.Name(),
  262. BlobSha: entry.ID.String(),
  263. }
  264. idxCount++
  265. }
  266. }
  267. return updates[:idxCount], nil
  268. }
  269. // genesisChanges get changes to add repo to the indexer for the first time
  270. func genesisChanges(repo *Repository, revision string) (*repoChanges, error) {
  271. var changes repoChanges
  272. stdout, err := git.NewCommand("ls-tree", "--full-tree", "-r", revision).
  273. RunInDirBytes(repo.RepoPath())
  274. if err != nil {
  275. return nil, err
  276. }
  277. changes.Updates, err = parseGitLsTreeOutput(stdout)
  278. return &changes, err
  279. }
  280. // nonGenesisChanges get changes since the previous indexer update
  281. func nonGenesisChanges(repo *Repository, revision string) (*repoChanges, error) {
  282. diffCmd := git.NewCommand("diff", "--name-status",
  283. repo.IndexerStatus.CommitSha, revision)
  284. stdout, err := diffCmd.RunInDir(repo.RepoPath())
  285. if err != nil {
  286. // previous commit sha may have been removed by a force push, so
  287. // try rebuilding from scratch
  288. log.Warn("git diff: %v", err)
  289. if err = indexer.DeleteRepoFromIndexer(repo.ID); err != nil {
  290. return nil, err
  291. }
  292. return genesisChanges(repo, revision)
  293. }
  294. var changes repoChanges
  295. updatedFilenames := make([]string, 0, 10)
  296. for _, line := range strings.Split(stdout, "\n") {
  297. line = strings.TrimSpace(line)
  298. if len(line) == 0 {
  299. continue
  300. }
  301. filename := strings.TrimSpace(line[1:])
  302. if len(filename) == 0 {
  303. continue
  304. } else if filename[0] == '"' {
  305. filename, err = strconv.Unquote(filename)
  306. if err != nil {
  307. return nil, err
  308. }
  309. }
  310. switch status := line[0]; status {
  311. case 'M', 'A':
  312. updatedFilenames = append(updatedFilenames, filename)
  313. case 'D':
  314. changes.RemovedFilenames = append(changes.RemovedFilenames, filename)
  315. default:
  316. log.Warn("Unrecognized status: %c (line=%s)", status, line)
  317. }
  318. }
  319. cmd := git.NewCommand("ls-tree", "--full-tree", revision, "--")
  320. cmd.AddArguments(updatedFilenames...)
  321. lsTreeStdout, err := cmd.RunInDirBytes(repo.RepoPath())
  322. if err != nil {
  323. return nil, err
  324. }
  325. changes.Updates, err = parseGitLsTreeOutput(lsTreeStdout)
  326. return &changes, err
  327. }
  328. func processRepoIndexerOperationQueue() {
  329. for {
  330. op := <-repoIndexerOperationQueue
  331. var err error
  332. if op.deleted {
  333. if err = indexer.DeleteRepoFromIndexer(op.repo.ID); err != nil {
  334. log.Error("DeleteRepoFromIndexer: %v", err)
  335. }
  336. } else {
  337. if err = updateRepoIndexer(op.repo); err != nil {
  338. log.Error("updateRepoIndexer: %v", err)
  339. }
  340. }
  341. for _, watcher := range op.watchers {
  342. watcher <- err
  343. }
  344. }
  345. }
  346. // DeleteRepoFromIndexer remove all of a repository's entries from the indexer
  347. func DeleteRepoFromIndexer(repo *Repository, watchers ...chan<- error) {
  348. addOperationToQueue(repoIndexerOperation{repo: repo, deleted: true, watchers: watchers})
  349. }
  350. // UpdateRepoIndexer update a repository's entries in the indexer
  351. func UpdateRepoIndexer(repo *Repository, watchers ...chan<- error) {
  352. addOperationToQueue(repoIndexerOperation{repo: repo, deleted: false, watchers: watchers})
  353. }
  354. func addOperationToQueue(op repoIndexerOperation) {
  355. if !setting.Indexer.RepoIndexerEnabled {
  356. return
  357. }
  358. select {
  359. case repoIndexerOperationQueue <- op:
  360. break
  361. default:
  362. go func() {
  363. repoIndexerOperationQueue <- op
  364. }()
  365. }
  366. }
上海开阖软件有限公司 沪ICP备12045867号-1