This repository has been archived by the owner on Sep 21, 2022. It is now read-only.
forked from vitessio/vitess
-
Notifications
You must be signed in to change notification settings - Fork 1
/
vtgate.go
453 lines (390 loc) · 16.6 KB
/
vtgate.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
/*
Copyright 2019 The Vitess Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// Package vtgate provides query routing rpc services
// for vttablets.
package vtgate
import (
"flag"
"fmt"
"net/http"
"os"
"strings"
"time"
"golang.org/x/net/context"
"vitess.io/vitess/go/acl"
"vitess.io/vitess/go/sqltypes"
"vitess.io/vitess/go/stats"
"vitess.io/vitess/go/tb"
"vitess.io/vitess/go/vt/discovery"
"vitess.io/vitess/go/vt/key"
"vitess.io/vitess/go/vt/log"
"vitess.io/vitess/go/vt/logutil"
"vitess.io/vitess/go/vt/servenv"
"vitess.io/vitess/go/vt/sqlparser"
"vitess.io/vitess/go/vt/srvtopo"
"vitess.io/vitess/go/vt/topo/topoproto"
"vitess.io/vitess/go/vt/vterrors"
"vitess.io/vitess/go/vt/vtgate/vtgateservice"
binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
querypb "vitess.io/vitess/go/vt/proto/query"
topodatapb "vitess.io/vitess/go/vt/proto/topodata"
vtgatepb "vitess.io/vitess/go/vt/proto/vtgate"
vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
)
var (
transactionMode = flag.String("transaction_mode", "MULTI", "SINGLE: disallow multi-db transactions, MULTI: allow multi-db transactions with best effort commit, TWOPC: allow multi-db transactions with 2pc commit")
normalizeQueries = flag.Bool("normalize_queries", true, "Rewrite queries with bind vars. Turn this off if the app itself sends normalized queries with bind vars.")
terseErrors = flag.Bool("vtgate-config-terse-errors", false, "prevent bind vars from escaping in returned errors")
streamBufferSize = flag.Int("stream_buffer_size", 32*1024, "the number of bytes sent from vtgate for each stream call. It's recommended to keep this value in sync with vttablet's query-server-config-stream-buffer-size.")
queryPlanCacheSize = flag.Int64("gate_query_cache_size", 10000, "gate server query cache size, maximum number of queries to be cached. vtgate analyzes every incoming query and generate a query plan, these plans are being cached in a lru cache. This config controls the capacity of the lru cache.")
_ = flag.Bool("disable_local_gateway", false, "deprecated: if specified, this process will not route any queries to local tablets in the local cell")
maxMemoryRows = flag.Int("max_memory_rows", 300000, "Maximum number of rows that will be held in memory for intermediate results as well as the final result.")
warnMemoryRows = flag.Int("warn_memory_rows", 30000, "Warning threshold for in-memory results. A row count higher than this amount will cause the VtGateWarnings.ResultsExceeded counter to be incremented.")
)
func getTxMode() vtgatepb.TransactionMode {
switch strings.ToLower(*transactionMode) {
case "single":
log.Infof("Transaction mode: '%s'", *transactionMode)
return vtgatepb.TransactionMode_SINGLE
case "multi":
log.Infof("Transaction mode: '%s'", *transactionMode)
return vtgatepb.TransactionMode_MULTI
case "twopc":
log.Infof("Transaction mode: '%s'", *transactionMode)
return vtgatepb.TransactionMode_TWOPC
default:
fmt.Printf("Invalid option: %v\n", *transactionMode)
fmt.Println("Usage: -transaction_mode {SINGLE | MULTI | TWOPC}")
os.Exit(1)
return -1
}
}
var (
rpcVTGate *VTGate
vschemaCounters *stats.CountersWithSingleLabel
// Error counters should be global so they can be set from anywhere
errorCounts *stats.CountersWithMultiLabels
warnings *stats.CountersWithSingleLabel
)
// VTGate is the rpc interface to vtgate. Only one instance
// can be created. It implements vtgateservice.VTGateService
// VTGate exposes multiple generations of interfaces.
type VTGate struct {
// Dependency: executor->resolver->scatterConn->txConn->gateway.
executor *Executor
resolver *Resolver
vsm *vstreamManager
txConn *TxConn
gw Gateway
// stats objects.
// TODO(sougou): This needs to be cleaned up. There
// are global vars that depend on this member var.
timings *stats.MultiTimings
rowsReturned *stats.CountersWithMultiLabels
// the throttled loggers for all errors, one per API entry
logExecute *logutil.ThrottledLogger
logStreamExecute *logutil.ThrottledLogger
}
// RegisterVTGate defines the type of registration mechanism.
type RegisterVTGate func(vtgateservice.VTGateService)
// RegisterVTGates stores register funcs for VTGate server.
var RegisterVTGates []RegisterVTGate
// Init initializes VTGate server.
func Init(ctx context.Context, hc discovery.HealthCheck, serv srvtopo.Server, cell string, retryCount int, tabletTypesToWait []topodatapb.TabletType) *VTGate {
if rpcVTGate != nil {
log.Fatalf("VTGate already initialized")
}
// vschemaCounters needs to be initialized before planner to
// catch the initial load stats.
vschemaCounters = stats.NewCountersWithSingleLabel("VtgateVSchemaCounts", "Vtgate vschema counts", "changes")
// Build objects from low to high level.
// Start with the gateway. If we can't reach the topology service,
// we can't go on much further, so we log.Fatal out.
gw := GatewayCreator()(ctx, hc, serv, cell, retryCount)
gw.RegisterStats()
if err := WaitForTablets(gw, tabletTypesToWait); err != nil {
log.Fatalf("gateway.WaitForTablets failed: %v", err)
}
// If we want to filter keyspaces replace the srvtopo.Server with a
// filtering server
if len(KeyspacesToWatch) > 0 {
log.Infof("Keyspace filtering enabled, selecting %v", KeyspacesToWatch)
var err error
serv, err = srvtopo.NewKeyspaceFilteringServer(serv, KeyspacesToWatch)
if err != nil {
log.Fatalf("Unable to construct SrvTopo server: %v", err.Error())
}
}
tc := NewTxConn(gw, getTxMode())
// ScatterConn depends on TxConn to perform forced rollbacks.
sc := NewScatterConn("VttabletCall", tc, gw, hc)
srvResolver := srvtopo.NewResolver(serv, gw, cell)
resolver := NewResolver(srvResolver, serv, cell, sc)
vsm := newVStreamManager(srvResolver, serv, cell)
rpcVTGate = &VTGate{
executor: NewExecutor(ctx, serv, cell, resolver, *normalizeQueries, *streamBufferSize, *queryPlanCacheSize),
resolver: resolver,
vsm: vsm,
txConn: tc,
gw: gw,
timings: stats.NewMultiTimings(
"VtgateApi",
"VtgateApi timings",
[]string{"Operation", "Keyspace", "DbType"}),
rowsReturned: stats.NewCountersWithMultiLabels(
"VtgateApiRowsReturned",
"Rows returned through the VTgate API",
[]string{"Operation", "Keyspace", "DbType"}),
logExecute: logutil.NewThrottledLogger("Execute", 5*time.Second),
logStreamExecute: logutil.NewThrottledLogger("StreamExecute", 5*time.Second),
}
errorCounts = stats.NewCountersWithMultiLabels("VtgateApiErrorCounts", "Vtgate API error counts per error type", []string{"Operation", "Keyspace", "DbType", "Code"})
_ = stats.NewRates("QPSByOperation", stats.CounterForDimension(rpcVTGate.timings, "Operation"), 15, 1*time.Minute)
_ = stats.NewRates("QPSByKeyspace", stats.CounterForDimension(rpcVTGate.timings, "Keyspace"), 15, 1*time.Minute)
_ = stats.NewRates("QPSByDbType", stats.CounterForDimension(rpcVTGate.timings, "DbType"), 15*60/5, 5*time.Second)
_ = stats.NewRates("ErrorsByOperation", stats.CounterForDimension(errorCounts, "Operation"), 15, 1*time.Minute)
_ = stats.NewRates("ErrorsByKeyspace", stats.CounterForDimension(errorCounts, "Keyspace"), 15, 1*time.Minute)
_ = stats.NewRates("ErrorsByDbType", stats.CounterForDimension(errorCounts, "DbType"), 15, 1*time.Minute)
_ = stats.NewRates("ErrorsByCode", stats.CounterForDimension(errorCounts, "Code"), 15, 1*time.Minute)
warnings = stats.NewCountersWithSingleLabel("VtGateWarnings", "Vtgate warnings", "type", "IgnoredSet", "ResultsExceeded")
servenv.OnRun(func() {
for _, f := range RegisterVTGates {
f(rpcVTGate)
}
})
rpcVTGate.registerDebugHealthHandler()
err := initQueryLogger(rpcVTGate)
if err != nil {
log.Fatalf("error initializing query logger: %v", err)
}
initAPI(ctx, hc)
return rpcVTGate
}
func (vtg *VTGate) registerDebugHealthHandler() {
http.HandleFunc("/debug/health", func(w http.ResponseWriter, r *http.Request) {
if err := acl.CheckAccessHTTP(r, acl.MONITORING); err != nil {
acl.SendError(w, err)
return
}
w.Header().Set("Content-Type", "text/plain")
if err := vtg.IsHealthy(); err != nil {
w.Write([]byte("not ok"))
return
}
w.Write([]byte("ok"))
})
}
// IsHealthy returns nil if server is healthy.
// Otherwise, it returns an error indicating the reason.
func (vtg *VTGate) IsHealthy() error {
return nil
}
// Gateway returns the current gateway implementation. Mostly used for tests.
func (vtg *VTGate) Gateway() Gateway {
return vtg.gw
}
// Execute executes a non-streaming query. This is a V3 function.
func (vtg *VTGate) Execute(ctx context.Context, session *vtgatepb.Session, sql string, bindVariables map[string]*querypb.BindVariable) (newSession *vtgatepb.Session, qr *sqltypes.Result, err error) {
// In this context, we don't care if we can't fully parse destination
destKeyspace, destTabletType, _, _ := vtg.executor.ParseDestinationTarget(session.TargetString)
statsKey := []string{"Execute", destKeyspace, topoproto.TabletTypeLString(destTabletType)}
defer vtg.timings.Record(statsKey, time.Now())
if bvErr := sqltypes.ValidateBindVariables(bindVariables); bvErr != nil {
err = vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "%v", bvErr)
goto handleError
}
qr, err = vtg.executor.Execute(ctx, "Execute", NewSafeSession(session), sql, bindVariables)
if err == nil {
vtg.rowsReturned.Add(statsKey, int64(len(qr.Rows)))
return session, qr, nil
}
handleError:
query := map[string]interface{}{
"Sql": sql,
"BindVariables": bindVariables,
"Session": session,
}
err = recordAndAnnotateError(err, statsKey, query, vtg.logExecute)
return session, nil, err
}
// ExecuteBatch executes a batch of queries. This is a V3 function.
func (vtg *VTGate) ExecuteBatch(ctx context.Context, session *vtgatepb.Session, sqlList []string, bindVariablesList []map[string]*querypb.BindVariable) (*vtgatepb.Session, []sqltypes.QueryResponse, error) {
// In this context, we don't care if we can't fully parse destination
destKeyspace, destTabletType, _, _ := vtg.executor.ParseDestinationTarget(session.TargetString)
statsKey := []string{"ExecuteBatch", destKeyspace, topoproto.TabletTypeLString(destTabletType)}
defer vtg.timings.Record(statsKey, time.Now())
for _, bindVariables := range bindVariablesList {
if bvErr := sqltypes.ValidateBindVariables(bindVariables); bvErr != nil {
return session, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "%v", bvErr)
}
}
qrl := make([]sqltypes.QueryResponse, len(sqlList))
for i, sql := range sqlList {
var bv map[string]*querypb.BindVariable
if len(bindVariablesList) != 0 {
bv = bindVariablesList[i]
}
session, qrl[i].QueryResult, qrl[i].QueryError = vtg.Execute(ctx, session, sql, bv)
if qr := qrl[i].QueryResult; qr != nil {
vtg.rowsReturned.Add(statsKey, int64(len(qr.Rows)))
}
}
return session, qrl, nil
}
// StreamExecute executes a streaming query. This is a V3 function.
// Note we guarantee the callback will not be called concurrently
// by multiple go routines.
func (vtg *VTGate) StreamExecute(ctx context.Context, session *vtgatepb.Session, sql string, bindVariables map[string]*querypb.BindVariable, callback func(*sqltypes.Result) error) error {
// In this context, we don't care if we can't fully parse destination
destKeyspace, destTabletType, dest, _ := vtg.executor.ParseDestinationTarget(session.TargetString)
statsKey := []string{"StreamExecute", destKeyspace, topoproto.TabletTypeLString(destTabletType)}
defer vtg.timings.Record(statsKey, time.Now())
var err error
if bvErr := sqltypes.ValidateBindVariables(bindVariables); bvErr != nil {
err = vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "%v", bvErr)
goto handleError
}
// TODO: This could be simplified to have a StreamExecute that takes
// a destTarget without explicit destination.
switch dest.(type) {
case key.DestinationShard:
err = vtg.resolver.StreamExecute(
ctx,
sql,
bindVariables,
destKeyspace,
destTabletType,
dest,
session.Options,
func(reply *sqltypes.Result) error {
vtg.rowsReturned.Add(statsKey, int64(len(reply.Rows)))
return callback(reply)
})
default:
err = vtg.executor.StreamExecute(
ctx,
"StreamExecute",
NewSafeSession(session),
sql,
bindVariables,
querypb.Target{
Keyspace: destKeyspace,
TabletType: destTabletType,
},
func(reply *sqltypes.Result) error {
vtg.rowsReturned.Add(statsKey, int64(len(reply.Rows)))
return callback(reply)
})
}
handleError:
if err != nil {
query := map[string]interface{}{
"Sql": sql,
"BindVariables": bindVariables,
"Session": session,
}
return recordAndAnnotateError(err, statsKey, query, vtg.logStreamExecute)
}
return nil
}
// ResolveTransaction resolves the specified 2PC transaction.
func (vtg *VTGate) ResolveTransaction(ctx context.Context, dtid string) error {
return formatError(vtg.txConn.Resolve(ctx, dtid))
}
// Prepare supports non-streaming prepare statement query with multi shards
func (vtg *VTGate) Prepare(ctx context.Context, session *vtgatepb.Session, sql string, bindVariables map[string]*querypb.BindVariable) (newSession *vtgatepb.Session, fld []*querypb.Field, err error) {
// In this context, we don't care if we can't fully parse destination
destKeyspace, destTabletType, _, _ := vtg.executor.ParseDestinationTarget(session.TargetString)
statsKey := []string{"Execute", destKeyspace, topoproto.TabletTypeLString(destTabletType)}
defer vtg.timings.Record(statsKey, time.Now())
if bvErr := sqltypes.ValidateBindVariables(bindVariables); bvErr != nil {
err = vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "%v", bvErr)
goto handleError
}
fld, err = vtg.executor.Prepare(ctx, "Prepare", NewSafeSession(session), sql, bindVariables)
if err == nil {
vtg.rowsReturned.Add(statsKey, int64(len(fld)))
return session, fld, nil
}
handleError:
query := map[string]interface{}{
"Sql": sql,
"BindVariables": bindVariables,
"Session": session,
}
err = recordAndAnnotateError(err, statsKey, query, vtg.logExecute)
return session, nil, err
}
// VStream streams binlog events.
func (vtg *VTGate) VStream(ctx context.Context, tabletType topodatapb.TabletType, vgtid *binlogdatapb.VGtid, filter *binlogdatapb.Filter, send func([]*binlogdatapb.VEvent) error) error {
return vtg.vsm.VStream(ctx, tabletType, vgtid, filter, send)
}
// GetGatewayCacheStatus returns a displayable version of the Gateway cache.
func (vtg *VTGate) GetGatewayCacheStatus() TabletCacheStatusList {
return vtg.resolver.GetGatewayCacheStatus()
}
// VSchemaStats returns the loaded vschema stats.
func (vtg *VTGate) VSchemaStats() *VSchemaStats {
return vtg.executor.VSchemaStats()
}
func truncateErrorStrings(data map[string]interface{}) map[string]interface{} {
ret := map[string]interface{}{}
if *terseErrors {
// request might have PII information. Return an empty map
return ret
}
for key, val := range data {
mapVal, ok := val.(map[string]interface{})
if ok {
ret[key] = truncateErrorStrings(mapVal)
} else {
strVal := fmt.Sprintf("%v", val)
ret[key] = sqlparser.TruncateForLog(strVal)
}
}
return ret
}
func recordAndAnnotateError(err error, statsKey []string, request map[string]interface{}, logger *logutil.ThrottledLogger) error {
ec := vterrors.Code(err)
fullKey := []string{
statsKey[0],
statsKey[1],
statsKey[2],
ec.String(),
}
// Traverse the request structure and truncate any long values
request = truncateErrorStrings(request)
errorCounts.Add(fullKey, 1)
// Most errors are not logged by vtgate because they're either too spammy or logged elsewhere.
switch ec {
case vtrpcpb.Code_UNKNOWN, vtrpcpb.Code_INTERNAL, vtrpcpb.Code_DATA_LOSS:
logger.Errorf("%v, request: %+v", err, request)
case vtrpcpb.Code_UNAVAILABLE:
logger.Infof("%v, request: %+v", err, request)
}
return vterrors.Wrapf(err, "vtgate: %s", servenv.ListeningURL.String())
}
func formatError(err error) error {
if err == nil {
return nil
}
return vterrors.Wrapf(err, "vtgate: %s", servenv.ListeningURL.String())
}
// HandlePanic recovers from panics, and logs / increment counters
func (vtg *VTGate) HandlePanic(err *error) {
if x := recover(); x != nil {
log.Errorf("Uncaught panic:\n%v\n%s", x, tb.Stack(4))
*err = fmt.Errorf("uncaught panic: %v, vtgate: %v", x, servenv.ListeningURL.String())
errorCounts.Add([]string{"Panic", "Unknown", "Unknown", vtrpcpb.Code_INTERNAL.String()}, 1)
}
}