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

sessions partitioning tests (#11734) 

* state: port KV and Tombstone tables to new pattern

* go fmt'ed

* handle wildcards for tombstones

* Fix graveyard ent vs oss

* fix oss compilation error

* add partition to tombstones and kv state store indexes

* refactor to use `indexWithEnterpriseIndexable`

* Apply suggestions from code review

Co-authored-by: Chris S. Kim <ckim@hashicorp.com>
Co-authored-by: R.B. Boyer <4903+rboyer@users.noreply.github.com>

* add `singleValueID` implementation assertions

* partition `tableSessions` table

* fix sessions to use UUID and fix prefix index

* fix oss build

* clean up unused functions

* fix oss compilation

* add a partition indexer for sessions

* Fix oss to not have partition index

* fix oss tests

* remove unused operations_ent.go and operations_oss.go func

* remove unused const

* convert `IndexID` of `session_checks` table

* convert `indexSession` of `session_checks` table

* convert `indexNodeCheck` of `session_checks` table

* partition `indexID` and `indexSession` of `tableSessionChecks`

* fix oss linter

* fix review comments

* remove partition for Checks as it's always use the session partition

* fix tests

* fix tests

* do not namespace nodeChecks index

Co-authored-by: Daniel Nephin <dnephin@hashicorp.com>
Co-authored-by: Chris S. Kim <ckim@hashicorp.com>
Co-authored-by: R.B. Boyer <4903+rboyer@users.noreply.github.com>
This commit is contained in:
Dhia Ayachi 2021-12-03 15:36:07 -05:00 committed by GitHub
parent b10e69ffda
commit a8874c65f7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 16 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
agent/consul/state/catalog.go
View File

@ -666,7 +666,7 @@ func (s *Store) deleteNodeTxn(tx WriteTxn, idx uint64, nodeName string, entMeta
} }
// Invalidate any sessions for this node. // Invalidate any sessions for this node.
toDelete, err := allNodeSessionsTxn(tx, nodeName) toDelete, err := allNodeSessionsTxn(tx, nodeName, entMeta.PartitionOrDefault())
if err != nil { if err != nil {
return err return err
} }
@ -2792,7 +2792,7 @@ func parseNodes(tx ReadTxn, ws memdb.WatchSet, idx uint64,
// checkSessionsTxn returns the IDs of all sessions associated with a health check // checkSessionsTxn returns the IDs of all sessions associated with a health check
func checkSessionsTxn(tx ReadTxn, hc *structs.HealthCheck) ([]*sessionCheck, error) { func checkSessionsTxn(tx ReadTxn, hc *structs.HealthCheck) ([]*sessionCheck, error) {
mappings, err := tx.Get(tableSessionChecks, indexNodeCheck, MultiQuery{Value: []string{hc.Node, string(hc.CheckID)}, mappings, err := tx.Get(tableSessionChecks, indexNodeCheck, MultiQuery{Value: []string{hc.Node, string(hc.CheckID)},
EnterpriseMeta: hc.EnterpriseMeta}) EnterpriseMeta: *structs.DefaultEnterpriseMetaInPartition(hc.PartitionOrDefault())})
if err != nil { if err != nil {
return nil, fmt.Errorf("failed session checks lookup: %s", err) return nil, fmt.Errorf("failed session checks lookup: %s", err)
} }

2
agent/consul/state/session.go
View File

@ -268,7 +268,7 @@ func sessionCreateTxn(tx WriteTxn, idx uint64, sess *structs.Session) error {
sess.ModifyIndex = idx sess.ModifyIndex = idx
// Check that the node exists // Check that the node exists
node, err := tx.First(tableNodes, indexID, Query{Value: sess.Node}) node, err := tx.First(tableNodes, indexID, Query{Value: sess.Node, EnterpriseMeta: *structs.DefaultEnterpriseMetaInPartition(sess.PartitionOrDefault())})
if err != nil { if err != nil {
return fmt.Errorf("failed node lookup: %s", err) return fmt.Errorf("failed node lookup: %s", err)
} }

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

@ -116,7 +116,7 @@ func insertSessionTxn(tx WriteTxn, session *structs.Session, idx uint64, updateM
return nil return nil
} }
func allNodeSessionsTxn(tx ReadTxn, node string) (structs.Sessions, error) { func allNodeSessionsTxn(tx ReadTxn, node string, _ string) (structs.Sessions, error) {
return nodeSessionsTxn(tx, nil, node, nil) return nodeSessionsTxn(tx, nil, node, nil)
} }

18
agent/consul/state/state_store_test.go
View File

@ -170,14 +170,20 @@ func testRegisterIngressService(t *testing.T, s *Store, idx uint64, nodeID, serv
t.Fatalf("bad service: %#v", result) t.Fatalf("bad service: %#v", result)
} }
} }
func testRegisterCheck(t *testing.T, s *Store, idx uint64, func testRegisterCheck(t *testing.T, s *Store, idx uint64,
nodeID string, serviceID string, checkID types.CheckID, state string) { nodeID string, serviceID string, checkID types.CheckID, state string) {
testRegisterCheckWithPartition(t, s, idx,
nodeID, serviceID, checkID, state, "")
}
func testRegisterCheckWithPartition(t *testing.T, s *Store, idx uint64,
nodeID string, serviceID string, checkID types.CheckID, state string, partition string) {
chk := &structs.HealthCheck{ chk := &structs.HealthCheck{
Node: nodeID, Node: nodeID,
CheckID: checkID, CheckID: checkID,
ServiceID: serviceID, ServiceID: serviceID,
Status: state, Status: state,
EnterpriseMeta: *structs.DefaultEnterpriseMetaInPartition(partition),
} }
if err := s.EnsureCheck(idx, chk); err != nil { if err := s.EnsureCheck(idx, chk); err != nil {
t.Fatalf("err: %s", err) t.Fatalf("err: %s", err)
@ -185,7 +191,7 @@ func testRegisterCheck(t *testing.T, s *Store, idx uint64,
tx := s.db.Txn(false) tx := s.db.Txn(false)
defer tx.Abort() defer tx.Abort()
c, err := tx.First(tableChecks, indexID, NodeCheckQuery{Node: nodeID, CheckID: string(checkID)}) c, err := tx.First(tableChecks, indexID, NodeCheckQuery{Node: nodeID, CheckID: string(checkID), EnterpriseMeta: *structs.DefaultEnterpriseMetaInPartition(partition)})
if err != nil { if err != nil {
t.Fatalf("err: %s", err) t.Fatalf("err: %s", err)
} }