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

Makes the iterator naming more consistent.

This commit is contained in:
James Phillips 2015-10-19 15:51:11 -07:00
parent 9a2fdff4c4
commit 22a46f7bf5
5 changed files with 86 additions and 91 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
consul/fsm.go
View File

@ -400,11 +400,11 @@ func (s *consulSnapshot) persistNodes(sink raft.SnapshotSink,
}
// Register each node
for ni := nodes.Next(); ni != nil; ni = nodes.Next() {
node := ni.(*structs.Node)
for node := nodes.Next(); node != nil; node = nodes.Next() {
n := node.(*structs.Node)
req := structs.RegisterRequest{
Node: node.Node,
Address: node.Address,
Node: n.Node,
Address: n.Address,
}
// Register the node itself
@ -414,13 +414,13 @@ func (s *consulSnapshot) persistNodes(sink raft.SnapshotSink,
}
// Register each service this node has
services, err := s.state.Services(node.Node)
services, err := s.state.Services(n.Node)
if err != nil {
return err
}
for si := services.Next(); si != nil; si = services.Next() {
req.Service = si.(*structs.ServiceNode).ToNodeService()
for service := services.Next(); service != nil; service = services.Next() {
sink.Write([]byte{byte(structs.RegisterRequestType)})
req.Service = service.(*structs.ServiceNode).ToNodeService()
if err := encoder.Encode(&req); err != nil {
return err
}
@ -428,13 +428,13 @@ func (s *consulSnapshot) persistNodes(sink raft.SnapshotSink,
// Register each check this node has
req.Service = nil
checks, err := s.state.Checks(node.Node)
checks, err := s.state.Checks(n.Node)
if err != nil {
return err
}
for ci := checks.Next(); ci != nil; ci = checks.Next() {
req.Check = ci.(*structs.HealthCheck)
for check := checks.Next(); check != nil; check = checks.Next() {
sink.Write([]byte{byte(structs.RegisterRequestType)})
req.Check = check.(*structs.HealthCheck)
if err := encoder.Encode(&req); err != nil {
return err
}
@ -445,14 +445,14 @@ func (s *consulSnapshot) persistNodes(sink raft.SnapshotSink,
func (s *consulSnapshot) persistSessions(sink raft.SnapshotSink,
encoder *codec.Encoder) error {
iter, err := s.state.Sessions()
sessions, err := s.state.Sessions()
if err != nil {
return err
}
for si := iter.Next(); si != nil; si = iter.Next() {
for session := sessions.Next(); session != nil; session = sessions.Next() {
sink.Write([]byte{byte(structs.SessionRequestType)})
if err := encoder.Encode(si.(*structs.Session)); err != nil {
if err := encoder.Encode(session.(*structs.Session)); err != nil {
return err
}
}
@ -461,14 +461,14 @@ func (s *consulSnapshot) persistSessions(sink raft.SnapshotSink,
func (s *consulSnapshot) persistACLs(sink raft.SnapshotSink,
encoder *codec.Encoder) error {
iter, err := s.state.ACLs()
acls, err := s.state.ACLs()
if err != nil {
return err
}
for ai := iter.Next(); ai != nil; ai = iter.Next() {
for acl := acls.Next(); acl != nil; acl = acls.Next() {
sink.Write([]byte{byte(structs.ACLRequestType)})
if err := encoder.Encode(ai.(*structs.ACL)); err != nil {
if err := encoder.Encode(acl.(*structs.ACL)); err != nil {
return err
}
}
@ -477,14 +477,14 @@ func (s *consulSnapshot) persistACLs(sink raft.SnapshotSink,
func (s *consulSnapshot) persistKVs(sink raft.SnapshotSink,
encoder *codec.Encoder) error {
iter, err := s.state.KVs()
entries, err := s.state.KVs()
if err != nil {
return err
}
for ki := iter.Next(); ki != nil; ki = iter.Next() {
for entry := entries.Next(); entry != nil; entry = entries.Next() {
sink.Write([]byte{byte(structs.KVSRequestType)})
if err := encoder.Encode(ki.(*structs.DirEntry)); err != nil {
if err := encoder.Encode(entry.(*structs.DirEntry)); err != nil {
return err
}
}
@ -493,22 +493,22 @@ func (s *consulSnapshot) persistKVs(sink raft.SnapshotSink,
func (s *consulSnapshot) persistTombstones(sink raft.SnapshotSink,
encoder *codec.Encoder) error {
iter, err := s.state.Tombstones()
stones, err := s.state.Tombstones()
if err != nil {
return err
}
for ti := iter.Next(); ti != nil; ti = iter.Next() {
for stone := stones.Next(); stone != nil; stone = stones.Next() {
sink.Write([]byte{byte(structs.TombstoneRequestType)})
// For historical reasons, these are serialized in the snapshots
// as KV entries. We want to keep the snapshot format compatible
// with pre-0.6 versions for now.
stone := ti.(*state.Tombstone)
s := stone.(*state.Tombstone)
fake := &structs.DirEntry{
Key: stone.Key,
Key: s.Key,
RaftIndex: structs.RaftIndex{
ModifyIndex: stone.Index,
ModifyIndex: s.Index,
},
}
if err := encoder.Encode(fake); err != nil {

10
consul/fsm_test.go
View File

@ -475,18 +475,18 @@ func TestFSM_SnapshotRestore(t *testing.T) {
func() {
snap := fsm2.state.Snapshot()
defer snap.Close()
iter, err := snap.Tombstones()
stones, err := snap.Tombstones()
if err != nil {
t.Fatalf("err: %s", err)
}
stone := iter.Next().(*state.Tombstone)
stone := stones.Next().(*state.Tombstone)
if stone == nil {
t.Fatalf("missing tombstone")
}
if stone.Key != "/remove" || stone.Index != 12 {
t.Fatalf("bad: %v", stone)
}
if iter.Next() != nil {
if stones.Next() != nil {
t.Fatalf("unexpected extra tombstones")
}
}()
@ -1023,11 +1023,11 @@ func TestFSM_TombstoneReap(t *testing.T) {
// Verify the tombstones are gone
snap := fsm.state.Snapshot()
defer snap.Close()
iter, err := snap.Tombstones()
stones, err := snap.Tombstones()
if err != nil {
t.Fatalf("err: %s", err)
}
if iter.Next() != nil {
if stones.Next() != nil {
t.Fatalf("unexpected extra tombstones")
}
}

10
consul/leader_test.go
View File

@ -590,14 +590,14 @@ func TestLeader_ReapTombstones(t *testing.T) {
func() {
snap := state.Snapshot()
defer snap.Close()
iter, err := snap.Tombstones()
stones, err := snap.Tombstones()
if err != nil {
t.Fatalf("err: %s", err)
}
if iter.Next() == nil {
if stones.Next() == nil {
t.Fatalf("missing tombstones")
}
if iter.Next() != nil {
if stones.Next() != nil {
t.Fatalf("unexpected extra tombstones")
}
}()
@ -607,11 +607,11 @@ func TestLeader_ReapTombstones(t *testing.T) {
testutil.WaitForResult(func() (bool, error) {
snap := state.Snapshot()
defer snap.Close()
iter, err := snap.Tombstones()
stones, err := snap.Tombstones()
if err != nil {
return false, err
}
return iter.Next() == nil, nil
return stones.Next() == nil, nil
}, func(err error) {
t.Fatalf("err: %v", err)
})

57
consul/state/state_store.go
View File

@ -464,8 +464,7 @@ func (s *StateStore) deleteNodeTxn(tx *memdb.Txn, idx uint64, nodeID string) err
}
var sids []string
for service := services.Next(); service != nil; service = services.Next() {
svc := service.(*structs.ServiceNode)
sids = append(sids, svc.ServiceID)
sids = append(sids, service.(*structs.ServiceNode).ServiceID)
}
// Do the delete in a separate loop so we don't trash the iterator.
@ -483,8 +482,7 @@ func (s *StateStore) deleteNodeTxn(tx *memdb.Txn, idx uint64, nodeID string) err
}
var cids []string
for check := checks.Next(); check != nil; check = checks.Next() {
hc := check.(*structs.HealthCheck)
cids = append(cids, hc.CheckID)
cids = append(cids, check.(*structs.HealthCheck).CheckID)
}
// Do the delete in a separate loop so we don't trash the iterator.
@ -596,13 +594,13 @@ func (s *StateStore) Services() (uint64, structs.Services, error) {
// tags.
unique := make(map[string]map[string]struct{})
for service := services.Next(); service != nil; service = services.Next() {
sn := service.(*structs.ServiceNode)
tags, ok := unique[sn.ServiceName]
svc := service.(*structs.ServiceNode)
tags, ok := unique[svc.ServiceName]
if !ok {
unique[sn.ServiceName] = make(map[string]struct{})
tags = unique[sn.ServiceName]
unique[svc.ServiceName] = make(map[string]struct{})
tags = unique[svc.ServiceName]
}
for _, tag := range sn.ServiceTags {
for _, tag := range svc.ServiceTags {
tags[tag] = struct{}{}
}
}
@ -618,8 +616,8 @@ func (s *StateStore) Services() (uint64, structs.Services, error) {
return idx, results, nil
}
// ServiceNodes returns the nodes associated with a given service.
func (s *StateStore) ServiceNodes(service string) (uint64, structs.ServiceNodes, error) {
// ServiceNodes returns the nodes associated with a given service name.
func (s *StateStore) ServiceNodes(serviceName string) (uint64, structs.ServiceNodes, error) {
tx := s.db.Txn(false)
defer tx.Abort()
@ -627,13 +625,13 @@ func (s *StateStore) ServiceNodes(service string) (uint64, structs.ServiceNodes,
idx := maxIndexTxn(tx, s.getWatchTables("ServiceNodes")...)
// List all the services.
services, err := tx.Get("services", "service", service)
services, err := tx.Get("services", "service", serviceName)
if err != nil {
return 0, nil, fmt.Errorf("failed service lookup: %s", err)
}
var results structs.ServiceNodes
for s := services.Next(); s != nil; s = services.Next() {
results = append(results, s.(*structs.ServiceNode))
for service := services.Next(); service != nil; service = services.Next() {
results = append(results, service.(*structs.ServiceNode))
}
// Fill in the address details.
@ -661,10 +659,10 @@ func (s *StateStore) ServiceTagNodes(service, tag string) (uint64, structs.Servi
// Gather all the services and apply the tag filter.
var results structs.ServiceNodes
for s := services.Next(); s != nil; s = services.Next() {
sn := s.(*structs.ServiceNode)
if !serviceTagFilter(sn, tag) {
results = append(results, sn)
for service := services.Next(); service != nil; service = services.Next() {
svc := service.(*structs.ServiceNode)
if !serviceTagFilter(svc, tag) {
results = append(results, svc)
}
}
@ -788,8 +786,7 @@ func (s *StateStore) deleteServiceTxn(tx *memdb.Txn, idx uint64, watches *DumbWa
}
var cids []string
for check := checks.Next(); check != nil; check = checks.Next() {
hc := check.(*structs.HealthCheck)
cids = append(cids, hc.CheckID)
cids = append(cids, check.(*structs.HealthCheck).CheckID)
}
// Do the delete in a separate loop so we don't trash the iterator.
@ -1046,7 +1043,7 @@ func (s *StateStore) deleteCheckTxn(tx *memdb.Txn, idx uint64, watches *DumbWatc
// The results are compounded into a CheckServiceNodes, and the index returned
// is the maximum index observed over any node, check, or service in the result
// set.
func (s *StateStore) CheckServiceNodes(service string) (uint64, structs.CheckServiceNodes, error) {
func (s *StateStore) CheckServiceNodes(serviceName string) (uint64, structs.CheckServiceNodes, error) {
tx := s.db.Txn(false)
defer tx.Abort()
@ -1054,15 +1051,15 @@ func (s *StateStore) CheckServiceNodes(service string) (uint64, structs.CheckSer
idx := maxIndexTxn(tx, s.getWatchTables("CheckServiceNodes")...)
// Query the state store for the service.
services, err := tx.Get("services", "service", service)
services, err := tx.Get("services", "service", serviceName)
if err != nil {
return 0, nil, fmt.Errorf("failed service lookup: %s", err)
}
// Return the results.
var results structs.ServiceNodes
for s := services.Next(); s != nil; s = services.Next() {
results = append(results, s.(*structs.ServiceNode))
for service := services.Next(); service != nil; service = services.Next() {
results = append(results, service.(*structs.ServiceNode))
}
return s.parseCheckServiceNodes(tx, idx, results, err)
}
@ -1071,7 +1068,7 @@ func (s *StateStore) CheckServiceNodes(service string) (uint64, structs.CheckSer
// service, filtering out services that don't contain the given tag. The results
// are compounded into a CheckServiceNodes, and the index returned is the maximum
// index observed over any node, check, or service in the result set.
func (s *StateStore) CheckServiceTagNodes(service, tag string) (uint64, structs.CheckServiceNodes, error) {
func (s *StateStore) CheckServiceTagNodes(serviceName, tag string) (uint64, structs.CheckServiceNodes, error) {
tx := s.db.Txn(false)
defer tx.Abort()
@ -1079,17 +1076,17 @@ func (s *StateStore) CheckServiceTagNodes(service, tag string) (uint64, structs.
idx := maxIndexTxn(tx, s.getWatchTables("CheckServiceNodes")...)
// Query the state store for the service.
services, err := tx.Get("services", "service", service)
services, err := tx.Get("services", "service", serviceName)
if err != nil {
return 0, nil, fmt.Errorf("failed service lookup: %s", err)
}
// Return the results, filtering by tag.
var results structs.ServiceNodes
for s := services.Next(); s != nil; s = services.Next() {
sn := s.(*structs.ServiceNode)
if !serviceTagFilter(sn, tag) {
results = append(results, sn)
for service := services.Next(); service != nil; service = services.Next() {
svc := service.(*structs.ServiceNode)
if !serviceTagFilter(svc, tag) {
results = append(results, svc)
}
}
return s.parseCheckServiceNodes(tx, idx, results, err)

52
consul/state/state_store_test.go
View File

@ -286,11 +286,11 @@ func TestStateStore_ReapTombstones(t *testing.T) {
// Make sure the tombstones are actually gone.
snap := s.Snapshot()
defer snap.Close()
iter, err := snap.Tombstones()
stones, err := snap.Tombstones()
if err != nil {
t.Fatalf("err: %s", err)
}
if iter.Next() != nil {
if stones.Next() != nil {
t.Fatalf("unexpected extra tombstones")
}
}
@ -686,8 +686,8 @@ func TestStateStore_DeleteNode(t *testing.T) {
if err != nil {
t.Fatalf("err: %s", err)
}
if s := services.Next(); s != nil {
t.Fatalf("bad: %#v", s)
if service := services.Next(); service != nil {
t.Fatalf("bad: %#v", service)
}
// Associated health check was removed.
@ -695,8 +695,8 @@ func TestStateStore_DeleteNode(t *testing.T) {
if err != nil {
t.Fatalf("err: %s", err)
}
if c := checks.Next(); c != nil {
t.Fatalf("bad: %#v", c)
if check := checks.Next(); check != nil {
t.Fatalf("bad: %#v", check)
}
// Indexes were updated.
@ -735,12 +735,12 @@ func TestStateStore_Node_Snapshot(t *testing.T) {
if idx := snap.LastIndex(); idx != 2 {
t.Fatalf("bad index: %d", idx)
}
iter, err := snap.Nodes()
nodes, err := snap.Nodes()
if err != nil {
t.Fatalf("err: %s", err)
}
for i := 0; i < 3; i++ {
node := iter.Next().(*structs.Node)
node := nodes.Next().(*structs.Node)
if node == nil {
t.Fatalf("unexpected end of nodes")
}
@ -752,7 +752,7 @@ func TestStateStore_Node_Snapshot(t *testing.T) {
t.Fatalf("bad: %#v", node)
}
}
if iter.Next() != nil {
if nodes.Next() != nil {
t.Fatalf("unexpected extra nodes")
}
}
@ -1276,12 +1276,12 @@ func TestStateStore_Service_Snapshot(t *testing.T) {
if idx := snap.LastIndex(); idx != 4 {
t.Fatalf("bad index: %d", idx)
}
iter, err := snap.Services("node1")
services, err := snap.Services("node1")
if err != nil {
t.Fatalf("err: %s", err)
}
for i := 0; i < len(ns); i++ {
svc := iter.Next().(*structs.ServiceNode)
svc := services.Next().(*structs.ServiceNode)
if svc == nil {
t.Fatalf("unexpected end of services")
}
@ -1291,7 +1291,7 @@ func TestStateStore_Service_Snapshot(t *testing.T) {
t.Fatalf("bad: %#v != %#v", svc, ns[i])
}
}
if iter.Next() != nil {
if services.Next() != nil {
t.Fatalf("unexpected extra services")
}
}
@ -3070,8 +3070,8 @@ func TestStateStore_KVS_Snapshot_Restore(t *testing.T) {
t.Fatalf("err: %s", err)
}
var dump structs.DirEntries
for ki := iter.Next(); ki != nil; ki = iter.Next() {
dump = append(dump, ki.(*structs.DirEntry))
for entry := iter.Next(); entry != nil; entry = iter.Next() {
dump = append(dump, entry.(*structs.DirEntry))
}
if !reflect.DeepEqual(dump, entries) {
t.Fatalf("bad: %#v", dump)
@ -3094,10 +3094,8 @@ func TestStateStore_KVS_Snapshot_Restore(t *testing.T) {
if idx != 7 {
t.Fatalf("bad index: %d", idx)
}
for i, entry := range res {
if !reflect.DeepEqual(entry, entries[i]) {
t.Fatalf("bad: %#v", entry)
}
if !reflect.DeepEqual(res, entries) {
t.Fatalf("bad: %#v", res)
}
// Check that the index was updated.
@ -3264,13 +3262,13 @@ func TestStateStore_Tombstone_Snapshot_Restore(t *testing.T) {
}
// Verify the snapshot.
iter, err := snap.Tombstones()
stones, err := snap.Tombstones()
if err != nil {
t.Fatalf("err: %s", err)
}
var dump []*Tombstone
for ti := iter.Next(); ti != nil; ti = iter.Next() {
dump = append(dump, ti.(*Tombstone))
for stone := stones.Next(); stone != nil; stone = stones.Next() {
dump = append(dump, stone.(*Tombstone))
}
if len(dump) != 1 {
t.Fatalf("bad %#v", dump)
@ -3316,11 +3314,11 @@ func TestStateStore_Tombstone_Snapshot_Restore(t *testing.T) {
// But make sure the tombstone is actually gone.
snap := s.Snapshot()
defer snap.Close()
iter, err := snap.Tombstones()
stones, err := snap.Tombstones()
if err != nil {
t.Fatalf("err: %s", err)
}
if iter.Next() != nil {
if stones.Next() != nil {
t.Fatalf("unexpected extra tombstones")
}
}()
@ -3665,8 +3663,8 @@ func TestStateStore_Session_Snapshot_Restore(t *testing.T) {
t.Fatalf("err: %s", err)
}
var dump structs.Sessions
for si := iter.Next(); si != nil; si = iter.Next() {
dump = append(dump, si.(*structs.Session))
for session := iter.Next(); session != nil; session = iter.Next() {
dump = append(dump, session.(*structs.Session))
}
if !reflect.DeepEqual(dump, sessions) {
t.Fatalf("bad: %#v", dump)
@ -4332,8 +4330,8 @@ func TestStateStore_ACL_Snapshot_Restore(t *testing.T) {
t.Fatalf("err: %s", err)
}
var dump structs.ACLs
for ai := iter.Next(); ai != nil; ai = iter.Next() {
dump = append(dump, ai.(*structs.ACL))
for acl := iter.Next(); acl != nil; acl = iter.Next() {
dump = append(dump, acl.(*structs.ACL))
}
if !reflect.DeepEqual(dump, acls) {
t.Fatalf("bad: %#v", dump)