Skip to content

Commit d1ac5aa

Browse files
dmygerdmyger
committed
aeon: connect implementation
Implement aeon console connection. The ‘Console’ module has been separate from the ‘Connect’ abstraction, to allow it being used independently of the transport layer. Closes #1050 @TarantoolBot document Title: Implement aeon console connection. Command allow connect to Enterprise Aeon database with specified URL. Available command options: - `sslkeyfile <private_key>` - path to private part of certificate. - `sslcertfile <pub_cert>` - path to public part of certificate. - `sslcafile <ca_file>` - path to root CA for self-signed certificate. - `transport [ssl|plain]` - connection mode.
1 parent e844f27 commit d1ac5aa

14 files changed

+897
-23
lines changed

‎CHANGELOG.md

+1
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
2323
* `sslcertfile` - path to an SSL certificate file,
2424
* `sslcafile` - path to a trusted certificate authorities (CA) file,
2525
* `sslciphers` - colon-separated list of SSL cipher suites the connection.
26+
- `tt aeon connect`: add support to connect Aeon database.
2627

2728
### Changed
2829

‎cli/aeon/client.go

+234
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,234 @@
1+
package aeon
2+
3+
import (
4+
"context"
5+
"crypto/tls"
6+
"crypto/x509"
7+
"errors"
8+
"fmt"
9+
"os"
10+
"strings"
11+
"time"
12+
13+
"github.com/apex/log"
14+
15+
"github.com/tarantool/go-prompt"
16+
"github.com/tarantool/tt/cli/aeon/cmd"
17+
"github.com/tarantool/tt/cli/aeon/pb"
18+
"github.com/tarantool/tt/cli/connector"
19+
"github.com/tarantool/tt/cli/console"
20+
"github.com/tarantool/tt/cli/formatter"
21+
"google.golang.org/grpc"
22+
"google.golang.org/grpc/credentials"
23+
"google.golang.org/grpc/credentials/insecure"
24+
)
25+
26+
// Client structure with parameters for gRPC connection to Aeon.
27+
type Client struct {
28+
title string
29+
conn *grpc.ClientConn
30+
client pb.SQLServiceClient
31+
}
32+
33+
func makeAddress(ctx cmd.ConnectCtx) string {
34+
if ctx.Network == connector.UnixNetwork {
35+
if strings.HasPrefix(ctx.Address, "@") {
36+
return "unix-abstract:" + (ctx.Address)[1:]
37+
}
38+
return "unix:" + ctx.Address
39+
}
40+
return ctx.Address
41+
}
42+
43+
func getCertificate(args cmd.Ssl) (tls.Certificate, error) {
44+
if args.CertFile == "" && args.KeyFile == "" {
45+
return tls.Certificate{}, nil
46+
}
47+
tls_cert, err := tls.LoadX509KeyPair(args.CertFile, args.KeyFile)
48+
if err != nil {
49+
return tls_cert, fmt.Errorf("could not load client key pair: %w", err)
50+
}
51+
return tls_cert, nil
52+
}
53+
54+
func getTlsConfig(args cmd.Ssl) (*tls.Config, error) {
55+
if args.CaFile == "" {
56+
return &tls.Config{
57+
ClientAuth: tls.NoClientCert,
58+
}, nil
59+
}
60+
61+
ca, err := os.ReadFile(args.CaFile)
62+
if err != nil {
63+
return nil, fmt.Errorf("failed to read CA file: %w", err)
64+
}
65+
certPool := x509.NewCertPool()
66+
if !certPool.AppendCertsFromPEM(ca) {
67+
return nil, errors.New("failed to append CA data")
68+
}
69+
cert, err := getCertificate(args)
70+
if err != nil {
71+
return nil, fmt.Errorf("failed get certificate: %w", err)
72+
}
73+
return &tls.Config{
74+
Certificates: []tls.Certificate{cert},
75+
ClientAuth: tls.RequireAndVerifyClientCert,
76+
RootCAs: certPool,
77+
}, nil
78+
}
79+
80+
func getDialOpts(ctx cmd.ConnectCtx) (grpc.DialOption, error) {
81+
var creds credentials.TransportCredentials
82+
if ctx.Transport == cmd.TransportSsl {
83+
config, err := getTlsConfig(ctx.Ssl)
84+
if err != nil {
85+
return nil, fmt.Errorf("not tls config: %w", err)
86+
}
87+
creds = credentials.NewTLS(config)
88+
} else {
89+
creds = insecure.NewCredentials()
90+
}
91+
return grpc.WithTransportCredentials(creds), nil
92+
}
93+
94+
// NewAeonHandler create new grpc connection to Aeon server.
95+
func NewAeonHandler(ctx cmd.ConnectCtx) (*Client, error) {
96+
c := Client{title: ctx.Address}
97+
target := makeAddress(ctx)
98+
// var err error
99+
opt, err := getDialOpts(ctx)
100+
if err != nil {
101+
return nil, fmt.Errorf("%w", err)
102+
}
103+
c.conn, err = grpc.NewClient(target, opt)
104+
if err != nil {
105+
return nil, fmt.Errorf("fail to dial: %w", err)
106+
}
107+
if err := c.ping(); err == nil {
108+
log.Infof("Aeon responses at %q", target)
109+
} else {
110+
return nil, fmt.Errorf("can't ping to Aeon at %q: %w", target, err)
111+
}
112+
113+
c.client = pb.NewSQLServiceClient(c.conn)
114+
return &c, nil
115+
}
116+
117+
func (c *Client) ping() error {
118+
log.Infof("Start ping aeon server")
119+
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
120+
defer cancel()
121+
122+
diag := pb.NewDiagServiceClient(c.conn)
123+
_, err := diag.Ping(ctx, &pb.PingRequest{})
124+
if err != nil {
125+
log.Warnf("Aeon ping %s", err)
126+
}
127+
return err
128+
}
129+
130+
// Title implements console.Handler interface.
131+
func (c *Client) Title() string {
132+
return c.title
133+
}
134+
135+
// Validate implements console.Handler interface.
136+
func (c *Client) Validate(input string) bool {
137+
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
138+
defer cancel()
139+
140+
check, err := c.client.SQLCheck(ctx, &pb.SQLRequest{Query: input})
141+
if err != nil {
142+
log.Warnf("Aeon validate %s\nFor request: %q", err, input)
143+
return false
144+
}
145+
146+
return check.Status == pb.SQLCheckStatus_SQL_QUERY_VALID
147+
}
148+
149+
// Execute implements console.Handler interface.
150+
func (c *Client) Execute(input string) any {
151+
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
152+
defer cancel()
153+
154+
resp, err := c.client.SQL(ctx, &pb.SQLRequest{Query: input})
155+
if err != nil {
156+
return err
157+
}
158+
return parseSQLResponse(resp)
159+
}
160+
161+
// Stop implements console.Handler interface.
162+
func (c *Client) Close() {
163+
c.conn.Close()
164+
}
165+
166+
// Complete implements console.Handler interface.
167+
func (c *Client) Complete(input prompt.Document) []prompt.Suggest {
168+
// TODO: waiting until there is support from Aeon side.
169+
return nil
170+
}
171+
172+
// parseSQLResponse returns result as table in map.
173+
// Where keys is name of columns. And body is array of values.
174+
// On any issue return an error.
175+
func parseSQLResponse(resp *pb.SQLResponse) any {
176+
if resp.Error != nil {
177+
return ResultError{resp.Error}
178+
}
179+
if resp.TupleFormat == nil {
180+
return ResultType{}
181+
}
182+
res := ResultType{
183+
data: make(map[string][]any, len(resp.TupleFormat.Names)),
184+
count: len(resp.Tuples),
185+
}
186+
// result := make(ResultType, len(resp.TupleFormat.Names))
187+
rows := len(resp.Tuples)
188+
for _, f := range resp.TupleFormat.Names {
189+
res.data[f] = make([]any, 0, rows)
190+
}
191+
192+
for r, row := range resp.Tuples {
193+
for i, v := range row.Fields {
194+
k := resp.TupleFormat.Names[i]
195+
val, err := decodeValue(v)
196+
if err != nil {
197+
return fmt.Errorf("tuple %d can't decode %s: %w", r, v.String(), err)
198+
}
199+
res.data[k] = append(res.data[k], val)
200+
}
201+
}
202+
return res
203+
}
204+
205+
// asYaml prepare results for formatter.MakeOutput.
206+
func (r ResultType) asYaml() string {
207+
yaml := "---\n"
208+
for i := range r.count {
209+
mark := "-"
210+
for k, v := range r.data {
211+
if i < len(v) {
212+
yaml += fmt.Sprintf("%s %s: %v\n", mark, k, v[i])
213+
mark = " "
214+
}
215+
}
216+
}
217+
return yaml
218+
}
219+
220+
// Format produce formatted string according required console.Format settings.
221+
func (r ResultType) Format(f console.Format) string {
222+
if r.count == 0 {
223+
return ""
224+
}
225+
output, err := formatter.MakeOutput(f.Mode, r.asYaml(), f.Opts)
226+
if err != nil {
227+
return fmt.Sprintf("can't format output: %s;\nResults:\n%v", err, r)
228+
}
229+
return output
230+
}
231+
232+
func (e *ResultError) Format(_ console.Format) string {
233+
return fmt.Sprintf("---\nError: %s\n%q", e.Name, e.Msg)
234+
}

