-
Notifications
You must be signed in to change notification settings - Fork 2.1k
/
discoverygateway.go
410 lines (357 loc) · 13.3 KB
/
discoverygateway.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
/*
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
import (
"fmt"
"math/rand"
"sort"
"strings"
"sync"
"time"
"context"
"vitess.io/vitess/go/stats"
"vitess.io/vitess/go/vt/discovery"
"vitess.io/vitess/go/vt/log"
"vitess.io/vitess/go/vt/srvtopo"
"vitess.io/vitess/go/vt/topo"
"vitess.io/vitess/go/vt/vterrors"
"vitess.io/vitess/go/vt/vtgate/buffer"
"vitess.io/vitess/go/vt/vttablet/queryservice"
querypb "vitess.io/vitess/go/vt/proto/query"
topodatapb "vitess.io/vitess/go/vt/proto/topodata"
vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
)
const (
// GatewayImplementationDiscovery defines the string value used as the implementation key for DiscoveryGateway
GatewayImplementationDiscovery = "discoverygateway"
)
//UsingLegacyGateway returns true when legacy
func UsingLegacyGateway() bool {
return *GatewayImplementation == GatewayImplementationDiscovery
}
func init() {
RegisterGatewayCreator(GatewayImplementationDiscovery, createDiscoveryGateway)
}
// DiscoveryGateway is the default Gateway implementation.
// This implementation uses the legacy healthcheck module.
type DiscoveryGateway struct {
queryservice.QueryService
hc discovery.LegacyHealthCheck
tsc *discovery.LegacyTabletStatsCache
srvTopoServer srvtopo.Server
localCell string
retryCount int
// tabletsWatchers contains a list of all the watchers we use.
// We create one per cell.
tabletsWatchers []*discovery.LegacyTopologyWatcher
// mu protects the fields of this group.
mu sync.RWMutex
// statusAggregators is a map indexed by the key
// keyspace/shard/tablet_type.
statusAggregators map[string]*TabletStatusAggregator
// buffer, if enabled, buffers requests during a detected MASTER failover.
buffer *buffer.Buffer
}
//TabletsCacheStatus is not implemented for this struct
func (dg *DiscoveryGateway) TabletsCacheStatus() discovery.TabletsCacheStatusList {
return nil
}
var _ Gateway = (*DiscoveryGateway)(nil)
func createDiscoveryGateway(ctx context.Context, hc discovery.LegacyHealthCheck, serv srvtopo.Server, cell string, retryCount int) Gateway {
return NewDiscoveryGateway(ctx, hc, serv, cell, retryCount)
}
// NewDiscoveryGateway creates a new DiscoveryGateway using the provided healthcheck and toposerver.
// cell is the cell where the gateway is located a.k.a localCell.
// This gateway can route to MASTER in any cell provided by the cells_to_watch command line argument.
// Other tablet type requests (REPLICA/RDONLY) are only routed to tablets in the same cell.
func NewDiscoveryGateway(ctx context.Context, hc discovery.LegacyHealthCheck, serv srvtopo.Server, cell string, retryCount int) *DiscoveryGateway {
var topoServer *topo.Server
if serv != nil {
var err error
topoServer, err = serv.GetTopoServer()
if err != nil {
log.Exitf("Unable to create new discoverygateway: %v", err)
}
}
dg := &DiscoveryGateway{
hc: hc,
tsc: discovery.NewTabletStatsCacheDoNotSetListener(topoServer, cell),
srvTopoServer: serv,
localCell: cell,
retryCount: retryCount,
tabletsWatchers: make([]*discovery.LegacyTopologyWatcher, 0, 1),
statusAggregators: make(map[string]*TabletStatusAggregator),
buffer: buffer.New(),
}
// Set listener which will update LegacyTabletStatsCache and MasterBuffer.
// We set sendDownEvents=true because it's required by LegacyTabletStatsCache.
hc.SetListener(dg, true /* sendDownEvents */)
cells := *CellsToWatch
log.Infof("loading tablets for cells: %v", cells)
for _, c := range strings.Split(cells, ",") {
if c == "" {
continue
}
var recorder discovery.LegacyTabletRecorder = dg.hc
if len(discovery.TabletFilters) > 0 {
if discovery.FilteringKeyspaces() {
log.Exitf("Only one of -keyspaces_to_watch and -tablet_filters may be specified at a time")
}
fbs, err := discovery.NewLegacyFilterByShard(recorder, discovery.TabletFilters)
if err != nil {
log.Exitf("Cannot parse tablet_filters parameter: %v", err)
}
recorder = fbs
} else if discovery.FilteringKeyspaces() {
recorder = discovery.NewLegacyFilterByKeyspace(recorder, discovery.KeyspacesToWatch)
}
ctw := discovery.NewLegacyCellTabletsWatcher(ctx, topoServer, recorder, c, *discovery.RefreshInterval, *discovery.RefreshKnownTablets, *discovery.TopoReadConcurrency)
dg.tabletsWatchers = append(dg.tabletsWatchers, ctw)
}
dg.QueryService = queryservice.Wrap(nil, dg.withRetry)
return dg
}
// RegisterStats registers the stats to export the lag since the last refresh
// and the checksum of the topology
func (dg *DiscoveryGateway) RegisterStats() {
stats.NewGaugeDurationFunc(
"TopologyWatcherMaxRefreshLag",
"maximum time since the topology watcher refreshed a cell",
dg.topologyWatcherMaxRefreshLag,
)
stats.NewGaugeFunc(
"TopologyWatcherChecksum",
"crc32 checksum of the topology watcher state",
dg.topologyWatcherChecksum,
)
}
// topologyWatcherMaxRefreshLag returns the maximum lag since the watched
// cells were refreshed from the topo server
func (dg *DiscoveryGateway) topologyWatcherMaxRefreshLag() time.Duration {
var lag time.Duration
for _, tw := range dg.tabletsWatchers {
cellLag := tw.RefreshLag()
if cellLag > lag {
lag = cellLag
}
}
return lag
}
// topologyWatcherChecksum returns a checksum of the topology watcher state
func (dg *DiscoveryGateway) topologyWatcherChecksum() int64 {
var checksum int64
for _, tw := range dg.tabletsWatchers {
checksum = checksum ^ int64(tw.TopoChecksum())
}
return checksum
}
// StatsUpdate forwards LegacyHealthCheck updates to LegacyTabletStatsCache and MasterBuffer.
// It is part of the discovery.LegacyHealthCheckStatsListener interface.
func (dg *DiscoveryGateway) StatsUpdate(ts *discovery.LegacyTabletStats) {
dg.tsc.StatsUpdate(ts)
if ts.Target.TabletType == topodatapb.TabletType_MASTER {
dg.buffer.StatsUpdate(ts)
}
}
// WaitForTablets is part of the gateway.Gateway interface.
func (dg *DiscoveryGateway) WaitForTablets(ctx context.Context, tabletTypesToWait []topodatapb.TabletType) error {
// Skip waiting for tablets if we are not told to do so.
if len(tabletTypesToWait) == 0 {
return nil
}
// Finds the targets to look for.
targets, err := srvtopo.FindAllTargets(ctx, dg.srvTopoServer, dg.localCell, tabletTypesToWait)
if err != nil {
return err
}
filteredTargets := discovery.FilterTargetsByKeyspaces(discovery.KeyspacesToWatch, targets)
return dg.tsc.WaitForAllServingTablets(ctx, filteredTargets)
}
// Close shuts down underlying connections.
// This function hides the inner implementation.
func (dg *DiscoveryGateway) Close(ctx context.Context) error {
dg.buffer.Shutdown()
for _, ctw := range dg.tabletsWatchers {
ctw.Stop()
}
return nil
}
// CacheStatus returns a list of TabletCacheStatus per
// keyspace/shard/tablet_type.
func (dg *DiscoveryGateway) CacheStatus() TabletCacheStatusList {
dg.mu.RLock()
res := make(TabletCacheStatusList, 0, len(dg.statusAggregators))
for _, aggr := range dg.statusAggregators {
res = append(res, aggr.GetCacheStatus())
}
dg.mu.RUnlock()
sort.Sort(res)
return res
}
// withRetry gets available connections and executes the action. If there are retryable errors,
// it retries retryCount times before failing. It does not retry if the connection is in
// the middle of a transaction. While returning the error check if it maybe a result of
// a resharding event, and set the re-resolve bit and let the upper layers
// re-resolve and retry.
func (dg *DiscoveryGateway) withRetry(ctx context.Context, target *querypb.Target, unused queryservice.QueryService, name string, inTransaction bool, inner func(ctx context.Context, target *querypb.Target, conn queryservice.QueryService) (bool, error)) error {
var err error
invalidTablets := make(map[string]bool)
if len(discovery.AllowedTabletTypes) > 0 {
var match bool
for _, allowed := range discovery.AllowedTabletTypes {
if allowed == target.TabletType {
match = true
break
}
}
if !match {
return vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "requested tablet type %v is not part of the allowed tablet types for this vtgate: %+v", target.TabletType.String(), discovery.AllowedTabletTypes)
}
}
bufferedOnce := false
for i := 0; i < dg.retryCount+1; i++ {
// Check if we should buffer MASTER queries which failed due to an ongoing
// failover.
// Note: We only buffer once and only "!inTransaction" queries i.e.
// a) no transaction is necessary (e.g. critical reads) or
// b) no transaction was created yet.
if !bufferedOnce && !inTransaction && target.TabletType == topodatapb.TabletType_MASTER {
// The next call blocks if we should buffer during a failover.
retryDone, bufferErr := dg.buffer.WaitForFailoverEnd(ctx, target.Keyspace, target.Shard, err)
if bufferErr != nil {
// Buffering failed e.g. buffer is already full. Do not retry.
err = vterrors.Errorf(
vterrors.Code(bufferErr),
"failed to automatically buffer and retry failed request during failover: %v original err (type=%T): %v",
bufferErr, err, err)
break
}
// Request may have been buffered.
if retryDone != nil {
// We're going to retry this request as part of a buffer drain.
// Notify the buffer after we retried.
defer retryDone()
bufferedOnce = true
}
}
tablets := dg.tsc.GetHealthyTabletStats(target.Keyspace, target.Shard, target.TabletType)
if len(tablets) == 0 {
// fail fast if there is no tablet
err = vterrors.Errorf(vtrpcpb.Code_UNAVAILABLE, "no healthy tablet available for '%s'", target.String())
break
}
shuffleTablets(dg.localCell, tablets)
// skip tablets we tried before
var ts *discovery.LegacyTabletStats
for _, t := range tablets {
if _, ok := invalidTablets[t.Key]; !ok {
ts = &t
break
}
}
if ts == nil {
if err == nil {
// do not override error from last attempt.
err = vterrors.New(vtrpcpb.Code_UNAVAILABLE, "no available connection")
}
break
}
// execute
conn := dg.hc.GetConnection(ts.Key)
if conn == nil {
err = vterrors.Errorf(vtrpcpb.Code_UNAVAILABLE, "no connection for key %v tablet %+v", ts.Key, ts.Tablet)
invalidTablets[ts.Key] = true
continue
}
startTime := time.Now()
var canRetry bool
canRetry, err = inner(ctx, ts.Target, conn)
dg.updateStats(target, startTime, err)
if canRetry {
invalidTablets[ts.Key] = true
continue
}
break
}
return NewShardError(err, target)
}
func shuffleTablets(cell string, tablets []discovery.LegacyTabletStats) {
sameCell, diffCell, sameCellMax := 0, 0, -1
length := len(tablets)
// move all same cell tablets to the front, this is O(n)
for {
sameCellMax = diffCell - 1
sameCell = nextTablet(cell, tablets, sameCell, length, true)
diffCell = nextTablet(cell, tablets, diffCell, length, false)
// either no more diffs or no more same cells should stop the iteration
if sameCell < 0 || diffCell < 0 {
break
}
if sameCell < diffCell {
// fast forward the `sameCell` lookup to `diffCell + 1`, `diffCell` unchanged
sameCell = diffCell + 1
} else {
// sameCell > diffCell, swap needed
tablets[sameCell], tablets[diffCell] = tablets[diffCell], tablets[sameCell]
sameCell++
diffCell++
}
}
//shuffle in same cell tablets
for i := sameCellMax; i > 0; i-- {
swap := rand.Intn(i + 1)
tablets[i], tablets[swap] = tablets[swap], tablets[i]
}
//shuffle in diff cell tablets
for i, diffCellMin := length-1, sameCellMax+1; i > diffCellMin; i-- {
swap := rand.Intn(i-sameCellMax) + diffCellMin
tablets[i], tablets[swap] = tablets[swap], tablets[i]
}
}
func nextTablet(cell string, tablets []discovery.LegacyTabletStats, offset, length int, sameCell bool) int {
for ; offset < length; offset++ {
if (tablets[offset].Tablet.Alias.Cell == cell) == sameCell {
return offset
}
}
return -1
}
func (dg *DiscoveryGateway) updateStats(target *querypb.Target, startTime time.Time, err error) {
elapsed := time.Since(startTime)
aggr := dg.getStatsAggregator(target)
aggr.UpdateQueryInfo("", target.TabletType, elapsed, err != nil)
}
func (dg *DiscoveryGateway) getStatsAggregator(target *querypb.Target) *TabletStatusAggregator {
key := fmt.Sprintf("%v/%v/%v", target.Keyspace, target.Shard, target.TabletType.String())
// get existing aggregator
dg.mu.RLock()
aggr, ok := dg.statusAggregators[key]
dg.mu.RUnlock()
if ok {
return aggr
}
// create a new one, but check again before the creation
dg.mu.Lock()
defer dg.mu.Unlock()
aggr, ok = dg.statusAggregators[key]
if ok {
return aggr
}
aggr = NewTabletStatusAggregator(target.Keyspace, target.Shard, target.TabletType, key)
dg.statusAggregators[key] = aggr
return aggr
}
// QueryServiceByAlias satisfies the Gateway interface
func (dg *DiscoveryGateway) QueryServiceByAlias(_ *topodatapb.TabletAlias, _ *querypb.Target) (queryservice.QueryService, error) {
return nil, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "DiscoveryGateway does not implement QueryServiceByAlias")
}