[WIP] migrate controller to core/errors TangoError

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",
     ],

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)
+}

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)

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) {

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,20 +120,17 @@ 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)
 			// Download the target graph based on treehash.
 			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)),

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 {