Skip to content

refactor(controller): GetChangedTargets takes the entity request and the repository #335

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
justinwon777 merged 1 commit into from
Oct 15, 2026
Merged

refactor(controller): GetChangedTargets takes the entity request and the repository #335

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion controller/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,6 @@ go_library(
"//internal/targetdiff",
"//internal/tgb",
"//internal/tgbdiff",
"//mapper/proto",
"//observability/metrics",
"//orchestrator",
"//tangopb",
Expand Down
7 changes: 3 additions & 4 deletions controller/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,10 +10,9 @@ response streaming.
The package is intentionally thin. It owns the cross-cutting concerns that
sit between the wire protocol and the rest of the system:

- **Request validation and translation.** GetChangedTargets validates its
inputs and normalizes them into the internal call shapes used downstream.
GetTargetGraph receives an entity request and the resolved repository
configuration from the handler, which maps and validates the request.
- **Repository input.** Each call receives an entity request and the
resolved repository configuration from the handler. The controller does
not map the request or look up the repository itself.
- **Read-through caching.** Where a request can be satisfied from previously
computed artifacts, the controller fetches them from storage and streams
them back without invoking the orchestrator. Cache misses fall through to
Expand Down
2 changes: 1 addition & 1 deletion controller/controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,7 @@ type controller struct {
// configured repository that the handler resolved for the request.
type Controller interface {
GetTargetGraph(request entity.GetTargetGraphRequest, outputConfig *pb.OutputConfig, stream pb.TangoServiceGetTargetGraphYARPCServer, repo config.RepositoryConfig) error
GetChangedTargets(request *pb.GetChangedTargetsRequest, stream pb.TangoServiceGetChangedTargetsYARPCServer) error
GetChangedTargets(request entity.GetChangedTargetsRequest, outputConfig *pb.OutputConfig, stream pb.TangoServiceGetChangedTargetsYARPCServer, repo config.RepositoryConfig) error
}

// NewController creates a new controller. appCtx is cancelled on process
Expand Down
8 changes: 4 additions & 4 deletions controller/controllermock/controllermock.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

26 changes: 8 additions & 18 deletions controller/getchangedtargets.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,6 @@ import (

"github.com/uber/tango/config"
"github.com/uber/tango/core/cachekey"
tangoerrors "github.com/uber/tango/core/errors"
"github.com/uber/tango/core/storage"
"github.com/uber/tango/entity"
"github.com/uber/tango/internal/mapper"
Expand All @@ -33,7 +32,6 @@ import (
"github.com/uber/tango/internal/targetdiff"
"github.com/uber/tango/internal/tgb"
"github.com/uber/tango/internal/tgbdiff"
"github.com/uber/tango/mapper/proto"
"github.com/uber/tango/observability/metrics"
pb "github.com/uber/tango/tangopb"
"go.uber.org/zap"
Expand Down Expand Up @@ -82,19 +80,11 @@ type job struct {
// GetChangedTargets returns the changed targets between two revisions. If the
// client disconnects, the stream's context is cancelled and the function
// returns with context.Canceled.
func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, stream pb.TangoServiceGetChangedTargetsYARPCServer) error {
entityReq, mappingErr := proto.ProtoToGetChangedTargetsRequest(request)
if mappingErr != nil {
mappingErr = tangoerrors.NewUser(fmt.Errorf("convert get changed targets request: %w", mappingErr))
}
repoCfg, repo, repositoryErr := c.resolveRequestRepository(entityReq.First.Remote, mappingErr)
e := c.emitter.Tagged(map[string]string{metrics.TagRepo: repo})
func (c *controller) GetChangedTargets(entityReq entity.GetChangedTargetsRequest, outputConfig *pb.OutputConfig, stream pb.TangoServiceGetChangedTargetsYARPCServer, repoCfg config.RepositoryConfig) error {
e := c.emitter.Tagged(map[string]string{metrics.TagRepo: repoCfg.RepositoryID})
logger := c.logger.WithLazy(
zap.String("repository", repo),
zap.String("repository", repoCfg.RepositoryID),
)
if repositoryErr != nil {
return repositoryErr
}
ctx, cancelLink := c.linkRequestCtx(stream.Context())
defer cancelLink()
start := time.Now()
Expand All @@ -105,13 +95,13 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str
// entirely. When OutputConfig is supplied, take max_distance at face value —
// see proto/tango.proto OutputConfig.max_distance for the wire-default caveat.
maxDist := int32(-1)
if request.GetOutputConfig() != nil {
maxDist = request.GetOutputConfig().GetMaxDistance()
if outputConfig != nil {
maxDist = outputConfig.GetMaxDistance()
}

// Fast path: stream a previously computed result straight from cache.
if !request.GetBypassCache() {
served, err := c.serveChangedTargetsFromCache(ctx, e, logger, entityReq, request.GetOutputConfig(), stream, repoCfg.RepositoryID, maxDist, start)
if !entityReq.BypassCache {
served, err := c.serveChangedTargetsFromCache(ctx, e, logger, entityReq, outputConfig, stream, repoCfg.RepositoryID, maxDist, start)
if err != nil {
return fmt.Errorf("serve from cache: %w", err)
}
Expand Down Expand Up @@ -141,7 +131,7 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str
c.cacheComparedTargets(logger, entityReq, repoCfg.RepositoryID, changedTargetsResponses)

sendStart := time.Now()
if err := sendTrimmedChangedTargets(stream, changedTargetsResponses, maxDist, request.GetOutputConfig()); err != nil {
if err := sendTrimmedChangedTargets(stream, changedTargetsResponses, maxDist, outputConfig); err != nil {
return fmt.Errorf("send response: %w", err)
}
sendDuration := time.Since(sendStart)
Expand Down
84 changes: 25 additions & 59 deletions controller/getchangedtargets_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -242,16 +242,6 @@ func TestCompareTargetGraphs_AllTargetsFileNoTrigger(t *testing.T) {
assert.Equal(t, 0, totalChanged, "no targets should be changed when AllTargetsFiles hashes match")
}

func TestGetChangedTargets_ValidationError(t *testing.T) {
ctrl := gomock.NewController(t)
stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl)

c := NewController(context.Background(), Params{RepoConfig: allowAnyRepositoryConfigProvider{}, Logger: zap.NewNop(), Orchestrator: orchestratormock.NewMockOrchestrator(ctrl)})

err := c.GetChangedTargets(nil, stream)
require.Error(t, err)
}

func TestGetChangedTargets_CacheHit(t *testing.T) {
ctrl := gomock.NewController(t)
stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl)
Expand Down Expand Up @@ -285,13 +275,10 @@ func TestGetChangedTargets_CacheHit(t *testing.T) {
Orchestrator: orchestratormock.NewMockOrchestrator(ctrl),
})

request := &pb.GetChangedTargetsRequest{
FirstRevision: &pb.BuildDescription{Strategy: pb.COMPUTATION_STRATEGY_UNSET, Remote: "repo:go-code", BaseSha: "sha1"},
SecondRevision: &pb.BuildDescription{Strategy: pb.COMPUTATION_STRATEGY_UNSET, Remote: "repo:go-code", BaseSha: "sha2"},
OutputConfig: &pb.OutputConfig{MaxDistance: -1},
}
request := changedTargetsRequest()
outputConfig := &pb.OutputConfig{MaxDistance: -1}

err := c.GetChangedTargets(request, stream)
err := c.GetChangedTargets(request, outputConfig, stream, config.RepositoryConfig{Remote: "repo:go-code", RepositoryID: testRepositoryID("repo:go-code")})
require.NoError(t, err)
}

Expand All @@ -317,13 +304,10 @@ func TestGetChangedTargets_TreehashReadError(t *testing.T) {
Orchestrator: orchestratormock.NewMockOrchestrator(ctrl),
})

request := &pb.GetChangedTargetsRequest{
FirstRevision: &pb.BuildDescription{Strategy: pb.COMPUTATION_STRATEGY_UNSET, Remote: "repo:go-code", BaseSha: "sha1"},
SecondRevision: &pb.BuildDescription{Strategy: pb.COMPUTATION_STRATEGY_UNSET, Remote: "repo:go-code", BaseSha: "sha2"},
OutputConfig: &pb.OutputConfig{MaxDistance: -1},
}
request := changedTargetsRequest()
outputConfig := &pb.OutputConfig{MaxDistance: -1}

err := c.GetChangedTargets(request, stream)
err := c.GetChangedTargets(request, outputConfig, stream, config.RepositoryConfig{Remote: "repo:go-code", RepositoryID: testRepositoryID("repo:go-code")})
require.Error(t, err)
assert.Contains(t, err.Error(), injected.Error())
}
Expand Down Expand Up @@ -404,13 +388,10 @@ func TestGetChangedTargets_StreamSendError(t *testing.T) {
Orchestrator: orchestratormock.NewMockOrchestrator(ctrl),
})

request := &pb.GetChangedTargetsRequest{
FirstRevision: &pb.BuildDescription{Strategy: pb.COMPUTATION_STRATEGY_UNSET, Remote: "repo:go-code", BaseSha: "sha1"},
SecondRevision: &pb.BuildDescription{Strategy: pb.COMPUTATION_STRATEGY_UNSET, Remote: "repo:go-code", BaseSha: "sha2"},
OutputConfig: &pb.OutputConfig{MaxDistance: -1},
}
request := changedTargetsRequest()
outputConfig := &pb.OutputConfig{MaxDistance: -1}

err := c.GetChangedTargets(request, stream)
err := c.GetChangedTargets(request, outputConfig, stream, config.RepositoryConfig{Remote: "repo:go-code", RepositoryID: testRepositoryID("repo:go-code")})
assert.Error(t, err)

select {
Expand Down Expand Up @@ -500,13 +481,10 @@ func TestGetChangedTargets_streamChunks(t *testing.T) {
Orchestrator: orchestratormock.NewMockOrchestrator(ctrl),
})

request := &pb.GetChangedTargetsRequest{
FirstRevision: &pb.BuildDescription{Strategy: pb.COMPUTATION_STRATEGY_UNSET, Remote: "repo:go-code", BaseSha: "sha1"},
SecondRevision: &pb.BuildDescription{Strategy: pb.COMPUTATION_STRATEGY_UNSET, Remote: "repo:go-code", BaseSha: "sha2"},
OutputConfig: &pb.OutputConfig{MaxDistance: -1, IncludeHashes: true, IncludeTags: true, IncludeAttributes: true},
}
request := changedTargetsRequest()
outputConfig := &pb.OutputConfig{MaxDistance: -1, IncludeHashes: true, IncludeTags: true, IncludeAttributes: true}

err := c.GetChangedTargets(request, stream)
err := c.GetChangedTargets(request, outputConfig, stream, config.RepositoryConfig{Remote: "repo:go-code", RepositoryID: testRepositoryID("repo:go-code")})
require.NoError(t, err)

select {
Expand Down Expand Up @@ -595,14 +573,13 @@ func TestGetChangedTargets_CacheWriteUsesAppCtx(t *testing.T) {
Orchestrator: orchestratormock.NewMockOrchestrator(ctrl),
})

request := &pb.GetChangedTargetsRequest{
FirstRevision: &pb.BuildDescription{Strategy: pb.COMPUTATION_STRATEGY_UNSET, Remote: "repo:go-code", BaseSha: "sha1"},
SecondRevision: &pb.BuildDescription{Strategy: pb.COMPUTATION_STRATEGY_UNSET, Remote: "repo:go-code", BaseSha: "sha2"},
OutputConfig: &pb.OutputConfig{MaxDistance: -1},
}
request := changedTargetsRequest()
outputConfig := &pb.OutputConfig{MaxDistance: -1}

handlerDone := make(chan error, 1)
go func() { handlerDone <- c.GetChangedTargets(request, stream) }()
go func() {
handlerDone <- c.GetChangedTargets(request, outputConfig, stream, config.RepositoryConfig{Remote: "repo:go-code", RepositoryID: testRepositoryID("repo:go-code")})
}()

var cacheCtx context.Context
select {
Expand Down Expand Up @@ -1089,13 +1066,10 @@ func TestGetChangedTargets_CacheHitWithDistanceFilter(t *testing.T) {
Orchestrator: orchestratormock.NewMockOrchestrator(ctrl),
})

request := &pb.GetChangedTargetsRequest{
FirstRevision: &pb.BuildDescription{Strategy: pb.COMPUTATION_STRATEGY_UNSET, Remote: "repo:go-code", BaseSha: "sha1"},
SecondRevision: &pb.BuildDescription{Strategy: pb.COMPUTATION_STRATEGY_UNSET, Remote: "repo:go-code", BaseSha: "sha2"},
OutputConfig: &pb.OutputConfig{MaxDistance: 1},
}
request := changedTargetsRequest()
outputConfig := &pb.OutputConfig{MaxDistance: 1}

err := c.GetChangedTargets(request, stream)
err := c.GetChangedTargets(request, outputConfig, stream, config.RepositoryConfig{Remote: "repo:go-code", RepositoryID: testRepositoryID("repo:go-code")})
require.NoError(t, err)

require.Len(t, sent, 2)
Expand Down Expand Up @@ -1337,15 +1311,7 @@ func TestSendTrimmedChangedTargets_RetainsDeletedAtMaxDistanceOne(t *testing.T)
assert.True(t, gotDeleted, "DELETED entry at distance 0 must survive max_distance=1")
}

func changedTargetsRequest() *pb.GetChangedTargetsRequest {
return &pb.GetChangedTargetsRequest{
FirstRevision: &pb.BuildDescription{Strategy: pb.COMPUTATION_STRATEGY_UNSET, Remote: "repo:go-code", BaseSha: "sha1"},
SecondRevision: &pb.BuildDescription{Strategy: pb.COMPUTATION_STRATEGY_UNSET, Remote: "repo:go-code", BaseSha: "sha2"},
OutputConfig: &pb.OutputConfig{MaxDistance: -1},
}
}

func changedTargetsEntityRequest() entity.GetChangedTargetsRequest {
func changedTargetsRequest() entity.GetChangedTargetsRequest {
return entity.GetChangedTargetsRequest{
First: entity.BuildDescription{Remote: "repo:go-code", BaseSha: "sha1"},
Second: entity.BuildDescription{Remote: "repo:go-code", BaseSha: "sha2"},
Expand All @@ -1364,7 +1330,7 @@ func TestServeChangedTargetsFromCache(t *testing.T) {
c.storage = st
stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl)

served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsEntityRequest(), &pb.OutputConfig{MaxDistance: -1}, stream, testRepositoryID("repo:go-code"), -1, time.Now())
served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), &pb.OutputConfig{MaxDistance: -1}, stream, testRepositoryID("repo:go-code"), -1, time.Now())
require.NoError(t, err)
assert.False(t, served, "a cache miss must not be served")
})
Expand Down Expand Up @@ -1403,7 +1369,7 @@ func TestServeChangedTargetsFromCache(t *testing.T) {
stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl)
// No Send expectation: a corrupt blob must not send anything to the client.

served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsEntityRequest(), &pb.OutputConfig{MaxDistance: -1}, stream, testRepositoryID("repo:go-code"), -1, time.Now())
served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), &pb.OutputConfig{MaxDistance: -1}, stream, testRepositoryID("repo:go-code"), -1, time.Now())
require.NoError(t, err)
assert.False(t, served, "a corrupt blob must trigger recompute, not a partial send")
})
Expand Down Expand Up @@ -1437,7 +1403,7 @@ func TestServeChangedTargetsFromCache(t *testing.T) {
stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl)
stream.EXPECT().Send(gomock.Any()).Return(nil).Times(2)

