Skip to content

Commit a5d4a6c

Browse files
CBistsCBists
authored
Change error resp (#415)
* Correction * Fix regress * Fix regress * Change client message * Update ops.go * Add new error codes * Replace spqrerror.New(fmt.Sprintf()) to spqrerror.Newf * fix regress * fix typo * Fix typo * Fix typo
1 parent 86d2b41 commit a5d4a6c

File tree

16 files changed

+160
-136
lines changed

16 files changed

+160
-136
lines changed

‎balancer/pkg/console.go

+3-3
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ package pkg
33
import (
44
"context"
55
"crypto/tls"
6-
"fmt"
6+
"github.com/pg-sharding/spqr/pkg/models/spqrerror"
77
"strconv"
88

99
"github.com/jackc/pgx/v5/pgproto3"
@@ -108,7 +108,7 @@ func (c *Console) ProcessQuery(ctx context.Context, q string, cl client.Client)
108108
Err(err).
109109
Msg("unknown stmt.Cmd")
110110

111-
return fmt.Errorf("Unknown show statement: %s", stmt.Cmd)
111+
return spqrerror.Newf(spqrerror.SPQR_COMPLEX_QUERY, "Unknown show statement: %s", stmt.Cmd)
112112
}
113113

114114
case *spqrparser.SplitKeyRange:
@@ -209,7 +209,7 @@ func (c *Console) ProcessQuery(ctx context.Context, q string, cl client.Client)
209209

210210
case *spqrparser.Shutdown:
211211
//t.stchan <- struct{}{}
212-
return fmt.Errorf("not implemented")
212+
return spqrerror.New(spqrerror.SPQR_NOT_IMPLEMENTED, "not implemented")
213213

214214
default:
215215
spqrlog.Zero.Error().

‎balancer/pkg/db.go

+5-4
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import (
55
"database/sql"
66
"errors"
77
"fmt"
8+
"github.com/pg-sharding/spqr/pkg/models/spqrerror"
89
"net"
910
"strconv"
1011
"strings"
@@ -139,7 +140,7 @@ func NewCluster(addrs []string, dbname, user, password, sslMode, sslRootCert str
139140

140141
db, err := sql.Open("pgx", connString)
141142
if err != nil {
142-
return nil, fmt.Errorf("failed to open pgx connection %v: %v", connString, err)
143+
return nil, spqrerror.Newf(spqrerror.SPQR_CONNECTION_ERROR, "failed to open pgx connection %v: %v", connString, err)
143144
}
144145
// TODO may be some connections settings here?
145146

@@ -155,7 +156,7 @@ func ConnString(addr, dbname, user, password, sslMode, sslRootCert string) (stri
155156

156157
host, portFromAddr, err := net.SplitHostPort(addr)
157158
if err != nil {
158-
return "", fmt.Errorf("invalid host spec: %s", err)
159+
return "", spqrerror.Newf(spqrerror.SPQR_CONNECTION_ERROR, "invalid host spec: %s", err)
159160
}
160161
connParams = append(connParams, "host="+host)
161162
connParams = append(connParams, "port="+portFromAddr)
@@ -194,7 +195,7 @@ func GetMasterConn(cluster *hasql.Cluster, retries int, sleepMS int) (*sql.Conn,
194195
defer cancel()
195196
node, err := cluster.WaitForPrimary(ctx)
196197
if err != nil {
197-
return nil, fmt.Errorf("there is no node with role master: %s", err)
198+
return nil, spqrerror.Newf(spqrerror.SPQR_CONNECTION_ERROR, "there is no node with role master: %s", err)
198199
}
199200
return GetNodeConn(context.TODO(), node, retries, sleepMS)
200201
}
@@ -205,7 +206,7 @@ func GetNodeConn(parentctx context.Context, node hasql.Node, retries int, sleepM
205206

206207
conn, err := node.DB().Conn(ctx)
207208
if err != nil {
208-
return nil, fmt.Errorf("failed to get connection with master node: %v", err)
209+
return nil, spqrerror.Newf(spqrerror.SPQR_CONNECTION_ERROR, "failed to get connection with master node: %v", err)
209210
}
210211

211212
return conn, nil

‎balancer/pkg/installation.go

+3-2
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"context"
55
"database/sql"
66
"fmt"
7+
"github.com/pg-sharding/spqr/pkg/models/spqrerror"
78
"math"
89
"math/big"
910
"strings"
@@ -193,7 +194,7 @@ func AddHostStats(stats *map[string]map[string]Stats, additionalStats *map[strin
193194
func (i *Installation) GetShardStats(shard Shard, keyRanges []KeyRange) (map[string]map[string]Stats, error) {
194195
cluster, ok := (*i.shardClusters)[shard.id]
195196
if !ok {
196-
return nil, fmt.Errorf("not known shard %d", shard.id)
197+
return nil, spqrerror.Newf(spqrerror.SPQR_NO_DATASHARD, "not known shard %d", shard.id)
197198
}
198199

199200
nodes := cluster.Nodes()
@@ -394,7 +395,7 @@ func (i *Installation) GetKeyDistanceByRange(conn *sql.Conn, keyRange KeyRange)
394395
}
395396

396397
if err := rows.Close(); err != nil {
397-
return nil, fmt.Errorf("failed to close keyRange rows: %w", err)
398+
return nil, spqrerror.Newf(spqrerror.SPQR_KEYRANGE_ERROR, "failed to close keyRange rows: %s", err.Error())
398399
}
399400

400401
kr := KeyRange{

‎coordinator/provider/coord_pool.go

+3-3
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
package provider
22

33
import (
4-
"fmt"
4+
"github.com/pg-sharding/spqr/pkg/models/spqrerror"
55

66
"github.com/pg-sharding/spqr/pkg/models/kr"
77
"github.com/pg-sharding/spqr/pkg/pool"
@@ -22,11 +22,11 @@ func NewCoordPool(info *protos.PoolInfo) *CoordPool {
2222
// TODO : unit tests
2323
// TODO : implement
2424
func (r *CoordPool) Connection(clid uint, shardKey kr.ShardKey) (shard.Shard, error) {
25-
return nil, fmt.Errorf("unimplemented")
25+
return nil, spqrerror.New(spqrerror.SPQR_NOT_IMPLEMENTED, "CoordPool.Connection method unimplemented")
2626
}
2727

2828
// TODO : unit tests
2929
// TODO : implement
3030
func (r *CoordPool) ForEach(cb func(p shard.Shardinfo) error) error {
31-
return fmt.Errorf("unimplemented")
31+
return spqrerror.New(spqrerror.SPQR_NOT_IMPLEMENTED, "CoordPool.ForEach method unimplemented")
3232
}

‎coordinator/provider/coordinator.go

+16-15
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package provider
33
import (
44
"context"
55
"fmt"
6+
"github.com/pg-sharding/spqr/pkg/models/spqrerror"
67
"net"
78
"time"
89

@@ -97,19 +98,19 @@ func (ci grpcConnectionIterator) ClientPoolForeach(cb func(client client.ClientI
9798
// TODO : implement
9899
// TODO : unit tests
99100
func (ci grpcConnectionIterator) Put(client client.Client) error {
100-
return fmt.Errorf("grpcConnectionIterator put not implemented")
101+
return spqrerror.New(spqrerror.SPQR_NOT_IMPLEMENTED, "grpcConnectionIterator put not implemented")
101102
}
102103

103104
// TODO : implement
104105
// TODO : unit tests
105106
func (ci grpcConnectionIterator) Pop(id uint) (bool, error) {
106-
return true, fmt.Errorf("grpcConnectionIterator pop not implemented")
107+
return true, spqrerror.New(spqrerror.SPQR_NOT_IMPLEMENTED, "grpcConnectionIterator pop not implemented")
107108
}
108109

109110
// TODO : implement
110111
// TODO : unit tests
111112
func (ci grpcConnectionIterator) Shutdown() error {
112-
return fmt.Errorf("grpcConnectionIterator shutdown not implemented")
113+
return spqrerror.New(spqrerror.SPQR_NOT_IMPLEMENTED, "grpcConnectionIterator shutdown not implemented")
113114
}
114115

115116
// TODO : unit tests
@@ -374,7 +375,7 @@ func (qc *qdbCoordinator) traverseRouters(ctx context.Context, cb func(cc *grpc.
374375
for _, rtr := range rtrs {
375376
if err := func() error {
376377
if rtr.State != qdb.OPENED {
377-
return fmt.Errorf("router is closed")
378+
return spqrerror.New(spqrerror.SPQR_ROUTER_ERROR, "router is closed")
378379
}
379380

380381
// TODO: run cb`s async
@@ -665,7 +666,7 @@ func (qc *qdbCoordinator) Split(ctx context.Context, req *kr.SplitKeyRange) erro
665666
Msg("split request is")
666667

667668
if _, err := qc.db.GetKeyRange(ctx, req.Krid); err == nil {
668-
return fmt.Errorf("key range %v already present in qdb", req.Krid)
669+
return spqrerror.Newf(spqrerror.SPQR_KEYRANGE_ERROR, "key range %v already present in qdb", req.Krid)
669670
}
670671

671672
krOld, err := qc.db.LockKeyRange(ctx, req.SourceID)
@@ -680,10 +681,10 @@ func (qc *qdbCoordinator) Split(ctx context.Context, req *kr.SplitKeyRange) erro
680681
}()
681682

682683
if kr.CmpRangesEqual(req.Bound, krOld.LowerBound) || kr.CmpRangesEqual(req.Bound, krOld.UpperBound) {
683-
return fmt.Errorf("failed to split because bound equals lower or upper bound of the key range")
684+
return spqrerror.New(spqrerror.SPQR_KEYRANGE_ERROR, "failed to split because bound equals lower or upper bound of the key range")
684685
}
685686
if kr.CmpRangesLess(req.Bound, krOld.LowerBound) || !kr.CmpRangesLess(req.Bound, krOld.UpperBound) {
686-
return fmt.Errorf("failed to split because bound is out of key range")
687+
return spqrerror.New(spqrerror.SPQR_KEYRANGE_ERROR, "failed to split because bound is out of key range")
687688
}
688689

689690
krNew := kr.KeyRangeFromDB(
@@ -709,7 +710,7 @@ func (qc *qdbCoordinator) Split(ctx context.Context, req *kr.SplitKeyRange) erro
709710
}
710711

711712
if err := ops.AddKeyRangeWithChecks(ctx, qc.db, krNew); err != nil {
712-
return fmt.Errorf("failed to add a new key range: %w", err)
713+
return spqrerror.Newf(spqrerror.SPQR_KEYRANGE_ERROR, "failed to add a new key range: %s", err.Error())
713714
}
714715

715716
if err := qc.traverseRouters(ctx, func(cc *grpc.ClientConn) error {
@@ -819,23 +820,23 @@ func (qc *qdbCoordinator) Unite(ctx context.Context, uniteKeyRange *kr.UniteKeyR
819820
}()
820821

821822
if krLeft.ShardID != krRight.ShardID {
822-
return fmt.Errorf("failed to unite key ranges routing different shards")
823+
return spqrerror.New(spqrerror.SPQR_KEYRANGE_ERROR, "failed to unite key ranges routing different shards")
823824
}
824825
if !kr.CmpRangesEqual(krLeft.UpperBound, krRight.LowerBound) {
825826
if !kr.CmpRangesEqual(krLeft.LowerBound, krRight.UpperBound) {
826-
return fmt.Errorf("failed to unite non-adjacent key ranges")
827+
return spqrerror.New(spqrerror.SPQR_KEYRANGE_ERROR, "failed to unite non-adjacent key ranges")
827828
}
828829
krLeft, krRight = krRight, krLeft
829830
}
830831

831832
krLeft.UpperBound = krRight.UpperBound
832833

833834
if err := qc.db.DropKeyRange(ctx, krRight.KeyRangeID); err != nil {
834-
return fmt.Errorf("failed to drop an old key range: %w", err)
835+
return spqrerror.Newf(spqrerror.SPQR_KEYRANGE_ERROR, "failed to drop an old key range: %s", err.Error())
835836
}
836837

837838
if err := ops.ModifyKeyRangeWithChecks(ctx, qc.db, kr.KeyRangeFromDB(krLeft)); err != nil {
838-
return fmt.Errorf("failed to update a new key range: %w", err)
839+
return spqrerror.Newf(spqrerror.SPQR_KEYRANGE_ERROR, "failed to update a new key range: %s", err.Error())
839840
}
840841

841842
if err := qc.traverseRouters(ctx, func(cc *grpc.ClientConn) error {
@@ -1054,13 +1055,13 @@ func (qc *qdbCoordinator) RegisterRouter(ctx context.Context, r *topology.Router
10541055
// ping router
10551056
conn, err := DialRouter(r)
10561057
if err != nil {
1057-
return fmt.Errorf("failed to ping router: %s", err)
1058+
return spqrerror.Newf(spqrerror.SPQR_CONNECTION_ERROR, "failed to ping router: %s", err)
10581059
}
10591060
defer conn.Close()
10601061
cl := routerproto.NewTopologyServiceClient(conn)
10611062
_, err = cl.GetRouterStatus(ctx, &routerproto.GetRouterStatusRequest{})
10621063
if err != nil {
1063-
return fmt.Errorf("failed to ping router: %s", err)
1064+
return spqrerror.Newf(spqrerror.SPQR_CONNECTION_ERROR, "failed to ping router: %s", err)
10641065
}
10651066

10661067
return qc.db.AddRouter(ctx, qdb.NewRouter(r.Address, r.ID, qdb.OPENED))
@@ -1157,7 +1158,7 @@ func (qc *qdbCoordinator) ProcClient(ctx context.Context, nconn net.Conn) error
11571158
spqrlog.Zero.Debug().Msg("processed OK")
11581159
}
11591160
default:
1160-
return cli.ReportError(fmt.Errorf("unsupported msg type %T", msg))
1161+
return spqrerror.Newf(spqrerror.SPQR_COMPLEX_QUERY, "unsupported msg type %T", msg)
11611162
}
11621163
}
11631164
}

‎coordinator/provider/keyranges.go

+3-3
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ package provider
33
import (
44
"bytes"
55
"context"
6-
"fmt"
6+
"github.com/pg-sharding/spqr/pkg/models/spqrerror"
77

88
"github.com/pg-sharding/spqr/pkg/spqrlog"
99

@@ -70,7 +70,7 @@ func (c *CoordinatorService) KeyRangeIDByBounds(ctx context.Context, keyRange *p
7070
}
7171
}
7272

73-
return "", fmt.Errorf("key range not found")
73+
return "", spqrerror.New(spqrerror.SPQR_KEYRANGE_ERROR, "key range not found")
7474
}
7575

7676
// TODO : unit tests
@@ -162,7 +162,7 @@ func (c *CoordinatorService) MergeKeyRange(ctx context.Context, request *protos.
162162
}
163163

164164
if err := c.impl.Unite(ctx, uniteKeyRange); err != nil {
165-
return nil, fmt.Errorf("failed to unite key ranges: %w", err)
165+
return nil, spqrerror.Newf(spqrerror.SPQR_KEYRANGE_ERROR, "failed to unite key ranges: %s", err.Error())
166166
}
167167

168168
return &protos.ModifyReply{}, nil

‎coordinator/provider/topology.go

+5-5
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
11
package provider
22

33
import (
4-
context "context"
5-
"fmt"
4+
"context"
5+
"github.com/pg-sharding/spqr/pkg/models/spqrerror"
66

77
"github.com/pg-sharding/spqr/coordinator"
88
protos "github.com/pg-sharding/spqr/pkg/protos"
@@ -16,17 +16,17 @@ type TopologyService struct {
1616

1717
// TODO : implement
1818
func (r *TopologyService) OpenRouter(ctx context.Context, request *protos.OpenRouterRequest) (*protos.OpenRouterReply, error) {
19-
return nil, fmt.Errorf("unimplemented")
19+
return nil, spqrerror.New(spqrerror.SPQR_NOT_IMPLEMENTED, "OpenRouter method unimplemented")
2020
}
2121

2222
// TODO : implement
2323
func (r *TopologyService) CloseRouter(ctx context.Context, request *protos.CloseRouterRequest) (*protos.CloseRouterReply, error) {
24-
return nil, fmt.Errorf("unimplemented")
24+
return nil, spqrerror.New(spqrerror.SPQR_NOT_IMPLEMENTED, "CloseRouter method unimplemented")
2525
}
2626

2727
// TODO : implement
2828
func (r *TopologyService) UpdateCoordinator(ctx context.Context, in *protos.UpdateCoordinatorRequest) (*protos.UpdateCoordinatorResponse, error) {
29-
return nil, fmt.Errorf("unimplemented")
29+
return nil, spqrerror.New(spqrerror.SPQR_NOT_IMPLEMENTED, "UpdateCoordinator method unimplemented")
3030
}
3131

3232
func NewTopologyService(impl coordinator.Coordinator) *TopologyService {

‎pkg/clientinteractor/interactor.go

+4-3
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package clientinteractor
33
import (
44
"context"
55
"fmt"
6+
"github.com/pg-sharding/spqr/pkg/models/spqrerror"
67
"net"
78
"strconv"
89
"strings"
@@ -373,18 +374,18 @@ func MatchRow(row []string, nameToIndex map[string]int, condition spqrparser.Whe
373374
}
374375
return right, nil
375376
default:
376-
return true, fmt.Errorf("not supported logic operation: %s", where.Op)
377+
return true, spqrerror.Newf(spqrerror.SPQR_COMPLEX_QUERY, "not supported logic operation: %s", where.Op)
377378
}
378379
case spqrparser.WhereClauseLeaf:
379380
switch where.Op {
380381
case "=":
381382
i, ok := nameToIndex[where.ColRef.ColName]
382383
if !ok {
383-
return true, fmt.Errorf("column %s not exists", where.ColRef.ColName)
384+
return true, spqrerror.Newf(spqrerror.SPQR_COMPLEX_QUERY, "column %s not exists", where.ColRef.ColName)
384385
}
385386
return row[i] == where.Value, nil
386387
default:
387-
return true, fmt.Errorf("not supported operation %s", where.Op)
388+
return true, spqrerror.Newf(spqrerror.SPQR_COMPLEX_QUERY, "not supported operation %s", where.Op)
388389
}
389390
default:
390391
return false, nil

0 commit comments

Comments
 (0)