sourcegraph: Remove HTTP APIs (#738) · boltless.me/zoekt@371a303

+322 -546

cmd/zoekt-sourcegraph-indexserver/index_test.go

··· 1 1 package main 2 2 3 3 import ( 4 - "bytes" 5 4 "context" 6 - "encoding/json" 5 + "errors" 7 6 "fmt" 8 - "net/http" 9 - "net/http/httptest" 10 7 "net/url" 11 8 "os" 12 9 "os/exec" 13 10 "path/filepath" 14 - "reflect" 15 11 "sort" 16 - "strconv" 17 12 "strings" 18 13 "testing" 19 14 "time" 20 15 21 16 "github.com/sourcegraph/log/logtest" 22 17 proto "github.com/sourcegraph/zoekt/cmd/zoekt-sourcegraph-indexserver/protos/sourcegraph/zoekt/configuration/v1" 18 + "github.com/sourcegraph/zoekt/ctags" 23 19 "google.golang.org/grpc" 24 20 "google.golang.org/protobuf/testing/protocmp" 25 21 "google.golang.org/protobuf/types/known/timestamppb" ··· 31 27 ) 32 28 33 29 func TestIterateIndexOptions_Fingerprint(t *testing.T) { 34 - t.Run("gRPC", func(t *testing.T) { 35 - fingerprintV0 := &proto.Fingerprint{ 36 - Identifier: 100, 37 - GeneratedAt: timestamppb.New(time.Unix(100, 0)), 38 - } 30 + fingerprintV0 := &proto.Fingerprint{ 31 + Identifier: 100, 32 + GeneratedAt: timestamppb.New(time.Unix(100, 0)), 33 + } 39 34 40 - fingerprintV1 := &proto.Fingerprint{ 41 - Identifier: 101, 42 - GeneratedAt: timestamppb.New(time.Unix(101, 0)), 43 - } 35 + fingerprintV1 := &proto.Fingerprint{ 36 + Identifier: 101, 37 + GeneratedAt: timestamppb.New(time.Unix(101, 0)), 38 + } 44 39 45 - fingerprintV2 := &proto.Fingerprint{ 46 - Identifier: 102, 47 - GeneratedAt: timestamppb.New(time.Unix(102, 0)), 48 - } 40 + fingerprintV2 := &proto.Fingerprint{ 41 + Identifier: 102, 42 + GeneratedAt: timestamppb.New(time.Unix(102, 0)), 43 + } 49 44 50 - mkSearchConfigurationResponse := func(fingerprint *proto.Fingerprint, repoIDs ...int32) *proto.SearchConfigurationResponse { 51 - repositories := make([]*proto.ZoektIndexOptions, 0, len(repoIDs)) 52 - for _, repoID := range repoIDs { 53 - repositories = append(repositories, &proto.ZoektIndexOptions{ 54 - RepoId: repoID, 55 - }) 56 - } 57 - 58 - return &proto.SearchConfigurationResponse{ 59 - UpdatedOptions: repositories, 60 - Fingerprint: fingerprint, 61 - } 45 + mkSearchConfigurationResponse := func(fingerprint *proto.Fingerprint, repoIDs ...int32) *proto.SearchConfigurationResponse { 46 + repositories := make([]*proto.ZoektIndexOptions, 0, len(repoIDs)) 47 + for _, repoID := range repoIDs { 48 + repositories = append(repositories, &proto.ZoektIndexOptions{ 49 + RepoId: repoID, 50 + }) 62 51 } 63 52 64 - grpcClient := &mockGRPCClient{ 65 - mockList: func(_ context.Context, in *proto.ListRequest, opts ...grpc.CallOption) (*proto.ListResponse, error) { 66 - return &proto.ListResponse{ 67 - RepoIds: []int32{1, 2, 3}, 68 - }, nil 69 - }, 53 + return &proto.SearchConfigurationResponse{ 54 + UpdatedOptions: repositories, 55 + Fingerprint: fingerprint, 70 56 } 57 + } 71 58 72 - clientOpts := []SourcegraphClientOption{ 73 - WithBatchSize(1), 74 - WithShouldUseGRPC(true), 75 - WithGRPCClient(grpcClient), 76 - } 59 + grpcClient := &mockGRPCClient{ 60 + mockList: func(_ context.Context, in *proto.ListRequest, opts ...grpc.CallOption) (*proto.ListResponse, error) { 61 + return &proto.ListResponse{ 62 + RepoIds: []int32{1, 2, 3}, 63 + }, nil 64 + }, 65 + } 77 66 78 - testURL := url.URL{Scheme: "http", Host: "does.not.matter", Path: "/"} 79 - sg := newSourcegraphClient(&testURL, "", clientOpts...) 67 + clientOpts := []SourcegraphClientOption{ 68 + WithBatchSize(1), 69 + } 80 70 81 - type step struct { 82 - name string 71 + testURL := url.URL{Scheme: "http", Host: "does.not.matter", Path: "/"} 72 + sg := newSourcegraphClient(&testURL, "", grpcClient, clientOpts...) 83 73 84 - wantFingerprint *proto.Fingerprint 85 - returnFingerprint *proto.Fingerprint 86 - returnErr error 87 - skipCheckingRepoIDs bool 88 - } 74 + type step struct { 75 + name string 89 76 90 - for _, step := range []step{ 91 - { 92 - name: "first call", 93 - wantFingerprint: nil, 94 - returnFingerprint: fingerprintV0, 95 - }, 96 - { 97 - name: "second call (should provide fingerprint from last time)", 98 - wantFingerprint: fingerprintV0, 99 - returnFingerprint: fingerprintV1, 100 - }, 101 - { 102 - name: "error", 103 - wantFingerprint: fingerprintV1, 104 - returnFingerprint: fingerprintV2, 77 + wantFingerprint *proto.Fingerprint 78 + returnFingerprint *proto.Fingerprint 79 + returnErr error 80 + skipCheckingRepoIDs bool 81 + } 105 82 106 - returnErr: fmt.Errorf("boom"), 107 - skipCheckingRepoIDs: true, // don't bother checking repoIDs if we expect an error 108 - }, 109 - { 110 - name: "call after error (should ignore fingerprint from last time, and provide the older one)", 111 - wantFingerprint: fingerprintV1, 112 - returnFingerprint: fingerprintV2, 113 - }, 114 - } { 115 - t.Run(step.name, func(t *testing.T) { 116 - called := false 117 - grpcClient.mockSearchConfiguration = func(_ context.Context, in *proto.SearchConfigurationRequest, opts ...grpc.CallOption) (*proto.SearchConfigurationResponse, error) { 118 - called = true 119 - 120 - diff := cmp.Diff(step.wantFingerprint, in.GetFingerprint(), protocmp.Transform()) 121 - if diff != "" { 122 - t.Fatalf("unexpected fingerprint (-want +got):\n%s", diff) 123 - } 124 - 125 - return mkSearchConfigurationResponse(step.returnFingerprint, in.RepoIds...), step.returnErr 126 - } 127 - 128 - result, err := sg.List(context.Background(), nil) 129 - if err != nil { 130 - t.Fatalf("unexpected error from List: %v", err) 131 - } 132 - 133 - var iteratedIDs []uint32 134 - result.IterateIndexOptions(func(options IndexOptions) { 135 - iteratedIDs = append(iteratedIDs, options.RepoID) 136 - }) 83 + for _, step := range []step{ 84 + { 85 + name: "first call", 86 + wantFingerprint: nil, 87 + returnFingerprint: fingerprintV0, 88 + }, 89 + { 90 + name: "second call (should provide fingerprint from last time)", 91 + wantFingerprint: fingerprintV0, 92 + returnFingerprint: fingerprintV1, 93 + }, 94 + { 95 + name: "error", 96 + wantFingerprint: fingerprintV1, 97 + returnFingerprint: fingerprintV2, 137 98 138 - if !called { 139 - t.Fatal("expected SearchConfiguration to be called") 140 - } 99 + returnErr: fmt.Errorf("boom"), 100 + skipCheckingRepoIDs: true, // don't bother checking repoIDs if we expect an error 101 + }, 102 + { 103 + name: "call after error (should ignore fingerprint from last time, and provide the older one)", 104 + wantFingerprint: fingerprintV1, 105 + returnFingerprint: fingerprintV2, 106 + }, 107 + } { 108 + t.Run(step.name, func(t *testing.T) { 109 + called := false 110 + grpcClient.mockSearchConfiguration = func(_ context.Context, in *proto.SearchConfigurationRequest, opts ...grpc.CallOption) (*proto.SearchConfigurationResponse, error) { 111 + called = true 141 112 142 - if step.skipCheckingRepoIDs { 143 - return 113 + diff := cmp.Diff(step.wantFingerprint, in.GetFingerprint(), protocmp.Transform()) 114 + if diff != "" { 115 + t.Fatalf("unexpected fingerprint (-want +got):\n%s", diff) 144 116 } 145 117 146 - sort.Slice(iteratedIDs, func(i, j int) bool { 147 - return iteratedIDs[i] < iteratedIDs[j] 148 - }) 118 + return mkSearchConfigurationResponse(step.returnFingerprint, in.RepoIds...), step.returnErr 119 + } 149 120 150 - expectedIDs := []uint32{1, 2, 3} 151 - sort.Slice(expectedIDs, func(i, j int) bool { 152 - return expectedIDs[i] < expectedIDs[j] 153 - }) 121 + result, err := sg.List(context.Background(), nil) 122 + if err != nil { 123 + t.Fatalf("unexpected error from List: %v", err) 124 + } 154 125 155 - if diff := cmp.Diff(expectedIDs, iteratedIDs); diff != "" { 156 - t.Fatalf("unexpected repo ids (-want +got):\n%s", diff) 157 - } 126 + var iteratedIDs []uint32 127 + result.IterateIndexOptions(func(options IndexOptions) { 128 + iteratedIDs = append(iteratedIDs, options.RepoID) 158 129 }) 159 - } 160 - }) 161 130 162 - t.Run("REST", func(t *testing.T) { 163 - fingerprintV0 := "v0" 164 - fingerprintV1 := "v1" 165 - fingerprintV2 := "v2" 166 - 167 - handleList := func(w http.ResponseWriter, _ *http.Request) { 168 - data := struct { 169 - RepoIDs []uint32 170 - }{ 171 - RepoIDs: []uint32{1, 2, 3}, 131 + if !called { 132 + t.Fatal("expected SearchConfiguration to be called") 172 133 } 173 134 174 - json.NewEncoder(w).Encode(data) 175 - } 176 - 177 - searchConfigurationHandler := func(w http.ResponseWriter, r *http.Request) { 178 - http.Error(w, "this search configuration handler hasn't been overridden", http.StatusForbidden) 179 - } 180 - 181 - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { 182 - switch r.URL.Path { 183 - case "/.internal/search/configuration": 184 - searchConfigurationHandler(w, r) 185 - case "/.internal/repos/index": 186 - handleList(w, r) 187 - default: 188 - t.Fatalf("unexpected path: %s", r.URL.Path) 135 + if step.skipCheckingRepoIDs { 136 + return 189 137 } 190 - })) 191 - defer server.Close() 192 138 193 - clientOpts := []SourcegraphClientOption{ 194 - WithBatchSize(1), 195 - WithShouldUseGRPC(false), 196 - } 139 + sort.Slice(iteratedIDs, func(i, j int) bool { 140 + return iteratedIDs[i] < iteratedIDs[j] 141 + }) 197 142 198 - testURL, err := url.Parse(server.URL) 199 - if err != nil { 200 - t.Fatalf("unexpected error parsing URL: %v", err) 201 - } 143 + expectedIDs := []uint32{1, 2, 3} 144 + sort.Slice(expectedIDs, func(i, j int) bool { 145 + return expectedIDs[i] < expectedIDs[j] 146 + }) 202 147 203 - sg := newSourcegraphClient(testURL, "", clientOpts...) 204 - 205 - type step struct { 206 - name string 207 - 208 - wantFingerprint string 209 - returnFingerprint string 210 - returnErr error 211 - skipCheckingRepoIDs bool 212 - } 213 - 214 - for _, step := range []step{ 215 - { 216 - name: "first call", 217 - wantFingerprint: "", 218 - returnFingerprint: fingerprintV0, 219 - }, 220 - { 221 - name: "second call (should provide fingerprint from last time)", 222 - wantFingerprint: fingerprintV0, 223 - returnFingerprint: fingerprintV1, 224 - }, 225 - { 226 - name: "error", 227 - wantFingerprint: fingerprintV1, 228 - returnFingerprint: fingerprintV2, 229 - 230 - returnErr: fmt.Errorf("boom"), 231 - skipCheckingRepoIDs: true, // don't bother checking repoIDs if we expect an error 232 - }, 233 - { 234 - name: "call after error (should ignore fingerprint from last time, and provide the older one)", 235 - wantFingerprint: fingerprintV1, 236 - returnFingerprint: fingerprintV2, 237 - }, 238 - } { 239 - t.Run(step.name, func(t *testing.T) { 240 - called := false 241 - 242 - searchConfigurationHandler = func(w http.ResponseWriter, r *http.Request) { 243 - called = true 244 - 245 - fingerprint := r.Header.Get(fingerprintHeader) 246 - if diff := cmp.Diff(step.wantFingerprint, fingerprint); diff != "" { 247 - t.Fatalf("unexpected fingerprint (-want +got):\n%s", diff) 248 - } 249 - 250 - w.Header().Set(fingerprintHeader, step.returnFingerprint) 251 - 252 - if step.returnErr != nil { 253 - // The status code is a bit of a hack, but it prevents 254 - // the retry logic from kicking in and stalling the test for 45 seconds. 255 - http.Error(w, step.returnErr.Error(), http.StatusBadRequest) 256 - return 257 - } 258 - 259 - if err := r.ParseForm(); err != nil { 260 - http.Error(w, fmt.Sprintf("unexpected error parsing form for repoIDs: %v", err.Error()), http.StatusBadRequest) 261 - return 262 - } 263 - 264 - repoIDs := make([]uint32, 0, len(r.Form["repoID"])) 265 - for _, idStr := range r.Form["repoID"] { 266 - id, err := strconv.Atoi(idStr) 267 - if err != nil { 268 - http.Error(w, fmt.Sprintf("invalid repo id %s: %s", idStr, err), http.StatusBadRequest) 269 - return 270 - } 271 - repoIDs = append(repoIDs, uint32(id)) 272 - } 273 - 274 - optionJSONSlice := make([][]byte, 0, len(repoIDs)) 275 - for _, repoID := range repoIDs { 276 - option := IndexOptions{ 277 - RepoID: repoID, 278 - } 279 - 280 - optionJSON, err := json.Marshal(option) 281 - if err != nil { 282 - t.Fatalf("unexpected error marshalling JSON: %v", err) 283 - } 284 - 285 - optionJSONSlice = append(optionJSONSlice, optionJSON) 286 - } 287 - 288 - w.Write(bytes.Join(optionJSONSlice, []byte("\n"))) 289 - } 290 - 291 - result, err := sg.List(context.Background(), nil) 292 - if err != nil { 293 - t.Fatalf("unexpected error from List: %v", err) 294 - } 295 - 296 - var iteratedIDs []uint32 297 - result.IterateIndexOptions(func(options IndexOptions) { 298 - iteratedIDs = append(iteratedIDs, options.RepoID) 299 - }) 300 - 301 - if !called { 302 - t.Fatal("expected SearchConfiguration to be called") 303 - } 304 - 305 - if step.skipCheckingRepoIDs { 306 - return 307 - } 308 - 309 - sort.Slice(iteratedIDs, func(i, j int) bool { 310 - return iteratedIDs[i] < iteratedIDs[j] 311 - }) 312 - 313 - expectedIDs := []uint32{1, 2, 3} 314 - sort.Slice(expectedIDs, func(i, j int) bool { 315 - return expectedIDs[i] < expectedIDs[j] 316 - }) 317 - 318 - if diff := cmp.Diff(expectedIDs, iteratedIDs); diff != "" { 319 - t.Fatalf("unexpected repo ids (-want +got):\n%s", diff) 320 - } 321 - }) 322 - } 323 - }) 148 + if diff := cmp.Diff(expectedIDs, iteratedIDs); diff != "" { 149 + t.Fatalf("unexpected repo ids (-want +got):\n%s", diff) 150 + } 151 + }) 152 + } 324 153 } 325 154 326 155 func TestGetIndexOptions(t *testing.T) { 327 - t.Run("gRPC", func(t *testing.T) { 328 - type testCase struct { 329 - name string 330 - response *proto.SearchConfigurationResponse 331 - want *IndexOptions 332 - wantErr string 333 - } 334 156 335 - for _, tc := range []testCase{ 336 - { 337 - name: "symbols, large files", 338 - response: &proto.SearchConfigurationResponse{ 339 - UpdatedOptions: []*proto.ZoektIndexOptions{ 340 - { 341 - Symbols: true, 342 - LargeFiles: []string{"foo", "bar"}, 343 - }, 157 + type testCase struct { 158 + name string 159 + response *proto.SearchConfigurationResponse 160 + want *IndexOptions 161 + wantErr string 162 + } 163 + 164 + for _, tc := range []testCase{ 165 + { 166 + name: "symbols, large files", 167 + response: &proto.SearchConfigurationResponse{ 168 + UpdatedOptions: []*proto.ZoektIndexOptions{ 169 + { 170 + Symbols: true, 171 + LargeFiles: []string{"foo", "bar"}, 344 172 }, 345 173 }, 346 - want: &IndexOptions{ 347 - Symbols: true, 348 - LargeFiles: []string{"foo", "bar"}, 349 - }, 350 174 }, 351 - { 352 - name: "no symbols , large files", 353 - response: &proto.SearchConfigurationResponse{ 354 - UpdatedOptions: []*proto.ZoektIndexOptions{ 355 - { 356 - Symbols: true, 357 - LargeFiles: []string{"foo", "bar"}, 358 - }, 175 + want: &IndexOptions{ 176 + Symbols: true, 177 + LargeFiles: []string{"foo", "bar"}, 178 + }, 179 + }, 180 + { 181 + name: "no symbols , large files", 182 + response: &proto.SearchConfigurationResponse{ 183 + UpdatedOptions: []*proto.ZoektIndexOptions{ 184 + { 185 + Symbols: true, 186 + LargeFiles: []string{"foo", "bar"}, 359 187 }, 360 - }, 361 - want: &IndexOptions{ 362 - Symbols: true, 363 - LargeFiles: []string{"foo", "bar"}, 364 188 }, 365 189 }, 366 - 367 - { 368 - name: "empty", 369 - response: nil, 370 - want: nil, 190 + want: &IndexOptions{ 191 + Symbols: true, 192 + LargeFiles: []string{"foo", "bar"}, 371 193 }, 194 + }, 372 195 373 - { 374 - name: "symbols", 375 - response: &proto.SearchConfigurationResponse{ 376 - UpdatedOptions: []*proto.ZoektIndexOptions{ 377 - { 378 - Symbols: true, 379 - }, 196 + { 197 + name: "empty", 198 + response: nil, 199 + want: nil, 200 + }, 201 + 202 + { 203 + name: "symbols", 204 + response: &proto.SearchConfigurationResponse{ 205 + UpdatedOptions: []*proto.ZoektIndexOptions{ 206 + { 207 + Symbols: true, 380 208 }, 381 209 }, 382 - want: &IndexOptions{ 383 - Symbols: true, 384 - }, 210 + }, 211 + want: &IndexOptions{ 212 + Symbols: true, 385 213 }, 386 - { 387 - name: "repoID", 388 - response: &proto.SearchConfigurationResponse{ 389 - UpdatedOptions: []*proto.ZoektIndexOptions{ 390 - { 391 - RepoId: 123, 392 - }, 214 + }, 215 + { 216 + name: "repoID", 217 + response: &proto.SearchConfigurationResponse{ 218 + UpdatedOptions: []*proto.ZoektIndexOptions{ 219 + { 220 + RepoId: 123, 393 221 }, 394 222 }, 395 - want: &IndexOptions{ 396 - RepoID: 123, 397 - }, 223 + }, 224 + want: &IndexOptions{ 225 + RepoID: 123, 398 226 }, 399 - { 400 - name: "error", 401 - response: &proto.SearchConfigurationResponse{ 402 - UpdatedOptions: []*proto.ZoektIndexOptions{ 403 - { 404 - Error: "boom", 405 - }, 227 + }, 228 + { 229 + name: "error", 230 + response: &proto.SearchConfigurationResponse{ 231 + UpdatedOptions: []*proto.ZoektIndexOptions{ 232 + { 233 + Error: "boom", 406 234 }, 407 235 }, 408 - want: nil, 409 - wantErr: "boom", 236 + }, 237 + want: nil, 238 + wantErr: "boom", 239 + }, 240 + } { 241 + called := false 242 + mockClient := &mockGRPCClient{ 243 + mockSearchConfiguration: func(_ context.Context, _ *proto.SearchConfigurationRequest, _ ...grpc.CallOption) (*proto.SearchConfigurationResponse, error) { 244 + called = true 245 + return tc.response, nil 410 246 }, 411 - } { 412 - called := false 413 - mockClient := &mockGRPCClient{ 414 - mockSearchConfiguration: func(_ context.Context, _ *proto.SearchConfigurationRequest, _ ...grpc.CallOption) (*proto.SearchConfigurationResponse, error) { 415 - called = true 416 - return tc.response, nil 417 - }, 418 - } 247 + } 419 248 420 - testURL := &url.URL{ 421 - Scheme: "http", 422 - Host: "does.not.matter", 423 - Path: "/", 424 - } 249 + testURL := &url.URL{ 250 + Scheme: "http", 251 + Host: "does.not.matter", 252 + Path: "/", 253 + } 425 254 426 - sg := newSourcegraphClient( 427 - testURL, 428 - "", 429 - WithShouldUseGRPC(true), 430 - WithGRPCClient(mockClient), 431 - ) 255 + sg := newSourcegraphClient( 256 + testURL, 257 + "", 258 + mockClient, 259 + ) 432 260 433 - var got IndexOptions 434 - var err error 435 - sg.ForceIterateIndexOptions(func(o IndexOptions) { 436 - got = o 437 - }, func(_ uint32, e error) { 438 - err = e 439 - }, 123) 261 + var got IndexOptions 262 + var err error 263 + sg.ForceIterateIndexOptions(func(o IndexOptions) { 264 + got = o 265 + }, func(_ uint32, e error) { 266 + err = e 267 + }, 123) 440 268 441 - if !called { 442 - t.Fatal("expected mock to be called") 443 - } 444 - 445 - if err != nil { 446 - if tc.wantErr == "" || !strings.Contains(err.Error(), tc.wantErr) { 447 - t.Fatalf("unexpected error: %v", err) 448 - } 449 - } 450 - 451 - if tc.want == nil { 452 - continue 453 - } 454 - 455 - tc.want.CloneURL = sg.getCloneURL(got.Name) 456 - 457 - if diff := cmp.Diff(tc.want, &got, cmpopts.EquateEmpty()); diff != "" { 458 - t.Errorf("mismatch (-want +got):\n%s", diff) 459 - } 269 + if !called { 270 + t.Fatal("expected mock to be called") 460 271 } 461 272 462 - // Mimic our fingerprint API, which doesn't return anything if the 463 - // repo hasn't changed. 464 - t.Run("unchanged", func(t *testing.T) { 465 - called := false 466 - mockClient := &mockGRPCClient{ 467 - mockSearchConfiguration: func(_ context.Context, _ *proto.SearchConfigurationRequest, _ ...grpc.CallOption) (*proto.SearchConfigurationResponse, error) { 468 - called = true 469 - return nil, nil 470 - }, 471 - } 472 - 473 - testURL := &url.URL{ 474 - Scheme: "http", 475 - Host: "does.not.matter", 476 - Path: "/", 477 - } 478 - 479 - sg := newSourcegraphClient( 480 - testURL, 481 - "", 482 - WithShouldUseGRPC(true), 483 - WithGRPCClient(mockClient)) 484 - 485 - gotAtLeastOneOption := false 486 - var err error 487 - sg.ForceIterateIndexOptions(func(_ IndexOptions) { 488 - gotAtLeastOneOption = true 489 - }, func(_ uint32, e error) { 490 - err = e 491 - }, 123) 492 - 493 - if !called { 494 - t.Fatal("expected mock to be called") 495 - } 496 - 497 - if err != nil { 273 + if err != nil { 274 + if tc.wantErr == "" || !strings.Contains(err.Error(), tc.wantErr) { 498 275 t.Fatalf("unexpected error: %v", err) 499 276 } 500 - 501 - if gotAtLeastOneOption { 502 - t.Fatalf("expected no options, got %v", gotAtLeastOneOption) 503 - } 504 - }) 505 - }) 506 - t.Run("REST", func(t *testing.T) { 507 - var response []byte 508 - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { 509 - if err := r.ParseForm(); err != nil { 510 - http.Error(w, err.Error(), http.StatusBadRequest) 511 - return 512 - } 513 - if got, want := r.URL.String(), "/.internal/search/configuration"; got != want { 514 - http.Error(w, fmt.Sprintf("got URL %v want %v", got, want), http.StatusBadRequest) 515 - return 516 - } 517 - if got, want := r.Form, (url.Values{"repoID": []string{"123"}}); !reflect.DeepEqual(got, want) { 518 - http.Error(w, fmt.Sprintf("got URL %v want %v", got, want), http.StatusBadRequest) 519 - return 520 - } 521 - _, _ = w.Write(response) 522 - })) 523 - defer server.Close() 277 + } 524 278 525 - u, err := url.Parse(server.URL) 526 - if err != nil { 527 - t.Fatal(err) 279 + if tc.want == nil { 280 + continue 528 281 } 529 282 530 - sg := newSourcegraphClient(u, "", WithBatchSize(0)) 531 - 532 - cases := map[string]*IndexOptions{ 533 - `{"Symbols": true, "LargeFiles": ["foo","bar"]}`: { 534 - Symbols: true, 535 - LargeFiles: []string{"foo", "bar"}, 536 - }, 537 - 538 - `{"Symbols": false, "LargeFiles": ["foo","bar"]}`: { 539 - LargeFiles: []string{"foo", "bar"}, 540 - }, 283 + tc.want.CloneURL = sg.getCloneURL(got.Name) 541 284 542 - `{}`: {}, 285 + if diff := cmp.Diff(tc.want, &got, cmpopts.EquateEmpty()); diff != "" { 286 + t.Errorf("mismatch (-want +got):\n%s", diff) 287 + } 288 + } 543 289 544 - `{"Symbols": true}`: { 545 - Symbols: true, 546 - }, 290 + // Mimic our fingerprint API, which doesn't return anything if the 291 + // repo hasn't changed. 292 + t.Run("unchanged", func(t *testing.T) { 547 293 548 - `{"RepoID": 123}`: { 549 - RepoID: 123, 294 + called := false 295 + mockClient := &mockGRPCClient{ 296 + mockSearchConfiguration: func(_ context.Context, _ *proto.SearchConfigurationRequest, _ ...grpc.CallOption) (*proto.SearchConfigurationResponse, error) { 297 + called = true 298 + return nil, nil 550 299 }, 300 + } 551 301 552 - `{"Error": "boom"}`: nil, 302 + testURL := &url.URL{ 303 + Scheme: "http", 304 + Host: "does.not.matter", 305 + Path: "/", 553 306 } 554 307 555 - for r, want := range cases { 556 - response = []byte(r) 308 + sg := newSourcegraphClient( 309 + testURL, 310 + "", 311 + mockClient, 312 + ) 313 + gotAtLeastOneOption := false 314 + var err error 315 + sg.ForceIterateIndexOptions(func(_ IndexOptions) { 316 + gotAtLeastOneOption = true 317 + }, func(_ uint32, e error) { 318 + err = e 319 + }, 123) 557 320 558 - var got IndexOptions 559 - var err error 560 - sg.ForceIterateIndexOptions(func(o IndexOptions) { 561 - got = o 562 - }, func(_ uint32, e error) { 563 - err = e 564 - }, 123) 321 + if !called { 322 + t.Fatal("expected mock to be called") 323 + } 565 324 566 - if err != nil && want != nil { 567 - t.Fatalf("unexpected error: %v", err) 568 - } 569 - if want == nil { 570 - continue 571 - } 325 + if err != nil { 326 + t.Fatalf("unexpected error: %v", err) 327 + } 572 328 573 - want.CloneURL = sg.getCloneURL(got.Name) 574 - 575 - if d := cmp.Diff(*want, got); d != "" { 576 - t.Log("response", r) 577 - t.Errorf("mismatch (-want +got):\n%s", d) 578 - } 329 + if gotAtLeastOneOption { 330 + t.Fatalf("expected no options, got %v", gotAtLeastOneOption) 579 331 } 580 - 581 - // Special case our fingerprint API which doesn't return anything if the 582 - // repo hasn't changed. 583 - t.Run("unchanged", func(t *testing.T) { 584 - response = []byte(``) 585 - 586 - got := false 587 - var err error 588 - sg.ForceIterateIndexOptions(func(_ IndexOptions) { 589 - got = true 590 - }, func(_ uint32, e error) { 591 - err = e 592 - }, 123) 593 - if err != nil { 594 - t.Fatalf("unexpected error: %v", err) 595 - } 596 - 597 - if got { 598 - t.Fatalf("expected no options, got %v", got) 599 - } 600 - }) 601 332 }) 602 333 603 - var response []byte 604 - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { 605 - if err := r.ParseForm(); err != nil { 606 - http.Error(w, err.Error(), http.StatusBadRequest) 607 - return 608 - } 609 - if got, want := r.URL.String(), "/.internal/search/configuration"; got != want { 610 - http.Error(w, fmt.Sprintf("got URL %v want %v", got, want), http.StatusBadRequest) 611 - return 612 - } 613 - if got, want := r.Form, (url.Values{"repoID": []string{"123"}}); !reflect.DeepEqual(got, want) { 614 - http.Error(w, fmt.Sprintf("got URL %v want %v", got, want), http.StatusBadRequest) 615 - return 616 - } 617 - _, _ = w.Write(response) 618 - })) 619 - defer server.Close() 620 - 621 - u, err := url.Parse(server.URL) 622 - if err != nil { 623 - t.Fatal(err) 334 + var response *proto.SearchConfigurationResponse 335 + mockClient := &mockGRPCClient{ 336 + mockSearchConfiguration: func(_ context.Context, req *proto.SearchConfigurationRequest, _ ...grpc.CallOption) (*proto.SearchConfigurationResponse, error) { 337 + if len(req.GetRepoIds()) == 0 || req.GetRepoIds()[0] != 123 { 338 + return nil, errors.New("invalid repo id") 339 + } 340 + return response, nil 341 + }, 624 342 } 625 343 626 - sg := newSourcegraphClient(u, "", WithBatchSize(0)) 344 + sg := newSourcegraphClient(&url.URL{Path: "/"}, "", mockClient, WithBatchSize(0)) 627 345 628 - cases := map[string]*IndexOptions{ 629 - `{"Symbols": true, "LargeFiles": ["foo","bar"]}`: { 630 - Symbols: true, 631 - LargeFiles: []string{"foo", "bar"}, 346 + cases := []struct { 347 + Response *proto.SearchConfigurationResponse 348 + *IndexOptions 349 + }{ 350 + { 351 + Response: &proto.SearchConfigurationResponse{ 352 + UpdatedOptions: []*proto.ZoektIndexOptions{ 353 + { 354 + Symbols: true, 355 + LargeFiles: []string{"foo", "bar"}, 356 + }, 357 + }, 358 + }, 359 + IndexOptions: &IndexOptions{ 360 + Symbols: true, 361 + LargeFiles: []string{"foo", "bar"}, 362 + Branches: []zoekt.RepositoryBranch{}, 363 + LanguageMap: map[string]ctags.CTagsParserType{}, 364 + }, 632 365 }, 633 366 634 - `{"Symbols": false, "LargeFiles": ["foo","bar"]}`: { 635 - LargeFiles: []string{"foo", "bar"}, 367 + { 368 + Response: &proto.SearchConfigurationResponse{ 369 + UpdatedOptions: []*proto.ZoektIndexOptions{ 370 + { 371 + Symbols: false, 372 + LargeFiles: []string{"foo", "bar"}, 373 + }, 374 + }, 375 + }, 376 + IndexOptions: &IndexOptions{ 377 + LargeFiles: []string{"foo", "bar"}, 378 + Branches: []zoekt.RepositoryBranch{}, 379 + LanguageMap: map[string]ctags.CTagsParserType{}, 380 + }, 636 381 }, 637 382 638 - `{}`: {}, 383 + { 384 + Response: &proto.SearchConfigurationResponse{}, 385 + }, 639 386 640 - `{"Symbols": true}`: { 641 - Symbols: true, 387 + { 388 + Response: &proto.SearchConfigurationResponse{ 389 + UpdatedOptions: []*proto.ZoektIndexOptions{ 390 + { 391 + Symbols: true, 392 + }, 393 + }, 394 + }, 395 + IndexOptions: &IndexOptions{ 396 + Symbols: true, 397 + Branches: []zoekt.RepositoryBranch{}, 398 + LanguageMap: map[string]ctags.CTagsParserType{}, 399 + }, 642 400 }, 643 401 644 - `{"RepoID": 123}`: { 645 - RepoID: 123, 402 + { 403 + Response: &proto.SearchConfigurationResponse{ 404 + UpdatedOptions: []*proto.ZoektIndexOptions{ 405 + { 406 + RepoId: 123, 407 + }, 408 + }, 409 + }, 410 + IndexOptions: &IndexOptions{ 411 + RepoID: 123, 412 + Branches: []zoekt.RepositoryBranch{}, 413 + LanguageMap: map[string]ctags.CTagsParserType{}, 414 + }, 646 415 }, 647 416 648 - `{"Error": "boom"}`: nil, 417 + { 418 + Response: &proto.SearchConfigurationResponse{ 419 + UpdatedOptions: []*proto.ZoektIndexOptions{ 420 + { 421 + Error: "boom", 422 + }, 423 + }, 424 + }, 425 + }, 649 426 } 650 427 651 - for r, want := range cases { 652 - response = []byte(r) 428 + for _, tc := range cases { 429 + response = tc.Response 653 430 654 431 var got IndexOptions 655 432 var err error ··· 659 436 err = e 660 437 }, 123) 661 438 662 - if err != nil && want != nil { 439 + if err != nil && tc.IndexOptions != nil { 663 440 t.Fatalf("unexpected error: %v", err) 664 441 } 665 - if want == nil { 442 + if tc.IndexOptions == nil { 666 443 continue 667 444 } 668 445 669 - want.CloneURL = sg.getCloneURL(got.Name) 446 + tc.IndexOptions.CloneURL = sg.getCloneURL(got.Name) 670 447 671 - if d := cmp.Diff(*want, got); d != "" { 672 - t.Log("response", r) 448 + if d := cmp.Diff(*tc.IndexOptions, got); d != "" { 673 449 t.Errorf("mismatch (-want +got):\n%s", d) 674 450 } 675 451 } ··· 677 453 // Special case our fingerprint API which doesn't return anything if the 678 454 // repo hasn't changed. 679 455 t.Run("unchanged", func(t *testing.T) { 680 - response = []byte(``) 456 + response = &proto.SearchConfigurationResponse{} 681 457 682 458 got := false 683 459 var err error

+2 -37

cmd/zoekt-sourcegraph-indexserver/main.go

··· 108 108 Help: "Number of indexed repos by code host", 109 109 }) 110 110 111 - metricNumAssigned = promauto.NewGauge(prometheus.GaugeOpts{ 112 - Name: "index_num_assigned", 113 - Help: "Number of repos assigned to this indexer by code host", 114 - }) 115 - 116 111 metricFailingTotal = promauto.NewCounter(prometheus.CounterOpts{ 117 112 Name: "index_failing_total", 118 113 Help: "Counts failures to index (indexing activity, should be used with rate())", ··· 1139 1134 return d 1140 1135 } 1141 1136 1142 - func getEnvWithDefaultBool(k string, defaultVal bool) bool { 1143 - v := os.Getenv(k) 1144 - if v == "" { 1145 - return defaultVal 1146 - } 1147 - 1148 - b, err := strconv.ParseBool(v) 1149 - if err != nil { 1150 - log.Fatalf("error parsing ENV %s to bool: %s", k, err) 1151 - } 1152 - return b 1153 - } 1154 - 1155 1137 func getEnvWithDefaultEmptySet(k string) map[string]struct{} { 1156 1138 set := map[string]struct{}{} 1157 1139 for _, v := range strings.Split(os.Getenv(k), ",") { ··· 1224 1206 // config values related to backoff indexing repos with one or more consecutive failures 1225 1207 backoffDuration time.Duration 1226 1208 maxBackoffDuration time.Duration 1227 - 1228 - // useGRPC is true if we should use the gRPC API to talk to Sourcegraph. 1229 - useGRPC bool 1230 1209 } 1231 1210 1232 1211 func (rc *rootConfig) registerRootFlags(fs *flag.FlagSet) { ··· 1240 1219 fs.IntVar(&rc.blockProfileRate, "block_profile_rate", getEnvWithDefaultInt("BLOCK_PROFILE_RATE", -1), "Sampling rate of Go's block profiler in nanoseconds. Values <=0 disable the blocking profiler Var(default). A value of 1 includes every blocking event. See https://pkg.go.dev/runtime#SetBlockProfileRate") 1241 1220 fs.DurationVar(&rc.backoffDuration, "backoff_duration", getEnvWithDefaultDuration("BACKOFF_DURATION", 10*time.Minute), "for the given duration we backoff from enqueue operations for a repository that's failed its previous indexing attempt. Consecutive failures increase the duration of the delay linearly up to the maxBackoffDuration. A negative value disables indexing backoff.") 1242 1221 fs.DurationVar(&rc.maxBackoffDuration, "max_backoff_duration", getEnvWithDefaultDuration("MAX_BACKOFF_DURATION", 120*time.Minute), "the maximum duration to backoff from enqueueing a repo for indexing. A negative value disables indexing backoff.") 1243 - fs.BoolVar(&rc.useGRPC, "use_grpc", mustGetBoolFromEnvironmentVariables([]string{"GRPC_ENABLED", "SG_FEATURE_FLAG_GRPC"}, true), "use the gRPC API to talk to Sourcegraph") 1244 1222 1245 1223 // flags related to shard merging 1246 1224 fs.DurationVar(&rc.vacuumInterval, "vacuum_interval", getEnvWithDefaultDuration("SRC_VACUUM_INTERVAL", 24*time.Hour), "run vacuum this often") ··· 1406 1384 if v := os.Getenv("SRC_REPO_CONFIG_BATCH_SIZE"); v != "" { 1407 1385 batchSize, err = strconv.Atoi(v) 1408 1386 if err != nil { 1409 - return nil, fmt.Errorf("Invalid value for SRC_REPO_CONFIG_BATCH_SIZE, must be int") 1387 + return nil, fmt.Errorf("invalid value for SRC_REPO_CONFIG_BATCH_SIZE, must be int") 1410 1388 } 1411 1389 } 1412 1390 1413 1391 opts := []SourcegraphClientOption{ 1414 1392 WithBatchSize(batchSize), 1415 - WithShouldUseGRPC(conf.useGRPC), 1416 1393 } 1417 1394 1418 1395 logger := sglog.Scoped("zoektConfigurationGRPCClient") ··· 1421 1398 return nil, fmt.Errorf("initializing gRPC connection to %q: %w", rootURL.Host, err) 1422 1399 } 1423 1400 1424 - opts = append(opts, WithGRPCClient(client)) 1425 - sg = newSourcegraphClient(rootURL, conf.hostname, opts...) 1401 + sg = newSourcegraphClient(rootURL, conf.hostname, client, opts...) 1426 1402 1427 1403 } else { 1428 1404 sg = sourcegraphFake{ ··· 1631 1607 if err := rootCmd().ParseAndRun(context.Background(), os.Args[1:]); err != nil { 1632 1608 log.Fatal(err) 1633 1609 } 1634 - } 1635 - 1636 - // mustGetBoolFromEnvironmentVariables is like getBoolFromEnvironmentVariables, but it panics 1637 - // if any of the provided environment variables fails to parse as a boolean. 1638 - func mustGetBoolFromEnvironmentVariables(envVarNames []string, defaultBool bool) bool { 1639 - value, err := getBoolFromEnvironmentVariables(envVarNames, defaultBool) 1640 - if err != nil { 1641 - panic(err) 1642 - } 1643 - 1644 - return value 1645 1610 } 1646 1611 1647 1612 // getBoolFromEnvironmentVariables returns the boolean defined by the first environment

+42 -116

cmd/zoekt-sourcegraph-indexserver/main_test.go

··· 6 6 "fmt" 7 7 "io" 8 8 "log" 9 - "net/http" 10 - "net/http/httptest" 11 9 "net/url" 12 10 "os" 13 11 "path/filepath" ··· 33 31 } 34 32 35 33 s := &Server{ 36 - Sourcegraph: newSourcegraphClient(root, "", WithBatchSize(0)), 34 + Sourcegraph: newSourcegraphClient(root, "", nil, WithBatchSize(0)), 37 35 IndexDir: "/testdata/index", 38 36 CPUCount: 6, 39 37 IndexConcurrency: 1, ··· 101 99 for _, tt := range cases { 102 100 t.Run(tt.name, func(t *testing.T) { 103 101 s := &Server{ 104 - Sourcegraph: newSourcegraphClient(root, "", WithBatchSize(0)), 102 + Sourcegraph: newSourcegraphClient(root, "", nil, WithBatchSize(0)), 105 103 IndexDir: "/testdata/index", 106 104 CPUCount: tt.cpuCount, 107 105 IndexConcurrency: tt.indexConcurrency, ··· 117 115 118 116 t.Run("index option is limited by available CPU", func(t *testing.T) { 119 117 s := &Server{ 120 - Sourcegraph: newSourcegraphClient(root, "", WithBatchSize(0)), 118 + Sourcegraph: newSourcegraphClient(root, "", nil, WithBatchSize(0)), 121 119 IndexDir: "/testdata/index", 122 120 IndexConcurrency: 1, 123 121 } ··· 133 131 } 134 132 135 133 func TestListRepoIDs(t *testing.T) { 136 - t.Run("gRPC", func(t *testing.T) { 137 - grpcClient := &mockGRPCClient{} 138 - 139 - clientOptions := []SourcegraphClientOption{ 140 - WithShouldUseGRPC(true), 141 - WithGRPCClient(grpcClient), 142 - WithBatchSize(0), 143 - } 134 + grpcClient := &mockGRPCClient{} 144 135 145 - testURL := url.URL{Scheme: "http", Host: "does.not.matter"} 146 - testHostname := "test-hostname" 147 - s := newSourcegraphClient(&testURL, testHostname, clientOptions...) 136 + clientOptions := []SourcegraphClientOption{ 137 + WithBatchSize(0), 138 + } 148 139 149 - listCalled := false 150 - grpcClient.mockList = func(ctx context.Context, in *proto.ListRequest, opts ...grpc.CallOption) (*proto.ListResponse, error) { 151 - listCalled = true 140 + testURL := url.URL{Scheme: "http", Host: "does.not.matter"} 141 + testHostname := "test-hostname" 142 + s := newSourcegraphClient(&testURL, testHostname, grpcClient, clientOptions...) 152 143 153 - gotRepoIDs := in.GetIndexedIds() 154 - sort.Slice(gotRepoIDs, func(i, j int) bool { 155 - return gotRepoIDs[i] < gotRepoIDs[j] 156 - }) 144 + listCalled := false 145 + grpcClient.mockList = func(ctx context.Context, in *proto.ListRequest, opts ...grpc.CallOption) (*proto.ListResponse, error) { 146 + listCalled = true 157 147 158 - wantRepoIDs := []int32{1, 3} 159 - sort.Slice(wantRepoIDs, func(i, j int) bool { 160 - return wantRepoIDs[i] < wantRepoIDs[j] 161 - }) 162 - 163 - if diff := cmp.Diff(wantRepoIDs, gotRepoIDs); diff != "" { 164 - t.Errorf("indexed repoIDs mismatch (-want +got):\n%s", diff) 165 - } 166 - 167 - hostname := in.GetHostname() 168 - if diff := cmp.Diff(testHostname, hostname); diff != "" { 169 - t.Errorf("hostname mismatch (-want +got):\n%s", diff) 170 - } 171 - 172 - return &proto.ListResponse{RepoIds: []int32{1, 2, 3}}, nil 173 - } 174 - 175 - ctx := context.Background() 176 - got, err := s.List(ctx, []uint32{1, 3}) 177 - if err != nil { 178 - t.Fatal(err) 179 - } 180 - 181 - if !listCalled { 182 - t.Fatalf("List was not called") 183 - } 184 - 185 - receivedRepoIDs := got.IDs 186 - sort.Slice(receivedRepoIDs, func(i, j int) bool { 187 - return receivedRepoIDs[i] < receivedRepoIDs[j] 148 + gotRepoIDs := in.GetIndexedIds() 149 + sort.Slice(gotRepoIDs, func(i, j int) bool { 150 + return gotRepoIDs[i] < gotRepoIDs[j] 188 151 }) 189 152 190 - expectedRepoIDs := []uint32{1, 2, 3} 191 - sort.Slice(expectedRepoIDs, func(i, j int) bool { 192 - return expectedRepoIDs[i] < expectedRepoIDs[j] 153 + wantRepoIDs := []int32{1, 3} 154 + sort.Slice(wantRepoIDs, func(i, j int) bool { 155 + return wantRepoIDs[i] < wantRepoIDs[j] 193 156 }) 194 157 195 - if diff := cmp.Diff(expectedRepoIDs, receivedRepoIDs); diff != "" { 196 - t.Errorf("mismatch in list of all repoIDs (-want +got):\n%s", diff) 197 - } 198 - }) 199 - 200 - t.Run("REST", func(t *testing.T) { 201 - var gotBody string 202 - var gotURL *url.URL 203 - ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { 204 - gotURL = r.URL 205 - 206 - b, err := io.ReadAll(r.Body) 207 - if err != nil { 208 - t.Fatal(err) 209 - } 210 - gotBody = string(b) 211 - 212 - _, err = w.Write([]byte(`{"RepoIDs": [1, 2, 3]}`)) 213 - if err != nil { 214 - t.Fatal(err) 215 - } 216 - })) 217 - defer ts.Close() 218 - 219 - u, err := url.Parse(ts.URL) 220 - if err != nil { 221 - t.Fatal(err) 158 + if diff := cmp.Diff(wantRepoIDs, gotRepoIDs); diff != "" { 159 + t.Errorf("indexed repoIDs mismatch (-want +got):\n%s", diff) 222 160 } 223 161 224 - s := newSourcegraphClient(u, "test-indexed-search-1", WithBatchSize(0)) 225 - 226 - gotRepos, err := s.List(context.Background(), []uint32{1, 3}) 227 - if err != nil { 228 - t.Fatal(err) 162 + hostname := in.GetHostname() 163 + if diff := cmp.Diff(testHostname, hostname); diff != "" { 164 + t.Errorf("hostname mismatch (-want +got):\n%s", diff) 229 165 } 230 166 231 - if want := []uint32{1, 2, 3}; !cmp.Equal(gotRepos.IDs, want) { 232 - t.Errorf("repos mismatch (-want +got):\n%s", cmp.Diff(want, gotRepos.IDs)) 233 - } 234 - if want := `{"Hostname":"test-indexed-search-1","IndexedIDs":[1,3]}`; gotBody != want { 235 - t.Errorf("body mismatch (-want +got):\n%s", cmp.Diff(want, gotBody)) 236 - } 237 - if want := "/.internal/repos/index"; gotURL.Path != want { 238 - t.Errorf("request path mismatch (-want +got):\n%s", cmp.Diff(want, gotURL.Path)) 239 - } 240 - }) 241 - } 242 - 243 - func TestListRepoIDs_Error_REST(t *testing.T) { 244 - // Note: There is no gRPC equivalent to this test because gRPC errors are 245 - // always returned as an error to the caller. 167 + return &proto.ListResponse{RepoIds: []int32{1, 2, 3}}, nil 168 + } 246 169 247 - msg := "deadbeaf deadbeaf" 248 - ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { 249 - // This is how Sourcegraph returns error messages to the caller. 250 - http.Error(w, msg, http.StatusInternalServerError) 251 - })) 252 - defer ts.Close() 253 - 254 - u, err := url.Parse(ts.URL) 170 + ctx := context.Background() 171 + got, err := s.List(ctx, []uint32{1, 3}) 255 172 if err != nil { 256 173 t.Fatal(err) 257 174 } 258 175 259 - s := newSourcegraphClient(u, "test-indexed-search-1", WithBatchSize(0)) 260 - s.restClient.RetryMax = 0 176 + if !listCalled { 177 + t.Fatalf("List was not called") 178 + } 179 + 180 + receivedRepoIDs := got.IDs 181 + sort.Slice(receivedRepoIDs, func(i, j int) bool { 182 + return receivedRepoIDs[i] < receivedRepoIDs[j] 183 + }) 261 184 262 - _, err = s.List(context.Background(), []uint32{1, 3}) 185 + expectedRepoIDs := []uint32{1, 2, 3} 186 + sort.Slice(expectedRepoIDs, func(i, j int) bool { 187 + return expectedRepoIDs[i] < expectedRepoIDs[j] 188 + }) 263 189 264 - if !strings.Contains(err.Error(), msg) { 265 - t.Fatalf("%s does not contain %s", err.Error(), msg) 190 + if diff := cmp.Diff(expectedRepoIDs, receivedRepoIDs); diff != "" { 191 + t.Errorf("mismatch in list of all repoIDs (-want +got):\n%s", diff) 266 192 } 267 193 } 268 194

-2

cmd/zoekt-sourcegraph-indexserver/merge.go

··· 16 16 "github.com/sourcegraph/zoekt" 17 17 ) 18 18 19 - var reCompound = regexp.MustCompile(`compound-.*\.zoekt`) 20 - 21 19 var metricShardMergingRunning = promauto.NewGauge(prometheus.GaugeOpts{ 22 20 Name: "index_shard_merging_running", 23 21 Help: "Set to 1 if indexserver's merge job is running.",

+10 -354

cmd/zoekt-sourcegraph-indexserver/sg.go

··· 4 4 "bufio" 5 5 "bytes" 6 6 "context" 7 - "encoding/json" 8 7 "errors" 9 8 "fmt" 10 9 "hash/crc32" 11 - "io" 12 10 "log" 13 11 "math/rand" 14 - "net/http" 15 12 "net/url" 16 13 "os" 17 14 "os/exec" ··· 22 19 "time" 23 20 24 21 "github.com/go-git/go-git/v5" 25 - retryablehttp "github.com/hashicorp/go-retryablehttp" 26 22 proto "github.com/sourcegraph/zoekt/cmd/zoekt-sourcegraph-indexserver/protos/sourcegraph/zoekt/configuration/v1" 27 23 "github.com/sourcegraph/zoekt/ctags" 28 24 "golang.org/x/net/trace" 29 - "google.golang.org/grpc" 30 25 31 26 "github.com/sourcegraph/zoekt" 32 27 ) ··· 93 88 } 94 89 } 95 90 96 - // WithShouldUseGRPC enables or disables the use of gRPC for communicating with Sourcegraph. 97 - func WithShouldUseGRPC(useGRPC bool) SourcegraphClientOption { 98 - return func(c *sourcegraphClient) { 99 - c.useGRPC = useGRPC 100 - } 101 - } 102 - 103 - // WithGRPCClient sets the gRPC client to use for communicating with Sourcegraph. 104 - func WithGRPCClient(client proto.ZoektConfigurationServiceClient) SourcegraphClientOption { 105 - return func(c *sourcegraphClient) { 106 - c.grpcClient = client 107 - } 108 - } 109 - 110 - func newSourcegraphClient(rootURL *url.URL, hostname string, opts ...SourcegraphClientOption) *sourcegraphClient { 111 - httpClient := retryablehttp.NewClient() 112 - httpClient.Logger = debug 113 - 114 - // Sourcegraph might return an error message in the body if StatusCode==500. The 115 - // default behavior of the go-retryablehttp restClient is to drain the body and not 116 - // to propagate the error. Hence, we call ErrorPropagatedRetryPolicy instead of 117 - // DefaultRetryPolicy and augment the error with the response body if possible. 118 - httpClient.CheckRetry = func(ctx context.Context, resp *http.Response, err error) (bool, error) { 119 - shouldRetry, checkErr := retryablehttp.ErrorPropagatedRetryPolicy(ctx, resp, err) 120 - 121 - if resp != nil && resp.StatusCode == http.StatusInternalServerError { 122 - if b, e := io.ReadAll(resp.Body); e == nil { 123 - checkErr = fmt.Errorf("%w: body=%q", checkErr, string(b)) 124 - } 125 - } 126 - 127 - return shouldRetry, checkErr 128 - } 129 - 91 + func newSourcegraphClient(rootURL *url.URL, hostname string, grpcClient proto.ZoektConfigurationServiceClient, opts ...SourcegraphClientOption) *sourcegraphClient { 130 92 client := &sourcegraphClient{ 131 93 Root: rootURL, 132 - restClient: httpClient, 133 94 Hostname: hostname, 134 95 BatchSize: 0, 135 - grpcClient: noopGRPCClient{}, 136 - useGRPC: false, // disable gRPC by default 96 + grpcClient: grpcClient, 137 97 } 138 98 139 99 for _, opt := range opts { ··· 157 117 // zero a value of 10000 is used. 158 118 BatchSize int 159 119 160 - // restClient is used to make requests to the Sourcegraph instance. Prefer to 161 - // use .doRequest() to ensure the appropriate headers are set. 162 - restClient *retryablehttp.Client 163 - 164 120 // grpcClient is used to make requests to the Sourcegraph instance if gRPC is enabled. 165 121 grpcClient proto.ZoektConfigurationServiceClient 166 - 167 - // configFingerprint is the last config fingerprint returned from 168 - // Sourcegraph. It can be used for future calls to the configuration 169 - // endpoint. 170 - // 171 - // configFingerprint is mutually exclusive with configFingerprintProto - this field 172 - // will only be used if gRPC is disabled. 173 - configFingerprint string 174 122 175 123 // configFingerprintProto is the last config fingerprint (as GRPC) returned from 176 124 // Sourcegraph. It can be used for future calls to the configuration ··· 185 133 // configFingerprint logic is faulty. When it is cleared out, we fallback to 186 134 // calculating everything. 187 135 configFingerprintReset time.Time 188 - 189 - // useGRPC indicates whether we should use a gRPC client to communicate with Sourcegraph. 190 - useGRPC bool 191 136 } 192 137 193 138 // GetDocumentRanks asks Sourcegraph for a mapping of file paths to rank 194 139 // vectors. 195 140 func (s *sourcegraphClient) GetDocumentRanks(ctx context.Context, repoName string) (RepoPathRanks, error) { 196 - if s.useGRPC { 197 - return s.getDocumentRanksGRPC(ctx, repoName) 198 - } 199 - 200 - return s.getDocumentRanksREST(ctx, repoName) 201 - } 202 - 203 - func (s *sourcegraphClient) getDocumentRanksGRPC(ctx context.Context, repoName string) (RepoPathRanks, error) { 204 141 resp, err := s.grpcClient.DocumentRanks(ctx, &proto.DocumentRanksRequest{Repository: repoName}) 205 142 if err != nil { 206 143 return RepoPathRanks{}, err ··· 212 149 return out, nil 213 150 } 214 151 215 - func (s *sourcegraphClient) getDocumentRanksREST(ctx context.Context, repoName string) (RepoPathRanks, error) { 216 - u := s.Root.ResolveReference(&url.URL{ 217 - Path: "/.internal/ranks/" + strings.Trim(repoName, "/") + "/documents", 218 - }) 219 - 220 - b, err := s.get(ctx, u) 221 - if err != nil { 222 - return RepoPathRanks{}, err 223 - } 224 - 225 - ranks := RepoPathRanks{} 226 - err = json.Unmarshal(b, &ranks) 227 - if err != nil { 228 - return RepoPathRanks{}, err 229 - } 230 - 231 - return ranks, nil 232 - } 233 - 234 - func (s *sourcegraphClient) get(ctx context.Context, u *url.URL) ([]byte, error) { 235 - req, err := retryablehttp.NewRequestWithContext(ctx, "GET", u.String(), nil) 236 - if err != nil { 237 - return nil, err 238 - } 239 - 240 - resp, err := s.doRequest(req) 241 - if err != nil { 242 - return nil, err 243 - } 244 - defer resp.Body.Close() 245 - 246 - if resp.StatusCode != http.StatusOK { 247 - b, err := io.ReadAll(io.LimitReader(resp.Body, 1024)) 248 - _ = resp.Body.Close() 249 - if err != nil { 250 - return nil, err 251 - } 252 - return nil, &url.Error{ 253 - Op: "Get", 254 - URL: u.String(), 255 - Err: fmt.Errorf("%s: %s", resp.Status, string(b)), 256 - } 257 - } 258 - 259 - b, err := io.ReadAll(resp.Body) 260 - if err != nil { 261 - return nil, err 262 - } 263 - 264 - return b, nil 265 - } 266 - 267 152 func (s *sourcegraphClient) List(ctx context.Context, indexed []uint32) (*SourcegraphListResult, error) { 268 153 repos, err := s.listRepoIDs(ctx, indexed) 269 154 if err != nil { ··· 287 172 s.configFingerprintReset = time.Now().Add(next) 288 173 289 174 s.configFingerprintProto = nil 290 - s.configFingerprint = "" 291 175 } 292 176 293 177 // getIndexOptionsFunc is a function that can be used to get the index ··· 299 183 // fail, the old fingerprint is restored. 300 184 type getIndexOptionsFunc func(repos ...uint32) ([]indexOptionsItem, error) 301 185 302 - // default to REST 303 186 mkGetIndexOptionsFunc := func(tr trace.Trace) getIndexOptionsFunc { 304 - startingFingerPrint := s.configFingerprint 305 - tr.LazyPrintf("fingerprint: %s", startingFingerPrint) 187 + startingFingerPrint := s.configFingerprintProto 188 + tr.LazyPrintf("fingerprint: %s", startingFingerPrint.String()) 306 189 307 190 first := true 308 191 return func(repos ...uint32) ([]indexOptionsItem, error) { 309 - options, nextFingerPrint, err := s.getIndexOptionsREST(startingFingerPrint, repos...) 192 + options, nextFingerPrint, err := s.getIndexOptions(ctx, startingFingerPrint, repos) 310 193 if err != nil { 311 194 first = false 312 - s.configFingerprint = startingFingerPrint 195 + s.configFingerprintProto = startingFingerPrint 313 196 314 197 return nil, err 315 198 } 316 199 317 200 if first { 318 201 first = false 319 - s.configFingerprint = nextFingerPrint 320 - 321 - tr.LazyPrintf("new fingerprint: %s", nextFingerPrint) 202 + s.configFingerprintProto = nextFingerPrint 203 + tr.LazyPrintf("new fingerprint: %s", nextFingerPrint.String()) 322 204 } 323 205 324 206 return options, nil 325 207 } 326 208 } 327 209 328 - // If we enabled GRPC, use our gRPC client instead. 329 - if s.useGRPC { 330 - mkGetIndexOptionsFunc = func(tr trace.Trace) getIndexOptionsFunc { 331 - startingFingerPrint := s.configFingerprintProto 332 - tr.LazyPrintf("fingerprint: %s", startingFingerPrint.String()) 333 - 334 - first := true 335 - return func(repos ...uint32) ([]indexOptionsItem, error) { 336 - options, nextFingerPrint, err := s.getIndexOptionsGRPC(ctx, startingFingerPrint, repos) 337 - if err != nil { 338 - first = false 339 - s.configFingerprintProto = startingFingerPrint 340 - 341 - return nil, err 342 - } 343 - 344 - if first { 345 - first = false 346 - s.configFingerprintProto = nextFingerPrint 347 - tr.LazyPrintf("new fingerprint: %s", nextFingerPrint.String()) 348 - } 349 - 350 - return options, nil 351 - } 352 - } 353 - } 354 - 355 210 iterate := func(f func(IndexOptions)) { 356 211 start := time.Now() 357 212 tr := trace.New("getIndexOptions", "") ··· 407 262 batchSize = 10_000 408 263 } 409 264 410 - getIndexOptions := func(repos ...uint32) ([]indexOptionsItem, error) { 411 - opts, _, err := s.getIndexOptionsREST("", repos...) 412 - return opts, err 413 - } 414 - 415 - if s.useGRPC { 416 - getIndexOptions = func(repos ...uint32) ([]indexOptionsItem, error) { 417 - opts, _, err := s.getIndexOptionsGRPC(context.Background(), nil, repos) 418 - return opts, err 419 - } 420 - } 421 - 422 265 for repos := range batched(repos, batchSize) { 423 - opts, err := getIndexOptions(repos...) 266 + opts, _, err := s.getIndexOptions(context.Background(), nil, repos) 424 267 if err != nil { 425 268 for _, id := range repos { 426 269 onError(id, err) ··· 525 368 } 526 369 } 527 370 528 - func (s *sourcegraphClient) getIndexOptionsGRPC(ctx context.Context, fingerprint *proto.Fingerprint, repos []uint32) ([]indexOptionsItem, *proto.Fingerprint, error) { 371 + func (s *sourcegraphClient) getIndexOptions(ctx context.Context, fingerprint *proto.Fingerprint, repos []uint32) ([]indexOptionsItem, *proto.Fingerprint, error) { 529 372 repoIDs := make([]int32, 0, len(repos)) 530 373 for _, id := range repos { 531 374 repoIDs = append(repoIDs, int32(id)) ··· 554 397 return items, response.GetFingerprint(), nil 555 398 } 556 399 557 - const fingerprintHeader = "X-Sourcegraph-Config-Fingerprint" 558 - 559 - func (s *sourcegraphClient) getIndexOptionsREST(fingerprint string, repos ...uint32) ([]indexOptionsItem, string, error) { 560 - u := s.Root.ResolveReference(&url.URL{ 561 - Path: "/.internal/search/configuration", 562 - }) 563 - 564 - repoIDs := make([]string, len(repos)) 565 - for i, id := range repos { 566 - repoIDs[i] = strconv.Itoa(int(id)) 567 - } 568 - data := url.Values{"repoID": repoIDs} 569 - req, err := retryablehttp.NewRequest("POST", u.String(), []byte(data.Encode())) 570 - if err != nil { 571 - return nil, "", err 572 - } 573 - req.Header.Set("Content-Type", "application/x-www-form-urlencoded") 574 - if fingerprint != "" { 575 - req.Header.Set(fingerprintHeader, fingerprint) 576 - } 577 - 578 - resp, err := s.doRequest(req) 579 - if err != nil { 580 - return nil, "", err 581 - } 582 - defer resp.Body.Close() 583 - 584 - if resp.StatusCode != http.StatusOK { 585 - b, err := io.ReadAll(io.LimitReader(resp.Body, 1024)) 586 - _ = resp.Body.Close() 587 - if err != nil { 588 - return nil, "", err 589 - } 590 - return nil, "", &url.Error{ 591 - Op: "Get", 592 - URL: u.String(), 593 - Err: fmt.Errorf("%s: %s", resp.Status, string(b)), 594 - } 595 - } 596 - 597 - dec := json.NewDecoder(resp.Body) 598 - var opts []indexOptionsItem 599 - for { 600 - var opt indexOptionsItem 601 - err := dec.Decode(&opt) 602 - if err == io.EOF { 603 - break 604 - } 605 - if err != nil { 606 - return nil, "", fmt.Errorf("error decoding body: %w", err) 607 - } 608 - opt.CloneURL = s.getCloneURL(opt.Name) 609 - opts = append(opts, opt) 610 - } 611 - 612 - return opts, resp.Header.Get(fingerprintHeader), nil 613 - } 614 - 615 400 func (s *sourcegraphClient) getCloneURL(name string) string { 616 401 return s.Root.ResolveReference(&url.URL{Path: path.Join("/.internal/git", name)}).String() 617 402 } 618 403 619 404 func (s *sourcegraphClient) listRepoIDs(ctx context.Context, indexed []uint32) ([]uint32, error) { 620 - if s.useGRPC { 621 - return s.listRepoIDsGRPC(ctx, indexed) 622 - } 623 - 624 - return s.listRepoIDsREST(ctx, indexed) 625 - } 626 - 627 - func (s *sourcegraphClient) listRepoIDsGRPC(ctx context.Context, indexed []uint32) ([]uint32, error) { 628 405 var request proto.ListRequest 629 406 request.Hostname = s.Hostname 630 407 request.IndexedIds = make([]int32, 0, len(indexed)) ··· 645 422 return repoIDs, nil 646 423 } 647 424 648 - func (s *sourcegraphClient) listRepoIDsREST(_ context.Context, indexed []uint32) ([]uint32, error) { 649 - body, err := json.Marshal(&struct { 650 - Hostname string 651 - IndexedIDs []uint32 652 - }{ 653 - Hostname: s.Hostname, 654 - IndexedIDs: indexed, 655 - }) 656 - if err != nil { 657 - return nil, err 658 - } 659 - 660 - u := s.Root.ResolveReference(&url.URL{Path: "/.internal/repos/index"}) 661 - req, err := retryablehttp.NewRequest(http.MethodPost, u.String(), bytes.NewReader(body)) 662 - if err != nil { 663 - return nil, err 664 - } 665 - req.Header.Set("Content-Type", "application/json; charset=utf-8") 666 - 667 - resp, err := s.doRequest(req) 668 - if err != nil { 669 - return nil, err 670 - } 671 - defer resp.Body.Close() 672 - 673 - if resp.StatusCode != http.StatusOK { 674 - return nil, fmt.Errorf("failed to list repositories: status %s", resp.Status) 675 - } 676 - 677 - var data struct { 678 - RepoIDs []uint32 679 - } 680 - err = json.NewDecoder(resp.Body).Decode(&data) 681 - if err != nil { 682 - return nil, err 683 - } 684 - 685 - return data.RepoIDs, nil 686 - } 687 - 688 425 type indexStatus struct { 689 426 RepoID uint32 690 427 Branches []zoekt.RepositoryBranch ··· 752 489 func (s *sourcegraphClient) UpdateIndexStatus(repositories []indexStatus) error { 753 490 r := updateIndexStatusRequest{Repositories: repositories} 754 491 755 - if s.useGRPC { 756 - return s.updateIndexStatusGRPC(r) 757 - } 758 - 759 - return s.updateIndexStatusREST(r) 760 - } 761 - 762 - func (s *sourcegraphClient) updateIndexStatusGRPC(r updateIndexStatusRequest) error { 763 492 request := r.ToProto() 764 493 _, err := s.grpcClient.UpdateIndexStatus(context.Background(), request) 765 494 if err != nil { ··· 769 498 return nil 770 499 } 771 500 772 - func (s *sourcegraphClient) updateIndexStatusREST(r updateIndexStatusRequest) error { 773 - payload, err := json.Marshal(r) 774 - if err != nil { 775 - return err 776 - } 777 - 778 - u := s.Root.ResolveReference(&url.URL{Path: "/.internal/search/index-status"}) 779 - req, err := retryablehttp.NewRequest(http.MethodPost, u.String(), bytes.NewReader(payload)) 780 - if err != nil { 781 - return err 782 - } 783 - req.Header.Set("Content-Type", "application/json; charset=utf-8") 784 - 785 - resp, err := s.doRequest(req) 786 - if err != nil { 787 - return err 788 - } 789 - defer resp.Body.Close() 790 - 791 - if resp.StatusCode != http.StatusOK { 792 - return fmt.Errorf("failed to update index status: status %s", resp.Status) 793 - } 794 - 795 - return nil 796 - } 797 - 798 - // doRequest executes the provided request after adding the appropriate headers 799 - // for interacting with a Sourcegraph instance. 800 - func (s *sourcegraphClient) doRequest(req *retryablehttp.Request) (*http.Response, error) { 801 - // Make all requests as an internal user. 802 - // 803 - // Should match github.com/sourcegraph/sourcegraph/internal/actor.headerKeyActorUID 804 - // and github.com/sourcegraph/sourcegraph/internal/actor.headerValueInternalActor 805 - req.Header.Set("X-Sourcegraph-Actor-UID", "internal") 806 - return s.restClient.Do(req) 807 - } 808 - 809 501 type sourcegraphFake struct { 810 502 RootDir string 811 503 Log *log.Logger ··· 842 534 843 535 ranks.MeanRank = sum / float64(count) 844 536 return ranks, nil 845 - } 846 - 847 - func floats64(s string) []float64 { 848 - parts := strings.Split(s, ",") 849 - 850 - var r []float64 851 - for _, rank := range parts { 852 - f, err := strconv.ParseFloat(rank, 64) 853 - if err != nil { 854 - return nil 855 - } 856 - r = append(r, f) 857 - } 858 - 859 - return r 860 537 } 861 538 862 539 func (sf sourcegraphFake) List(ctx context.Context, indexed []uint32) (*SourcegraphListResult, error) { ··· 1084 761 } 1085 762 1086 763 func (s sourcegraphNop) ForceIterateIndexOptions(onSuccess func(IndexOptions), onError func(uint32, error), repos ...uint32) { 1087 - return 1088 764 } 1089 765 1090 766 func (s sourcegraphNop) GetDocumentRanks(ctx context.Context, repoName string) (RepoPathRanks, error) { ··· 1124 800 Paths: paths, 1125 801 } 1126 802 } 1127 - 1128 - type noopGRPCClient struct{} 1129 - 1130 - func (n noopGRPCClient) SearchConfiguration(ctx context.Context, in *proto.SearchConfigurationRequest, opts ...grpc.CallOption) (*proto.SearchConfigurationResponse, error) { 1131 - return nil, fmt.Errorf("grpc client not enabled") 1132 - } 1133 - 1134 - func (n noopGRPCClient) List(ctx context.Context, in *proto.ListRequest, opts ...grpc.CallOption) (*proto.ListResponse, error) { 1135 - return nil, fmt.Errorf("grpc client not enabled") 1136 - } 1137 - 1138 - func (n noopGRPCClient) DocumentRanks(ctx context.Context, in *proto.DocumentRanksRequest, opts ...grpc.CallOption) (*proto.DocumentRanksResponse, error) { 1139 - return nil, fmt.Errorf("grpc client not enabled") 1140 - } 1141 - 1142 - func (n noopGRPCClient) UpdateIndexStatus(ctx context.Context, in *proto.UpdateIndexStatusRequest, opts ...grpc.CallOption) (*proto.UpdateIndexStatusResponse, error) { 1143 - return nil, fmt.Errorf("grpc client not enabled") 1144 - } 1145 - 1146 - var _ proto.ZoektConfigurationServiceClient = noopGRPCClient{}

+1 -1

go.mod

··· 19 19 github.com/grafana/regexp v0.0.0-20221123153739-15dc172cd2db 20 20 github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.0.0-rc.0 21 21 github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.0.0 22 - github.com/hashicorp/go-retryablehttp v0.7.4 23 22 github.com/keegancsmith/rpc v1.3.0 24 23 github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f 25 24 github.com/opentracing/opentracing-go v1.2.0 ··· 97 96 github.com/grpc-ecosystem/grpc-gateway/v2 v2.19.1 // indirect 98 97 github.com/hashicorp/go-cleanhttp v0.5.2 // indirect 99 98 github.com/hashicorp/go-hclog v0.16.2 // indirect 99 + github.com/hashicorp/go-retryablehttp v0.7.4 // indirect 100 100 github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 // indirect 101 101 github.com/kevinburke/ssh_config v1.2.0 // indirect 102 102 github.com/kr/pretty v0.3.1 // indirect

Configure Feed

Configure Feed