Skip to content

Commit 7330a14

Browse files
api: support iproto feature discovery
Since version 2.10.0 Tarantool supports feature discovery [1]. Client can send client protocol version and supported features and receive server protocol version and supported features information to tune its behavior. After this patch, the request will be sent on `dial`, before authentication is performed. Connector stores server info in connection internals. User can also set option RequiredProtocolVersion and RequiredFeatures to fast fail on connect in server (or even client) does not provide some expected feature. Feature check iterates over lists to check if feature is enabled. It seems that iterating over a small list is way faster than building a map, see [2]. Benchmark tests show that this check is rather fast (0.5 ns for both client and server check on HP ProBook 440 G5) so it is not necessary to cache it in any way. Traces of IPROTO_FEATURE_GRACEFUL_SHUTDOWN flag and protocol version 4 could be found in Tarantool source code but they were removed in the following commits before the release and treated like they never existed. We also ignore them here too. See [3] for more info. 1. tarantool/tarantool#6253 2. https://stackoverflow.com/a/52710077/11646599 3. tarantool/tarantool-python#262 Closes #120
1 parent c367122 commit 7330a14

12 files changed

+558
-7
lines changed

CHANGELOG.md

+2
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,8 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
1010

1111
### Added
1212

13+
- Support iproto feature discovery (#120).
14+
1315
### Changed
1416

1517
### Fixed

connection.go

+123
Original file line numberDiff line numberDiff line change
@@ -146,6 +146,11 @@ type Connection struct {
146146
lenbuf [PacketLengthBytes]byte
147147

148148
lastStreamId uint64
149+
150+
clientProtocolVersion ProtocolVersion
151+
clientFeatures []Feature
152+
serverProtocolVersion ProtocolVersion
153+
serverFeatures []Feature
149154
}
150155

151156
var _ = Connector(&Connection{}) // Check compatibility with connector interface.
@@ -269,6 +274,14 @@ type Opts struct {
269274
Transport string
270275
// SslOpts is used only if the Transport == 'ssl' is set.
271276
Ssl SslOpts
277+
// Minimal protocol version that should be supported by
278+
// Go connection client and Tarantool server. By default
279+
// it is equal to tarantool.NoProtocolVersion (no restrictions)
280+
RequiredProtocolVersion ProtocolVersion
281+
// List of features that should be supported by
282+
// Go connection client and Tarantool server. By default
283+
// it is an empty list (no restrictions)
284+
RequiredFeatures []Feature
272285
}
273286

274287
// SslOpts is a way to configure ssl transport.
@@ -351,6 +364,17 @@ func Connect(addr string, opts Opts) (conn *Connection, err error) {
351364
}
352365
}
353366

