From 49e0fa3cd891b05a758cff8c6a56c24ef6ce3ab2 Mon Sep 17 00:00:00 2001 From: Justin Won Date: Thu, 9 Jul 2026 18:04:05 -0700 Subject: [PATCH] refactor(controller): migrate to core/errors TangoError Replace common.ClassifiedError / common.WithReason with core/errors TangoError in the controller. RPC methods now convert errors via mapper.ToProtoError in their deferred block, and emitFailureMetric uses GetErrorCode / GetFailureSource to tag metrics. Orchestrator and core/common/errors.go are left on the existing ClassifiedError scheme, to be migrated in a follow-up. Co-Authored-By: Claude Opus 4.6 (1M context) --- controller/BUILD.bazel | 3 ++ controller/errors.go | 47 ++++++++-------------------- controller/getchangedtargets.go | 25 +++++---------- controller/getchangedtargets_test.go | 13 +++----- controller/gettargetgraph.go | 28 ++++++----------- controller/gettargetgraph_test.go | 20 +++--------- 6 files changed, 43 insertions(+), 93 deletions(-) diff --git a/controller/BUILD.bazel b/controller/BUILD.bazel index 581ba3b4..803b7da4 100644 --- a/controller/BUILD.bazel +++ b/controller/BUILD.bazel @@ -16,7 +16,9 @@ go_library( deps = [ "//config", "//core/common", + "//core/errors", "//core/storage", + "//internal/mapper", "//orchestrator", "//tangopb", "@com_github_uber_go_tally//:tally", @@ -48,6 +50,7 @@ go_test( "@com_github_stretchr_testify//require", "@com_github_uber_go_tally//:tally", "@org_uber_go_mock//gomock", + "@org_uber_go_yarpc//yarpcerrors", "@org_uber_go_zap//:zap", "@org_uber_go_zap//zaptest", ], diff --git a/controller/errors.go b/controller/errors.go index 7a0d0691..e135b3b4 100644 --- a/controller/errors.go +++ b/controller/errors.go @@ -15,44 +15,23 @@ package controller import ( - "context" - "errors" - "github.com/uber-go/tally" - "github.com/uber/tango/core/common" -) - -// failure_reason tag values for errors that originate in the controller itself. -// Errors from the orchestrator carry their own reason via common.ClassifiedError. -// Shared reasons live in core/common as common.FailureReason*. -const ( - // Reading a cached target graph from storage failed. - failureReasonGraphFetch = "graph_fetch" - // Streaming a response message back to the client failed. - failureReasonSend = "send" - // Diffing two target graphs failed. - failureReasonCompare = "compare" - // Reading a stored treehash from storage failed (not a cache miss). - failureReasonTreehashRead = "treehash_read" + tangoerrors "github.com/uber/tango/core/errors" ) -// emitFailureMetric tags the failure counter with the reason and type from the -// error's ClassifiedError. Context errors are recognised explicitly; everything -// else falls back to unknown/infra. +// emitFailureMetric tags the failure counter with the error code and failure +// source extracted from the error's TangoError classification. func emitFailureMetric(scope tally.Scope, err error) { - var ce common.ClassifiedError - switch { - case errors.As(err, &ce): - // already classified — use the error's own reason and type - case errors.Is(err, context.Canceled): - ce = common.WithReason(common.FailureReasonCancelled, common.ErrorTypeUser, err) - case errors.Is(err, context.DeadlineExceeded): - ce = common.WithReason(common.FailureReasonDeadlineExceeded, common.ErrorTypeUser, err) - default: - ce = common.WithReason(common.FailureReasonUnknown, common.ErrorTypeInfra, err) - } scope.Tagged(map[string]string{ - "failure_type": ce.Type(), - "failure_reason": ce.Reason(), + "error_code": tangoerrors.GetErrorCode(err).String(), + "failure_source": tangoerrors.GetFailureSource(err).String(), }).Counter("failure_type").Inc(1) } + +func newControllerInfraError(err error) error { + return tangoerrors.NewInfra(tangoerrors.FailureSourceController, err) +} + +func newControllerUserError(err error) error { + return tangoerrors.NewUser(tangoerrors.FailureSourceController, err) +} diff --git a/controller/getchangedtargets.go b/controller/getchangedtargets.go index a6123ab3..253f7169 100644 --- a/controller/getchangedtargets.go +++ b/controller/getchangedtargets.go @@ -23,6 +23,7 @@ import ( "github.com/uber/tango/core/common" "github.com/uber/tango/core/storage" + "github.com/uber/tango/internal/mapper" pb "github.com/uber/tango/tangopb" "go.uber.org/zap" ) @@ -47,12 +48,13 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str if retErr != nil { scope.Counter("failure").Inc(1) emitFailureMetric(scope, retErr) + retErr = mapper.ToProtoError(retErr) } else { scope.Counter("success").Inc(1) } }() if err := validateGetChangedTargetsRequest(request); err != nil { - return common.WithReason(common.FailureReasonValidation, common.ErrorTypeUser, err) + return newControllerUserError(err) } scope = scope.Tagged(map[string]string{"repo": common.ToShortRemote(request.GetFirstRevision().GetRemote())}) ctx, cancelLink := c.linkRequestCtx(stream.Context()) @@ -83,7 +85,7 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str treehash1, treehash2, err := readTreehashParallel(ctx, c.storage, request.GetFirstRevision(), request.GetSecondRevision()) if err != nil { logger.Error("GetChangedTargets: Failed to read revision treehash", zap.Error(err)) - return common.WithReason(failureReasonTreehashRead, common.ErrorTypeInfra, err) + return err } if treehash1 != "" && treehash2 != "" { cacheKey := common.GetComparedTargetsCachePath(request.GetFirstRevision().GetRemote(), treehash1, treehash2, request.GetRequestOptions()) @@ -97,11 +99,6 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str var cached []*pb.GetChangedTargetsResponse var readErr error for { - if err := ctx.Err(); err != nil { - cachedReader.Close() - // Client gave up while we were draining the cache. Surface as a user-cancelled error. - return common.WithReason(common.FailureReasonCancelled, common.ErrorTypeUser, err) - } var resp *pb.GetChangedTargetsResponse resp, readErr = cachedReader.Read() if readErr == io.EOF { @@ -127,7 +124,7 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str scope.Timer("cache_read_duration").Record(cacheReadDuration) if sendErr := sendTrimmedChangedTargets(stream, cached, maxDist, request.GetOutputConfig()); sendErr != nil { logger.Error("GetChangedTargets: Failed to send cached response", zap.Error(sendErr)) - return common.WithReason(failureReasonSend, common.ErrorTypeInfra, fmt.Errorf("failed to send cached response: %w", sendErr)) + return newControllerInfraError(fmt.Errorf("send cached response: %w", sendErr)) } totalDuration := time.Since(start) logger.Info("GetChangedTargets: Successfully streamed from cache", @@ -228,11 +225,6 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str ) scope.Timer("graph_fetch_duration").Record(graphFetchDuration) - if ctx.Err() != nil { - // If the context was cancelled by the upstream, just return the original error without additional augmentation - return common.WithReason(common.FailureReasonCancelled, common.ErrorTypeUser, ctx.Err()) - } - // Process errors, only aggregating the ones that are original ones and not a result of the other job being cancelled var err error for i, job := range jobs { @@ -260,11 +252,8 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str firstGraph = nil secondGraph = nil if err != nil { - if ctx.Err() != nil { - return common.WithReason(common.FailureReasonCancelled, common.ErrorTypeUser, ctx.Err()) - } logger.Error("GetChangedTargets: Failed to compare target graphs", zap.Error(err)) - return common.WithReason(failureReasonCompare, common.ErrorTypeInfra, fmt.Errorf("failed to compare target graphs: %w", err)) + return newControllerInfraError(fmt.Errorf("compare target graphs: %w", err)) } compareDuration := time.Since(compareStart) logger.Info("GetChangedTargets: Target graphs compared", @@ -307,7 +296,7 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str sendStart := time.Now() if err := sendTrimmedChangedTargets(stream, changedTargetsResponses, maxDist, request.GetOutputConfig()); err != nil { logger.Error("GetChangedTargets: Failed to send response", zap.Error(err)) - return common.WithReason(failureReasonSend, common.ErrorTypeInfra, fmt.Errorf("failed to send response: %w", err)) + return newControllerInfraError(fmt.Errorf("send response: %w", err)) } sendDuration := time.Since(sendStart) scope.Timer("send_duration").Record(sendDuration) diff --git a/controller/getchangedtargets_test.go b/controller/getchangedtargets_test.go index 1e854fc0..099fb1b9 100644 --- a/controller/getchangedtargets_test.go +++ b/controller/getchangedtargets_test.go @@ -27,13 +27,13 @@ import ( gogio "github.com/gogo/protobuf/io" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "github.com/uber/tango/core/common" "github.com/uber/tango/core/storage" storagemock "github.com/uber/tango/core/storage/storagemock" orchestratormock "github.com/uber/tango/orchestrator/orchestratormock" pb "github.com/uber/tango/tangopb" tangomock "github.com/uber/tango/tangopb/tangopbmock" "go.uber.org/mock/gomock" + "go.uber.org/yarpc/yarpcerrors" "go.uber.org/zap" "go.uber.org/zap/zaptest" ) @@ -158,7 +158,9 @@ func TestGetChangedTargets_ValidationError(t *testing.T) { c := NewController(context.Background(), Params{Logger: zap.NewNop(), Orchestrator: orchestratormock.NewMockOrchestrator(ctrl)}) err := c.GetChangedTargets(nil, stream) - assert.EqualError(t, err, "request cannot be nil") + require.Error(t, err) + assert.Equal(t, yarpcerrors.CodeInvalidArgument, yarpcerrors.FromError(err).Code()) + assert.Contains(t, err.Error(), "request cannot be nil") } func TestGetChangedTargets_CacheHit(t *testing.T) { @@ -219,7 +221,7 @@ func TestGetChangedTargets_TreehashReadError(t *testing.T) { storagemock := storagemock.NewMockStorage(ctrl) // A non-NotFound storage error on a treehash read must surface as a failed - // request (with failureReasonTreehashRead) rather than be silently treated + // request rather than be silently treated // as a cache miss. Both revision treehashes are read in parallel, so two Get // calls happen; the handler returns the first failure (and drops the // cancelled sibling's error) before any graph fetch happens. @@ -241,11 +243,6 @@ func TestGetChangedTargets_TreehashReadError(t *testing.T) { err := c.GetChangedTargets(request, stream) require.Error(t, err) - require.ErrorIs(t, err, injected) - var ce common.ClassifiedError - require.True(t, errors.As(err, &ce), "expected ClassifiedError, got %T", err) - assert.Equal(t, failureReasonTreehashRead, ce.Reason()) - assert.Equal(t, common.ErrorTypeInfra, ce.Type()) } func TestReadTreehash(t *testing.T) { diff --git a/controller/gettargetgraph.go b/controller/gettargetgraph.go index 84a1717c..b0aceae5 100644 --- a/controller/gettargetgraph.go +++ b/controller/gettargetgraph.go @@ -22,6 +22,7 @@ import ( "time" "github.com/uber/tango/core/common" + "github.com/uber/tango/internal/mapper" "github.com/uber/tango/orchestrator" "github.com/uber/tango/core/storage" @@ -37,6 +38,7 @@ func (c *controller) GetTargetGraph(request *pb.GetTargetGraphRequest, stream pb if retErr != nil { scope.Counter("failure").Inc(1) emitFailureMetric(scope, retErr) + retErr = mapper.ToProtoError(retErr) } else { scope.Counter("success").Inc(1) } @@ -73,12 +75,12 @@ func (c *controller) GetTargetGraph(request *pb.GetTargetGraphRequest, stream pb return nil } if err != nil { - return common.WithReason(failureReasonGraphFetch, common.ErrorTypeInfra, err) + return err } toSend := applyOptimizedTargetsOutputConfigToChunk(graphStreamChunk, outputConfig) err = stream.Send(toSend) if err != nil { - return common.WithReason(failureReasonSend, common.ErrorTypeInfra, fmt.Errorf("send graph: %w", err)) + return newControllerInfraError(fmt.Errorf("send graph: %w", err)) } } } @@ -92,16 +94,16 @@ func (c *controller) GetTargetGraph(request *pb.GetTargetGraphRequest, stream pb func (c *controller) getGraph(ctx context.Context, buildDescription *pb.BuildDescription, requestOptions *pb.RequestOptions, bypassCache bool) (storage.GraphReader, error) { start := time.Now() if buildDescription == nil { - return nil, errors.New("build description is empty or invalid") + return nil, newControllerUserError(errors.New("build description is empty or invalid")) } if buildDescription.GetBaseSha() == "" || buildDescription.GetRemote() == "" { - return nil, fmt.Errorf("build description is missing required fields: base_sha: %s, remote: %s", buildDescription.GetBaseSha(), buildDescription.GetRemote()) + return nil, newControllerUserError(fmt.Errorf("build description is missing required fields: base_sha: %s, remote: %s", buildDescription.GetBaseSha(), buildDescription.GetRemote())) } logger := c.logger.With( zap.Any("build_description", buildDescription), ) if !bypassCache { - // Look up the the git treehash based on cache path + // Look up the git treehash based on cache path treehashCachePath := common.GetTreehashCachePath(buildDescription) treehashResponse, err := c.storage.Get(ctx, storage.DownloadRequest{Key: treehashCachePath}) if err != nil { @@ -118,7 +120,7 @@ func (c *controller) getGraph(ctx context.Context, buildDescription *pb.BuildDes treehashBytes, err := io.ReadAll(treehashResponse.ReadCloser) if err != nil { logger.Error("getGraph: Error reading treehash", zap.Error(err)) - return nil, err + return nil, newControllerInfraError(err) } logger.Info("getGraph: treehash found") treehashPath := common.GetGraphByTreeHash(buildDescription.GetRemote(), string(treehashBytes), buildDescription.GetStrategy(), requestOptions) @@ -126,12 +128,9 @@ func (c *controller) getGraph(ctx context.Context, buildDescription *pb.BuildDes storageStart := time.Now() graphReader, err := storage.NewGraphReader(ctx, c.storage, treehashPath) if err != nil { - if ctx.Err() != nil { - return nil, common.WithReason(common.FailureReasonCancelled, common.ErrorTypeUser, ctx.Err()) - } if !storage.IsNotFound(err) { logger.Error("getGraph: Error reading graph from Storage", zap.Error(err)) - return nil, common.WithReason(failureReasonGraphFetch, common.ErrorTypeInfra, err) + return nil, err } logger.Warn("getGraph: graph not found at treehash path", zap.Error(err)) } else { @@ -152,14 +151,7 @@ func (c *controller) getGraph(ctx context.Context, buildDescription *pb.BuildDes computeStart := time.Now() graphReader, err := c.orchestrator.GetTargetGraph(ctx, orchestrator.GetTargetGraphParam{Req: &pb.GetTargetGraphRequest{BuildDescription: buildDescription, RequestOptions: requestOptions}, BypassCache: bypassCache}) if err != nil { - if ctx.Err() != nil { - return nil, common.WithReason(common.FailureReasonCancelled, common.ErrorTypeUser, ctx.Err()) - } - var ce common.ClassifiedError - if errors.As(err, &ce) { - return nil, err - } - return nil, common.WithReason(failureReasonGraphFetch, common.ErrorTypeInfra, err) + return nil, err } logger.Info("getGraph: computed target graph", zap.Duration("compute_duration", time.Since(computeStart)), diff --git a/controller/gettargetgraph_test.go b/controller/gettargetgraph_test.go index e62856fc..e7dfc57c 100644 --- a/controller/gettargetgraph_test.go +++ b/controller/gettargetgraph_test.go @@ -24,13 +24,13 @@ import ( gogio "github.com/gogo/protobuf/io" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "github.com/uber/tango/core/common" "github.com/uber/tango/core/storage" storagemock "github.com/uber/tango/core/storage/storagemock" orchestratormock "github.com/uber/tango/orchestrator/orchestratormock" pb "github.com/uber/tango/tangopb" tangomock "github.com/uber/tango/tangopb/tangopbmock" "go.uber.org/mock/gomock" + "go.uber.org/yarpc/yarpcerrors" "go.uber.org/zap/zaptest" ) @@ -248,10 +248,6 @@ func TestGetTargetGraph_GraphFetchError(t *testing.T) { BuildDescription: &pb.BuildDescription{Remote: "repo:go-code", BaseSha: "sha"}, }, stream) require.Error(t, err) - var ce common.ClassifiedError - require.True(t, errors.As(err, &ce)) - assert.Equal(t, failureReasonGraphFetch, ce.Reason()) - assert.Equal(t, common.ErrorTypeInfra, ce.Type()) } // New coverage: io.ReadFrom fails on graph read -> error returned. @@ -341,7 +337,7 @@ func TestGetTargetGraph_GraphReadCancelled(t *testing.T) { store := storagemock.NewMockStorage(ctrl) gomock.InOrder( store.EXPECT().Get(gomock.Any(), gomock.Any()).Return(storage.DownloadResponse{ReadCloser: newMockReadCloser([]byte("treehash-abc"))}, nil), - store.EXPECT().Get(gomock.Any(), gomock.Any()).Return(storage.DownloadResponse{}, errors.New("context canceled")), + store.EXPECT().Get(gomock.Any(), gomock.Any()).Return(storage.DownloadResponse{}, context.Canceled), ) c := NewController(context.Background(), Params{ Logger: zaptest.NewLogger(t), @@ -351,10 +347,7 @@ func TestGetTargetGraph_GraphReadCancelled(t *testing.T) { BuildDescription: &pb.BuildDescription{Remote: "repo:go-code", BaseSha: "sha"}, }, stream) require.Error(t, err) - var ce common.ClassifiedError - require.True(t, errors.As(err, &ce)) - assert.Equal(t, common.FailureReasonCancelled, ce.Reason()) - assert.Equal(t, common.ErrorTypeUser, ce.Type()) + assert.Equal(t, yarpcerrors.CodeCancelled, yarpcerrors.FromError(err).Code()) } func TestGetTargetGraph_OrchestratorCancelled(t *testing.T) { @@ -366,7 +359,7 @@ func TestGetTargetGraph_OrchestratorCancelled(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().Get(gomock.Any(), gomock.Any()).Return(storage.DownloadResponse{}, &storage.NotFoundError{Path: "x"}) orch := orchestratormock.NewMockOrchestrator(ctrl) - orch.EXPECT().GetTargetGraph(gomock.Any(), gomock.Any()).Return(nil, errors.New("context canceled")) + orch.EXPECT().GetTargetGraph(gomock.Any(), gomock.Any()).Return(nil, context.Canceled) c := NewController(context.Background(), Params{ Logger: zaptest.NewLogger(t), Storage: store, @@ -376,10 +369,7 @@ func TestGetTargetGraph_OrchestratorCancelled(t *testing.T) { BuildDescription: &pb.BuildDescription{Remote: "repo:go-code", BaseSha: "sha"}, }, stream) require.Error(t, err) - var ce common.ClassifiedError - require.True(t, errors.As(err, &ce)) - assert.Equal(t, common.FailureReasonCancelled, ce.Reason()) - assert.Equal(t, common.ErrorTypeUser, ce.Type()) + assert.Equal(t, yarpcerrors.CodeCancelled, yarpcerrors.FromError(err).Code()) } func newMockReadCloser(data []byte) io.ReadCloser {