luxolus/open-consul
2
0
Fork 0
Code Issues 1 Pull requests Packages Releases Wiki Activity

Add a paramter in state store methods to indicate whether a resource insertion is from a snapshot restoration (#9156)

Browse source 
The Catalog, Config Entry, KV and Session resources potentially re-validate the input as its coming in. We need to prevent snapshot restoration failures due to missing namespaces or namespaces that are being deleted in enterprise.
This commit is contained in:
Matt Keeler 2020-11-11 11:21:42 -05:00 committed by GitHub
parent 4c2ca3a516
commit 2badb01d30
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 85 additions and 79 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
.changelog/9156.txt Normal file
View file

@ -0,0 +1,3 @@
```release-note:bug
namespace: **(Enterprise Only)** Fixed a bug that could case snapshot restoration to fail when it contained a namespace marked for deletion while still containing other resources in that namespace.
```

3
.changelog/_666.txt Normal file
View file

@ -0,0 +1,3 @@
```release-note:bug
namespace: **(Enterprise Only)** Fixed an issue where namespaced services and checks were not being deleted when the containing namespace was deleted.
```

8
agent/consul/state/catalog.go
View file

@ -263,7 +263,7 @@ func (s *Snapshot) Checks(node string) (memdb.ResultIterator, error) {
// performed within a single transaction to avoid race conditions on state
// updates.
func (s *Restore) Registration(idx uint64, req *structs.RegisterRequest) error {
return s.store.ensureRegistrationTxn(s.tx, idx, true, req)
return s.store.ensureRegistrationTxn(s.tx, idx, true, req, true)
}
// EnsureRegistration is used to make sure a node, service, and check
@ -273,7 +273,7 @@ func (s *Store) EnsureRegistration(idx uint64, req *structs.RegisterRequest) err
tx := s.db.WriteTxn(idx)
defer tx.Abort()
if err := s.ensureRegistrationTxn(tx, idx, false, req); err != nil {
if err := s.ensureRegistrationTxn(tx, idx, false, req, false); err != nil {
return err
}
@ -294,8 +294,8 @@ func (s *Store) ensureCheckIfNodeMatches(tx WriteTxn, idx uint64, preserveIndexe
// ensureRegistrationTxn is used to make sure a node, service, and check
// registration is performed within a single transaction to avoid race
// conditions on state updates.
func (s *Store) ensureRegistrationTxn(tx WriteTxn, idx uint64, preserveIndexes bool, req *structs.RegisterRequest) error {
if _, err := validateRegisterRequestTxn(tx, req); err != nil {
func (s *Store) ensureRegistrationTxn(tx WriteTxn, idx uint64, preserveIndexes bool, req *structs.RegisterRequest, restore bool) error {
if _, err := validateRegisterRequestTxn(tx, req, restore); err != nil {
return err
}

136
agent/consul/state/catalog_events_test.go
View file

@ -40,7 +40,7 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
Name: "service reg, new node",
Mutate: func(s *Store, tx *txn) error {
return s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web"))
testServiceRegistration(t, "web"), false)
},
WantEvents: []stream.Event{
testServiceHealthEvent(t, "web"),
@ -51,11 +51,11 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
Name: "service reg, existing node",
Setup: func(s *Store, tx *txn) error {
return s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "db"))
testServiceRegistration(t, "db"), false)
},
Mutate: func(s *Store, tx *txn) error {
return s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web"))
testServiceRegistration(t, "web"), false)
},
WantEvents: []stream.Event{
// Should only publish new service
@ -67,11 +67,11 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
Name: "service dereg, existing node",
Setup: func(s *Store, tx *txn) error {
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "db")); err != nil {
testServiceRegistration(t, "db"), false); err != nil {
return err
}
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web")); err != nil {
testServiceRegistration(t, "web"), false); err != nil {
return err
}
return nil
@ -88,10 +88,10 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
{
Name: "node dereg",
Setup: func(s *Store, tx *txn) error {
if err := s.ensureRegistrationTxn(tx, tx.Index, false, testServiceRegistration(t, "db")); err != nil {
if err := s.ensureRegistrationTxn(tx, tx.Index, false, testServiceRegistration(t, "db"), false); err != nil {
return err
}
if err := s.ensureRegistrationTxn(tx, tx.Index, false, testServiceRegistration(t, "web")); err != nil {
if err := s.ensureRegistrationTxn(tx, tx.Index, false, testServiceRegistration(t, "web"), false); err != nil {
return err
}
return nil
@ -109,7 +109,7 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
{
Name: "connect native reg, new node",
Mutate: func(s *Store, tx *txn) error {
return s.ensureRegistrationTxn(tx, tx.Index, false, testServiceRegistration(t, "web", regConnectNative))
return s.ensureRegistrationTxn(tx, tx.Index, false, testServiceRegistration(t, "web", regConnectNative), false)
},
WantEvents: []stream.Event{
// We should see both a regular service health event as well as a connect
@ -122,10 +122,10 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
{
Name: "connect native reg, existing node",
Setup: func(s *Store, tx *txn) error {
return s.ensureRegistrationTxn(tx, tx.Index, false, testServiceRegistration(t, "db"))
return s.ensureRegistrationTxn(tx, tx.Index, false, testServiceRegistration(t, "db"), false)
},
Mutate: func(s *Store, tx *txn) error {
return s.ensureRegistrationTxn(tx, tx.Index, false, testServiceRegistration(t, "web", regConnectNative))
return s.ensureRegistrationTxn(tx, tx.Index, false, testServiceRegistration(t, "web", regConnectNative), false)
},
WantEvents: []stream.Event{
// We should see both a regular service health event as well as a connect
@ -143,11 +143,11 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
{
Name: "connect native dereg, existing node",
Setup: func(s *Store, tx *txn) error {
if err := s.ensureRegistrationTxn(tx, tx.Index, false, testServiceRegistration(t, "db")); err != nil {
if err := s.ensureRegistrationTxn(tx, tx.Index, false, testServiceRegistration(t, "db"), false); err != nil {
return err
}
return s.ensureRegistrationTxn(tx, tx.Index, false, testServiceRegistration(t, "web", regConnectNative))
return s.ensureRegistrationTxn(tx, tx.Index, false, testServiceRegistration(t, "web", regConnectNative), false)
},
Mutate: func(s *Store, tx *txn) error {
return s.deleteServiceTxn(tx, tx.Index, "node1", "web", nil)
@ -162,10 +162,10 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
{
Name: "connect sidecar reg, new node",
Mutate: func(s *Store, tx *txn) error {
if err := s.ensureRegistrationTxn(tx, tx.Index, false, testServiceRegistration(t, "web")); err != nil {
if err := s.ensureRegistrationTxn(tx, tx.Index, false, testServiceRegistration(t, "web"), false); err != nil {
return err
}
return s.ensureRegistrationTxn(tx, tx.Index, false, testServiceRegistration(t, "web", regSidecar))
return s.ensureRegistrationTxn(tx, tx.Index, false, testServiceRegistration(t, "web", regSidecar), false)
},
WantEvents: []stream.Event{
// We should see both a regular service health event for the web service
@ -180,15 +180,15 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
Name: "connect sidecar reg, existing node",
Setup: func(s *Store, tx *txn) error {
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "db")); err != nil {
testServiceRegistration(t, "db"), false); err != nil {
return err
}
return s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web"))
testServiceRegistration(t, "web"), false)
},
Mutate: func(s *Store, tx *txn) error {
return s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regSidecar))
testServiceRegistration(t, "web", regSidecar), false)
},
WantEvents: []stream.Event{
// We should see both a regular service health event for the proxy
@ -202,15 +202,15 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
Name: "connect sidecar dereg, existing node",
Setup: func(s *Store, tx *txn) error {
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "db")); err != nil {
testServiceRegistration(t, "db"), false); err != nil {
return err
}
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web")); err != nil {
testServiceRegistration(t, "web"), false); err != nil {
return err
}
return s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regSidecar))
testServiceRegistration(t, "web", regSidecar), false)
},
Mutate: func(s *Store, tx *txn) error {
// Delete only the sidecar
@ -227,20 +227,20 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
Name: "connect sidecar mutate svc",
Setup: func(s *Store, tx *txn) error {
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "db")); err != nil {
testServiceRegistration(t, "db"), false); err != nil {
return err
}
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web")); err != nil {
testServiceRegistration(t, "web"), false); err != nil {
return err
}
return s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regSidecar))
testServiceRegistration(t, "web", regSidecar), false)
},
Mutate: func(s *Store, tx *txn) error {
// Change port of the target service instance
return s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regMutatePort))
testServiceRegistration(t, "web", regMutatePort), false)
},
WantEvents: []stream.Event{
// We should see the service topic update but not connect since proxy
@ -258,20 +258,20 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
Name: "connect sidecar mutate sidecar",
Setup: func(s *Store, tx *txn) error {
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "db")); err != nil {
testServiceRegistration(t, "db"), false); err != nil {
return err
}
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web")); err != nil {
testServiceRegistration(t, "web"), false); err != nil {
return err
}
return s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regSidecar))
testServiceRegistration(t, "web", regSidecar), false)
},
Mutate: func(s *Store, tx *txn) error {
// Change port of the sidecar service instance
return s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regSidecar, regMutatePort))
testServiceRegistration(t, "web", regSidecar, regMutatePort), false)
},
WantEvents: []stream.Event{
// We should see the proxy service topic update and a connect update
@ -295,24 +295,24 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
Name: "connect sidecar rename service",
Setup: func(s *Store, tx *txn) error {
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "db")); err != nil {
testServiceRegistration(t, "db"), false); err != nil {
return err
}
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web")); err != nil {
testServiceRegistration(t, "web"), false); err != nil {
return err
}
return s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regSidecar))
testServiceRegistration(t, "web", regSidecar), false)
},
Mutate: func(s *Store, tx *txn) error {
// Change service name but not ID, update proxy too
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regRenameService)); err != nil {
testServiceRegistration(t, "web", regRenameService), false); err != nil {
return err
}
return s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regSidecar, regRenameService))
testServiceRegistration(t, "web", regSidecar, regRenameService), false)
},
WantEvents: []stream.Event{
// We should see events to deregister the old service instance and the
@ -353,25 +353,25 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
Setup: func(s *Store, tx *txn) error {
// Register a web_changed service
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web_changed")); err != nil {
testServiceRegistration(t, "web_changed"), false); err != nil {
return err
}
// Also a web
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web")); err != nil {
testServiceRegistration(t, "web"), false); err != nil {
return err
}
// And a sidecar initially for web, will be moved to target web_changed
// in Mutate.
return s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regSidecar))
testServiceRegistration(t, "web", regSidecar), false)
},
Mutate: func(s *Store, tx *txn) error {
// Change only the destination service of the proxy without a service
// rename or deleting and recreating the proxy. This is far fetched but
// still valid.
return s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regSidecar, regRenameService))
testServiceRegistration(t, "web", regSidecar, regRenameService), false)
},
WantEvents: []stream.Event{
// We should only see service health events for the sidecar service
@ -405,17 +405,17 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
Setup: func(s *Store, tx *txn) error {
// Register a db service
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "db")); err != nil {
testServiceRegistration(t, "db"), false); err != nil {
return err
}
// Also a web
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web")); err != nil {
testServiceRegistration(t, "web"), false); err != nil {
return err
}
// With a connect sidecar
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regSidecar)); err != nil {
testServiceRegistration(t, "web", regSidecar), false); err != nil {
return err
}
return nil
@ -423,7 +423,7 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
Mutate: func(s *Store, tx *txn) error {
// Change only the node meta.
return s.ensureRegistrationTxn(tx, tx.Index, false,
testNodeRegistration(t, regNodeMeta))
testNodeRegistration(t, regNodeMeta), false)
},
WantEvents: []stream.Event{
// We should see updates for all services and a connect update for the
@ -463,17 +463,17 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
Setup: func(s *Store, tx *txn) error {
// Register a db service
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "db")); err != nil {
testServiceRegistration(t, "db"), false); err != nil {
return err
}
// Also a web
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web")); err != nil {
testServiceRegistration(t, "web"), false); err != nil {
return err
}
// With a connect sidecar
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regSidecar)); err != nil {
testServiceRegistration(t, "web", regSidecar), false); err != nil {
return err
}
return nil
@ -485,17 +485,17 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
// services registered afterwards.
// Register a db service
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "db", regRenameNode)); err != nil {
testServiceRegistration(t, "db", regRenameNode), false); err != nil {
return err
}
// Also a web
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regRenameNode)); err != nil {
testServiceRegistration(t, "web", regRenameNode), false); err != nil {
return err
}
// With a connect sidecar
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regSidecar, regRenameNode)); err != nil {
testServiceRegistration(t, "web", regSidecar, regRenameNode), false); err != nil {
return err
}
return nil
@ -540,17 +540,17 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
Setup: func(s *Store, tx *txn) error {
// Register a db service
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "db")); err != nil {
testServiceRegistration(t, "db"), false); err != nil {
return err
}
// Also a web
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web")); err != nil {
testServiceRegistration(t, "web"), false); err != nil {
return err
}
// With a connect sidecar
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regSidecar)); err != nil {
testServiceRegistration(t, "web", regSidecar), false); err != nil {
return err
}
return nil
@ -558,7 +558,7 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
Mutate: func(s *Store, tx *txn) error {
// Change only the node-level check status
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regNodeCheckFail)); err != nil {
testServiceRegistration(t, "web", regNodeCheckFail), false); err != nil {
return err
}
return nil
@ -600,17 +600,17 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
Setup: func(s *Store, tx *txn) error {
// Register a db service
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "db")); err != nil {
testServiceRegistration(t, "db"), false); err != nil {
return err
}
// Also a web
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web")); err != nil {
testServiceRegistration(t, "web"), false); err != nil {
return err
}
// With a connect sidecar
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regSidecar)); err != nil {
testServiceRegistration(t, "web", regSidecar), false); err != nil {
return err
}
return nil
@ -618,7 +618,7 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
Mutate: func(s *Store, tx *txn) error {
// Change the service-level check status
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regServiceCheckFail)); err != nil {
testServiceRegistration(t, "web", regServiceCheckFail), false); err != nil {
return err
}
// Also change the service-level check status for the proxy. This is
@ -626,7 +626,7 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
// - the proxies check would get updated at roughly the same time as the
// target service check updates.
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regSidecar, regServiceCheckFail)); err != nil {
testServiceRegistration(t, "web", regSidecar, regServiceCheckFail), false); err != nil {
return err
}
return nil
@ -663,17 +663,17 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
Setup: func(s *Store, tx *txn) error {
// Register a db service
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "db")); err != nil {
testServiceRegistration(t, "db"), false); err != nil {
return err
}
// Also a web
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web")); err != nil {
testServiceRegistration(t, "web"), false); err != nil {
return err
}
// With a connect sidecar
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regSidecar)); err != nil {
testServiceRegistration(t, "web", regSidecar), false); err != nil {
return err
}
return nil
@ -717,17 +717,17 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
Setup: func(s *Store, tx *txn) error {
// Register a db service
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "db")); err != nil {
testServiceRegistration(t, "db"), false); err != nil {
return err
}
// Also a web
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web")); err != nil {
testServiceRegistration(t, "web"), false); err != nil {
return err
}
// With a connect sidecar
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regSidecar)); err != nil {
testServiceRegistration(t, "web", regSidecar), false); err != nil {
return err
}
return nil
@ -777,19 +777,19 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
// Register a db service
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "db")); err != nil {
testServiceRegistration(t, "db"), false); err != nil {
return err
}
// Node2
// Also a web
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regNode2)); err != nil {
testServiceRegistration(t, "web", regNode2), false); err != nil {
return err
}
// With a connect sidecar
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regSidecar, regNode2)); err != nil {
testServiceRegistration(t, "web", regSidecar, regNode2), false); err != nil {
return err
}
@ -808,17 +808,17 @@ func TestServiceHealthEventsFromChanges(t *testing.T) {
// Register those on node1
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web")); err != nil {
testServiceRegistration(t, "web"), false); err != nil {
return err
}
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "web", regSidecar)); err != nil {
testServiceRegistration(t, "web", regSidecar), false); err != nil {
return err
}
// And for good measure, add a new connect-native service to node2
if err := s.ensureRegistrationTxn(tx, tx.Index, false,
testServiceRegistration(t, "api", regConnectNative, regNode2)); err != nil {
testServiceRegistration(t, "api", regConnectNative, regNode2), false); err != nil {
return err
}

