-
Notifications
You must be signed in to change notification settings - Fork 3.8k
/
stores.go
307 lines (281 loc) · 10.1 KB
/
stores.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
// Copyright 2014 The Cockroach 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.
//
// Author: Spencer Kimball (spencer.kimball@gmail.com)
package storage
import (
"fmt"
"sync"
"github.com/gogo/protobuf/proto"
"golang.org/x/net/context"
"github.com/cockroachdb/cockroach/client"
"github.com/cockroachdb/cockroach/gossip"
"github.com/cockroachdb/cockroach/keys"
"github.com/cockroachdb/cockroach/roachpb"
"github.com/cockroachdb/cockroach/storage/engine"
"github.com/cockroachdb/cockroach/util/hlc"
"github.com/cockroachdb/cockroach/util/log"
"github.com/cockroachdb/cockroach/util/tracer"
)
// A Stores provides methods to access a collection of stores. There's
// a visitor pattern and also an implementation of the client.Sender
// interface which directs a call to the appropriate store based on
// the call's key range. Stores also implements the gossip.Storage
// interface, which allows gossip bootstrap information to be
// persisted consistently to every store and the most recent bootstrap
// information to be read at node startup.
type Stores struct {
clock *hlc.Clock
mu sync.RWMutex // Protects storeMap and addrs
storeMap map[roachpb.StoreID]*Store // Map from StoreID to Store
biLatestTS roachpb.Timestamp // Timestamp of gossip bootstrap info
latestBI *gossip.BootstrapInfo // Latest cached bootstrap info
}
var _ client.Sender = &Stores{} // Stores implements the client.Sender interface
var _ gossip.Storage = &Stores{} // Stores implements the gossip.Storage interface
// NewStores returns a local-only sender which directly accesses
// a collection of stores.
func NewStores(clock *hlc.Clock) *Stores {
return &Stores{
clock: clock,
storeMap: map[roachpb.StoreID]*Store{},
}
}
// GetStoreCount returns the number of stores this node is exporting.
func (ls *Stores) GetStoreCount() int {
ls.mu.RLock()
defer ls.mu.RUnlock()
return len(ls.storeMap)
}
// HasStore returns true if the specified store is owned by this Stores.
func (ls *Stores) HasStore(storeID roachpb.StoreID) bool {
ls.mu.RLock()
defer ls.mu.RUnlock()
_, ok := ls.storeMap[storeID]
return ok
}
// GetStore looks up the store by store ID. Returns an error
// if not found.
func (ls *Stores) GetStore(storeID roachpb.StoreID) (*Store, *roachpb.Error) {
ls.mu.RLock()
store, ok := ls.storeMap[storeID]
ls.mu.RUnlock()
if !ok {
return nil, roachpb.NewErrorf("store %d not found", storeID)
}
return store, nil
}
// AddStore adds the specified store to the store map.
func (ls *Stores) AddStore(s *Store) {
ls.mu.Lock()
defer ls.mu.Unlock()
if _, ok := ls.storeMap[s.Ident.StoreID]; ok {
panic(fmt.Sprintf("cannot add store twice: %+v", s.Ident))
}
ls.storeMap[s.Ident.StoreID] = s
// If we've already read the gossip bootstrap info, ensure that
// all stores have the most recent values.
if !ls.biLatestTS.Equal(roachpb.ZeroTimestamp) {
if err := ls.updateBootstrapInfo(ls.latestBI); err != nil {
log.Errorf("failed to update bootstrap info on newly added store: %s", err)
}
}
}
// RemoveStore removes the specified store from the store map.
func (ls *Stores) RemoveStore(s *Store) {
ls.mu.Lock()
defer ls.mu.Unlock()
delete(ls.storeMap, s.Ident.StoreID)
}
// VisitStores implements a visitor pattern over stores in the storeMap.
// The specified function is invoked with each store in turn. Stores are
// visited in a random order.
func (ls *Stores) VisitStores(visitor func(s *Store) error) error {
ls.mu.RLock()
defer ls.mu.RUnlock()
for _, s := range ls.storeMap {
if err := visitor(s); err != nil {
return err
}
}
return nil
}
// Send implements the client.Sender interface. The store is looked up from the
// store map if specified by the request; otherwise, the command is being
// executed locally, and the replica is determined via lookup through each
// store's LookupRange method. The latter path is taken only by unit tests.
func (ls *Stores) Send(ctx context.Context, ba roachpb.BatchRequest) (*roachpb.BatchResponse, *roachpb.Error) {
trace := tracer.FromCtx(ctx)
var store *Store
var pErr *roachpb.Error
// If we aren't given a Replica, then a little bending over
// backwards here. This case applies exclusively to unittests.
if ba.RangeID == 0 || ba.Replica.StoreID == 0 {
var repl *roachpb.ReplicaDescriptor
var rangeID roachpb.RangeID
rs := keys.Range(ba)
rangeID, repl, pErr = ls.lookupReplica(rs.Key, rs.EndKey)
if pErr == nil {
ba.RangeID = rangeID
ba.Replica = *repl
}
}
ctx = log.Add(ctx,
log.RangeID, ba.RangeID)
if pErr == nil {
store, pErr = ls.GetStore(ba.Replica.StoreID)
}
var br *roachpb.BatchResponse
if pErr != nil {
return nil, pErr
}
// For calls that read data within a txn, we can avoid uncertainty
// related retries in certain situations. If the node is in
// "CertainNodes", we need not worry about uncertain reads any
// more. Setting MaxTimestamp=Timestamp for the operation
// accomplishes that. See roachpb.Transaction.CertainNodes for details.
if ba.Txn != nil && ba.Txn.CertainNodes.Contains(ba.Replica.NodeID) {
// MaxTimestamp = Timestamp corresponds to no clock uncertainty.
trace.Event("read has no clock uncertainty")
ba.Txn.MaxTimestamp = ba.Txn.Timestamp
}
br, pErr = store.Send(ctx, ba)
if br != nil && br.Error != nil {
panic(roachpb.ErrorUnexpectedlySet(store, br))
}
return br, pErr
}
// lookupReplica looks up replica by key [range]. Lookups are done
// by consulting each store in turn via Store.LookupRange(key).
// Returns RangeID and replica on success; RangeKeyMismatch error
// if not found.
// This is only for testing usage; performance doesn't matter.
func (ls *Stores) lookupReplica(start, end roachpb.RKey) (rangeID roachpb.RangeID, replica *roachpb.ReplicaDescriptor, pErr *roachpb.Error) {
ls.mu.RLock()
defer ls.mu.RUnlock()
var rng *Replica
for _, store := range ls.storeMap {
rng = store.LookupReplica(start, end)
if rng == nil {
if tmpRng := store.LookupReplica(start, nil); tmpRng != nil {
log.Warningf("range not contained in one range: [%s,%s), but have [%s,%s)",
start, end, tmpRng.Desc().StartKey, tmpRng.Desc().EndKey)
}
continue
}
if replica == nil {
rangeID = rng.RangeID
replica = rng.GetReplica()
continue
}
// Should never happen outside of tests.
return 0, nil, roachpb.NewErrorf(
"range %+v exists on additional store: %+v", rng, store)
}
if replica == nil {
pErr = roachpb.NewError(roachpb.NewRangeKeyMismatchError(start.AsRawKey(), end.AsRawKey(), nil))
}
return rangeID, replica, pErr
}
// FirstRange implements the RangeDescriptorDB interface. It returns the
// range descriptor which contains KeyMin.
func (ls *Stores) FirstRange() (*roachpb.RangeDescriptor, *roachpb.Error) {
_, replica, pErr := ls.lookupReplica(roachpb.RKeyMin, nil)
if pErr != nil {
return nil, pErr
}
store, pErr := ls.GetStore(replica.StoreID)
if pErr != nil {
return nil, pErr
}
rpl := store.LookupReplica(roachpb.RKeyMin, nil)
if rpl == nil {
panic("firstRange found no first range")
}
return rpl.Desc(), nil
}
// RangeLookup implements the RangeDescriptorDB interface. It looks up
// the descriptors for the given (meta) key.
func (ls *Stores) RangeLookup(key roachpb.RKey, _ *roachpb.RangeDescriptor, considerIntents, useReverseScan bool) ([]roachpb.RangeDescriptor, *roachpb.Error) {
ba := roachpb.BatchRequest{}
ba.ReadConsistency = roachpb.INCONSISTENT
ba.Add(&roachpb.RangeLookupRequest{
Span: roachpb.Span{
// key is a meta key, so it's guaranteed not local-prefixed.
Key: key.AsRawKey(),
},
MaxRanges: 1,
ConsiderIntents: considerIntents,
Reverse: useReverseScan,
})
br, pErr := ls.Send(context.Background(), ba)
if pErr != nil {
return nil, pErr
}
return br.Responses[0].GetInner().(*roachpb.RangeLookupResponse).Ranges, nil
}
// ReadBootstrapInfo implements the gossip.Storage interface. Read
// attempts to read gossip bootstrap info from every known store and
// finds the most recent from all stores to initialize the bootstrap
// info argument. Returns an error on any issues reading data for the
// stores (but excluding the case in which no data has been persisted
// yet).
func (ls *Stores) ReadBootstrapInfo(bi *gossip.BootstrapInfo) error {
ls.mu.RLock()
defer ls.mu.RUnlock()
latestTS := roachpb.ZeroTimestamp
// Find the most recent bootstrap info.
for _, s := range ls.storeMap {
var storeBI gossip.BootstrapInfo
ok, err := engine.MVCCGetProto(s.engine, keys.StoreGossipKey(), roachpb.ZeroTimestamp, true, nil, &storeBI)
if err != nil {
return err
}
if ok && latestTS.Less(storeBI.Timestamp) {
latestTS = storeBI.Timestamp
*bi = storeBI
}
}
log.Infof("read %d node addresses from persistent storage", len(bi.Addresses))
return ls.updateBootstrapInfo(bi)
}
// WriteBootstrapInfo implements the gossip.Storage interface. Write
// persists the supplied bootstrap info to every known store. Returns
// nil on success; otherwise returns first error encountered writing
// to the stores.
func (ls *Stores) WriteBootstrapInfo(bi *gossip.BootstrapInfo) error {
ls.mu.RLock()
defer ls.mu.RUnlock()
bi.Timestamp = ls.clock.Now()
if err := ls.updateBootstrapInfo(bi); err != nil {
return err
}
log.Infof("wrote %d node addresses to persistent storage", len(bi.Addresses))
return nil
}
func (ls *Stores) updateBootstrapInfo(bi *gossip.BootstrapInfo) error {
if bi.Timestamp.Less(ls.biLatestTS) {
return nil
}
// Update the latest timestamp and set cached version.
ls.biLatestTS = bi.Timestamp
ls.latestBI = proto.Clone(bi).(*gossip.BootstrapInfo)
// Update all stores.
for _, s := range ls.storeMap {
if err := engine.MVCCPutProto(s.engine, nil, keys.StoreGossipKey(), roachpb.ZeroTimestamp, nil, bi); err != nil {
return err
}
}
return nil
}