1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
|
package repoutil
import (
"bytes"
"context"
"fmt"
"io"
"io/fs"
"os"
"path/filepath"
"time"
"gitlab.com/gitlab-org/gitaly/v16/internal/git"
"gitlab.com/gitlab-org/gitaly/v16/internal/git/repository"
"gitlab.com/gitlab-org/gitaly/v16/internal/git/stats"
"gitlab.com/gitlab-org/gitaly/v16/internal/gitaly/storage"
"gitlab.com/gitlab-org/gitaly/v16/internal/gitaly/transaction"
"gitlab.com/gitlab-org/gitaly/v16/internal/structerr"
"gitlab.com/gitlab-org/gitaly/v16/internal/tempdir"
"gitlab.com/gitlab-org/gitaly/v16/internal/transaction/voting"
"gitlab.com/gitlab-org/gitaly/v16/proto/go/gitalypb"
)
type createConfig struct {
gitOptions []git.Option
skipInit bool
}
// CreateOption is an option that can be passed to Create.
type CreateOption func(cfg *createConfig)
// WithBranchName overrides the default branch name that is to be used when creating the repository.
// If called with an empty string then the default branch name will not be changed.
func WithBranchName(branch string) CreateOption {
return func(cfg *createConfig) {
if branch == "" {
return
}
cfg.gitOptions = append(cfg.gitOptions, git.ValueFlag{Name: "--initial-branch", Value: branch})
}
}
// WithObjectHash overrides the default object hash of the created repository.
func WithObjectHash(hash git.ObjectHash) CreateOption {
return func(cfg *createConfig) {
cfg.gitOptions = append(cfg.gitOptions, git.ValueFlag{Name: "--object-format", Value: hash.Format})
}
}
// WithSkipInit causes Create to skip calling git-init(1) so that the seeding function will be
// called with a nonexistent target directory. This can be useful when using git-clone(1) to seed
// the repository.
func WithSkipInit() CreateOption {
return func(cfg *createConfig) {
cfg.skipInit = true
}
}
// Create will create a new repository in a race-free way with proper transactional semantics. The
// repository will only be created if it doesn't yet exist and if nodes which take part in the
// transaction reach quorum. Otherwise, the target path of the new repository will not be modified.
// The repository can optionally be seeded with contents
func Create(
ctx context.Context,
locator storage.Locator,
gitCmdFactory git.CommandFactory,
txManager transaction.Manager,
repository repository.GitRepo,
seedRepository func(repository *gitalypb.Repository) error,
options ...CreateOption,
) error {
targetPath, err := locator.GetPath(repository)
if err != nil {
return structerr.NewInvalidArgument("locate repository: %w", err)
}
// The repository must not exist on disk already, or otherwise we won't be able to
// create it with atomic semantics.
if _, err := os.Stat(targetPath); !os.IsNotExist(err) {
return structerr.NewAlreadyExists("repository exists already")
}
newRepo, newRepoDir, err := tempdir.NewRepository(ctx, repository.GetStorageName(), locator)
if err != nil {
return fmt.Errorf("creating temporary repository: %w", err)
}
defer func() {
// We don't really care about whether this succeeds or not. It will either get
// cleaned up after the context is done, or eventually by the tempdir walker when
// it's old enough.
_ = os.RemoveAll(newRepoDir.Path())
}()
// Note that we do not create the repository directly in its target location, but
// instead create it in a temporary directory, first. This is done such that we can
// guarantee atomicity and roll back the change easily in case an error happens.
var cfg createConfig
for _, option := range options {
option(&cfg)
}
if !cfg.skipInit {
stderr := &bytes.Buffer{}
cmd, err := gitCmdFactory.NewWithoutRepo(ctx, git.Command{
Name: "init",
Flags: append([]git.Option{
git.Flag{Name: "--bare"},
git.Flag{Name: "--quiet"},
}, cfg.gitOptions...),
Args: []string{newRepoDir.Path()},
}, git.WithStderr(stderr))
if err != nil {
return fmt.Errorf("spawning git-init: %w", err)
}
if err := cmd.Wait(); err != nil {
return fmt.Errorf("creating repository: %w, stderr: %q", err, stderr.String())
}
} else {
if err := os.Remove(newRepoDir.Path()); err != nil {
return fmt.Errorf("removing precreated directory: %w", err)
}
}
if err := seedRepository(newRepo); err != nil {
// Return the error returned by the callback function as-is so we don't clobber any
// potential returned gRPC error codes.
return err
}
// In order to guarantee that the repository is going to be the same across all
// Gitalies in case we're behind Praefect, we walk the repository and hash all of
// its files.
voteHash := voting.NewVoteHash()
if err := filepath.WalkDir(newRepoDir.Path(), func(path string, entry fs.DirEntry, err error) error {
if err != nil {
return err
}
switch path {
// The way packfiles are generated may not be deterministic, so we skip over the
// object database.
case filepath.Join(newRepoDir.Path(), "objects"):
return fs.SkipDir
// FETCH_HEAD refers to the remote we're fetching from. This URL may not be
// deterministic, e.g. when fetching from a temporary file like we do in
// CreateRepositoryFromBundle.
case filepath.Join(newRepoDir.Path(), "FETCH_HEAD"):
return nil
}
// We do not care about directories.
if entry.IsDir() {
return nil
}
file, err := os.Open(path)
if err != nil {
return fmt.Errorf("opening %q: %w", entry.Name(), err)
}
defer file.Close()
if _, err := io.Copy(voteHash, file); err != nil {
return fmt.Errorf("hashing %q: %w", entry.Name(), err)
}
return nil
}); err != nil {
return fmt.Errorf("walking repository: %w", err)
}
vote, err := voteHash.Vote()
if err != nil {
return fmt.Errorf("computing vote: %w", err)
}
// Create the full-repack timestamp in the new repository. This is done so that we don't
// consider new repositories to have never been repacked yet, which would cause repository
// housekeeping to perform a full repack right away. And in general, this would not really
// be needed as the end result for most of the repository-creating RPCs would be a either an
// empty or a neatly-packed repository anyway.
//
// As this timestamp should never impact the user-observable state of a repository we do not
// include it in the voting hash.
if err := stats.UpdateFullRepackTimestamp(newRepoDir.Path(), time.Now()); err != nil {
return fmt.Errorf("creating full-repack timestamp: %w", err)
}
// We're now entering the critical section where we want to have exclusive access
// over creation of the repository. So we:
//
// 1. Lock the repository path such that no other process can create it at the same
// time.
// 2. Vote on the new repository's state.
// 3. Move the repository into place.
// 4. Do another confirmatory vote to signal that we performed the change.
// 5. Unlock the repository again.
//
// This sequence guarantees that the change is atomic and can trivially be rolled
// back in case we fail to either lock the repository or reach quorum in the initial
// vote.
unlock, err := Lock(ctx, locator, repository)
if err != nil {
return fmt.Errorf("locking repository: %w", err)
}
defer unlock()
// Now that the repository is locked, we must assert that it _still_ doesn't exist.
// Otherwise, it could have happened that a concurrent RPC call created it while we created
// and seeded our temporary repository. While we would notice this at the point of moving
// the repository into place, we want to be as sure as possible that the action will succeed
// previous to the first transactional vote.
if _, err := os.Stat(targetPath); !os.IsNotExist(err) {
return structerr.NewAlreadyExists("repository exists already")
}
if err := transaction.VoteOnContext(ctx, txManager, vote, voting.Prepared); err != nil {
return structerr.NewFailedPrecondition("preparatory vote: %w", err)
}
// Now that we have locked the repository and all Gitalies have agreed that they
// want to do the same change we can move the repository into place.
if err := os.Rename(newRepoDir.Path(), targetPath); err != nil {
return fmt.Errorf("moving repository into place: %w", err)
}
if err := transaction.VoteOnContext(ctx, txManager, vote, voting.Committed); err != nil {
return structerr.NewFailedPrecondition("committing vote: %w", err)
}
// We unlock the repository implicitly via the deferred `Close()` call.
return nil
}
|