diff --git a/go.mod b/go.mod index 80d8e7aee..a283db244 100644 --- a/go.mod +++ b/go.mod @@ -43,6 +43,7 @@ require ( github.com/rogpeppe/go-internal v1.14.1 // indirect github.com/spf13/pflag v1.0.9 // indirect github.com/twmb/murmur3 v1.1.8 // indirect + github.com/uber-go/mapdecode v1.0.0 // indirect github.com/uber/tchannel-go v1.34.4 // indirect go.uber.org/atomic v1.11.0 // indirect go.uber.org/dig v1.19.0 // indirect @@ -57,5 +58,6 @@ require ( golang.org/x/text v0.37.0 // indirect golang.org/x/tools v0.44.0 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect + gopkg.in/yaml.v2 v2.4.0 // indirect honnef.co/go/tools v0.4.3 // indirect ) diff --git a/service/submitqueue/orchestrator/server/BUILD.bazel b/service/submitqueue/orchestrator/server/BUILD.bazel index f6a1cc452..6e186511c 100644 --- a/service/submitqueue/orchestrator/server/BUILD.bazel +++ b/service/submitqueue/orchestrator/server/BUILD.bazel @@ -30,6 +30,8 @@ go_library( "//platform/extension/consumergate/noop:go_default_library", "//platform/extension/counter:go_default_library", "//platform/extension/counter/mysql:go_default_library", + "//platform/extension/gitworkspace:go_default_library", + "//platform/extension/gitworkspace/local:go_default_library", "//platform/extension/hook:go_default_library", "//platform/extension/hook/noop:go_default_library", "//platform/extension/messagequeue/mysql:go_default_library", @@ -56,6 +58,10 @@ go_library( "//submitqueue/extension/conflict/fake:go_default_library", "//submitqueue/extension/conflict/none:go_default_library", "//submitqueue/extension/conflict/pathoverlap:go_default_library", + "//submitqueue/extension/conflict/tango:go_default_library", + "//submitqueue/extension/conflict/tango/baserevision:go_default_library", + "//submitqueue/extension/conflict/tango/mysql:go_default_library", + "//submitqueue/extension/conflict/tango/tangoclient:go_default_library", "//submitqueue/extension/queuepause:go_default_library", "//submitqueue/extension/speculation/allocator/sticky:go_default_library", "//submitqueue/extension/speculation/generator/bestfirst:go_default_library", @@ -142,11 +148,14 @@ go_test( "//submitqueue/extension/buildrunner:go_default_library", "//submitqueue/extension/changeprovider:go_default_library", "//submitqueue/extension/conflict:go_default_library", + "//submitqueue/extension/conflict/tango:go_default_library", + "//submitqueue/extension/conflict/tango/tangoclient:go_default_library", "//submitqueue/extension/queuepause/fake:go_default_library", "//submitqueue/extension/speculation/scorer:go_default_library", "//submitqueue/extension/speculation/speculator:go_default_library", "//submitqueue/extension/validator:go_default_library", "//submitqueue/orchestrator/extension/storage:go_default_library", + "@com_github_data_dog_go_sqlmock//:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", "@com_github_uber_go_tally//:go_default_library", diff --git a/service/submitqueue/orchestrator/server/config.go b/service/submitqueue/orchestrator/server/config.go index 5b95832c6..30e0aef64 100644 --- a/service/submitqueue/orchestrator/server/config.go +++ b/service/submitqueue/orchestrator/server/config.go @@ -22,6 +22,8 @@ import ( "time" yamlv3 "gopkg.in/yaml.v3" + + "github.com/uber/submitqueue/submitqueue/extension/conflict/tango/tangoclient" ) // Change provider types selectable from configuration. @@ -51,6 +53,7 @@ const ( analyzerTypeAll = "all" analyzerTypeNone = "none" analyzerTypePathOverlap = "pathoverlap" + analyzerTypeTango = "tango" ) // Granularities a pathoverlap analyzer compares paths at. @@ -117,12 +120,26 @@ const ( // closed set in the schema, so supporting a new provider is a new value here and // a new implementation behind it — not a change to the shape of this file. type profilesConfig struct { + // Tango is the process-level setup shared by every queue that uses the + // tango analyzer. It is required only when one does. + Tango tangoConfig `yaml:"tango"` // Defaults applies to any queue without an override for a given extension. Defaults queueProfileConfig `yaml:"defaults"` // Queues holds per-queue overrides. Queues []namedQueueProfileConfig `yaml:"queues"` } +// tangoConfig is how this process reaches Tango and how much memory the +// signature cache may use. +type tangoConfig struct { + // Address is the host:port of the Tango service's gRPC inbound. + Address string `yaml:"address"` + // SignatureCacheBytes bounds the in-memory signature cache shared by every + // queue. It has no default, because a zero budget would silently cache + // nothing. + SignatureCacheBytes int64 `yaml:"signatureCacheBytes"` +} + // namedQueueProfileConfig is one queue's overrides. Each extension is // independently optional: an absent block inherits that one extension from the // defaults, so a queue that differs only in its analyzer says only that. @@ -244,6 +261,18 @@ type analyzerConfig struct { // FailAlways makes every analysis return an error, for exercising the // analyzer failure path. Test-only. FailAlways bool `yaml:"failAlways"` + // Tango names the repository Tango analyzes for this queue (tango only). + Tango *tangoAnalyzerConfig `yaml:"tango"` +} + +// tangoAnalyzerConfig is what Tango needs to know about one queue's repository. +type tangoAnalyzerConfig struct { + // Remote identifies the repository to Tango. It must equal, exactly, a + // repository remote in Tango's own configuration. + Remote string `yaml:"remote"` + // Strategy is how Tango computes the base target graph: "unset", "shell" + // or "native". Defaults to "unset". + Strategy string `yaml:"strategy"` } // scorerConfig selects how a queue ranks candidate speculation paths. The @@ -352,7 +381,62 @@ func (c *profilesConfig) normalizeAndValidate() error { } } } - return c.validateGitRepoPaths() + if err := c.validateGitRepoPaths(); err != nil { + return err + } + return c.validateTango() +} + +type namedProfile struct { + where string + profile queueProfileConfig +} + +// namedProfiles lists the defaults followed by every queue's resolved profile, +// in config order. +func (c profilesConfig) namedProfiles() []namedProfile { + profiles := []namedProfile{{where: "defaults", profile: c.Defaults}} + for _, q := range c.Queues { + profiles = append(profiles, namedProfile{where: fmt.Sprintf("queue %q", q.Name), profile: c.resolve(q)}) + } + return profiles +} + +// usesTango reports whether any profile selects the tango analyzer. +func (c profilesConfig) usesTango() bool { + for _, p := range c.namedProfiles() { + if p.profile.Analyzer.Type == analyzerTypeTango { + return true + } + } + return false +} + +// validateTango rejects a tango analyzer on a queue that has no repository +// clone to read the base revision from, and a tango analyzer without the +// process-level setup it needs. The defaults count as a queue, since queues +// without an override inherit them. +func (c profilesConfig) validateTango() error { + used := false + for _, p := range c.namedProfiles() { + if p.profile.Analyzer.Type != analyzerTypeTango { + continue + } + used = true + if p.profile.ChangeProvider.Type != changeProviderTypeGit { + return fmt.Errorf("%s: the tango analyzer needs the git change provider, got %q", p.where, p.profile.ChangeProvider.Type) + } + } + if !used { + return nil + } + if c.Tango.Address == "" { + return fmt.Errorf("a queue uses the tango analyzer, so tango.address is required") + } + if c.Tango.SignatureCacheBytes <= 0 { + return fmt.Errorf("a queue uses the tango analyzer, so tango.signatureCacheBytes must be positive") + } + return nil } // validateGitRepoPaths rejects two queues keeping their copies in one directory @@ -614,6 +698,17 @@ func (a *analyzerConfig) normalizeAndValidate(where string) error { return fmt.Errorf("%s: unknown pathoverlap granularity %q", where, a.By) } return nil + case analyzerTypeTango: + if a.Tango == nil || a.Tango.Remote == "" { + return fmt.Errorf("%s: tango analyzer needs a tango block with a remote", where) + } + if a.Tango.Strategy == "" { + a.Tango.Strategy = "unset" + } + if !tangoclient.ValidStrategy(a.Tango.Strategy) { + return fmt.Errorf("%s: unknown tango strategy %q", where, a.Tango.Strategy) + } + return nil default: return fmt.Errorf("%s: unknown analyzer type %q", where, a.Type) } diff --git a/service/submitqueue/orchestrator/server/config_test.go b/service/submitqueue/orchestrator/server/config_test.go index 64ca8b487..b84846df7 100644 --- a/service/submitqueue/orchestrator/server/config_test.go +++ b/service/submitqueue/orchestrator/server/config_test.go @@ -24,9 +24,11 @@ import ( "strings" "testing" + "github.com/DATA-DOG/go-sqlmock" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/uber-go/tally" + gitrepo "github.com/uber/submitqueue/platform/git/repo" "go.uber.org/zap/zaptest" "github.com/uber/submitqueue/platform/base/change" @@ -34,6 +36,8 @@ import ( "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/changeprovider" "github.com/uber/submitqueue/submitqueue/extension/conflict" + "github.com/uber/submitqueue/submitqueue/extension/conflict/tango" + "github.com/uber/submitqueue/submitqueue/extension/conflict/tango/tangoclient" "github.com/uber/submitqueue/submitqueue/extension/speculation/speculator" "github.com/uber/submitqueue/submitqueue/extension/validator" ) @@ -291,7 +295,7 @@ func TestNewProfiles_ResolvesPerQueueAnalyzers(t *testing.T) { // Asserted on behavior rather than instance identity: the analyzers are // stateless value types, so what matters is that each queue got the type it // asked for. - profiles, err := newProfiles(context.Background(), zaptest.NewLogger(t), tally.NoopScope, nil, nil, defaultProfilesConfig()) + profiles, err := newProfiles(context.Background(), zaptest.NewLogger(t), tally.NoopScope, nil, nil, tangoDeps{}, defaultProfilesConfig()) require.NoError(t, err) tests := []struct { @@ -343,7 +347,7 @@ func TestNewProfiles_FailsWhenAForgeTokenIsMissing(t *testing.T) { cfg, err := loadProfilesConfig(writeProfiles(t, tt.contents)) require.NoError(t, err) - _, err = newProfiles(context.Background(), zaptest.NewLogger(t), tally.NoopScope, nil, nil, cfg) + _, err = newProfiles(context.Background(), zaptest.NewLogger(t), tally.NoopScope, nil, nil, tangoDeps{}, cfg) require.Error(t, err) }) } @@ -360,7 +364,7 @@ queues: `) cfg, err := loadProfilesConfig(path) require.NoError(t, err) - profiles, err := newProfiles(context.Background(), zaptest.NewLogger(t), tally.NoopScope, nil, nil, cfg) + profiles, err := newProfiles(context.Background(), zaptest.NewLogger(t), tally.NoopScope, nil, nil, tangoDeps{}, cfg) require.NoError(t, err) foreignHost := entity.Request{Change: change.Change{URIs: []string{ @@ -402,7 +406,7 @@ defaults: cfg, err := loadProfilesConfig(path) require.NoError(t, err) - profiles, err := newProfiles(context.Background(), zaptest.NewLogger(t), tally.NoopScope, nil, nil, cfg) + profiles, err := newProfiles(context.Background(), zaptest.NewLogger(t), tally.NoopScope, nil, nil, tangoDeps{}, cfg) require.NoError(t, err) assert.NotNil(t, profiles.For("anything").ChangeProvider) } @@ -448,7 +452,7 @@ func keysOf(m map[string]scorerConfig) []string { func TestNewProfiles_ComposesASpeculatorPerQueue(t *testing.T) { // Speculation is only "on" if every queue resolves to a real speculator — // a nil one leaves the stage inert and nothing ever gets built. - profiles, err := newProfiles(context.Background(), zaptest.NewLogger(t), tally.NoopScope, nil, nil, defaultProfilesConfig()) + profiles, err := newProfiles(context.Background(), zaptest.NewLogger(t), tally.NoopScope, nil, nil, tangoDeps{}, defaultProfilesConfig()) require.NoError(t, err) for _, queue := range []string{"test-queue", "e2e-test-queue", "file-overlap-queue", "unlisted"} { @@ -478,7 +482,7 @@ queues: `) cfg, err := loadProfilesConfig(path) require.NoError(t, err) - profiles, err := newProfiles(context.Background(), zaptest.NewLogger(t), tally.NoopScope, nil, nil, cfg) + profiles, err := newProfiles(context.Background(), zaptest.NewLogger(t), tally.NoopScope, nil, nil, tangoDeps{}, cfg) require.NoError(t, err) batches := make([]entity.Batch, 0, 8) @@ -508,6 +512,95 @@ queues: } } +func TestLoadProfilesConfig_Tango(t *testing.T) { + const git = ` + changeProvider: + type: git + git: + remoteUrl: file:///srv/git/x.git + target: main + repoPath: /var/repos/x +` + tests := []struct { + name string + contents string + wantErr bool + }{ + { + name: "accepted with a git change provider and a connection", + contents: ` +tango: {address: "127.0.0.1:8081", signatureCacheBytes: 1048576} +defaults:` + git + ` analyzer: {type: tango, tango: {remote: origin}} +`, + }, + { + name: "rejected without an analyzer block", + contents: ` +tango: {address: "127.0.0.1:8081", signatureCacheBytes: 1048576} +defaults:` + git + ` analyzer: {type: tango} +`, + wantErr: true, + }, + { + name: "rejected with an unknown strategy", + contents: ` +tango: {address: "127.0.0.1:8081", signatureCacheBytes: 1048576} +defaults:` + git + ` analyzer: {type: tango, tango: {remote: origin, strategy: magic}} +`, + wantErr: true, + }, + { + name: "rejected without a connection address", + contents: ` +tango: {signatureCacheBytes: 1048576} +defaults:` + git + ` analyzer: {type: tango, tango: {remote: origin}} +`, + wantErr: true, + }, + { + name: "rejected without a cache size", + contents: ` +tango: {address: "127.0.0.1:8081"} +defaults:` + git + ` analyzer: {type: tango, tango: {remote: origin}} +`, + wantErr: true, + }, + { + name: "rejected on a queue that does not use the git change provider", + contents: ` +tango: {address: "127.0.0.1:8081", signatureCacheBytes: 1048576} +defaults:` + git + ` analyzer: {type: none} +queues: + - name: a + changeProvider: {type: fake} + analyzer: {type: tango, tango: {remote: origin}} +`, + wantErr: true, + }, + { + name: "connection not needed when no queue uses tango", + contents: ` +defaults: + analyzer: {type: none} +`, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg, err := loadProfilesConfig(writeProfiles(t, tt.contents)) + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + if cfg.Defaults.Analyzer.Type == analyzerTypeTango { + assert.Equal(t, "unset", cfg.Defaults.Analyzer.Tango.Strategy) + } + }) + } +} + func TestLoadProfilesConfig_GitChangeProvider(t *testing.T) { t.Run("defaults what it can and keeps what is given", func(t *testing.T) { path := writeProfiles(t, ` @@ -659,7 +752,7 @@ defaults: cfg, err := loadProfilesConfig(path) require.NoError(t, err) - profiles, err := newProfiles(context.Background(), zaptest.NewLogger(t), tally.NoopScope, nil, nil, cfg) + profiles, err := newProfiles(context.Background(), zaptest.NewLogger(t), tally.NoopScope, nil, nil, tangoDeps{}, cfg) require.NoError(t, err) provider, err := profiles.ChangeProviderFactory().For(changeprovider.Config{QueueName: "q"}) @@ -695,7 +788,7 @@ defaults: cfg, err := loadProfilesConfig(path) require.NoError(t, err) - _, err = newProfiles(context.Background(), zaptest.NewLogger(t), tally.NoopScope, nil, nil, cfg) + _, err = newProfiles(context.Background(), zaptest.NewLogger(t), tally.NoopScope, nil, nil, tangoDeps{}, cfg) require.Error(t, err) } @@ -840,3 +933,100 @@ func TestLoadProfilesConfig_QueueScorerBaseReplacesDefaultBase(t *testing.T) { require.NotNil(t, cfg.Defaults.Scorer.Base) assert.Equal(t, 0.4, cfg.Defaults.Scorer.Base.Buckets[0].Score) } + +func TestNewTangoAnalyzerFactory(t *testing.T) { + key := tangoAnalyzerKey{ + Analyzer: analyzerConfig{Type: analyzerTypeTango, Tango: &tangoAnalyzerConfig{Remote: "origin", Strategy: "unset"}}, + Git: gitProviderConfig{RepoPath: "/var/repos/x", Remote: "origin", Target: "main"}, + } + connection, err := tangoclient.Dial("127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { _ = connection.Close() }) + db, _, err := sqlmock.New() + require.NoError(t, err) + t.Cleanup(func() { _ = db.Close() }) + + tests := []struct { + name string + deps tangoDeps + wantErr bool + }{ + {name: "without process-level pieces", wantErr: true}, + { + name: "with process-level pieces", + deps: tangoDeps{DB: db, Connection: connection, Cache: tango.NewSignatureCache(1 << 20)}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + repo, err := gitrepo.NewRepo(gitrepo.RepoConfig{Path: key.Git.RepoPath, RemoteURL: "file:///srv/git/x.git", Remote: "origin", Target: "main"}) + require.NoError(t, err) + b := &profileBuilder{scope: tally.NoopScope, tango: tt.deps, repos: map[string]*gitrepo.Repo{key.Git.RepoPath: repo}} + factory, err := b.newTangoAnalyzerFactory(key, "defaults") + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + analyzer, err := factory.For(conflict.Config{QueueName: "q"}) + require.NoError(t, err) + assert.NotNil(t, analyzer) + }) + } +} + +func TestNewProfiles_TangoQueuesShareTheirCloneWithTheGitProvider(t *testing.T) { + git := gitexectest.Git(t) + root := t.TempDir() + remote := filepath.Join(root, "remote.git") + work := filepath.Join(root, "work") + + run := func(dir string, args ...string) { + t.Helper() + cmd := exec.Command(git, args...) + cmd.Dir = dir + out, err := cmd.CombinedOutput() + require.NoError(t, err, "git %v: %s", args, out) + } + run("", "init", "--bare", "-b", "main", remote) + run("", "clone", remote, work) + run(work, "config", "user.name", "Test") + run(work, "config", "user.email", "test@example.invalid") + require.NoError(t, os.WriteFile(filepath.Join(work, "seed.txt"), []byte("seed\n"), 0o644)) + run(work, "add", ".") + run(work, "commit", "--no-gpg-sign", "-m", "seed") + run(work, "push", "origin", "main") + + cfg, err := loadProfilesConfig(writeProfiles(t, fmt.Sprintf(` +tango: {address: "127.0.0.1:0", signatureCacheBytes: 1048576} +defaults: + changeProvider: + type: git + git: + remoteUrl: %s + target: main + repoPath: %s + analyzer: {type: tango, tango: {remote: origin}} +queues: + - name: a + - name: b +`, remote, filepath.Join(root, "copy.git")))) + require.NoError(t, err) + + connection, err := tangoclient.Dial("127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { _ = connection.Close() }) + db, _, err := sqlmock.New() + require.NoError(t, err) + t.Cleanup(func() { _ = db.Close() }) + deps := tangoDeps{DB: db, Connection: connection, Cache: tango.NewSignatureCache(1 << 20)} + + profiles, err := newProfiles(context.Background(), zaptest.NewLogger(t), tally.NoopScope, nil, nil, deps, cfg) + require.NoError(t, err) + + for _, queue := range []string{"a", "b"} { + analyzer, err := profiles.For(queue).Analyzer.For(conflict.Config{QueueName: queue}) + require.NoError(t, err) + assert.NotNil(t, analyzer) + } +} diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go index 3d3df9ebe..86474a0bf 100644 --- a/service/submitqueue/orchestrator/server/main.go +++ b/service/submitqueue/orchestrator/server/main.go @@ -50,6 +50,8 @@ import ( servicemq "github.com/uber/submitqueue/service/messagequeue" "github.com/uber/submitqueue/service/submitqueue/core" "github.com/uber/submitqueue/submitqueue/core/changeset" + "github.com/uber/submitqueue/submitqueue/extension/conflict/tango" + "github.com/uber/submitqueue/submitqueue/extension/conflict/tango/tangoclient" "github.com/uber/submitqueue/submitqueue/orchestrator" mysqlstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage/mysql" "go.uber.org/zap" @@ -199,9 +201,23 @@ func run() error { subscriberName = fmt.Sprintf("orchestrator-%d", time.Now().Unix()) } + var tangoShared tangoDeps + if profilesCfg.usesTango() { + connection, err := tangoclient.Dial(profilesCfg.Tango.Address) + if err != nil { + return fmt.Errorf("failed to connect to tango: %w", err) + } + defer connection.Close() + tangoShared = tangoDeps{ + DB: appDB, + Connection: connection, + Cache: tango.NewSignatureCache(profilesCfg.Tango.SignatureCacheBytes), + } + } + profiles, err := newProfiles(ctx, logger, scope, changeset.New(func(queue string) (changeset.Stores, error) { return storageFty.For(orchstorage.Config{QueueName: queue}) - }), storageFty, profilesCfg) + }), storageFty, tangoShared, profilesCfg) if err != nil { return fmt.Errorf("failed to build profiles: %w", err) } diff --git a/service/submitqueue/orchestrator/server/profiles.go b/service/submitqueue/orchestrator/server/profiles.go index 24cfd4903..184c73225 100644 --- a/service/submitqueue/orchestrator/server/profiles.go +++ b/service/submitqueue/orchestrator/server/profiles.go @@ -15,6 +15,7 @@ package main import ( + "database/sql" "fmt" nethttp "net/http" @@ -28,6 +29,9 @@ import ( "context" platformbuildkite "github.com/uber/submitqueue/platform/buildkite" + "github.com/uber/submitqueue/platform/extension/gitworkspace" + gitworkspacelocal "github.com/uber/submitqueue/platform/extension/gitworkspace/local" + gitrepo "github.com/uber/submitqueue/platform/git/repo" platformgithubactions "github.com/uber/submitqueue/platform/githubactions" "github.com/uber/submitqueue/platform/http" "github.com/uber/submitqueue/submitqueue/core/changeset" @@ -47,6 +51,10 @@ import ( conflictfake "github.com/uber/submitqueue/submitqueue/extension/conflict/fake" "github.com/uber/submitqueue/submitqueue/extension/conflict/none" "github.com/uber/submitqueue/submitqueue/extension/conflict/pathoverlap" + "github.com/uber/submitqueue/submitqueue/extension/conflict/tango" + "github.com/uber/submitqueue/submitqueue/extension/conflict/tango/baserevision" + tangomysql "github.com/uber/submitqueue/submitqueue/extension/conflict/tango/mysql" + "github.com/uber/submitqueue/submitqueue/extension/conflict/tango/tangoclient" "github.com/uber/submitqueue/submitqueue/extension/speculation/allocator/sticky" "github.com/uber/submitqueue/submitqueue/extension/speculation/generator/bestfirst" "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer" @@ -213,6 +221,7 @@ func newProfiles( scope tally.Scope, resolver changeset.Resolver, stores orchstorage.Factory, + tangoDeps tangoDeps, cfg profilesConfig, ) (Profiles, error) { b := &profileBuilder{ @@ -221,7 +230,9 @@ func newProfiles( scope: scope, resolver: resolver, stores: stores, + tango: tangoDeps, built: make(map[string]any), + repos: make(map[string]*gitrepo.Repo), } defaultProfile, err := b.build(cfg.Defaults, "defaults") @@ -267,7 +278,19 @@ type profileBuilder struct { scope tally.Scope resolver changeset.Resolver stores orchstorage.Factory + tango tangoDeps built map[string]any + // repos are the provisioned clones by directory, so the tango analyzer + // locks the same Repo as the git change provider that owns the clone. + repos map[string]*gitrepo.Repo +} + +// tangoDeps are the process-level pieces the tango analyzer shares across +// queues. They are zero when no queue uses it. +type tangoDeps struct { + DB *sql.DB + Connection *tangoclient.Connection + Cache *tango.SignatureCache } func (b *profileBuilder) build(cfg queueProfileConfig, where string) (Profile, error) { @@ -283,7 +306,12 @@ func (b *profileBuilder) build(cfg queueProfileConfig, where string) (Profile, e if err != nil { return Profile{}, err } - analyzer, err := reuse(b, cfg.Analyzer, where, b.newAnalyzerFactory) + var analyzer conflict.Factory + if cfg.Analyzer.Type == analyzerTypeTango { + analyzer, err = reuse(b, tangoAnalyzerKey{Analyzer: cfg.Analyzer, Git: *cfg.ChangeProvider.Git}, where, b.newTangoAnalyzerFactory) + } else { + analyzer, err = reuse(b, cfg.Analyzer, where, b.newAnalyzerFactory) + } if err != nil { return Profile{}, err } @@ -442,6 +470,7 @@ func (b *profileBuilder) newChangeProviderFactory(cfg changeProviderConfig, wher if err != nil { return nil, fmt.Errorf("%s: %w", where, err) } + b.repos[cfg.Git.RepoPath] = repo return changeProviderFunc(func(c changeprovider.Config) (changeprovider.ChangeProvider, error) { return gitprovider.New(gitprovider.Params{ Config: c, @@ -705,6 +734,56 @@ func (b *profileBuilder) newAnalyzerFactory(cfg analyzerConfig, where string) (c }), nil } +// tangoAnalyzerKey keys the tango analyzer on both the analyzer block and the +// git block of its queue, because the base revision comes from that clone. +type tangoAnalyzerKey struct { + Analyzer analyzerConfig + Git gitProviderConfig +} + +// newTangoAnalyzerFactory builds a factory for tango analyzers that read the +// base revision from the queue's own clone. Validation guarantees the queue +// uses the git change provider and that the process-level pieces are set. +func (b *profileBuilder) newTangoAnalyzerFactory(key tangoAnalyzerKey, where string) (conflict.Factory, error) { + if b.tango.Connection == nil || b.tango.Cache == nil || b.tango.DB == nil { + return nil, fmt.Errorf("%s: the tango analyzer needs a tango connection, signature cache and database", where) + } + client, err := b.tango.Connection.ChangedTargetsClient(key.Analyzer.Tango.Remote, key.Analyzer.Tango.Strategy) + if err != nil { + return nil, fmt.Errorf("%s: %w", where, err) + } + + repo, ok := b.repos[key.Git.RepoPath] + if !ok { + return nil, fmt.Errorf("%s: no change repository provisioned at %q", where, key.Git.RepoPath) + } + workspaces := cloneWorkspaces{} + target := baserevision.Target{Repo: key.Git.RepoPath, Remote: key.Git.Remote, Branch: key.Git.Target} + + return analyzerFunc(func(c conflict.Config) (conflict.Analyzer, error) { + fromGit := baserevision.FromGit(workspaces, map[string]baserevision.Target{c.QueueName: target}) + return tango.New(tango.Params{ + Signatures: tangomysql.NewSignatureStore(b.tango.DB, b.scope.SubScope("tango"), c.QueueName), + Cache: b.tango.Cache, + Client: client, + BaseRevision: func(ctx context.Context, queue string) (string, error) { + repo.Lock() + defer repo.Unlock() + return fromGit(ctx, queue) + }, + Changes: b.resolver, + }), nil + }), nil +} + +// cloneWorkspaces serves each repo, named by its clone directory, as a local +// git workspace. +type cloneWorkspaces struct{} + +func (cloneWorkspaces) For(ctx context.Context, repo string) (gitworkspace.Workspace, error) { + return gitworkspacelocal.NewFactory(gitworkspacelocal.Params{Dir: repo}).For(ctx, repo) +} + // pathOverlapKey maps the configured granularity onto the projection the // analyzer keys on. The value is validated when the config is loaded, so an // unrecognized one here would be a programming error; it falls back to the diff --git a/submitqueue/extension/conflict/tango/README.md b/submitqueue/extension/conflict/tango/README.md index f235784be..d94cac1ba 100644 --- a/submitqueue/extension/conflict/tango/README.md +++ b/submitqueue/extension/conflict/tango/README.md @@ -21,3 +21,10 @@ Each signature is computed against trunk at its own analysis time, without the o ## Injected pieces The analyzer owns no connection. A deployment supplies the Tango client adapter, which collects the response and resolves each target's label, and a function that returns the commit SHA of the target branch tip a new batch is analyzed against. `baserevision.FromGit` builds that function over any `gitworkspace.Factory`. The changes of a batch come from the changeset resolver, the signature store is bound to one queue, and one signature cache can be shared by every queue's analyzer. The MySQL implementations of the signature and dictionary stores are in `mysql/`. + +## Reference orchestrator + +A queue selects this analyzer with `analyzer: {type: tango, tango: {remote, strategy}}`. `remote` must equal the repository remote Tango is configured with, and `strategy` is `unset`, `shell` or `native` (default `unset`). The top-level `tango` block sets `address`, the Tango gRPC endpoint, and `signatureCacheBytes`, the size of the one signature cache every queue shares. Two limits apply: + +- The queue must use the git change provider, because the base revision is read from its clone. GitHub-backed queues are not supported until a base revision can be read from the GitHub API. +- Tango's native strategy resolves changes only from canonical `github://` URIs, while the git change provider emits `git://` URIs, so `native` does not work on a git-provider queue. diff --git a/submitqueue/extension/conflict/tango/tangoclient/BUILD.bazel b/submitqueue/extension/conflict/tango/tangoclient/BUILD.bazel index 03d900b79..638f7720e 100644 --- a/submitqueue/extension/conflict/tango/tangoclient/BUILD.bazel +++ b/submitqueue/extension/conflict/tango/tangoclient/BUILD.bazel @@ -2,7 +2,10 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", - srcs = ["tangoclient.go"], + srcs = [ + "connection.go", + "tangoclient.go", + ], importpath = "github.com/uber/submitqueue/submitqueue/extension/conflict/tango/tangoclient", visibility = [ "//service/submitqueue/orchestrator:__subpackages__", @@ -12,12 +15,17 @@ go_library( deps = [ "//submitqueue/extension/conflict/tango:go_default_library", "@com_github_uber_tango//tangopb:go_default_library", + "@org_uber_go_yarpc//:go_default_library", + "@org_uber_go_yarpc//transport/grpc:go_default_library", ], ) go_test( name = "go_default_test", - srcs = ["tangoclient_test.go"], + srcs = [ + "connection_test.go", + "tangoclient_test.go", + ], embed = [":go_default_library"], deps = [ "//submitqueue/extension/conflict/tango:go_default_library", diff --git a/submitqueue/extension/conflict/tango/tangoclient/connection.go b/submitqueue/extension/conflict/tango/tangoclient/connection.go new file mode 100644 index 000000000..11772acfc --- /dev/null +++ b/submitqueue/extension/conflict/tango/tangoclient/connection.go @@ -0,0 +1,86 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package tangoclient + +import ( + "fmt" + + "github.com/uber/tango/tangopb" + "go.uber.org/yarpc" + yarpcgrpc "go.uber.org/yarpc/transport/grpc" + + "github.com/uber/submitqueue/submitqueue/extension/conflict/tango" +) + +// tangoServiceName is the name Tango's server registers under; a call is +// routed to the outbound with this name. +const tangoServiceName = "tango" + +var strategies = map[string]tangopb.ComputationStrategy{ + "unset": tangopb.COMPUTATION_STRATEGY_UNSET, + "shell": tangopb.COMPUTATION_STRATEGY_SHELL, + "native": tangopb.COMPUTATION_STRATEGY_NATIVE, +} + +// ValidStrategy reports whether name is a computation strategy this package can +// request: unset, shell or native. +func ValidStrategy(name string) bool { + _, ok := strategies[name] + return ok +} + +// Connection is a gRPC connection to one Tango service, shared by every +// adapter built from it. It is safe for concurrent use. +type Connection struct { + dispatcher *yarpc.Dispatcher + client tangopb.TangoYARPCClient +} + +// Dial opens a connection to the Tango service at address, a host:port. The +// connection is established on first use, so a service that is not up yet does +// not fail Dial. Close releases it. +func Dial(address string) (*Connection, error) { + dispatcher := yarpc.NewDispatcher(yarpc.Config{ + Name: "submitqueue", + Outbounds: yarpc.Outbounds{ + tangoServiceName: {Stream: yarpcgrpc.NewTransport().NewSingleOutbound(address)}, + }, + }) + if err := dispatcher.Start(); err != nil { + return nil, fmt.Errorf("failed to start the connection to Tango at %s: %w", address, err) + } + return &Connection{ + dispatcher: dispatcher, + client: tangopb.NewTangoYARPCClient(dispatcher.ClientConfig(tangoServiceName)), + }, nil +} + +// ChangedTargetsClient returns an adapter for the repository Tango knows as +// remote. An empty strategy means unset. +func (c *Connection) ChangedTargetsClient(remote, strategy string) (tango.ChangedTargetsClient, error) { + if strategy == "" { + strategy = "unset" + } + computation, ok := strategies[strategy] + if !ok { + return nil, fmt.Errorf("unknown Tango computation strategy %q", strategy) + } + return New(c.client, Config{Remote: remote, Strategy: computation}), nil +} + +// Close stops the connection. +func (c *Connection) Close() error { + return c.dispatcher.Stop() +} diff --git a/submitqueue/extension/conflict/tango/tangoclient/connection_test.go b/submitqueue/extension/conflict/tango/tangoclient/connection_test.go new file mode 100644 index 000000000..df246b3ff --- /dev/null +++ b/submitqueue/extension/conflict/tango/tangoclient/connection_test.go @@ -0,0 +1,70 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package tangoclient + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestValidStrategy(t *testing.T) { + tests := []struct { + name string + strategy string + want bool + }{ + {name: "unset", strategy: "unset", want: true}, + {name: "shell", strategy: "shell", want: true}, + {name: "native", strategy: "native", want: true}, + {name: "unknown", strategy: "fast"}, + {name: "empty", strategy: ""}, + {name: "wrong case", strategy: "Native"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, ValidStrategy(tt.strategy)) + }) + } +} + +func TestConnection_ChangedTargetsClient(t *testing.T) { + tests := []struct { + name string + strategy string + wantError bool + }{ + {name: "empty strategy means unset"}, + {name: "named strategy", strategy: "native"}, + {name: "unknown strategy fails", strategy: "fast", wantError: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + connection, err := Dial("127.0.0.1:0") + require.NoError(t, err) + defer func() { require.NoError(t, connection.Close()) }() + + client, err := connection.ChangedTargetsClient("org/repo", tt.strategy) + + if tt.wantError { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.NotNil(t, client) + }) + } +}