served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsEntityRequest(), &pb.OutputConfig{MaxDistance: -1}, stream, testRepositoryID("repo:go-code"), -1, time.Now())
served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), &pb.OutputConfig{MaxDistance: -1}, stream, testRepositoryID("repo:go-code"), -1, time.Now())
require.NoError(t, err)
assert.True(t, served, "a clean cache hit must be served")
})
Expand All @@ -1447,7 +1413,7 @@ func TestFetchTargetGraphs(t *testing.T) {
// BypassCache=true keeps getGraph on the orchestrator path only, so these
// tests need no storage mock.
bypassRequest := func() entity.GetChangedTargetsRequest {
r := changedTargetsEntityRequest()
r := changedTargetsRequest()
r.BypassCache = true
return r
}
Expand Down
20 changes: 10 additions & 10 deletions controller/getchangedtargets_tgb_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,8 +133,8 @@ func TestGetChangedTargets_TGBNativePath(t *testing.T) {
})

request := changedTargetsRequest()
request.OutputConfig = &pb.OutputConfig{MaxDistance: -1, IncludeHashes: true, IncludeTags: true, IncludeAttributes: true}
require.NoError(t, c.GetChangedTargets(request, stream))
outputConfig := &pb.OutputConfig{MaxDistance: -1, IncludeHashes: true, IncludeTags: true, IncludeAttributes: true}
require.NoError(t, c.GetChangedTargets(request, outputConfig, stream, config.RepositoryConfig{Remote: "repo:go-code", RepositoryID: testRepositoryID("repo:go-code")}))

