forked from hashicorp/consul
-
Notifications
You must be signed in to change notification settings - Fork 0
/
state_store.go
222 lines (187 loc) · 6.41 KB
/
state_store.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
package state
import (
"errors"
"fmt"
"github.com/hashicorp/consul/types"
"github.com/hashicorp/go-memdb"
)
var (
// ErrMissingNode is the error returned when trying an operation
// which requires a node registration but none exists.
ErrMissingNode = errors.New("Missing node registration")
// ErrMissingService is the error we return if trying an
// operation which requires a service but none exists.
ErrMissingService = errors.New("Missing service registration")
// ErrMissingSessionID is returned when a session registration
// is attempted with an empty session ID.
ErrMissingSessionID = errors.New("Missing session ID")
// ErrMissingACLID is returned when an ACL set is called on
// an ACL with an empty ID.
ErrMissingACLID = errors.New("Missing ACL ID")
// ErrMissingQueryID is returned when a Query set is called on
// a Query with an empty ID.
ErrMissingQueryID = errors.New("Missing Query ID")
// ErrMissingCARootID is returned when an CARoot set is called
// with an CARoot with an empty ID.
ErrMissingCARootID = errors.New("Missing CA Root ID")
// ErrMissingIntentionID is returned when an Intention set is called
// with an Intention with an empty ID.
ErrMissingIntentionID = errors.New("Missing Intention ID")
)
const (
// watchLimit is used as a soft limit to cap how many watches we allow
// for a given blocking query. If this is exceeded, then we will use a
// higher-level watch that's less fine-grained. This isn't as bad as it
// seems since we have made the main culprits (nodes and services) more
// efficient by diffing before we update via register requests.
//
// Given the current size of aFew == 32 in memdb's watch_few.go, this
// will allow for up to ~64 goroutines per blocking query.
watchLimit = 2048
)
// Store is where we store all of Consul's state, including
// records of node registrations, services, checks, key/value
// pairs and more. The DB is entirely in-memory and is constructed
// from the Raft log through the FSM.
type Store struct {
schema *memdb.DBSchema
db *memdb.MemDB
// abandonCh is used to signal watchers that this state store has been
// abandoned (usually during a restore). This is only ever closed.
abandonCh chan struct{}
// kvsGraveyard manages tombstones for the key value store.
kvsGraveyard *Graveyard
// lockDelay holds expiration times for locks associated with keys.
lockDelay *Delay
}
// Snapshot is used to provide a point-in-time snapshot. It
// works by starting a read transaction against the whole state store.
type Snapshot struct {
store *Store
tx *memdb.Txn
lastIndex uint64
}
// Restore is used to efficiently manage restoring a large amount of
// data to a state store.
type Restore struct {
store *Store
tx *memdb.Txn
}
// IndexEntry keeps a record of the last index per-table.
type IndexEntry struct {
Key string
Value uint64
}
// sessionCheck is used to create a many-to-one table such that
// each check registered by a session can be mapped back to the
// session table. This is only used internally in the state
// store and thus it is not exported.
type sessionCheck struct {
Node string
CheckID types.CheckID
Session string
}
// NewStateStore creates a new in-memory state storage layer.
func NewStateStore(gc *TombstoneGC) (*Store, error) {
// Create the in-memory DB.
schema := stateStoreSchema()
db, err := memdb.NewMemDB(schema)
if err != nil {
return nil, fmt.Errorf("Failed setting up state store: %s", err)
}
// Create and return the state store.
s := &Store{
schema: schema,
db: db,
abandonCh: make(chan struct{}),
kvsGraveyard: NewGraveyard(gc),
lockDelay: NewDelay(),
}
return s, nil
}
// Snapshot is used to create a point-in-time snapshot of the entire db.
func (s *Store) Snapshot() *Snapshot {
tx := s.db.Txn(false)
var tables []string
for table := range s.schema.Tables {
tables = append(tables, table)
}
idx := maxIndexTxn(tx, tables...)
return &Snapshot{s, tx, idx}
}
// LastIndex returns that last index that affects the snapshotted data.
func (s *Snapshot) LastIndex() uint64 {
return s.lastIndex
}
// Close performs cleanup of a state snapshot.
func (s *Snapshot) Close() {
s.tx.Abort()
}
// Restore is used to efficiently manage restoring a large amount of data into
// the state store. It works by doing all the restores inside of a single
// transaction.
func (s *Store) Restore() *Restore {
tx := s.db.Txn(true)
return &Restore{s, tx}
}
// Abort abandons the changes made by a restore. This or Commit should always be
// called.
func (s *Restore) Abort() {
s.tx.Abort()
}
// Commit commits the changes made by a restore. This or Abort should always be
// called.
func (s *Restore) Commit() {
s.tx.Commit()
}
// AbandonCh returns a channel you can wait on to know if the state store was
// abandoned.
func (s *Store) AbandonCh() <-chan struct{} {
return s.abandonCh
}
// Abandon is used to signal that the given state store has been abandoned.
// Calling this more than one time will panic.
func (s *Store) Abandon() {
close(s.abandonCh)
}
// maxIndex is a helper used to retrieve the highest known index
// amongst a set of tables in the db.
func (s *Store) maxIndex(tables ...string) uint64 {
tx := s.db.Txn(false)
defer tx.Abort()
return maxIndexTxn(tx, tables...)
}
// maxIndexTxn is a helper used to retrieve the highest known index
// amongst a set of tables in the db.
func maxIndexTxn(tx *memdb.Txn, tables ...string) uint64 {
var lindex uint64
for _, table := range tables {
ti, err := tx.First("index", "id", table)
if err != nil {
panic(fmt.Sprintf("unknown index: %s err: %s", table, err))
}
if idx, ok := ti.(*IndexEntry); ok && idx.Value > lindex {
lindex = idx.Value
}
}
return lindex
}
// indexUpdateMaxTxn is used when restoring entries and sets the table's index to
// the given idx only if it's greater than the current index.
func indexUpdateMaxTxn(tx *memdb.Txn, idx uint64, table string) error {
ti, err := tx.First("index", "id", table)
if err != nil {
return fmt.Errorf("failed to retrieve existing index: %s", err)
}
// Always take the first update, otherwise do the > check.
if ti == nil {
if err := tx.Insert("index", &IndexEntry{table, idx}); err != nil {
return fmt.Errorf("failed updating index %s", err)
}
} else if cur, ok := ti.(*IndexEntry); ok && idx > cur.Value {
if err := tx.Insert("index", &IndexEntry{table, idx}); err != nil {
return fmt.Errorf("failed updating index %s", err)
}
}
return nil
}