fork of https://github.com/sourcegraph/zoekt
indexserver: log name of freshly minted compound shard (#849)
This updates zoekt-merge-index to print the name of a new compound shard to stdout. Indexserver picks it up and logs it. This has the nice property that indexserver now has all the info. If we want to log this to a file in the future, we don't have to worry as much about competing writes to the file.
Together with a new log line in vacuum we can now follow the full lifecycle of a compound shard in the logs.
Test plan:
updated unit tests
author
Stefan Hengl
committer
GitHub
date
(Oct 24, 2024, 3:26 PM +0200)
commit
b3b74a2a
b3b74a2a34b5dc40da0725e461f9a6cdc9b526f0
parent
9bdc5625
9bdc5625fc4c15a2f1bef73da5d8aba156de2287
+61
-72
+16
-11
cmd/zoekt-merge-index/main.go
+16
-11
cmd/zoekt-merge-index/main.go
···
11
11
"github.com/sourcegraph/zoekt"
12
12
)
13
13
14
-
func merge(dstDir string, names []string) error {
14
+
// merge merges the input shards into a compound shard in dstDir. It returns the
15
+
// full path to the compound shard. The input shards are removed on success.
16
+
func merge(dstDir string, names []string) (string, error) {
15
17
var files []zoekt.IndexFile
16
18
for _, fn := range names {
17
19
f, err := os.Open(fn)
18
20
if err != nil {
19
-
return err
21
+
return "", nil
20
22
}
21
23
defer f.Close()
22
24
23
25
indexFile, err := zoekt.NewIndexFile(f)
24
26
if err != nil {
25
-
return err
27
+
return "", err
26
28
}
27
29
defer indexFile.Close()
28
30
···
31
33
32
34
tmpName, dstName, err := zoekt.Merge(dstDir, files...)
33
35
if err != nil {
34
-
return err
36
+
return "", err
35
37
}
36
38
37
39
// Delete input shards.
38
40
for _, name := range names {
39
41
paths, err := zoekt.IndexFilePaths(name)
40
42
if err != nil {
41
-
return fmt.Errorf("zoekt-merge-index: %w", err)
43
+
return "", fmt.Errorf("zoekt-merge-index: %w", err)
42
44
}
43
45
for _, p := range paths {
44
46
if err := os.Remove(p); err != nil {
45
-
return fmt.Errorf("zoekt-merge-index: failed to remove simple shard: %w", err)
47
+
return "", fmt.Errorf("zoekt-merge-index: failed to remove simple shard: %w", err)
46
48
}
47
49
}
48
50
}
···
50
52
// We only rename the compound shard if all simple shards could be deleted in the
51
53
// previous step. This guarantees we won't have duplicate indexes.
52
54
if err := os.Rename(tmpName, dstName); err != nil {
53
-
return fmt.Errorf("zoekt-merge-index: failed to rename compound shard: %w", err)
55
+
return "", fmt.Errorf("zoekt-merge-index: failed to rename compound shard: %w", err)
54
56
}
55
-
return nil
57
+
58
+
return dstName, nil
56
59
}
57
60
58
-
func mergeCmd(paths []string) error {
61
+
func mergeCmd(paths []string) (string, error) {
59
62
if paths[0] == "-" {
60
63
paths = []string{}
61
64
scanner := bufio.NewScanner(os.Stdin)
···
63
66
paths = append(paths, strings.TrimSpace(scanner.Text()))
64
67
}
65
68
if err := scanner.Err(); err != nil {
66
-
return err
69
+
return "", err
67
70
}
68
71
log.Printf("merging %d paths from stdin", len(paths))
69
72
}
···
129
132
func main() {
130
133
switch subCommand := os.Args[1]; subCommand {
131
134
case "merge":
132
-
if err := mergeCmd(os.Args[2:]); err != nil {
135
+
compoundShardPath, err := mergeCmd(os.Args[2:])
136
+
if err != nil {
133
137
log.Fatal(err)
134
138
}
139
+
fmt.Println(compoundShardPath)
135
140
case "explode":
136
141
if err := explodeCmd(os.Args[2]); err != nil {
137
142
log.Fatal(err)
+26
-56
cmd/zoekt-merge-index/main_test.go
+26
-56
cmd/zoekt-merge-index/main_test.go
···
8
8
"sort"
9
9
"testing"
10
10
11
+
"github.com/stretchr/testify/require"
12
+
11
13
"github.com/sourcegraph/zoekt"
12
14
"github.com/sourcegraph/zoekt/query"
13
15
"github.com/sourcegraph/zoekt/shards"
···
15
17
16
18
func TestMerge(t *testing.T) {
17
19
v16Shards, err := filepath.Glob("../../testdata/shards/*_v16.*.zoekt")
18
-
if err != nil {
19
-
t.Fatal(err)
20
-
}
20
+
require.NoError(t, err)
21
21
sort.Strings(v16Shards)
22
22
23
23
testShards, err := copyTestShards(t.TempDir(), v16Shards)
24
-
if err != nil {
25
-
t.Fatal(err)
26
-
}
24
+
require.NoError(t, err)
27
25
t.Log(testShards)
28
26
29
27
dir := t.TempDir()
30
-
err = merge(dir, testShards)
31
-
if err != nil {
32
-
t.Fatal(err)
33
-
}
28
+
cs, err := merge(dir, testShards)
29
+
require.NoError(t, err)
30
+
// The name of the compound shard is based on the merged repos, so it should be
31
+
// stable
32
+
require.Equal(t, filepath.Base(cs), "compound-ea9613e2ffba7d7361856aebfca75fb714856509_v17.00000.zoekt")
34
33
35
34
ss, err := shards.NewDirectorySearcher(dir)
36
-
if err != nil {
37
-
t.Fatalf("NewDirectorySearcher(%s): %v", dir, err)
38
-
}
35
+
require.NoError(t, err)
39
36
defer ss.Close()
40
37
41
38
q, err := query.Parse("hello")
42
-
if err != nil {
43
-
t.Fatalf("Parse(hello): %v", err)
44
-
}
39
+
require.NoError(t, err)
45
40
46
41
var sOpts zoekt.SearchOptions
47
42
ctx := context.Background()
48
43
result, err := ss.Search(ctx, q, &sOpts)
49
-
if err != nil {
50
-
t.Fatalf("Search(%v): %v", q, err)
51
-
}
44
+
require.NoError(t, err)
52
45
53
46
// we are merging the same shard twice, so we expect the same file twice.
54
-
if len(result.Files) != 2 {
55
-
t.Errorf("got %v, want 2 files.", result.Files)
56
-
}
47
+
require.Len(t, result.Files, 2)
57
48
}
58
49
59
50
// Merge 2 simple shards and then explode them.
60
51
func TestExplode(t *testing.T) {
61
52
v16Shards, err := filepath.Glob("../../testdata/shards/repo*_v16.*.zoekt")
62
-
if err != nil {
63
-
t.Fatal(err)
64
-
}
53
+
require.NoError(t, err)
65
54
sort.Strings(v16Shards)
66
55
67
56
testShards, err := copyTestShards(t.TempDir(), v16Shards)
68
-
if err != nil {
69
-
t.Fatal(err)
70
-
}
57
+
require.NoError(t, err)
71
58
t.Log(testShards)
72
59
73
60
dir := t.TempDir()
74
-
err = merge(dir, testShards)
75
-
if err != nil {
76
-
t.Fatal(err)
77
-
}
61
+
_, err = merge(dir, testShards)
62
+
require.NoError(t, err)
78
63
79
64
cs, err := filepath.Glob(filepath.Join(dir, "compound-*.zoekt"))
80
-
if err != nil {
81
-
t.Fatal(err)
82
-
}
65
+
require.NoError(t, err)
83
66
err = explode(dir, cs[0])
84
-
if err != nil {
85
-
t.Fatal(err)
86
-
}
67
+
require.NoError(t, err)
87
68
88
69
cs, err = filepath.Glob(filepath.Join(dir, "compound-*.zoekt"))
89
-
if err != nil {
90
-
t.Fatal(err)
91
-
}
70
+
require.NoError(t, err)
71
+
92
72
if len(cs) != 0 {
93
73
t.Fatalf("explode should have deleted the compound shard if it returned without error")
94
74
}
95
75
96
76
exploded, err := filepath.Glob(filepath.Join(dir, "*.zoekt"))
97
-
if err != nil {
98
-
t.Fatal(err)
99
-
}
77
+
require.NoError(t, err)
100
78
101
79
if len(exploded) != len(testShards) {
102
80
t.Fatalf("the number of simple shards before %d and after %d should be the same", len(testShards), len(exploded))
103
81
}
104
82
105
83
ss, err := shards.NewDirectorySearcher(dir)
106
-
if err != nil {
107
-
t.Fatalf("NewDirectorySearcher(%s): %v", dir, err)
108
-
}
84
+
require.NoError(t, err)
109
85
defer ss.Close()
110
86
111
87
var sOpts zoekt.SearchOptions
···
132
108
for _, c := range cases {
133
109
t.Run(c.searchLiteral, func(t *testing.T) {
134
110
q, err := query.Parse(c.searchLiteral)
135
-
if err != nil {
136
-
t.Fatalf("Parse(%s): %v", c.searchLiteral, err)
137
-
}
111
+
require.NoError(t, err)
138
112
result, err := ss.Search(ctx, q, &sOpts)
139
-
if err != nil {
140
-
t.Fatalf("Search(%v): %v", q, err)
141
-
}
142
-
if got := len(result.Files); got != c.wantResults {
143
-
t.Fatalf("wanted %d results, got %d", c.wantResults, got)
144
-
}
113
+
require.NoError(t, err)
114
+
require.Len(t, result.Files, c.wantResults)
145
115
})
146
116
}
147
117
}
+3
-2
cmd/zoekt-sourcegraph-indexserver/cleanup.go
+3
-2
cmd/zoekt-sourcegraph-indexserver/cleanup.go
···
422
422
})
423
423
424
424
if err != nil {
425
-
log.Printf("failed to explode compound shard %s: %s", path, string(b))
425
+
log.Printf("failed to explode compound shard: shard=%s out=%s err=%s", path, string(b), err)
426
426
}
427
+
log.Printf("exploded compound shard: shard=%s", path)
427
428
continue
428
429
}
429
430
···
432
433
})
433
434
434
435
if err != nil {
435
-
log.Printf("error while removing tombstones in %s: %s", fn, err)
436
+
log.Printf("error while removing tombstones in %s: %s", path, err)
436
437
}
437
438
}
438
439
}
+16
-3
cmd/zoekt-sourcegraph-indexserver/merge.go
+16
-3
cmd/zoekt-sourcegraph-indexserver/merge.go
···
1
1
package main
2
2
3
3
import (
4
+
"bytes"
4
5
"log"
5
6
"os"
6
7
"os/exec"
···
88
89
}
89
90
90
91
start := time.Now()
91
-
out, err := mergeCmd(paths...).CombinedOutput()
92
+
93
+
cmd := mergeCmd(paths...)
92
94
93
-
metricShardMergingDuration.WithLabelValues(strconv.FormatBool(err != nil)).Observe(time.Since(start).Seconds())
95
+
// zoekt-merge-index writes the full path of the new compound shard to stdout.
96
+
stdoutBuf := &bytes.Buffer{}
97
+
stderrBuf := &bytes.Buffer{}
98
+
cmd.Stdout = stdoutBuf
99
+
cmd.Stderr = stderrBuf
100
+
101
+
err := cmd.Run()
102
+
103
+
durationSeconds := time.Since(start).Seconds()
104
+
metricShardMergingDuration.WithLabelValues(strconv.FormatBool(err != nil)).Observe(durationSeconds)
94
105
if err != nil {
95
-
log.Printf("mergeCmd: out=%s, err=%s", out, err)
106
+
log.Printf("error merging shards: stdout=%s, stderr=%s, durationSeconds=%.2f err=%s", stdoutBuf.String(), stderrBuf.String(), durationSeconds, err)
96
107
return
97
108
}
109
+
110
+
log.Printf("finished merging: shard=%s durationSeconds=%.2f", stdoutBuf.String(), durationSeconds)
98
111
99
112
next = true
100
113
})