changed, idToName := changedTargetsSent(t, sent)
require.Len(t, changed, 1, "should detect exactly the hash-flipped target")
Expand Down Expand Up @@ -243,8 +243,8 @@ func TestGetChangedTargets_TGBAllTargetsTrigger(t *testing.T) {
})

request := changedTargetsRequest()
request.OutputConfig = &pb.OutputConfig{MaxDistance: -1, IncludeHashes: true}
require.NoError(t, c.GetChangedTargets(request, stream))
outputConfig := &pb.OutputConfig{MaxDistance: -1, IncludeHashes: true}
require.NoError(t, c.GetChangedTargets(request, outputConfig, stream, config.RepositoryConfig{Remote: "repo:go-code", RepositoryID: testRepositoryID("repo:go-code")}))

changed, _ := changedTargetsSent(t, sent)
require.Len(t, changed, 4, "all targets from second graph should be reported as changed")
Expand Down Expand Up @@ -288,8 +288,8 @@ func TestGetChangedTargets_TGBAllTargetsTriggerPreservesMembershipChanges(t *tes
})

request := changedTargetsRequest()
request.OutputConfig = &pb.OutputConfig{MaxDistance: -1, IncludeHashes: true}
require.NoError(t, c.GetChangedTargets(request, stream))
outputConfig := &pb.OutputConfig{MaxDistance: -1, IncludeHashes: true}
require.NoError(t, c.GetChangedTargets(request, outputConfig, stream, config.RepositoryConfig{Remote: "repo:go-code", RepositoryID: testRepositoryID("repo:go-code")}))

