fork of https://github.com/sourcegraph/zoekt
indexserver: Send list of indexed repos when asking for repos to index
Sourcegraph will now take into account rebalancing when deciding which
repositories to index. As such we need to tell it which repositories we have
indexed such that it can tell us to keep repositories that are not yet indexed
by other replicas.
We also move to using a new endpoint targetted at indexing. Previously we used
the generic "repoList" endpoint.
Additionally we remove requesting the list from sourcegraph out of the debug
request code path. This should improve performance of the page for large
installations.
Change-Id: I1e5872c7a525e0a4734885622bd07f4f5acbc66f
author
Keegan Carruthers-Smith
date
(Oct 31, 2019, 10:50 AM +0200)
commit
5bd7e847
5bd7e84795f082a96956f6ffd1969f2d2c6dc9a2
parent
c1011d80
c1011d80f8fa2b27f19672f1735732b6cc91d109
+40
-29
+31
-20
cmd/zoekt-sourcegraph-indexserver/main.go
+31
-20
cmd/zoekt-sourcegraph-indexserver/main.go
···
4
4
5
5
import (
6
6
"bytes"
7
+
"context"
7
8
"encoding/json"
8
9
"errors"
9
10
"flag"
···
19
20
"os/exec"
20
21
"path/filepath"
21
22
"runtime"
23
+
"sort"
22
24
"strconv"
25
+
"sync"
23
26
"time"
24
27
25
28
"go.uber.org/automaxprocs/maxprocs"
···
72
75
// CPUCount is the amount of parallelism to use when indexing a
73
76
// repository.
74
77
CPUCount int
78
+
79
+
mu sync.Mutex
80
+
lastListRepos []string
75
81
}
76
82
77
83
var debug = log.New(ioutil.Discard, "", log.LstdFlags)
···
109
115
go func() {
110
116
t := time.NewTicker(s.Interval)
111
117
for {
112
-
repos, err := listRepos(s.Hostname, s.Root)
118
+
repos, err := listRepos(context.Background(), s.Hostname, s.Root, listIndexed(s.IndexDir))
113
119
if err != nil {
114
120
log.Println(err)
115
121
<-t.C
116
122
continue
117
123
}
124
+
125
+
s.mu.Lock()
126
+
s.lastListRepos = repos
127
+
s.mu.Unlock()
118
128
119
129
debug.Printf("updating index queue with %d repositories", len(repos))
120
130
···
388
398
}
389
399
}
390
400
391
-
var err error
392
-
data.Repos, err = listRepos(s.Hostname, s.Root)
393
-
if err != nil {
394
-
http.Error(w, err.Error(), http.StatusInternalServerError)
395
-
return
396
-
}
401
+
s.mu.Lock()
402
+
data.Repos = s.lastListRepos
403
+
s.mu.Unlock()
397
404
398
405
repoTmpl.Execute(w, data)
399
406
}
400
407
401
-
func listRepos(hostname string, root *url.URL) ([]string, error) {
408
+
func listIndexed(indexDir string) []string {
409
+
index := getShards(indexDir)
410
+
repoNames := make([]string, 0, len(index))
411
+
for name := range index {
412
+
repoNames = append(repoNames, name)
413
+
}
414
+
sort.Strings(repoNames)
415
+
return repoNames
416
+
}
417
+
418
+
func listRepos(ctx context.Context, hostname string, root *url.URL, indexed []string) ([]string, error) {
402
419
c := retryablehttp.NewClient()
403
420
c.Logger = debug
404
421
405
422
body, err := json.Marshal(&struct {
406
423
Hostname string
407
-
Enabled bool
408
-
Index bool
424
+
Indexed []string
409
425
}{
410
426
Hostname: hostname,
411
-
Enabled: true,
412
-
Index: true,
427
+
Indexed: indexed,
413
428
})
414
429
if err != nil {
415
430
return nil, err
416
431
}
417
432
418
-
u := root.ResolveReference(&url.URL{Path: "/.internal/repos/list"})
433
+
u := root.ResolveReference(&url.URL{Path: "/.internal/repos/index"})
419
434
resp, err := c.Post(u.String(), "application/json; charset=utf8", bytes.NewReader(body))
420
435
if err != nil {
421
436
return nil, err
···
426
441
return nil, fmt.Errorf("failed to list repositories: status %s", resp.Status)
427
442
}
428
443
429
-
var data []struct {
430
-
URI string
444
+
var data struct {
445
+
RepoNames []string
431
446
}
432
447
err = json.NewDecoder(resp.Body).Decode(&data)
433
448
if err != nil {
434
449
return nil, err
435
450
}
436
451
437
-
repos := make([]string, len(data))
438
-
for i, r := range data {
439
-
repos[i] = r.URI
440
-
}
441
-
return repos, nil
452
+
return data.RepoNames, nil
442
453
}
443
454
444
455
func resolveRevision(root *url.URL, repo, spec string) (string, error) {
+9
-9
cmd/zoekt-sourcegraph-indexserver/main_test.go
+9
-9
cmd/zoekt-sourcegraph-indexserver/main_test.go
···
1
1
package main
2
2
3
3
import (
4
+
"context"
4
5
"fmt"
5
6
"io/ioutil"
6
7
"net/http"
7
8
"net/http/httptest"
8
9
"net/url"
9
-
"reflect"
10
10
"strings"
11
11
"testing"
12
12
"time"
···
147
147
}
148
148
gotBody = string(b)
149
149
150
-
_, err = w.Write([]byte(`[{"uri":"foo"}, {"uri":"bar"}, {"uri":"baz"}]`))
150
+
_, err = w.Write([]byte(`{"RepoNames": ["foo", "bar", "baz"]}`))
151
151
if err != nil {
152
152
t.Fatal(err)
153
153
}
···
159
159
t.Fatal(err)
160
160
}
161
161
162
-
gotRepos, err := listRepos("test-indexed-search-1", u)
162
+
gotRepos, err := listRepos(context.Background(), "test-indexed-search-1", u, []string{"foo", "bam"})
163
163
if err != nil {
164
164
t.Fatal(err)
165
165
}
166
166
167
-
if want := []string{"foo", "bar", "baz"}; !reflect.DeepEqual(gotRepos, want) {
168
-
t.Fatalf("unexpected repos. got %v, want %v", gotRepos, want)
167
+
if want := []string{"foo", "bar", "baz"}; !cmp.Equal(gotRepos, want) {
168
+
t.Errorf("repos mismatch (-want +got):\n%s", cmp.Diff(want, gotRepos))
169
169
}
170
-
if want := `{"Hostname":"test-indexed-search-1","Enabled":true,"Index":true}`; gotBody != want {
171
-
t.Fatalf("unexpected request body. got %q, want %q", gotBody, want)
170
+
if want := `{"Hostname":"test-indexed-search-1","Indexed":["foo","bam"]}`; gotBody != want {
171
+
t.Errorf("body mismatch (-want +got):\n%s", cmp.Diff(want, gotBody))
172
172
}
173
-
if want := "/.internal/repos/list"; gotURL.Path != want {
174
-
t.Fatalf("unexpected request path. got %q, want %q", gotURL.Path, want)
173
+
if want := "/.internal/repos/index"; gotURL.Path != want {
174
+
t.Errorf("request path mismatch (-want +got):\n%s", cmp.Diff(want, gotURL.Path))
175
175
}
176
176
}
177
177