diff --git a/config/service_config.go b/config/service_config.go index 2245f35..c551346 100644 --- a/config/service_config.go +++ b/config/service_config.go @@ -14,6 +14,24 @@ package config +const ( + // DefaultTargetChunkSize is the default number of OptimizedTarget entries per stream message. + // Sized conservatively: at ~40KB/target worst-case (target with ~10K direct deps × 4 bytes), + // 250 targets ≈ 10MB — well under the 64MB default gRPC per-message limit. + DefaultTargetChunkSize = 250 + + // DefaultChangedTargetChunkSize is the default number of ChangedTarget entries per stream message. + // A ChangedTarget carries both old_target and new_target (2× an OptimizedTarget), so we use + // half the regular chunk size to stay within the same byte budget. + DefaultChangedTargetChunkSize = 125 + + // DefaultMetadataMapChunkSize is the max entries per metadata message chunk. + // target_id_mapping and attribute_string_value_mapping scale with repo size and can exceed + // the 64MB gRPC message limit for large monorepos, so they are split across multiple messages. + // At ~85 bytes/entry (60-char avg target name + proto overhead), 50 000 entries ≈ 4.25MB per chunk. + DefaultMetadataMapChunkSize = 50_000 +) + // ServiceConfig holds operational configuration for the Tango service. type ServiceConfig struct { WorkerPoolSize int `yaml:"worker_pool_size"` // number of worker workspaces per repo diff --git a/controller/BUILD.bazel b/controller/BUILD.bazel index 581ba3b..23609fd 100644 --- a/controller/BUILD.bazel +++ b/controller/BUILD.bazel @@ -16,6 +16,7 @@ go_library( deps = [ "//config", "//core/common", + "//core/errors", "//core/storage", "//orchestrator", "//tangopb", @@ -37,6 +38,7 @@ go_test( ], embed = [":controller"], deps = [ + "//config", "//core/common", "//core/storage", "//core/storage/storagemock", diff --git a/controller/controller.go b/controller/controller.go index 34ca33c..1c786c4 100644 --- a/controller/controller.go +++ b/controller/controller.go @@ -20,7 +20,6 @@ import ( "github.com/uber-go/tally" "github.com/uber/tango/config" - "github.com/uber/tango/core/common" "github.com/uber/tango/core/storage" "github.com/uber/tango/orchestrator" pb "github.com/uber/tango/tangopb" @@ -66,15 +65,15 @@ func NewController(appCtx context.Context, p Params) pb.TangoYARPCServer { } targetChunkSize := p.ChunkConfig.TargetChunkSize if targetChunkSize <= 0 { - targetChunkSize = common.DefaultTargetChunkSize + targetChunkSize = config.DefaultTargetChunkSize } changedTargetChunkSize := p.ChunkConfig.ChangedTargetChunkSize if changedTargetChunkSize <= 0 { - changedTargetChunkSize = common.DefaultChangedTargetChunkSize + changedTargetChunkSize = config.DefaultChangedTargetChunkSize } metadataMapChunkSize := p.ChunkConfig.MetadataMapChunkSize if metadataMapChunkSize <= 0 { - metadataMapChunkSize = common.DefaultMetadataMapChunkSize + metadataMapChunkSize = config.DefaultMetadataMapChunkSize } return &controller{ logger: p.Logger, diff --git a/controller/getchangedtargets.go b/controller/getchangedtargets.go index feac4c3..42ae142 100644 --- a/controller/getchangedtargets.go +++ b/controller/getchangedtargets.go @@ -22,6 +22,7 @@ import ( "time" "github.com/uber/tango/core/common" + tgerrors "github.com/uber/tango/core/errors" "github.com/uber/tango/core/storage" pb "github.com/uber/tango/tangopb" "go.uber.org/zap" @@ -560,9 +561,6 @@ func (c *controller) compareTargetGraphs(ctx context.Context, logger *zap.Logger return results, nil } -// cancelCheckInterval is how often long-running loops check ctx.Err(). -const cancelCheckInterval = 4096 - // getTargetsAndMetadata builds ID->target maps and merges metadata from a target graph stream. // Metadata may arrive in multiple chunks (e.g. when target_id_mapping exceeds the gRPC message // size limit); all chunks are merged into a single Metadata so callers can use it uniformly. @@ -611,7 +609,7 @@ func buildNameIndex(ctx context.Context, targetsByID map[int32]*pb.OptimizedTarg byName := make(map[string]*pb.OptimizedTarget, len(targetsByID)) i := 0 for id, t := range targetsByID { - if i%cancelCheckInterval == 0 && ctx.Err() != nil { + if i%tgerrors.CancelCheckInterval == 0 && ctx.Err() != nil { return nil, ctx.Err() } i++ @@ -963,7 +961,7 @@ func computeDistances(ctx context.Context, changedByName map[string]*pb.ChangedT reverseDeps := make(map[string][]string, len(targetsByName)) revDepIter := 0 for name, t := range targetsByName { - if revDepIter%cancelCheckInterval == 0 && ctx.Err() != nil { + if revDepIter%tgerrors.CancelCheckInterval == 0 && ctx.Err() != nil { return ctx.Err() } revDepIter++ @@ -991,7 +989,7 @@ func computeDistances(ctx context.Context, changedByName map[string]*pb.ChangedT // BFS from seeds through reverseDeps. Shortest distance wins. bfsIter := 0 for len(queue) > 0 { - if bfsIter%cancelCheckInterval == 0 && ctx.Err() != nil { + if bfsIter%tgerrors.CancelCheckInterval == 0 && ctx.Err() != nil { return ctx.Err() } bfsIter++ diff --git a/controller/testhelper_test.go b/controller/testhelper_test.go index 56ff439..2a11419 100644 --- a/controller/testhelper_test.go +++ b/controller/testhelper_test.go @@ -18,7 +18,7 @@ import ( "context" "github.com/uber-go/tally" - "github.com/uber/tango/core/common" + "github.com/uber/tango/config" "go.uber.org/zap" ) @@ -26,9 +26,9 @@ func newTestController(logger *zap.Logger) *controller { return &controller{ logger: logger, scope: tally.NoopScope, - targetChunkSize: common.DefaultTargetChunkSize, - changedTargetChunkSize: common.DefaultChangedTargetChunkSize, - metadataMapChunkSize: common.DefaultMetadataMapChunkSize, + targetChunkSize: config.DefaultTargetChunkSize, + changedTargetChunkSize: config.DefaultChangedTargetChunkSize, + metadataMapChunkSize: config.DefaultMetadataMapChunkSize, totalDurationBuckets: _totalDurationBuckets, appCtx: context.Background(), } diff --git a/core/bazel/BUILD.bazel b/core/bazel/BUILD.bazel index 3fa695b..8772704 100644 --- a/core/bazel/BUILD.bazel +++ b/core/bazel/BUILD.bazel @@ -11,6 +11,7 @@ go_library( importpath = "github.com/uber/tango/core/bazel", visibility = ["//visibility:public"], deps = [ + "//core/errors", "//core/execcmd", "@com_github_bazelbuild_buildtools//build_proto", "@org_golang_google_protobuf//encoding/protodelim", diff --git a/core/bazel/stream.go b/core/bazel/stream.go index 2e78fd7..4da83cd 100644 --- a/core/bazel/stream.go +++ b/core/bazel/stream.go @@ -21,6 +21,8 @@ import ( buildpb "github.com/bazelbuild/buildtools/build_proto" "google.golang.org/protobuf/encoding/protodelim" + + tgerrors "github.com/uber/tango/core/errors" ) func streamOutput(ctx context.Context, src io.Reader, dst io.Writer) error { @@ -58,11 +60,6 @@ func streamAndParseTargets(ctx context.Context, src io.Reader, dst io.Writer) (* } } -// cancelCheckInterval is how often we poll ctx.Err() inside per-target hot loops. -// Picked to keep overhead negligible while still surfacing cancellation in <100ms -// for typical target rates. -const cancelCheckInterval = 1024 - // getQueryResult reads a QueryResult containing targets from the stream and returns it. func getQueryResult(ctx context.Context, src io.Reader, dst io.Writer) (*buildpb.QueryResult, error) { result := &buildpb.QueryResult{ @@ -75,7 +72,7 @@ func getQueryResult(ctx context.Context, src io.Reader, dst io.Writer) (*buildpb } var parseErr error for i := 0; ; i++ { - if i%cancelCheckInterval == 0 { + if i%tgerrors.CancelCheckInterval == 0 { if err := ctx.Err(); err != nil { return result, err } diff --git a/core/common/BUILD.bazel b/core/common/BUILD.bazel index 0daeb3c..35fc9a2 100644 --- a/core/common/BUILD.bazel +++ b/core/common/BUILD.bazel @@ -10,6 +10,8 @@ go_library( importpath = "github.com/uber/tango/core/common", visibility = ["//visibility:public"], deps = [ + "//config", + "//core/errors", "//core/targethasher", "//tangopb", "@com_github_bazelbuild_buildtools//build_proto", @@ -24,6 +26,7 @@ go_test( ], embed = [":common"], deps = [ + "//config", "//core/targethasher", "//tangopb", "@com_github_stretchr_testify//assert", diff --git a/core/common/utils.go b/core/common/utils.go index 90df802..3edcc04 100644 --- a/core/common/utils.go +++ b/core/common/utils.go @@ -24,28 +24,12 @@ import ( "strings" buildpb "github.com/bazelbuild/buildtools/build_proto" + "github.com/uber/tango/config" + tgerrors "github.com/uber/tango/core/errors" "github.com/uber/tango/core/targethasher" "github.com/uber/tango/tangopb" ) -const ( - // DefaultTargetChunkSize is the default number of OptimizedTarget entries per stream message. - // Sized conservatively: at ~40KB/target worst-case (target with ~10K direct deps × 4 bytes), - // 250 targets ≈ 10MB — well under the 64MB default gRPC per-message limit. - DefaultTargetChunkSize = 250 - - // DefaultChangedTargetChunkSize is the default number of ChangedTarget entries per stream message. - // A ChangedTarget carries both old_target and new_target (2× an OptimizedTarget), so we use - // half the regular chunk size to stay within the same byte budget. - DefaultChangedTargetChunkSize = 125 - - // DefaultMetadataMapChunkSize is the max entries per metadata message chunk. - // target_id_mapping and attribute_string_value_mapping scale with repo size and can exceed - // the 64MB gRPC message limit for large monorepos, so they are split across multiple messages. - // At ~85 bytes/entry (60-char avg target name + proto overhead), 50 000 entries ≈ 4.25MB per chunk. - DefaultMetadataMapChunkSize = 50_000 -) - // ToShortRemote returns the short remote name given a git ssh remote string. // For example, "git@github:uber/tango" will return "uber/tango". func ToShortRemote(remote string) string { @@ -132,13 +116,8 @@ func HashRequestOptions(opts *tangopb.RequestOptions) string { return fmt.Sprintf("%x", h.Sum(nil)) } -// cancelCheckInterval is how often we poll ctx.Err() inside per-target hot loops. -// Picked to keep overhead negligible while still surfacing cancellation in <100ms -// for typical target rates. -const cancelCheckInterval = 4096 - // ResultToGetTargetGraphResponse converts a Result to a GetTargetGraphResponse -func ResultToGetTargetGraphResponse(ctx context.Context, result targethasher.Result) ([]*tangopb.GetTargetGraphResponse, error) { +func ResultToGetTargetGraphResponse(ctx context.Context, result targethasher.Result, targetChunkSize, metadataMapChunkSize int) ([]*tangopb.GetTargetGraphResponse, error) { // Map target names to ids. This list is topologically sorted, so the ids are stable. // IDs start at 1 — 0 is reserved as the proto3 "unset" sentinel so consumers using // encoding/json (which honors `omitempty` on int32 fields) never silently lose a target. @@ -157,7 +136,7 @@ func ResultToGetTargetGraphResponse(ctx context.Context, result targethasher.Res n := 0 for _, t := range result.Targets { - if n%cancelCheckInterval == 0 { + if n%tgerrors.CancelCheckInterval == 0 { if err := ctx.Err(); err != nil { return nil, err } @@ -224,14 +203,14 @@ func ResultToGetTargetGraphResponse(ctx context.Context, result targethasher.Res attrStrValIDToVal := attrStrValMapper.Invert() // chunk targets into multiple messages for streaming - responses := chunkTargets(optimizedTargets, DefaultTargetChunkSize) + responses := chunkTargets(optimizedTargets, targetChunkSize) for _, meta := range ChunkMetadata( targetIDToName, ruleTypeIDToName, tagIDToName, attrNameIDToName, attrStrValIDToVal, - DefaultMetadataMapChunkSize, + metadataMapChunkSize, ) { responses = append(responses, &tangopb.GetTargetGraphResponse{ Item: &tangopb.GetTargetGraphResponse_Metadata{Metadata: meta}, @@ -243,7 +222,7 @@ func ResultToGetTargetGraphResponse(ctx context.Context, result targethasher.Res func chunkTargets(targets []*tangopb.OptimizedTarget, chunkSize int) []*tangopb.GetTargetGraphResponse { if chunkSize <= 0 { - chunkSize = DefaultTargetChunkSize + chunkSize = config.DefaultTargetChunkSize } // at least one chunk @@ -294,7 +273,7 @@ func ChunkMetadata( chunkSize int, ) []*tangopb.Metadata { if chunkSize <= 0 { - chunkSize = DefaultMetadataMapChunkSize + chunkSize = config.DefaultMetadataMapChunkSize } targetChunks := splitMap(targetIDToName, chunkSize) diff --git a/core/common/utils_test.go b/core/common/utils_test.go index 4172da0..6ff5aa1 100644 --- a/core/common/utils_test.go +++ b/core/common/utils_test.go @@ -23,6 +23,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/uber/tango/config" "github.com/uber/tango/core/targethasher" pb "github.com/uber/tango/tangopb" ) @@ -198,7 +199,7 @@ func TestResultToGetTargetGraphResponse_Chunking(t *testing.T) { result.Targets[name] = &targethasher.Target{Name: name, Hash: []byte{0}, RuleType: "go_library"} } - responses, err := ResultToGetTargetGraphResponse(context.Background(), result) + responses, err := ResultToGetTargetGraphResponse(context.Background(), result, config.DefaultTargetChunkSize, config.DefaultMetadataMapChunkSize) require.NoError(t, err) // 2 target chunks + 1 metadata chunk (500 targets well under DefaultMetadataMapChunkSize) diff --git a/core/errors/BUILD.bazel b/core/errors/BUILD.bazel new file mode 100644 index 0000000..71c7646 --- /dev/null +++ b/core/errors/BUILD.bazel @@ -0,0 +1,8 @@ +load("@rules_go//go:def.bzl", "go_library") + +go_library( + name = "errors", + srcs = ["constants.go"], + importpath = "github.com/uber/tango/core/errors", + visibility = ["//visibility:public"], +) diff --git a/core/errors/constants.go b/core/errors/constants.go new file mode 100644 index 0000000..4483969 --- /dev/null +++ b/core/errors/constants.go @@ -0,0 +1,8 @@ +package errors + +const ( + // CancelCheckInterval is how often we poll ctx.Err() inside per-target hot loops. + // Picked to keep overhead negligible while still surfacing cancellation in <100ms + // for typical target rates. + CancelCheckInterval = 1024 +) diff --git a/example/cmd/query-bench/BUILD.bazel b/example/cmd/query-bench/BUILD.bazel index ba71ce1..6c798b8 100644 --- a/example/cmd/query-bench/BUILD.bazel +++ b/example/cmd/query-bench/BUILD.bazel @@ -6,6 +6,7 @@ go_library( importpath = "github.com/uber/tango/example/cmd/query-bench", visibility = ["//visibility:private"], deps = [ + "//config", "//core/bazel", "//core/common", "//core/targethasher", diff --git a/example/cmd/query-bench/main.go b/example/cmd/query-bench/main.go index 3071b01..a766386 100644 --- a/example/cmd/query-bench/main.go +++ b/example/cmd/query-bench/main.go @@ -31,6 +31,7 @@ import ( "time" "github.com/gogo/protobuf/jsonpb" + "github.com/uber/tango/config" "github.com/uber/tango/core/bazel" "github.com/uber/tango/core/common" "github.com/uber/tango/core/targethasher" @@ -112,7 +113,7 @@ func run() error { totalDuration += elapsed fmt.Printf("run %d: targethasher: %v (%d targets)\n", i+1, elapsed.Round(time.Millisecond), len(targethasherResult.TargetNames)) start = time.Now() - response, err := common.ResultToGetTargetGraphResponse(ctx, targethasherResult) // also print the time to convert to GetTargetGraphResponse format + response, err := common.ResultToGetTargetGraphResponse(ctx, targethasherResult, config.DefaultTargetChunkSize, config.DefaultMetadataMapChunkSize) if err != nil { return fmt.Errorf("run %d: converting to GetTargetGraphResponse: %w", i+1, err) } diff --git a/orchestrator/native_orchestrator.go b/orchestrator/native_orchestrator.go index 4558cff..4c45987 100644 --- a/orchestrator/native_orchestrator.go +++ b/orchestrator/native_orchestrator.go @@ -208,7 +208,8 @@ func (b *nativeOrchestrator) GetTargetGraph(ctx context.Context, param GetTarget if err != nil { return nil, fmt.Errorf("compute target graph: %w", err) } - responses, err := common.ResultToGetTargetGraphResponse(ctx, result) + + responses, err := common.ResultToGetTargetGraphResponse(ctx, result, b.config.Service.Chunking.TargetChunkSize, b.config.Service.Chunking.MetadataMapChunkSize) if err != nil { return nil, fmt.Errorf("convert target graph to response: %w", err) }