2
agent/consul/state/catalog_oss.go
View file

@ -323,7 +323,7 @@ func catalogChecksForNodeService(tx ReadTxn, node string, service string, entMet
return tx.Get("checks", "node_service", node, service)
}
func validateRegisterRequestTxn(_ ReadTxn, _ *structs.RegisterRequest) (*structs.EnterpriseMeta, error) {
func validateRegisterRequestTxn(_ ReadTxn, _ *structs.RegisterRequest, _ bool) (*structs.EnterpriseMeta, error) {
return nil, nil
}

4
agent/consul/state/kvs.go
View file

@ -69,7 +69,7 @@ func (s *Snapshot) Tombstones() (memdb.ResultIterator, error) {
// KVS is used when restoring from a snapshot. Use KVSSet for general inserts.
func (s *Restore) KVS(entry *structs.DirEntry) error {
if err := insertKVTxn(s.tx, entry, true); err != nil {
if err := insertKVTxn(s.tx, entry, true, true); err != nil {
return fmt.Errorf("failed inserting kvs entry: %s", err)
}
@ -153,7 +153,7 @@ func kvsSetTxn(tx WriteTxn, idx uint64, entry *structs.DirEntry, updateSession b
entry.ModifyIndex = idx
// Store the kv pair in the state store and update the index.
if err := insertKVTxn(tx, entry, false); err != nil {
if err := insertKVTxn(tx, entry, false, false); err != nil {
return fmt.Errorf("failed inserting kvs entry: %s", err)
}

2
agent/consul/state/kvs_oss.go
View file

@ -16,7 +16,7 @@ func kvsIndexer() *memdb.StringFieldIndex {
}
}
func insertKVTxn(tx WriteTxn, entry *structs.DirEntry, updateMax bool) error {
func insertKVTxn(tx WriteTxn, entry *structs.DirEntry, updateMax bool, _ bool) error {
if err := tx.Insert("kvs", entry); err != nil {
return err
}

4
agent/consul/state/session.go
View file

@ -146,7 +146,7 @@ func (s *Snapshot) Sessions() (memdb.ResultIterator, error) {
// Session is used when restoring from a snapshot. For general inserts, use
// SessionCreate.
func (s *Restore) Session(sess *structs.Session) error {
if err := insertSessionTxn(s.tx, sess, sess.ModifyIndex, true); err != nil {
if err := insertSessionTxn(s.tx, sess, sess.ModifyIndex, true, true); err != nil {
return fmt.Errorf("failed inserting session: %s", err)
}
@ -213,7 +213,7 @@ func sessionCreateTxn(tx *txn, idx uint64, sess *structs.Session) error {
}
// Insert the session
if err := insertSessionTxn(tx, sess, idx, false); err != nil {
if err := insertSessionTxn(tx, sess, idx, false, false); err != nil {
return fmt.Errorf("failed inserting session: %s", err)
}

2
agent/consul/state/session_oss.go
View file

@ -48,7 +48,7 @@ func sessionDeleteWithSession(tx WriteTxn, session *structs.Session, idx uint64)
return nil
}
func insertSessionTxn(tx *txn, session *structs.Session, idx uint64, updateMax bool) error {
func insertSessionTxn(tx *txn, session *structs.Session, idx uint64, updateMax bool, _ bool) error {
if err := tx.Insert("sessions", session); err != nil {
return err
}