Refactor replace errors wrap with join (#2001)

* replace for internal pkgs

* fix want error message

* fix test

* fix test for pkg

* replace Wrap with Join for pkg packages

* fix test to adopt to errors.Join

* replace some left errors.Wrap with errors.Join

* fix error handling

* revert to errors.Wrap to pass the grpc stream test

---------

Co-authored-by: Yusuke Kato <[email protected]>

hack/benchmark/internal/operation/operation.go

@@ -20,7 +20,8 @@ import (
 	"github.com/vdaas/vald/apis/grpc/v1/payload"
 	"github.com/vdaas/vald/hack/benchmark/internal/assets"
 	"github.com/vdaas/vald/internal/client/v1/client"
-	"github.com/vdaas/vald/internal/errors"
+	"github.com/vdaas/vald/internal/net/grpc/status"
+	"google.golang.org/grpc/codes"
 )
 
 type Operation interface {
@@ -56,8 +57,11 @@ func (o *operation) CreateIndex(ctx context.Context, b *testing.B) {
 	b.Run("CreateIndex", func(b *testing.B) {
 		for i := 0; i < b.N; i++ {
 			_, err := o.indexerC.CreateIndex(ctx, req)
-			if err != nil && !errors.Is(err, errors.ErrUncommittedIndexNotFound) {
-				b.Error(err)
+			if err != nil {
+				st, _ := status.FromError(err)
+				if st.Code() != codes.FailedPrecondition {
+					b.Error(err)
+				}
 			}
 		}
 	})

internal/backoff/backoff.go

@@ -130,7 +130,7 @@ func (b *backoff) Do(ctx context.Context, f func(ctx context.Context) (val inter
 				log.Debugf("[backoff]\tfor: "+name+",\tCanceled\terror: %v", err.Error())
 				return nil, err
 			default:
-				return nil, errors.Wrap(err, dctx.Err().Error())
+				return nil, errors.Join(err, dctx.Err())
 			}
 		default:
 			res, ret, err = func() (val interface{}, retryable bool, err error) {
@@ -174,7 +174,7 @@ func (b *backoff) Do(ctx context.Context, f func(ctx context.Context) (val inter
 					log.Debugf("[backoff]\tfor: "+name+",\tCanceled\terror: %v", err.Error())
 					return nil, err
 				default:
-					return nil, errors.Wrap(dctx.Err(), err.Error())
+					return nil, errors.Join(dctx.Err(), err)
 				}
 			case <-timer.C:
 				if dur >= b.durationLimit {

internal/circuitbreaker/manager.go

@@ -97,11 +97,11 @@ func (bm *breakerManager) Do(ctx context.Context, key string, fn func(ctx contex
 			err = errors.Wrapf(err, "circuitbreaker state is %s, this error is not caused by circuitbreaker", st.String())
 		case StateOpen:
 			if !errors.Is(err, errors.ErrCircuitBreakerOpenState) {
-				err = errors.Wrap(err, errors.ErrCircuitBreakerOpenState.Error())
+				err = errors.Join(err, errors.ErrCircuitBreakerOpenState)
 			}
 		case StateHalfOpen:
 			if !errors.Is(err, errors.ErrCircuitBreakerHalfOpenFlowLimitation) {
-				err = errors.Wrap(err, errors.ErrCircuitBreakerHalfOpenFlowLimitation.Error())
+				err = errors.Join(err, errors.ErrCircuitBreakerHalfOpenFlowLimitation)
 			}
 		}
 		return val, err

internal/client/v1/client/discoverer/discover.go

@@ -103,7 +103,7 @@ func (c *client) Start(ctx context.Context) (<-chan error, error) {
 	err = c.discover(ctx, ech)
 	if err != nil {
 		close(ech)
-		return nil, errors.Wrap(c.dscClient.Close(ctx), err.Error())
+		return nil, errors.Join(c.dscClient.Close(ctx), err)
 	}
 
 	c.eg.Go(safety.RecoverFunc(func() (err error) {
@@ -114,17 +114,17 @@ func (c *client) Start(ctx context.Context) (<-chan error, error) {
 			var errs error
 			err = c.dscClient.Close(ctx)
 			if err != nil {
-				errs = errors.Wrap(errs, err.Error())
+				errs = errors.Join(errs, err)
 			}
 			if c.autoconn && c.client != nil {
 				err = c.client.Close(ctx)
 				if err != nil {
-					errs = errors.Wrap(errs, err.Error())
+					errs = errors.Join(errs, err)
 				}
 			}
 			err = ctx.Err()
 			if err != nil && err != context.Canceled {
-				errs = errors.Wrap(errs, err.Error())
+				errs = errors.Join(errs, err)
 			}
 			return errs
 		}
@@ -373,7 +373,7 @@ func (c *client) disconnectOldAddrs(ctx context.Context, oldAddrs, connectedAddr
 				if err != nil {
 					select {
 					case <-ctx.Done():
-						return errors.Wrap(ctx.Err(), err.Error())
+						return errors.Join(ctx.Err(), err)
 					case ech <- err:
 						return err
 					}
@@ -383,7 +383,7 @@ func (c *client) disconnectOldAddrs(ctx context.Context, oldAddrs, connectedAddr
 		}); err != nil {
 			select {
 			case <-ctx.Done():
-				return errors.Wrap(ctx.Err(), err.Error())
+				return errors.Join(ctx.Err(), err)
 			case ech <- err:
 				return err
 			}

internal/compress/gzip.go

@@ -133,7 +133,7 @@ func (g *gzipReader) Read(p []byte) (n int, err error) {
 func (g *gzipReader) Close() (err error) {
 	err = g.r.Close()
 	if err != nil {
-		return errors.Wrap(g.src.Close(), err.Error())
+		return errors.Join(g.src.Close(), err)
 	}
 
 	return g.src.Close()
@@ -153,7 +153,7 @@ func (g *gzipWriter) Write(p []byte) (n int, err error) {
 func (g *gzipWriter) Close() (err error) {
 	err = g.w.Close()
 	if err != nil {
-		return errors.Wrap(g.dst.Close(), err.Error())
+		return errors.Join(g.dst.Close(), err)
 	}
 
 	return g.dst.Close()

internal/compress/gzip_test.go

@@ -853,7 +853,7 @@ func Test_gzipReader_Close(t *testing.T) {
 				},
 			},
 			want: want{
-				err: errors.Wrap(errors.New("serr"), errors.New("rerr").Error()),
+				err: errors.Join(errors.New("serr"), errors.New("rerr")),
 			},
 		},
 
@@ -872,7 +872,7 @@ func Test_gzipReader_Close(t *testing.T) {
 				},
 			},
 			want: want{
-				err: errors.Wrap(nil, errors.New("rerr").Error()),
+				err: errors.New("rerr"),
 			},
 		},
 
@@ -1055,7 +1055,7 @@ func Test_gzipWriter_Close(t *testing.T) {
 				},
 			},
 			want: want{
-				err: errors.Wrap(errors.New("derr"), errors.New("werr").Error()),
+				err: errors.Join(errors.New("derr"), errors.New("werr")),
 			},
 		},
 
@@ -1074,7 +1074,7 @@ func Test_gzipWriter_Close(t *testing.T) {
 				},
 			},
 			want: want{
-				err: errors.Wrap(nil, errors.New("werr").Error()),
+				err: errors.New("werr"),
 			},
 		},
 

internal/compress/lz4.go

@@ -60,7 +60,7 @@ func (l *lz4Compressor) CompressVector(vector []float32) (b []byte, err error) {
 		cerr := zw.Close()
 		if cerr != nil {
 			b = nil
-			err = errors.Wrap(err, cerr.Error())
+			err = errors.Join(err, cerr)
 		}
 	}()
 
@@ -142,7 +142,7 @@ func (l *lz4Writer) Write(p []byte) (n int, err error) {
 func (l *lz4Writer) Close() (err error) {
 	err = l.w.Close()
 	if err != nil {
-		return errors.Wrap(l.dst.Close(), err.Error())
+		return errors.Join(l.dst.Close(), err)
 	}
 
 	return l.dst.Close()

internal/compress/zstd.go

@@ -153,7 +153,7 @@ func (z *zstdWriter) Write(p []byte) (n int, err error) {
 func (z *zstdWriter) Close() (err error) {
 	err = z.w.Close()
 	if err != nil {
-		return errors.Wrap(z.dst.Close(), err.Error())
+		return errors.Join(z.dst.Close(), err)
 	}
 
 	return z.dst.Close()

internal/compress/zstd_test.go

@@ -1151,7 +1151,7 @@ func Test_zstdWriter_Close(t *testing.T) {
 				},
 			},
 			want: want{
-				err: errors.Wrap(errors.New("dst close err"), "w close err"),
+				err: errors.Join(errors.New("dst close err"), errors.New("w close err")),
 			},
 		},
 	}

internal/config/config.go

@@ -69,7 +69,7 @@ func Read(path string, cfg interface{}) (err error) {
 	defer func() {
 		if f != nil {
 			if err != nil {
-				err = errors.Wrap(f.Close(), err.Error())
+				err = errors.Join(f.Close(), err)
 				return
 			}
 			err = f.Close()

internal/db/kvs/redis/redis.go

@@ -283,7 +283,7 @@ func (rc *redisClient) ping(ctx context.Context) (r Redis, err error) {
 	for {
 		select {
 		case <-pctx.Done():
-			err = errors.Wrap(errors.Wrap(err, errors.ErrRedisConnectionPingFailed.Error()), pctx.Err().Error())
+			err = errors.Join(err, errors.ErrRedisConnectionPingFailed, pctx.Err())
 			log.Error(err)
 			return nil, err
 		case <-tick.C:

internal/db/kvs/redis/redis_test.go

@@ -216,7 +216,7 @@ func Test_redisClient_ping(t *testing.T) {
 				},
 				want: want{
 					wantR: nil,
-					err:   errors.Wrap(errors.Wrap(err, errors.ErrRedisConnectionPingFailed.Error()), context.DeadlineExceeded.Error()),
+					err:   errors.Join(err, errors.ErrRedisConnectionPingFailed, context.DeadlineExceeded),
 				},
 			}
 		}(),
@@ -993,7 +993,7 @@ func Test_redisClient_Connect(t *testing.T) {
 					dialer:               dialer,
 				},
 				want: want{
-					err: errors.Wrap(errors.Wrap(nil, errors.ErrRedisConnectionPingFailed.Error()), context.DeadlineExceeded.Error()),
+					err: errors.Join(nil, errors.ErrRedisConnectionPingFailed, context.DeadlineExceeded),
 				},
 				checkFunc: defaultCheckFunc,
 			}

internal/db/nosql/cassandra/option_test.go

@@ -535,7 +535,7 @@ func TestWithConnectTimeout(t *testing.T) {
 				dur: "dummy",
 			},
 			want: want{
-				err: errors.NewErrCriticalOption("connectTimeout", "dummy", errors.New("invalid timeout value: dummy	:timeout parse error out put failed: time: invalid duration \"dummy\"")),
+				err: errors.NewErrCriticalOption("connectTimeout", "dummy", errors.Join(errors.New("time: invalid duration \"dummy\""), errors.ErrTimeoutParseFailed("dummy"))),
 				obj: &T{},
 			},
 		},
@@ -1521,7 +1521,7 @@ func TestWithRetryPolicyMinDuration(t *testing.T) {
 				err: errors.NewErrCriticalOption(
 					"retryPolicyMinDuration",
 					"dummy",
-					errors.New("invalid timeout value: dummy	:timeout parse error out put failed: time: invalid duration \"dummy\""),
+					errors.Join(errors.New("time: invalid duration \"dummy\""), errors.ErrTimeoutParseFailed("dummy")),
 				),
 				obj: &T{},
 			},
@@ -1613,7 +1613,7 @@ func TestWithRetryPolicyMaxDuration(t *testing.T) {
 				err: errors.NewErrCriticalOption(
 					"retryPolicyMaxDuration",
 					"dummy",
-					errors.New("invalid timeout value: dummy	:timeout parse error out put failed: time: invalid duration \"dummy\""),
+					errors.Join(errors.New("time: invalid duration \"dummy\""), errors.ErrTimeoutParseFailed("dummy")),
 				),
 				obj: &T{},
 			},
@@ -1705,7 +1705,7 @@ func TestWithReconnectionPolicyInitialInterval(t *testing.T) {
 				err: errors.NewErrCriticalOption(
 					"reconnectionPolicyInitialInterval",
 					"dummy",
-					errors.New("invalid timeout value: dummy	:timeout parse error out put failed: time: invalid duration \"dummy\""),
+					errors.Join(errors.New("time: invalid duration \"dummy\""), errors.ErrTimeoutParseFailed("dummy")),
 				),
 				obj: &T{},
 			},
@@ -1869,7 +1869,11 @@ func TestWithSocketKeepalive(t *testing.T) {
 				socketKeepalive: "dummy",
 			},
 			want: want{
-				err: errors.NewErrCriticalOption("socketKeepalive", "dummy", errors.New("invalid timeout value: dummy	:timeout parse error out put failed: time: invalid duration \"dummy\"")),
+				err: errors.NewErrCriticalOption(
+					"socketKeepalive",
+					"dummy",
+					errors.Join(errors.New("time: invalid duration \"dummy\""), errors.ErrTimeoutParseFailed("dummy")),
+				),
 				obj: &T{},
 			},
 		},
@@ -2973,7 +2977,7 @@ func TestWithMaxWaitSchemaAgreement(t *testing.T) {
 				err: errors.NewErrCriticalOption(
 					"maxWaitSchemaAgreement",
 					"dummy",
-					errors.New("invalid timeout value: dummy	:timeout parse error out put failed: time: invalid duration \"dummy\""),
+					errors.Join(errors.New("time: invalid duration \"dummy\""), errors.ErrTimeoutParseFailed("dummy")),
 				),
 				obj: &T{},
 			},
@@ -3060,7 +3064,11 @@ func TestWithReconnectInterval(t *testing.T) {
 				reconnectInterval: "dummy",
 			},
 			want: want{
-				err: errors.NewErrCriticalOption("reconnectInterval", "dummy", errors.New("invalid timeout value: dummy	:timeout parse error out put failed: time: invalid duration \"dummy\"")),
+				err: errors.NewErrCriticalOption(
+					"reconnectInterval",
+					"dummy",
+					errors.Join(errors.New("time: invalid duration \"dummy\""), errors.ErrTimeoutParseFailed("dummy")),
+				),
 				obj: &T{},
 			},
 		},
@@ -3911,7 +3919,11 @@ func TestWithWriteCoalesceWaitTime(t *testing.T) {
 				writeCoalesceWaitTime: "dummy",
 			},
 			want: want{
-				err: errors.NewErrCriticalOption("writeCoalesceWaitTime", "dummy", errors.New("invalid timeout value: dummy	:timeout parse error out put failed: time: invalid duration \"dummy\"")),
+				err: errors.NewErrCriticalOption(
+					"writeCoalesceWaitTime",
+					"dummy",
+					errors.Join(errors.New("time: invalid duration \"dummy\""), errors.ErrTimeoutParseFailed("dummy")),
+				),
 				obj: &T{},
 			},
 		},

internal/db/rdb/mysql/mysql.go

@@ -159,13 +159,13 @@ func (m *mySQLClient) Ping(ctx context.Context) (err error) {
 		select {
 		case <-pctx.Done():
 			if err != nil {
-				err = errors.Wrap(errors.ErrMySQLConnectionPingFailed, err.Error())
+				err = errors.Join(errors.ErrMySQLConnectionPingFailed, err)
 			} else {
 				err = errors.ErrMySQLConnectionPingFailed
 			}
 			cerr := pctx.Err()
 			if cerr != nil {
-				err = errors.Wrap(err, cerr.Error())
+				err = errors.Join(err, cerr)
 			}
 			return err
 		case <-tick.C:

internal/db/rdb/mysql/mysql_test.go

@@ -401,7 +401,7 @@ func Test_mySQLClient_Open(t *testing.T) {
 					cancel()
 				},
 				want: want{
-					err: errors.Wrap(errors.ErrMySQLConnectionPingFailed, err.Error()),
+					err: errors.Join(errors.ErrMySQLConnectionPingFailed, err, context.DeadlineExceeded),
 				},
 			}
 		}(),
@@ -550,7 +550,7 @@ func Test_mySQLClient_Ping(t *testing.T) {
 					},
 				},
 				want: want{
-					err: errors.Wrap(errors.Wrap(errors.ErrMySQLConnectionPingFailed, err.Error()), context.DeadlineExceeded.Error()),
+					err: errors.Join(errors.ErrMySQLConnectionPingFailed, err, context.DeadlineExceeded),
 				},
 				afterFunc: func(t *testing.T, _ args) {
 					t.Helper()

internal/db/storage/blob/s3/option_test.go

@@ -338,7 +338,7 @@ func TestWithMaxPartSize(t *testing.T) {
 					maxPartSize: 0,
 				},
 				err: func() (err error) {
-					err = errors.Wrap(errors.New("byte quantity must be a positive integer with a unit of measurement like M, MB, MiB, G, GiB, or GB"), errors.ErrParseUnitFailed("a").Error())
+					err = errors.Join(errors.New("byte quantity must be a positive integer with a unit of measurement like M, MB, MiB, G, GiB, or GB"), errors.ErrParseUnitFailed("a"))
 					return
 				}(),
 			},
@@ -435,7 +435,7 @@ func TestWithMaxChunkSize(t *testing.T) {
 					maxChunkSize: 0,
 				},
 				err: func() (err error) {
-					err = errors.Wrap(errors.New("byte quantity must be a positive integer with a unit of measurement like M, MB, MiB, G, GiB, or GB"), errors.ErrParseUnitFailed("a").Error())
+					err = errors.Join(errors.New("byte quantity must be a positive integer with a unit of measurement like M, MB, MiB, G, GiB, or GB"), errors.ErrParseUnitFailed("a"))
 					return
 				}(),
 			},

internal/errgroup/group.go

@@ -174,7 +174,7 @@ func (g *group) Wait() error {
 	default:
 		g.err = g.errs[0]
 		for _, err := range g.errs[1:] {
-			g.err = errors.Wrap(g.err, err.Error())
+			g.err = errors.Join(g.err, err)
 		}
 	}
 	g.mu.RUnlock()

internal/errgroup/group_test.go

@@ -766,7 +766,7 @@ func Test_group_Wait(t *testing.T) {
 				},
 			},
 			want: want{
-				err: errors.Wrap(errors.New("err1"), errors.New("err2").Error()),
+				err: errors.Join(errors.New("err1"), errors.New("err2")),
 			},
 		},
 	}