fork of https://github.com/sourcegraph/zoekt
indexserver: index new repos (#212)
We only calculate options of repositories that have changed since our
last check. However, if we are informed of a new repository and the
repository hasn't changed, we will never fetch its options. This can
happen if our cluster grows or shrinks.
This commit expands the interface to have a "force" option to fetch
options. I don't like this design, and will address it in follow-up
commits. However, for now I'd like to fix the bug.
I think the reason this is becoming complicated is the state stored by
"IterateIndexOptions" around search configuration fingerprint. I think a
better solution is to add an API call to calculate a new fingerprint and
then make the setting/updating of the fingerprint more explicit.
author
Keegan Carruthers-Smith
committer
GitHub
date
(Nov 12, 2021, 3:49 PM +0200)
commit
3c5cea34
3c5cea345db99fa04213451d9f895e9ccb6adcc8
parent
8ea6c201
8ea6c2019cbf078dcb2db1f2f8b1f123c00534ab
+52
-7
+5
-2
cmd/zoekt-sourcegraph-indexserver/main.go
+5
-2
cmd/zoekt-sourcegraph-indexserver/main.go
···
292
292
// IterateIndexOptions will only iterate over repositories that have
293
293
// changed since we last called list. However, we want to add all IDs
294
294
// back onto the queue just to check that what is on disk is still
295
-
// correct. This will use the last IndexOptions we stored in the queue.
296
-
s.queue.Bump(repos.IDs)
295
+
// correct. This will use the last IndexOptions we stored in the
296
+
// queue. The repositories not on the queue (missing) need a forced
297
+
// fetch of IndexOptions.
298
+
missing := s.queue.Bump(repos.IDs)
299
+
s.Sourcegraph.ForceIterateIndexOptions(s.queue.AddOrUpdate, missing...)
297
300
298
301
<-cleanupDone
299
302
}
+9
-3
cmd/zoekt-sourcegraph-indexserver/queue.go
+9
-3
cmd/zoekt-sourcegraph-indexserver/queue.go
···
90
90
}
91
91
92
92
// Bump will take any repository in ids which is not on the queue and
93
-
// re-insert it with the last known IndexOptions.
94
-
func (q *Queue) Bump(ids []uint32) {
93
+
// re-insert it with the last known IndexOptions. Bump returns ids that are
94
+
// unknown to the queue.
95
+
func (q *Queue) Bump(ids []uint32) []uint32 {
95
96
q.mu.Lock()
96
97
defer q.mu.Unlock()
97
98
···
99
100
q.init()
100
101
}
101
102
103
+
var missing []uint32
102
104
for _, id := range ids {
103
105
item, ok := q.items[id]
104
-
if ok && item.heapIdx < 0 {
106
+
if !ok {
107
+
missing = append(missing, id)
108
+
} else if item.heapIdx < 0 {
105
109
q.seq++
106
110
item.seq = q.seq
107
111
heap.Push(&q.pq, item)
···
109
113
metricQueueCap.Set(float64(len(q.items)))
110
114
}
111
115
}
116
+
117
+
return missing
112
118
}
113
119
114
120
// Iterate will call f on each item known to the queue, including items that
+5
-2
cmd/zoekt-sourcegraph-indexserver/queue_test.go
+5
-2
cmd/zoekt-sourcegraph-indexserver/queue_test.go
···
100
100
}
101
101
102
102
// Bump 2 and 3. 3 doesn't exist, so only 2 should exist.
103
-
queue.Bump([]uint32{2, 3})
103
+
missing := queue.Bump([]uint32{2, 3})
104
+
if d := cmp.Diff([]uint32{3}, missing); d != "" {
105
+
t.Errorf("unexpected missing (-want, +got):\n%s", d)
106
+
}
104
107
105
108
want := []IndexOptions{{RepoID: 2, Name: "bar"}}
106
109
var got []IndexOptions
···
113
116
}
114
117
115
118
if d := cmp.Diff(want, got); d != "" {
116
-
t.Fatalf("(-want, +got):\n%s", d)
119
+
t.Errorf("unexpected items bumped into the queue (-want, +got):\n%s", d)
117
120
}
118
121
}
119
122
+33
cmd/zoekt-sourcegraph-indexserver/sg.go
+33
cmd/zoekt-sourcegraph-indexserver/sg.go
···
31
31
32
32
// IterateIndexOptions best effort resolves the IndexOptions for RepoIDs. If
33
33
// any repository fails it internally logs.
34
+
//
35
+
// Note: this has a side-effect of setting a the "config fingerprint". The
36
+
// config fingerprint means we only calculate index options for repositories
37
+
// that have changed since the last call to IterateIndexOptions. If you want
38
+
// to force calculation of index options use Sourcegraph.GetIndexOptions.
34
39
IterateIndexOptions func(func(IndexOptions))
35
40
}
36
41
37
42
type Sourcegraph interface {
38
43
List(ctx context.Context, indexed []uint32) (*SourcegraphListResult, error)
44
+
45
+
// ForceIterateIndexOptions will best-effort calculate the index options for
46
+
// all of ids. If any repository fails it internally logs.
47
+
ForceIterateIndexOptions(func(IndexOptions), ...uint32)
39
48
40
49
// GetIndexOptions is deprecated but kept around until we improve our
41
50
// forceIndex code.
···
131
140
IDs: repos,
132
141
IterateIndexOptions: iterate,
133
142
}, nil
143
+
}
144
+
145
+
func (s *sourcegraphClient) ForceIterateIndexOptions(f func(IndexOptions), repos ...uint32) {
146
+
opts, err := s.GetIndexOptions(repos...)
147
+
if err != nil {
148
+
return
149
+
}
150
+
for _, o := range opts {
151
+
if o.Error == "" {
152
+
f(o.IndexOptions)
153
+
}
154
+
}
134
155
}
135
156
136
157
// indexOptionsItem wraps IndexOptions to also include an error returned by
···
270
291
IDs: repos,
271
292
IterateIndexOptions: iterate,
272
293
}, nil
294
+
}
295
+
296
+
func (sf sourcegraphFake) ForceIterateIndexOptions(f func(IndexOptions), repos ...uint32) {
297
+
opts, err := sf.GetIndexOptions(repos...)
298
+
if err != nil {
299
+
return
300
+
}
301
+
for _, o := range opts {
302
+
if o.Error == "" {
303
+
f(o.IndexOptions)
304
+
}
305
+
}
273
306
}
274
307
275
308
func (sf sourcegraphFake) GetIndexOptions(repos ...uint32) ([]indexOptionsItem, error) {