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
|
package catfile
import (
"context"
"fmt"
"io"
"sync"
"time"
"github.com/prometheus/client_golang/prometheus"
"gitlab.com/gitlab-org/gitaly-proto/go/gitalypb"
"gitlab.com/gitlab-org/gitaly/internal/git/alternates"
"gitlab.com/gitlab-org/gitaly/internal/metadata"
"gitlab.com/gitlab-org/gitaly/internal/metadata/featureflag"
)
var catfileCacheHitOrMiss = prometheus.NewCounterVec(
prometheus.CounterOpts{
Name: "gitaly_catfile_cache_total",
Help: "Counter of catfile cache hit/miss",
},
[]string{"type"},
)
var currentCatfileProcesses = prometheus.NewGauge(
prometheus.GaugeOpts{
Name: "gitaly_catfile_processes",
Help: "Gauge of active catfile processes",
},
)
var totalCatfileProcesses = prometheus.NewCounter(
prometheus.CounterOpts{
Name: "gitaly_catfile_processes_total",
Help: "Counter of catfile processes",
},
)
// DefaultBatchfileTTL is the default ttl for batch files to live in the cache
var DefaultBatchfileTTL = 10 * time.Second
func init() {
prometheus.MustRegister(catfileCacheHitOrMiss)
prometheus.MustRegister(currentCatfileProcesses)
prometheus.MustRegister(totalCatfileProcesses)
}
// Batch abstracts 'git cat-file --batch' and 'git cat-file --batch-check'.
// It lets you retrieve object metadata and raw objects from a Git repo.
//
// A Batch instance can only serve single request at a time. If you want to
// use it across multiple goroutines you need to add your own locking.
type Batch struct {
sync.Mutex
*batchCheck
*batch
cancel func()
closed bool
}
// Info returns an ObjectInfo if spec exists. If spec does not exist the
// error is of type NotFoundError.
func (c *Batch) Info(revspec string) (*ObjectInfo, error) {
return c.batchCheck.info(revspec)
}
// Tree returns a raw tree object. It is an error if revspec does not
// point to a tree. To prevent this firstuse Info to resolve the revspec
// and check the object type. Caller must consume the Reader before
// making another call on C.
func (c *Batch) Tree(revspec string) (io.Reader, error) {
return c.batch.reader(revspec, "tree")
}
// Commit returns a raw commit object. It is an error if revspec does not
// point to a commit. To prevent this first use Info to resolve the revspec
// and check the object type. Caller must consume the Reader before
// making another call on C.
func (c *Batch) Commit(revspec string) (io.Reader, error) {
return c.batch.reader(revspec, "commit")
}
// Blob returns a reader for the requested blob. The entire blob must be
// read before any new objects can be requested from this Batch instance.
//
// It is an error if revspec does not point to a blob. To prevent this
// first use Info to resolve the revspec and check the object type.
func (c *Batch) Blob(revspec string) (io.Reader, error) {
return c.batch.reader(revspec, "blob")
}
// Tag returns a raw tag object. Caller must consume the Reader before
// making another call on C.
func (c *Batch) Tag(revspec string) (io.Reader, error) {
return c.batch.reader(revspec, "tag")
}
// Close closes the writers for batchCheck and batch. This is only used for
// cached Batches
func (c *Batch) Close() {
c.Lock()
defer c.Unlock()
if c.closed {
return
}
c.closed = true
if c.cancel != nil {
// both c.batch and c.batchCheck have goroutines that listen on <ctx.Done()
// when this is cancelled, it will cause those goroutines to close both writers
c.cancel()
}
}
func (c *Batch) isClosed() bool {
c.Lock()
defer c.Unlock()
return c.closed
}
// HasUnreadData returns a boolean specifying whether or not the Batch has more
// data still to be read
func (c *Batch) HasUnreadData() bool {
return c.n > 1
}
// New returns a new Batch instance. It is important that ctx gets canceled
// somewhere, because if it doesn't the cat-file processes spawned by
// New() never terminate.
func New(ctx context.Context, repo *gitalypb.Repository) (*Batch, error) {
if ctx.Done() == nil {
panic("empty ctx.Done() in catfile.Batch.New()")
}
repoPath, env, err := alternates.PathAndEnv(repo)
if err != nil {
return nil, err
}
sessionID := metadata.GetValue(ctx, "gitaly-session-id")
if featureflag.IsDisabled(ctx, CacheFeatureFlagKey) || sessionID == "" {
// if caching us used, the caller is responsible for putting the catfile
// into the cache
batch, err := newBatch(ctx, repoPath, env)
if err != nil {
return nil, err
}
batchCheck, err := newBatchCheck(ctx, repoPath, env)
if err != nil {
return nil, err
}
return &Batch{batch: batch, batchCheck: batchCheck}, nil
}
cacheKey := NewCacheKey(sessionID, repo)
c := cache.Get(cacheKey)
defer func() {
go cache.returnToCache(ctx, cacheKey, c)
}()
if c != nil {
catfileCacheHitOrMiss.WithLabelValues("hit").Inc()
cache.Del(cacheKey)
return c, nil
}
catfileCacheHitOrMiss.WithLabelValues("miss").Inc()
// if we are using caching, create a fresh context for the new batch
// and initialize the new batch with a cache key and cancel function
cacheCtx, cacheCancel := context.WithCancel(context.Background())
c = &Batch{cancel: cacheCancel}
c.batch, err = newBatch(cacheCtx, repoPath, env)
if err != nil {
return nil, fmt.Errorf("error when creating new batch: %v", err)
}
c.batchCheck, err = newBatchCheck(cacheCtx, repoPath, env)
if err != nil {
return nil, fmt.Errorf("error when creating new batch check: %v", err)
}
return c, nil
}
|