changed, idToName := changedTargetsSent(t, sent)
require.Len(t, changed, 4)
Expand Down Expand Up @@ -366,8 +366,8 @@ func TestGetChangedTargets_TGBAllTargetsNoTrigger(t *testing.T) {
})

request := changedTargetsRequest()
request.OutputConfig = &pb.OutputConfig{MaxDistance: -1, IncludeHashes: true}
require.NoError(t, c.GetChangedTargets(request, stream))
outputConfig := &pb.OutputConfig{MaxDistance: -1, IncludeHashes: true}
require.NoError(t, c.GetChangedTargets(request, outputConfig, stream, config.RepositoryConfig{Remote: "repo:go-code", RepositoryID: testRepositoryID("repo:go-code")}))

changed, _ := changedTargetsSent(t, sent)
require.Len(t, changed, 1, "only the hash-flipped target should be changed")
Expand Down Expand Up @@ -412,8 +412,8 @@ func TestGetChangedTargets_TGBMixedFormatFallsBack(t *testing.T) {
})

request := changedTargetsRequest()
request.OutputConfig = &pb.OutputConfig{MaxDistance: -1, IncludeHashes: true, IncludeTags: true, IncludeAttributes: true}
require.NoError(t, c.GetChangedTargets(request, stream))
outputConfig := &pb.OutputConfig{MaxDistance: -1, IncludeHashes: true, IncludeTags: true, IncludeAttributes: true}
require.NoError(t, c.GetChangedTargets(request, outputConfig, stream, config.RepositoryConfig{Remote: "repo:go-code", RepositoryID: testRepositoryID("repo:go-code")}))

changed, idToName := changedTargetsSent(t, sent)
require.Len(t, changed, 1)
Expand Down
Loading
Loading