367+
conn.clientProtocolVersion = ClientProtocolVersion
368+
conn.clientFeatures = ClientFeatures
369+
370+
if err = checkProtocolVersion(opts.RequiredProtocolVersion, conn.clientProtocolVersion); err != nil {
371+
return nil, fmt.Errorf("client: %w", err)
372+
}
373+
374+
if err = checkFeatures(opts.RequiredFeatures, conn.clientFeatures); err != nil {
375+
return nil, fmt.Errorf("client: %w", err)
376+
}
377+
354378
if conn.opts.Logger == nil {
355379
conn.opts.Logger = defaultLogger{}
356380
}
@@ -502,6 +526,23 @@ func (conn *Connection) dial() (err error) {
502526
conn.Greeting.Version = bytes.NewBuffer(greeting[:64]).String()
503527
conn.Greeting.auth = bytes.NewBuffer(greeting[64:108]).String()
504528

529+
// IPROTO_ID requests can be processed without authentication.
530+
// https://www.tarantool.io/en/doc/latest/dev_guide/internals/iproto/requests/#iproto-id
531+
if err = conn.identify(w, r); err != nil {
532+
connection.Close()
533+
return err
534+
}
535+
536+
if err = checkProtocolVersion(opts.RequiredProtocolVersion, conn.serverProtocolVersion); err != nil {
537+
connection.Close()
538+
return fmt.Errorf("server: %w", err)
539+
}
540+
541+
if err = checkFeatures(opts.RequiredFeatures, conn.serverFeatures); err != nil {
542+
connection.Close()
543+
return fmt.Errorf("server: %w", err)
544+
}
545+
505546
// Auth
506547
if opts.User != "" {
507548
scr, err := scramble(conn.Greeting.auth, opts.Pass)
@@ -608,6 +649,18 @@ func (conn *Connection) writeAuthRequest(w *bufio.Writer, scramble []byte) (err
608649
return nil
609650
}
610651

652+
func (conn *Connection) writeIdRequest(w *bufio.Writer, version ProtocolVersion,
653+
features []Feature) (err error) {
654+
req := newIdRequest(version, features)
655+
656+
err = conn.writeRequest(w, req)
657+
if err != nil {
658+
return fmt.Errorf("id: %w", err)
659+
}
660+
661+
return nil
662+
}
663+
611664
func (conn *Connection) readResponse(r io.Reader) (resp Response, err error) {
612665
respBytes, err := conn.read(r)
613666
if err != nil {
@@ -639,6 +692,15 @@ func (conn *Connection) readAuthResponse(r io.Reader) (err error) {
639692
return nil
640693
}
641694

695+
func (conn *Connection) readIdResponse(r io.Reader) (resp Response, err error) {
696+
resp, err = conn.readResponse(r)
697+
if err != nil {
698+
return resp, fmt.Errorf("id: %w", err)
699+
}
700+
701+
return resp, nil
702+
}
703+
642704
func (conn *Connection) createConnection(reconnect bool) (err error) {
643705
var reconnects uint
644706
for conn.c == nil && conn.state == connDisconnected {
@@ -1182,3 +1244,64 @@ func (conn *Connection) NewStream() (*Stream, error) {
11821244
Conn: conn,
11831245
}, nil
11841246
}
1247+
1248+
// identify sends info about client protocol, receives info
1249+
// about server protocol in response and store it in the connection.
1250+
func (conn *Connection) identify(w *bufio.Writer, r *bufio.Reader) error {
1251+
werr := conn.writeIdRequest(w, conn.clientProtocolVersion, conn.clientFeatures)
1252+
if werr != nil {
1253+
return werr
1254+
}
1255+
1256+
resp, rerr := conn.readIdResponse(r)
1257+
if rerr != nil {
1258+
if resp.Code == ErrUnknownRequestType {
1259+
// IPROTO_ID requests are not supported by server.
1260+
conn.serverProtocolVersion = NoProtocolVersion
1261+
conn.serverFeatures = []Feature{}
1262+
1263+
return nil
1264+
}
1265+
1266+
return rerr
1267+
}
1268+
1269+
if len(resp.Data) == 0 {
1270+
return fmt.Errorf("identify: unexpected response: no data")
1271+
}
1272+
1273+
serverIdInfo, ok := resp.Data[0].(idInfo)
1274+
if !ok {
1275+
return fmt.Errorf("identify: unexpected response: wrong data")
1276+
}
1277+
conn.serverProtocolVersion = serverIdInfo.version
1278+
conn.serverFeatures = serverIdInfo.features
1279+
1280+
return nil
1281+
}
1282+
1283+
// ServerProtocolVersion returns protocol version supported by
1284+
// connected Tarantool server.
1285+
// Since 1.10.0
1286+
func (conn *Connection) ServerProtocolVersion() ProtocolVersion {
1287+
return conn.serverProtocolVersion
1288+
}
1289+
1290+
// ClientProtocolVersion returns protocol version supported by
1291+
// Go connection client.
1292+
// Since 1.10.0
1293+
func (conn *Connection) ClientProtocolVersion() ProtocolVersion {
1294+
return conn.clientProtocolVersion
1295+
}
1296+
1297+
// ServerFeatures returns features supported by connected Tarantool server.
1298+
// Since 1.10.0
1299+
func (conn *Connection) ServerFeatures() []Feature {
1300+
return conn.serverFeatures
1301+
}
1302+
1303+
// ClientFeatures returns features supported by Go connection client.
1304+
// Since 1.10.0
1305+
func (conn *Connection) ClientFeatures() []Feature {
1306+
return conn.clientFeatures
1307+
}

connection_pool/example_test.go

+37-3
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,19 @@ func examplePool(roles []bool) (*connection_pool.ConnectionPool, error) {
3232
return connPool, nil
3333
}
3434

35+
func examplePoolWithOpts(roles []bool, connOpts tarantool.Opts) (*connection_pool.ConnectionPool, error) {
36+
err := test_helpers.SetClusterRO(servers, connOpts, roles)
37+
if err != nil {
38+
return nil, fmt.Errorf("ConnectionPool is not established")
39+
}
40+
connPool, err := connection_pool.Connect(servers, connOpts)
41+
if err != nil || connPool == nil {
42+
return nil, fmt.Errorf("ConnectionPool is not established")
43+
}
44+
45+
return connPool, nil
46+
}
47+
3548
func ExampleConnectionPool_Select() {
3649
pool, err := examplePool(testRoles)
3750
if err != nil {
@@ -586,7 +599,14 @@ func ExampleCommitRequest() {
586599
return
587600
}
588601

589-
pool, err := examplePool(testRoles)
602+
// Assert that server supports expected features
603+
txnOpts := connOpts
604+
txnOpts.RequiredFeatures = []tarantool.Feature{
605+
tarantool.StreamsFeature,
606+
tarantool.TransactionsFeature,
607+
}
608+
609+
pool, err := examplePoolWithOpts(testRoles, txnOpts)
590610
if err != nil {
591611
fmt.Println(err)
592612
return
@@ -672,8 +692,15 @@ func ExampleRollbackRequest() {
672692
return
673693
}
674694

695+
// Assert that server supports expected features
696+
txnOpts := connOpts
697+
txnOpts.RequiredFeatures = []tarantool.Feature{
698+
tarantool.StreamsFeature,
699+
tarantool.TransactionsFeature,
700+
}
701+
675702
// example pool has only one rw instance
676-
pool, err := examplePool(testRoles)
703+
pool, err := examplePoolWithOpts(testRoles, txnOpts)
677704
if err != nil {
678705
fmt.Println(err)
679706
return
@@ -758,8 +785,15 @@ func ExampleBeginRequest_TxnIsolation() {
758785
return
759786
}
760787

788+
// Assert that server supports expected features
789+
txnOpts := connOpts
790+
txnOpts.RequiredFeatures = []tarantool.Feature{
791+
tarantool.StreamsFeature,
792+
tarantool.TransactionsFeature,
793+
}
794+
761795
// example pool has only one rw instance
762-
pool, err := examplePool(testRoles)
796+
pool, err := examplePoolWithOpts(testRoles, txnOpts)
763797
if err != nil {
764798
fmt.Println(err)
765799
return

const.go

+3
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ const (
1818
RollbackRequestCode = 16
1919
PingRequestCode = 64
2020
SubscribeRequestCode = 66
21+
IdRequestCode = 73
2122

2223
KeyCode = 0x00
2324
KeySync = 0x01
@@ -41,6 +42,8 @@ const (
4142
KeySQLBind = 0x41
4243
KeySQLInfo = 0x42
4344
KeyStmtID = 0x43
45+
KeyVersion = 0x54
46+
KeyFeatures = 0x55
4447
KeyTimeout = 0x56
4548
KeyTxnIsolation = 0x59
4649

example_test.go

+32-3
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,14 @@ func example_connect() *tarantool.Connection {
2626
return conn
2727
}
2828

29+
func example_connect_with_opts(opts tarantool.Opts) *tarantool.Connection {
30+
conn, err := tarantool.Connect(server, opts)
31+
if err != nil {
32+
panic("Connection is not established: " + err.Error())
33+
}
34+
return conn
35+
}
36+
2937
// Example demonstrates how to use SSL transport.
3038
func ExampleSslOpts() {
3139
var opts = tarantool.Opts{
@@ -331,7 +339,14 @@ func ExampleCommitRequest() {
331339
return
332340
}
333341

334-
conn := example_connect()
342+
// Assert that server supports expected features
343+
txnOpts := opts
344+
txnOpts.RequiredFeatures = []tarantool.Feature{
345+
tarantool.StreamsFeature,
346+
tarantool.TransactionsFeature,
347+
}
348+
349+
conn := example_connect_with_opts(txnOpts)
335350
defer conn.Close()
336351

337352
stream, _ := conn.NewStream()
@@ -407,7 +422,14 @@ func ExampleRollbackRequest() {
407422
return
408423
}
409424

410-
conn := example_connect()
425+
// Assert that server supports expected features
426+
txnOpts := opts
427+
txnOpts.RequiredFeatures = []tarantool.Feature{
428+
tarantool.StreamsFeature,
429+
tarantool.TransactionsFeature,
430+
}
431+
432+
conn := example_connect_with_opts(txnOpts)
411433
defer conn.Close()
412434

413435
stream, _ := conn.NewStream()
@@ -483,7 +505,14 @@ func ExampleBeginRequest_TxnIsolation() {
483505
return
484506
}
485507

486-
conn := example_connect()
508+
// Assert that server supports expected features
509+
txnOpts := opts
510+
txnOpts.RequiredFeatures = []tarantool.Feature{
511+
tarantool.StreamsFeature,
512+
tarantool.TransactionsFeature,
513+
}
514+
515+
conn := example_connect_with_opts(txnOpts)
487516
defer conn.Close()
488517

489518
stream, _ := conn.NewStream()

export_test.go

+6
Original file line numberDiff line numberDiff line change
@@ -111,6 +111,12 @@ func RefImplRollbackBody(enc *encoder) error {
111111
return fillRollback(enc)
112112
}
113113

114+
// RefImplIdBody is reference implementation for filling of an id
115+
// request's body.
116+
func RefImplIdBody(enc *encoder, version ProtocolVersion, features []Feature) error {
117+
return fillId(enc, version, features)
118+
}
119+
114120
func NewEncoder(w io.Writer) *encoder {
115121
return newEncoder(w)
116122
}

0 commit comments

Comments
 (0)