‎cli/aeon/cmd/connect.go

+4
Original file line numberDiff line numberDiff line change
@@ -16,4 +16,8 @@ type ConnectCtx struct {
1616
Ssl Ssl
1717
// Transport is a connection mode.
1818
Transport Transport
19+
// Network is kind of transport layer.
20+
Network string
21+
// Address is a connection URL, unix socket address and etc.
22+
Address string
1923
}

‎cli/aeon/decode.go

+100
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,100 @@
1+
package aeon
2+
3+
import (
4+
"fmt"
5+
"time"
6+
7+
"github.com/google/uuid"
8+
"github.com/tarantool/go-tarantool/v2/datetime"
9+
"github.com/tarantool/go-tarantool/v2/decimal"
10+
"github.com/tarantool/tt/cli/aeon/pb"
11+
)
12+
13+
// decodeValue convert a value obtained from protobuf into a value that can be used as an
14+
// argument to Tarantool functions.
15+
//
16+
// Copy from https://github.com/tarantool/aeon/blob/master/aeon/grpc/server/pb/decode.go
17+
func decodeValue(val *pb.Value) (any, error) {
18+
switch casted := val.Kind.(type) {
19+
case *pb.Value_UnsignedValue:
20+
return val.GetUnsignedValue(), nil
21+
case *pb.Value_StringValue:
22+
return val.GetStringValue(), nil
23+
case *pb.Value_NumberValue:
24+
return val.GetNumberValue(), nil
25+
case *pb.Value_IntegerValue:
26+
return val.GetIntegerValue(), nil
27+
case *pb.Value_BooleanValue:
28+
return val.GetBooleanValue(), nil
29+
case *pb.Value_VarbinaryValue:
30+
return val.GetVarbinaryValue(), nil
31+
case *pb.Value_DecimalValue:
32+
decStr := val.GetDecimalValue()
33+
res, err := decimal.MakeDecimalFromString(decStr)
34+
if err != nil {
35+
return nil, err
36+
}
37+
return res, nil
38+
case *pb.Value_UuidValue:
39+
uuidStr := val.GetUuidValue()
40+
res, err := uuid.Parse(uuidStr)
41+
if err != nil {
42+
return nil, err
43+
}
44+
return res, nil
45+
case *pb.Value_DatetimeValue:
46+
sec := casted.DatetimeValue.Seconds
47+
nsec := casted.DatetimeValue.Nsec
48+
t := time.Unix(sec, nsec)
49+
if len(casted.DatetimeValue.Location) > 0 {
50+
locStr := casted.DatetimeValue.Location
51+
loc, err := time.LoadLocation(locStr)
52+
if err != nil {
53+
return nil, err
54+
}
55+
t = t.In(loc)
56+
}
57+
res, err := datetime.MakeDatetime(t)
58+
if err != nil {
59+
return nil, err
60+
}
61+
return res, nil
62+
case *pb.Value_IntervalValue:
63+
res := datetime.Interval{
64+
Year: casted.IntervalValue.Year,
65+
Month: casted.IntervalValue.Month,
66+
Week: casted.IntervalValue.Week,
67+
Day: casted.IntervalValue.Day,
68+
Hour: casted.IntervalValue.Hour,
69+
Min: casted.IntervalValue.Min,
70+
Sec: casted.IntervalValue.Sec,
71+
Nsec: casted.IntervalValue.Nsec,
72+
Adjust: datetime.Adjust(casted.IntervalValue.Adjust)}
73+
return res, nil
74+
case *pb.Value_ArrayValue:
75+
array := val.GetArrayValue()
76+
res := make([]any, len(array.Fields))
77+
for k, v := range array.Fields {
78+
field, err := decodeValue(v)
79+
if err != nil {
80+
return nil, err
81+
}
82+
res[k] = field
83+
}
84+
return res, nil
85+
case *pb.Value_MapValue:
86+
res := make(map[any]any, len(casted.MapValue.Fields))
87+
for k, v := range casted.MapValue.Fields {
88+
item, err := decodeValue(v)
89+
if err != nil {
90+
return nil, err
91+
}
92+
res[k] = item
93+
}
94+
return res, nil
95+
case *pb.Value_NullValue:
96+
return nil, nil
97+
default:
98+
return nil, fmt.Errorf("unsupported type for value")
99+
}
100+
}

‎cli/aeon/results.go

+14
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
package aeon
2+
3+
import "github.com/tarantool/tt/cli/aeon/pb"
4+
5+
// ResultType is a custom type to format output with console.Formatter interface.
6+
type ResultType struct {
7+
data map[string][]any
8+
count int
9+
}
10+
11+
// ResultError wraps pb.Error to implement console.Formatter interface.
12+
type ResultError struct {
13+
*pb.Error
14+
}

0 commit comments

Comments
 (0)