github
/
consul
mirror of https://github.com/hashicorp/consul
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

consul: updating state store to associate changes with raft index

pull/19/head
Armon Dadgar 2014-02-04 18:33:15 -08:00
parent 49378a0323
commit 67a7d25e1c
11 changed files with 478 additions and 258 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
consul/catalog_endpoint.go
View File

@ -93,7 +93,7 @@ func (c *Catalog) ListNodes(dc string, reply *structs.Nodes) error {
// Get the current nodes // Get the current nodes
state := c.srv.fsm.State() state := c.srv.fsm.State()
nodes := state.Nodes() _, nodes := state.Nodes()
*reply = nodes *reply = nodes
return nil return nil
@ -107,7 +107,7 @@ func (c *Catalog) ListServices(dc string, reply *structs.Services) error {
// Get the current nodes // Get the current nodes
state := c.srv.fsm.State() state := c.srv.fsm.State()
services := state.Services() _, services := state.Services()
*reply = services *reply = services
return nil return nil
@ -128,9 +128,9 @@ func (c *Catalog) ServiceNodes(args *structs.ServiceSpecificRequest, reply *stru
state := c.srv.fsm.State() state := c.srv.fsm.State()
var nodes structs.ServiceNodes var nodes structs.ServiceNodes
if args.TagFilter { if args.TagFilter {
nodes = state.ServiceTagNodes(args.ServiceName, args.ServiceTag) _, nodes = state.ServiceTagNodes(args.ServiceName, args.ServiceTag)
} else { } else {
nodes = state.ServiceNodes(args.ServiceName) _, nodes = state.ServiceNodes(args.ServiceName)
} }
*reply = nodes *reply = nodes
@ -150,7 +150,7 @@ func (c *Catalog) NodeServices(args *structs.NodeSpecificRequest, reply *structs
// Get the node services // Get the node services
state := c.srv.fsm.State() state := c.srv.fsm.State()
services := state.NodeServices(args.Node) _, services := state.NodeServices(args.Node)
*reply = *services *reply = *services
return nil return nil

18
consul/catalog_endpoint_test.go
View File

@ -207,7 +207,7 @@ func TestCatalogListNodes(t *testing.T) {
time.Sleep(100 * time.Millisecond) time.Sleep(100 * time.Millisecond)
// Just add a node // Just add a node
s1.fsm.State().EnsureNode(structs.Node{"foo", "127.0.0.1"}) s1.fsm.State().EnsureNode(1, structs.Node{"foo", "127.0.0.1"})
if err := client.Call("Catalog.ListNodes", "dc1", &out); err != nil { if err := client.Call("Catalog.ListNodes", "dc1", &out); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
@ -240,7 +240,7 @@ func BenchmarkCatalogListNodes(t *testing.B) {
time.Sleep(100 * time.Millisecond) time.Sleep(100 * time.Millisecond)
// Just add a node // Just add a node
s1.fsm.State().EnsureNode(structs.Node{"foo", "127.0.0.1"}) s1.fsm.State().EnsureNode(1, structs.Node{"foo", "127.0.0.1"})
for i := 0; i < t.N; i++ { for i := 0; i < t.N; i++ {
var out structs.Nodes var out structs.Nodes
@ -267,8 +267,8 @@ func TestCatalogListServices(t *testing.T) {
time.Sleep(100 * time.Millisecond) time.Sleep(100 * time.Millisecond)
// Just add a node // Just add a node
s1.fsm.State().EnsureNode(structs.Node{"foo", "127.0.0.1"}) s1.fsm.State().EnsureNode(1, structs.Node{"foo", "127.0.0.1"})
s1.fsm.State().EnsureService("foo", "db", "db", "primary", 5000) s1.fsm.State().EnsureService(2, "foo", &structs.NodeService{"db", "db", "primary", 5000})
if err := client.Call("Catalog.ListServices", "dc1", &out); err != nil { if err := client.Call("Catalog.ListServices", "dc1", &out); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
@ -312,8 +312,8 @@ func TestCatalogListServiceNodes(t *testing.T) {
time.Sleep(100 * time.Millisecond) time.Sleep(100 * time.Millisecond)
// Just add a node // Just add a node
s1.fsm.State().EnsureNode(structs.Node{"foo", "127.0.0.1"}) s1.fsm.State().EnsureNode(1, structs.Node{"foo", "127.0.0.1"})
s1.fsm.State().EnsureService("foo", "db", "db", "primary", 5000) s1.fsm.State().EnsureService(2, "foo", &structs.NodeService{"db", "db", "primary", 5000})
if err := client.Call("Catalog.ServiceNodes", &args, &out); err != nil { if err := client.Call("Catalog.ServiceNodes", &args, &out); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
@ -356,9 +356,9 @@ func TestCatalogNodeServices(t *testing.T) {
time.Sleep(100 * time.Millisecond) time.Sleep(100 * time.Millisecond)
// Just add a node // Just add a node
s1.fsm.State().EnsureNode(structs.Node{"foo", "127.0.0.1"}) s1.fsm.State().EnsureNode(1, structs.Node{"foo", "127.0.0.1"})
s1.fsm.State().EnsureService("foo", "db", "db", "primary", 5000) s1.fsm.State().EnsureService(2, "foo", &structs.NodeService{"db", "db", "primary", 5000})
s1.fsm.State().EnsureService("foo", "web", "web", "", 80) s1.fsm.State().EnsureService(3, "foo", &structs.NodeService{"web", "web", "", 80})
if err := client.Call("Catalog.NodeServices", &args, &out); err != nil { if err := client.Call("Catalog.NodeServices", &args, &out); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)

55
consul/fsm.go
View File

@ -25,6 +25,13 @@ type consulSnapshot struct {
state *StateSnapshot state *StateSnapshot
} }
// snapshotHeader is the first entry in our snapshot
type snapshotHeader struct {
// LastIndex is the last index that affects the data.
// This is used when we do the restore for watchers.
LastIndex uint64
}
// NewFSM is used to construct a new FSM with a blank state // NewFSM is used to construct a new FSM with a blank state
func NewFSM(logOutput io.Writer) (*consulFSM, error) { func NewFSM(logOutput io.Writer) (*consulFSM, error) {
state, err := NewStateStore() state, err := NewStateStore()
@ -48,34 +55,33 @@ func (c *consulFSM) Apply(log *raft.Log) interface{} {
buf := log.Data buf := log.Data
switch structs.MessageType(buf[0]) { switch structs.MessageType(buf[0]) {
case structs.RegisterRequestType: case structs.RegisterRequestType:
return c.decodeRegister(buf[1:]) return c.decodeRegister(buf[1:], log.Index)
case structs.DeregisterRequestType: case structs.DeregisterRequestType:
return c.applyDeregister(buf[1:]) return c.applyDeregister(buf[1:], log.Index)
default: default:
panic(fmt.Errorf("failed to apply request: %#v", buf)) panic(fmt.Errorf("failed to apply request: %#v", buf))
} }
} }
func (c *consulFSM) decodeRegister(buf []byte) interface{} { func (c *consulFSM) decodeRegister(buf []byte, index uint64) interface{} {
var req structs.RegisterRequest var req structs.RegisterRequest
if err := structs.Decode(buf, &req); err != nil { if err := structs.Decode(buf, &req); err != nil {
panic(fmt.Errorf("failed to decode request: %v", err)) panic(fmt.Errorf("failed to decode request: %v", err))
} }
return c.applyRegister(&req) return c.applyRegister(&req, index)
} }
func (c *consulFSM) applyRegister(req *structs.RegisterRequest) interface{} { func (c *consulFSM) applyRegister(req *structs.RegisterRequest, index uint64) interface{} {
// Ensure the node // Ensure the node
node := structs.Node{req.Node, req.Address} node := structs.Node{req.Node, req.Address}
if err := c.state.EnsureNode(node); err != nil { if err := c.state.EnsureNode(index, node); err != nil {
c.logger.Printf("[INFO] consul.fsm: EnsureNode failed: %v", err) c.logger.Printf("[INFO] consul.fsm: EnsureNode failed: %v", err)
return err return err
} }
// Ensure the service if provided // Ensure the service if provided
if req.Service != nil { if req.Service != nil {
if err := c.state.EnsureService(req.Node, req.Service.ID, req.Service.Service, if err := c.state.EnsureService(index, req.Node, req.Service); err != nil {
req.Service.Tag, req.Service.Port); err != nil {
c.logger.Printf("[INFO] consul.fsm: EnsureService failed: %v", err) c.logger.Printf("[INFO] consul.fsm: EnsureService failed: %v", err)
return err return err
} }
@ -83,7 +89,7 @@ func (c *consulFSM) applyRegister(req *structs.RegisterRequest) interface{} {
// Ensure the check if provided // Ensure the check if provided
if req.Check != nil { if req.Check != nil {
if err := c.state.EnsureCheck(req.Check); err != nil { if err := c.state.EnsureCheck(index, req.Check); err != nil {
c.logger.Printf("[INFO] consul.fsm: EnsureCheck failed: %v", err) c.logger.Printf("[INFO] consul.fsm: EnsureCheck failed: %v", err)
return err return err
} }
@ -92,7 +98,7 @@ func (c *consulFSM) applyRegister(req *structs.RegisterRequest) interface{} {
return nil return nil
} }
func (c *consulFSM) applyDeregister(buf []byte) interface{} { func (c *consulFSM) applyDeregister(buf []byte, index uint64) interface{} {
var req structs.DeregisterRequest var req structs.DeregisterRequest
if err := structs.Decode(buf, &req); err != nil { if err := structs.Decode(buf, &req); err != nil {
panic(fmt.Errorf("failed to decode request: %v", err)) panic(fmt.Errorf("failed to decode request: %v", err))
@ -100,17 +106,17 @@ func (c *consulFSM) applyDeregister(buf []byte) interface{} {
// Either remove the service entry or the whole node // Either remove the service entry or the whole node
if req.ServiceID != "" { if req.ServiceID != "" {
if err := c.state.DeleteNodeService(req.Node, req.ServiceID); err != nil { if err := c.state.DeleteNodeService(index, req.Node, req.ServiceID); err != nil {
c.logger.Printf("[INFO] consul.fsm: DeleteNodeService failed: %v", err) c.logger.Printf("[INFO] consul.fsm: DeleteNodeService failed: %v", err)
return err return err
} }
} else if req.CheckID != "" { } else if req.CheckID != "" {
if err := c.state.DeleteNodeCheck(req.Node, req.CheckID); err != nil { if err := c.state.DeleteNodeCheck(index, req.Node, req.CheckID); err != nil {
c.logger.Printf("[INFO] consul.fsm: DeleteNodeCheck failed: %v", err) c.logger.Printf("[INFO] consul.fsm: DeleteNodeCheck failed: %v", err)
return err return err
} }
} else { } else {
if err := c.state.DeleteNode(req.Node); err != nil { if err := c.state.DeleteNode(index, req.Node); err != nil {
c.logger.Printf("[INFO] consul.fsm: DeleteNode failed: %v", err) c.logger.Printf("[INFO] consul.fsm: DeleteNode failed: %v", err)
return err return err
} }
@ -145,6 +151,12 @@ func (c *consulFSM) Restore(old io.ReadCloser) error {
var handle codec.MsgpackHandle var handle codec.MsgpackHandle
dec := codec.NewDecoder(old, &handle) dec := codec.NewDecoder(old, &handle)
// Read in the header
var header snapshotHeader
if err := dec.Decode(&header); err != nil {
return err
}
// Populate the new state // Populate the new state
msgType := make([]byte, 1) msgType := make([]byte, 1)
for { for {
@ -163,7 +175,7 @@ func (c *consulFSM) Restore(old io.ReadCloser) error {
if err := dec.Decode(&req); err != nil { if err := dec.Decode(&req); err != nil {
return err return err
} }
c.applyRegister(&req) c.applyRegister(&req, header.LastIndex)
default: default:
return fmt.Errorf("Unrecognized msg type: %v", msgType) return fmt.Errorf("Unrecognized msg type: %v", msgType)
@ -174,13 +186,22 @@ func (c *consulFSM) Restore(old io.ReadCloser) error {
} }
func (s *consulSnapshot) Persist(sink raft.SnapshotSink) error { func (s *consulSnapshot) Persist(sink raft.SnapshotSink) error {
// Get all the nodes
nodes := s.state.Nodes()
// Register the nodes // Register the nodes
handle := codec.MsgpackHandle{} handle := codec.MsgpackHandle{}
encoder := codec.NewEncoder(sink, &handle) encoder := codec.NewEncoder(sink, &handle)
// Write the header
header := snapshotHeader{
LastIndex: s.state.LastIndex(),
}
if err := encoder.Encode(&header); err != nil {
sink.Cancel()
return err
}
// Get all the nodes
nodes := s.state.Nodes()
// Register each node // Register each node
var req structs.RegisterRequest var req structs.RegisterRequest
for i := 0; i < len(nodes); i++ { for i := 0; i < len(nodes); i++ {

46
consul/fsm_test.go
View File

@ -57,12 +57,14 @@ func TestFSM_RegisterNode(t *testing.T) {
} }
// Verify we are registered // Verify we are registered
if found, _ := fsm.state.GetNode("foo"); !found { if idx, found, _ := fsm.state.GetNode("foo"); !found {
t.Fatalf("not found!") t.Fatalf("not found!")
} else if idx != 1 {
t.Fatalf("bad index: %d", idx)
} }
// Verify service registered // Verify service registered
services := fsm.state.NodeServices("foo") _, services := fsm.state.NodeServices("foo")
if len(services.Services) != 0 { if len(services.Services) != 0 {
t.Fatalf("Services: %v", services) t.Fatalf("Services: %v", services)
} }
@ -103,18 +105,18 @@ func TestFSM_RegisterNode_Service(t *testing.T) {
} }
// Verify we are registered // Verify we are registered
if found, _ := fsm.state.GetNode("foo"); !found { if _, found, _ := fsm.state.GetNode("foo"); !found {
t.Fatalf("not found!") t.Fatalf("not found!")
} }
// Verify service registered // Verify service registered
services := fsm.state.NodeServices("foo") _, services := fsm.state.NodeServices("foo")
if _, ok := services.Services["db"]; !ok { if _, ok := services.Services["db"]; !ok {
t.Fatalf("not registered!") t.Fatalf("not registered!")
} }
// Verify check // Verify check
checks := fsm.state.NodeChecks("foo") _, checks := fsm.state.NodeChecks("foo")
if checks[0].CheckID != "db" { if checks[0].CheckID != "db" {
t.Fatalf("not registered!") t.Fatalf("not registered!")
} }
@ -163,12 +165,12 @@ func TestFSM_DeregisterService(t *testing.T) {
} }
// Verify we are registered // Verify we are registered
if found, _ := fsm.state.GetNode("foo"); !found { if _, found, _ := fsm.state.GetNode("foo"); !found {
t.Fatalf("not found!") t.Fatalf("not found!")
} }
// Verify service not registered // Verify service not registered
services := fsm.state.NodeServices("foo") _, services := fsm.state.NodeServices("foo")
if _, ok := services.Services["db"]; ok { if _, ok := services.Services["db"]; ok {
t.Fatalf("db registered!") t.Fatalf("db registered!")
} }
@ -217,12 +219,12 @@ func TestFSM_DeregisterCheck(t *testing.T) {
} }
// Verify we are registered // Verify we are registered
if found, _ := fsm.state.GetNode("foo"); !found { if _, found, _ := fsm.state.GetNode("foo"); !found {
t.Fatalf("not found!") t.Fatalf("not found!")
} }
// Verify check not registered // Verify check not registered
checks := fsm.state.NodeChecks("foo") _, checks := fsm.state.NodeChecks("foo")
if len(checks) != 0 { if len(checks) != 0 {
t.Fatalf("check registered!") t.Fatalf("check registered!")
} }
@ -277,18 +279,18 @@ func TestFSM_DeregisterNode(t *testing.T) {
} }
// Verify we are registered // Verify we are registered
if found, _ := fsm.state.GetNode("foo"); found { if _, found, _ := fsm.state.GetNode("foo"); found {
t.Fatalf("found!") t.Fatalf("found!")
} }
// Verify service not registered // Verify service not registered
services := fsm.state.NodeServices("foo") _, services := fsm.state.NodeServices("foo")
if len(services.Services) != 0 { if len(services.Services) != 0 {
t.Fatalf("Services: %v", services) t.Fatalf("Services: %v", services)
} }
// Verify checks not registered // Verify checks not registered
checks := fsm.state.NodeChecks("foo") _, checks := fsm.state.NodeChecks("foo")
if len(checks) != 0 { if len(checks) != 0 {
t.Fatalf("Services: %v", services) t.Fatalf("Services: %v", services)
} }
@ -301,13 +303,13 @@ func TestFSM_SnapshotRestore(t *testing.T) {
} }
// Add some state // Add some state
fsm.state.EnsureNode(structs.Node{"foo", "127.0.0.1"}) fsm.state.EnsureNode(1, structs.Node{"foo", "127.0.0.1"})
fsm.state.EnsureNode(structs.Node{"baz", "127.0.0.2"}) fsm.state.EnsureNode(2, structs.Node{"baz", "127.0.0.2"})
fsm.state.EnsureService("foo", "web", "web", "", 80) fsm.state.EnsureService(3, "foo", &structs.NodeService{"web", "web", "", 80})
fsm.state.EnsureService("foo", "db", "db", "primary", 5000) fsm.state.EnsureService(4, "foo", &structs.NodeService{"db", "db", "primary", 5000})
fsm.state.EnsureService("baz", "web", "web", "", 80) fsm.state.EnsureService(5, "baz", &structs.NodeService{"web", "web", "", 80})
fsm.state.EnsureService("baz", "db", "db", "secondary", 5000) fsm.state.EnsureService(6, "baz", &structs.NodeService{"db", "db", "secondary", 5000})
fsm.state.EnsureCheck(&structs.HealthCheck{ fsm.state.EnsureCheck(7, &structs.HealthCheck{
Node: "foo", Node: "foo",
CheckID: "web", CheckID: "web",
Name: "web connectivity", Name: "web connectivity",
@ -341,12 +343,12 @@ func TestFSM_SnapshotRestore(t *testing.T) {
} }
// Verify the contents // Verify the contents
nodes := fsm2.state.Nodes() _, nodes := fsm2.state.Nodes()
if len(nodes) != 2 { if len(nodes) != 2 {
t.Fatalf("Bad: %v", nodes) t.Fatalf("Bad: %v", nodes)
} }
fooSrv := fsm2.state.NodeServices("foo") _, fooSrv := fsm2.state.NodeServices("foo")
if len(fooSrv.Services) != 2 { if len(fooSrv.Services) != 2 {
t.Fatalf("Bad: %v", fooSrv) t.Fatalf("Bad: %v", fooSrv)
} }
@ -357,7 +359,7 @@ func TestFSM_SnapshotRestore(t *testing.T) {
t.Fatalf("Bad: %v", fooSrv) t.Fatalf("Bad: %v", fooSrv)
} }
checks := fsm2.state.NodeChecks("foo") _, checks := fsm2.state.NodeChecks("foo")
if len(checks) != 1 { if len(checks) != 1 {
t.Fatalf("Bad: %v", checks) t.Fatalf("Bad: %v", checks)
} }

10
consul/health_endpoint.go
View File

@ -19,7 +19,7 @@ func (h *Health) ChecksInState(args *structs.ChecksInStateRequest,
// Get the state specific checks // Get the state specific checks
state := h.srv.fsm.State() state := h.srv.fsm.State()
checks := state.ChecksInState(args.State) _, checks := state.ChecksInState(args.State)
*reply = checks *reply = checks
return nil return nil
} }
@ -33,7 +33,7 @@ func (h *Health) NodeChecks(args *structs.NodeSpecificRequest,
// Get the node checks // Get the node checks
state := h.srv.fsm.State() state := h.srv.fsm.State()
checks := state.NodeChecks(args.Node) _, checks := state.NodeChecks(args.Node)
*reply = checks *reply = checks
return nil return nil
} }
@ -53,7 +53,7 @@ func (h *Health) ServiceChecks(args *structs.ServiceSpecificRequest,
// Get the service checks // Get the service checks
state := h.srv.fsm.State() state := h.srv.fsm.State()
checks := state.ServiceChecks(args.ServiceName) _, checks := state.ServiceChecks(args.ServiceName)
*reply = checks *reply = checks
return nil return nil
} }
@ -73,9 +73,9 @@ func (h *Health) ServiceNodes(args *structs.ServiceSpecificRequest, reply *struc
state := h.srv.fsm.State() state := h.srv.fsm.State()
var nodes structs.CheckServiceNodes var nodes structs.CheckServiceNodes
if args.TagFilter { if args.TagFilter {
nodes = state.CheckServiceTagNodes(args.ServiceName, args.ServiceTag) _, nodes = state.CheckServiceTagNodes(args.ServiceName, args.ServiceTag)
} else { } else {
nodes = state.CheckServiceNodes(args.ServiceName) _, nodes = state.CheckServiceNodes(args.ServiceName)
} }
*reply = nodes *reply = nodes

12
consul/leader.go
View File

@ -156,12 +156,12 @@ func (s *Server) handleAliveMember(member serf.Member) error {
} }
// Check if the node exists // Check if the node exists
found, addr := state.GetNode(member.Name) _, found, addr := state.GetNode(member.Name)
if found && addr == member.Addr.String() { if found && addr == member.Addr.String() {
// Check if the associated service is available // Check if the associated service is available
if service != nil { if service != nil {
match := false match := false
services := state.NodeServices(member.Name) _, services := state.NodeServices(member.Name)
for id, _ := range services.Services { for id, _ := range services.Services {
if id == service.ID { if id == service.ID {
match = true match = true
@ -173,7 +173,7 @@ func (s *Server) handleAliveMember(member serf.Member) error {
} }
// Check if the serfCheck is in the passing state // Check if the serfCheck is in the passing state
checks := state.NodeChecks(member.Name) _, checks := state.NodeChecks(member.Name)
for _, check := range checks { for _, check := range checks {
if check.CheckID == SerfCheckID && check.Status == structs.HealthPassing { if check.CheckID == SerfCheckID && check.Status == structs.HealthPassing {
return nil return nil
@ -206,10 +206,10 @@ func (s *Server) handleFailedMember(member serf.Member) error {
state := s.fsm.State() state := s.fsm.State()
// Check if the node exists // Check if the node exists
found, addr := state.GetNode(member.Name) _, found, addr := state.GetNode(member.Name)
if found && addr == member.Addr.String() { if found && addr == member.Addr.String() {
// Check if the serfCheck is in the critical state // Check if the serfCheck is in the critical state
checks := state.NodeChecks(member.Name) _, checks := state.NodeChecks(member.Name)
for _, check := range checks { for _, check := range checks {
if check.CheckID == SerfCheckID && check.Status == structs.HealthCritical { if check.CheckID == SerfCheckID && check.Status == structs.HealthCritical {
return nil return nil
@ -240,7 +240,7 @@ func (s *Server) handleLeftMember(member serf.Member) error {
state := s.fsm.State() state := s.fsm.State()
// Check if the node does not exists // Check if the node does not exists
found, _ := state.GetNode(member.Name) _, found, _ := state.GetNode(member.Name)
if !found { if !found {
return nil return nil
} }

20
consul/leader_test.go
View File

@ -32,13 +32,13 @@ func TestLeader_RegisterMember(t *testing.T) {
// Client should be registered // Client should be registered
state := s1.fsm.State() state := s1.fsm.State()
found, _ := state.GetNode(c1.config.NodeName) _, found, _ := state.GetNode(c1.config.NodeName)
if !found { if !found {
t.Fatalf("client not registered") t.Fatalf("client not registered")
} }
// Should have a check // Should have a check
checks := state.NodeChecks(c1.config.NodeName) _, checks := state.NodeChecks(c1.config.NodeName)
if len(checks) != 1 { if len(checks) != 1 {
t.Fatalf("client missing check") t.Fatalf("client missing check")
} }
@ -53,13 +53,13 @@ func TestLeader_RegisterMember(t *testing.T) {
} }
// Server should be registered // Server should be registered
found, _ = state.GetNode(s1.config.NodeName) _, found, _ = state.GetNode(s1.config.NodeName)
if !found { if !found {
t.Fatalf("server not registered") t.Fatalf("server not registered")
} }
// Service should be registered // Service should be registered
services := state.NodeServices(s1.config.NodeName) _, services := state.NodeServices(s1.config.NodeName)
if _, ok := services.Services["consul"]; !ok { if _, ok := services.Services["consul"]; !ok {
t.Fatalf("consul service not registered: %v", services) t.Fatalf("consul service not registered: %v", services)
} }
@ -92,13 +92,13 @@ func TestLeader_FailedMember(t *testing.T) {
// Should be registered // Should be registered
state := s1.fsm.State() state := s1.fsm.State()
found, _ := state.GetNode(c1.config.NodeName) _, found, _ := state.GetNode(c1.config.NodeName)
if !found { if !found {
t.Fatalf("client not registered") t.Fatalf("client not registered")
} }
// Should have a check // Should have a check
checks := state.NodeChecks(c1.config.NodeName) _, checks := state.NodeChecks(c1.config.NodeName)
if len(checks) != 1 { if len(checks) != 1 {
t.Fatalf("client missing check") t.Fatalf("client missing check")
} }
@ -137,7 +137,7 @@ func TestLeader_LeftMember(t *testing.T) {
// Should be registered // Should be registered
state := s1.fsm.State() state := s1.fsm.State()
found, _ := state.GetNode(c1.config.NodeName) _, found, _ := state.GetNode(c1.config.NodeName)
if !found { if !found {
t.Fatalf("client not registered") t.Fatalf("client not registered")
} }
@ -150,7 +150,7 @@ func TestLeader_LeftMember(t *testing.T) {
time.Sleep(500 * time.Millisecond) time.Sleep(500 * time.Millisecond)
// Should be deregistered // Should be deregistered
found, _ = state.GetNode(c1.config.NodeName) _, found, _ = state.GetNode(c1.config.NodeName)
if found { if found {
t.Fatalf("client registered") t.Fatalf("client registered")
} }
@ -174,7 +174,7 @@ func TestLeader_Reconcile(t *testing.T) {
// Should not be registered // Should not be registered
state := s1.fsm.State() state := s1.fsm.State()
found, _ := state.GetNode(c1.config.NodeName) _, found, _ := state.GetNode(c1.config.NodeName)
if found { if found {
t.Fatalf("client registered") t.Fatalf("client registered")
} }
@ -183,7 +183,7 @@ func TestLeader_Reconcile(t *testing.T) {
time.Sleep(100 * time.Millisecond) time.Sleep(100 * time.Millisecond)
// Should be registered // Should be registered
found, _ = state.GetNode(c1.config.NodeName) _, found, _ = state.GetNode(c1.config.NodeName)
if !found { if !found {
t.Fatalf("client not registered") t.Fatalf("client not registered")
} }

58
consul/mdb_table.go
View File

@ -311,14 +311,23 @@ AFTER_DELETE:
// Get is used to lookup one or more rows. An index an appropriate // Get is used to lookup one or more rows. An index an appropriate
// fields are specified. The fields can be a prefix of the index. // fields are specified. The fields can be a prefix of the index.
func (t *MDBTable) Get(index string, parts ...string) ([]interface{}, error) { func (t *MDBTable) Get(index string, parts ...string) (uint64, []interface{}, error) {
// Start a readonly txn // Start a readonly txn
tx, err := t.StartTxn(true, nil) tx, err := t.StartTxn(true, nil)
if err != nil { if err != nil {
return nil, err return 0, nil, err
} }
defer tx.Abort() defer tx.Abort()
return t.GetTxn(tx, index, parts...)
// Get the last associated index
idx, err := t.LastIndexTxn(tx)
if err != nil {
return 0, nil, err
}
// Get the actual results
res, err := t.GetTxn(tx, index, parts...)
return idx, res, err
} }
// GetTxn is like Get but it operates within a specific transaction. // GetTxn is like Get but it operates within a specific transaction.
@ -572,20 +581,6 @@ func (i *MDBIndex) iterate(tx *MDBTxn, prefix []byte,
return nil return nil
} }
// StartTxn is used to create a transaction that spans a list of tables
func (t MDBTables) StartTxn(readonly bool) (*MDBTxn, error) {
var tx *MDBTxn
for _, table := range t {
newTx, err := table.StartTxn(readonly, tx)
if err != nil {
tx.Abort()
return nil, err
}
tx = newTx
}
return tx, nil
}
// LastIndex is get the last index that updated the table // LastIndex is get the last index that updated the table
func (t *MDBTable) LastIndex() (uint64, error) { func (t *MDBTable) LastIndex() (uint64, error) {
// Start a readonly txn // Start a readonly txn
@ -631,3 +626,32 @@ func (t *MDBTable) SetLastIndexTxn(tx *MDBTxn, index uint64) error {
encIndex := uint64ToBytes(index) encIndex := uint64ToBytes(index)
return tx.tx.Put(tx.dbis[t.Name], encRowId, encIndex, 0) return tx.tx.Put(tx.dbis[t.Name], encRowId, encIndex, 0)
} }
// StartTxn is used to create a transaction that spans a list of tables
func (t MDBTables) StartTxn(readonly bool) (*MDBTxn, error) {
var tx *MDBTxn
for _, table := range t {
newTx, err := table.StartTxn(readonly, tx)
if err != nil {
tx.Abort()
return nil, err
}
tx = newTx
}
return tx, nil
}
// LastIndexTxn is used to get the last transaction from all of the tables
func (t MDBTables) LastIndexTxn(tx *MDBTxn) (uint64, error) {
var index uint64
for _, table := range t {
idx, err := table.LastIndexTxn(tx)
if err != nil {
return index, err
}
if idx > index {
index = idx
}
}
return index, nil
}

41
consul/mdb_table_test.go
View File

@ -119,14 +119,17 @@ func TestMDBTableInsert(t *testing.T) {
} }
// Insert some mock objects // Insert some mock objects
for _, obj := range objs { for idx, obj := range objs {
if err := table.Insert(obj); err != nil { if err := table.Insert(obj); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
if err := table.SetLastIndex(uint64(idx + 1)); err != nil {
t.Fatalf("err: %v", err)
}
} }
// Verify with some gets // Verify with some gets
res, err := table.Get("id", "1") idx, res, err := table.Get("id", "1")
if err != nil { if err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
@ -136,8 +139,11 @@ func TestMDBTableInsert(t *testing.T) {
if !reflect.DeepEqual(res[0], objs[0]) { if !reflect.DeepEqual(res[0], objs[0]) {
t.Fatalf("bad: %#v", res[0]) t.Fatalf("bad: %#v", res[0])
} }
if idx != 3 {
t.Fatalf("bad index: %d", idx)
}
res, err = table.Get("name", "Kevin") idx, res, err = table.Get("name", "Kevin")
if err != nil { if err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
@ -150,8 +156,11 @@ func TestMDBTableInsert(t *testing.T) {
if !reflect.DeepEqual(res[1], objs[1]) { if !reflect.DeepEqual(res[1], objs[1]) {
t.Fatalf("bad: %#v", res[1]) t.Fatalf("bad: %#v", res[1])
} }
if idx != 3 {
t.Fatalf("bad index: %d", idx)
}
res, err = table.Get("country", "Mexico") idx, res, err = table.Get("country", "Mexico")
if err != nil { if err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
@ -161,8 +170,11 @@ func TestMDBTableInsert(t *testing.T) {
if !reflect.DeepEqual(res[0], objs[2]) { if !reflect.DeepEqual(res[0], objs[2]) {
t.Fatalf("bad: %#v", res[2]) t.Fatalf("bad: %#v", res[2])
} }
if idx != 3 {
t.Fatalf("bad index: %d", idx)
}
res, err = table.Get("id") idx, res, err = table.Get("id")
if err != nil { if err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
@ -178,6 +190,9 @@ func TestMDBTableInsert(t *testing.T) {
if !reflect.DeepEqual(res[2], objs[2]) { if !reflect.DeepEqual(res[2], objs[2]) {
t.Fatalf("bad: %#v", res[2]) t.Fatalf("bad: %#v", res[2])
} }
if idx != 3 {
t.Fatalf("bad index: %d", idx)
}
} }
func TestMDBTableInsert_MissingFields(t *testing.T) { func TestMDBTableInsert_MissingFields(t *testing.T) {
@ -338,7 +353,7 @@ func TestMDBTableDelete(t *testing.T) {
} }
} }
_, err := table.Get("id", "3") _, _, err := table.Get("id", "3")
if err != nil { if err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
@ -351,7 +366,7 @@ func TestMDBTableDelete(t *testing.T) {
if num != 1 { if num != 1 {
t.Fatalf("expect 1 delete: %#v", num) t.Fatalf("expect 1 delete: %#v", num)
} }
res, err := table.Get("id", "3") _, res, err := table.Get("id", "3")
if err != nil { if err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
@ -366,7 +381,7 @@ func TestMDBTableDelete(t *testing.T) {
if num != 2 { if num != 2 {
t.Fatalf("expect 2 deletes: %#v", num) t.Fatalf("expect 2 deletes: %#v", num)
} }
res, err = table.Get("name", "Kevin") _, res, err = table.Get("name", "Kevin")
if err != nil { if err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
@ -449,7 +464,7 @@ func TestMDBTableUpdate(t *testing.T) {
} }
// Verify with some gets // Verify with some gets
res, err := table.Get("id", "1") _, res, err := table.Get("id", "1")
if err != nil { if err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
@ -460,7 +475,7 @@ func TestMDBTableUpdate(t *testing.T) {
t.Fatalf("bad: %#v", res[0]) t.Fatalf("bad: %#v", res[0])
} }
res, err = table.Get("name", "Kevin") _, res, err = table.Get("name", "Kevin")
if err != nil { if err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
@ -468,7 +483,7 @@ func TestMDBTableUpdate(t *testing.T) {
t.Fatalf("expect 0 result: %#v", res) t.Fatalf("expect 0 result: %#v", res)
} }
res, err = table.Get("name", "Ahmad") _, res, err = table.Get("name", "Ahmad")
if err != nil { if err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
@ -479,7 +494,7 @@ func TestMDBTableUpdate(t *testing.T) {
t.Fatalf("bad: %#v", res[0]) t.Fatalf("bad: %#v", res[0])
} }
res, err = table.Get("country", "Mexico") _, res, err = table.Get("country", "Mexico")
if err != nil { if err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
@ -490,7 +505,7 @@ func TestMDBTableUpdate(t *testing.T) {
t.Fatalf("bad: %#v", res[0]) t.Fatalf("bad: %#v", res[0])
} }
res, err = table.Get("id") _, res, err = table.Get("id")
if err != nil { if err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }

214
consul/state_store.go
View File

@ -34,8 +34,9 @@ type StateStore struct {
// StateSnapshot is used to provide a point-in-time snapshot // StateSnapshot is used to provide a point-in-time snapshot
// It works by starting a readonly transaction against all tables. // It works by starting a readonly transaction against all tables.
type StateSnapshot struct { type StateSnapshot struct {
store *StateStore store *StateStore
tx *MDBTxn tx *MDBTxn
lastIndex uint64
} }
// Close is used to abort the transaction and allow for cleanup // Close is used to abort the transaction and allow for cleanup
@ -188,26 +189,39 @@ func (s *StateStore) initialize() error {
} }
// EnsureNode is used to ensure a given node exists, with the provided address // EnsureNode is used to ensure a given node exists, with the provided address
func (s *StateStore) EnsureNode(node structs.Node) error { func (s *StateStore) EnsureNode(index uint64, node structs.Node) error {
return s.nodeTable.Insert(node) // Start a new txn
tx, err := s.nodeTable.StartTxn(false, nil)
if err != nil {
return err
}
defer tx.Abort()
if err := s.nodeTable.InsertTxn(tx, node); err != nil {
return err
}
if err := s.nodeTable.SetLastIndexTxn(tx, index); err != nil {
return err
}
return tx.Commit()
} }
// GetNode returns all the address of the known and if it was found // GetNode returns all the address of the known and if it was found
func (s *StateStore) GetNode(name string) (bool, string) { func (s *StateStore) GetNode(name string) (uint64, bool, string) {
res, err := s.nodeTable.Get("id", name) idx, res, err := s.nodeTable.Get("id", name)
if err != nil { if err != nil {
panic(fmt.Errorf("Failed to get node: %v", err)) panic(fmt.Errorf("Failed to get node: %v", err))
} }
if len(res) == 0 { if len(res) == 0 {
return false, "" return idx, false, ""
} }
return true, res[0].(*structs.Node).Address return idx, true, res[0].(*structs.Node).Address
} }
// GetNodes returns all the known nodes, the slice alternates between // GetNodes returns all the known nodes, the slice alternates between
// the node name and address // the node name and address
func (s *StateStore) Nodes() structs.Nodes { func (s *StateStore) Nodes() (uint64, structs.Nodes) {
res, err := s.nodeTable.Get("id") idx, res, err := s.nodeTable.Get("id")
if err != nil { if err != nil {
panic(fmt.Errorf("Failed to get nodes: %v", err)) panic(fmt.Errorf("Failed to get nodes: %v", err))
} }
@ -215,11 +229,11 @@ func (s *StateStore) Nodes() structs.Nodes {
for i, r := range res { for i, r := range res {
results[i] = *r.(*structs.Node) results[i] = *r.(*structs.Node)
} }
return results return idx, results
} }
// EnsureService is used to ensure a given node exposes a service // EnsureService is used to ensure a given node exposes a service
func (s *StateStore) EnsureService(name, id, service, tag string, port int) error { func (s *StateStore) EnsureService(index uint64, node string, ns *structs.NodeService) error {
tx, err := s.tables.StartTxn(false) tx, err := s.tables.StartTxn(false)
if err != nil { if err != nil {
panic(fmt.Errorf("Failed to start txn: %v", err)) panic(fmt.Errorf("Failed to start txn: %v", err))
@ -227,7 +241,7 @@ func (s *StateStore) EnsureService(name, id, service, tag string, port int) erro
defer tx.Abort() defer tx.Abort()
// Ensure the node exists // Ensure the node exists
res, err := s.nodeTable.GetTxn(tx, "id", name) res, err := s.nodeTable.GetTxn(tx, "id", node)
if err != nil { if err != nil {
return err return err
} }
@ -237,22 +251,25 @@ func (s *StateStore) EnsureService(name, id, service, tag string, port int) erro
// Create the entry // Create the entry
entry := structs.ServiceNode{ entry := structs.ServiceNode{
Node: name, Node: node,
ServiceID: id, ServiceID: ns.ID,
ServiceName: service, ServiceName: ns.Service,
ServiceTag: tag, ServiceTag: ns.Tag,
ServicePort: port, ServicePort: ns.Port,
} }
// Ensure the service entry is set // Ensure the service entry is set
if err := s.serviceTable.InsertTxn(tx, &entry); err != nil { if err := s.serviceTable.InsertTxn(tx, &entry); err != nil {
return err return err
} }
if err := s.serviceTable.SetLastIndexTxn(tx, index); err != nil {
return err
}
return tx.Commit() return tx.Commit()
} }
// NodeServices is used to return all the services of a given node // NodeServices is used to return all the services of a given node
func (s *StateStore) NodeServices(name string) *structs.NodeServices { func (s *StateStore) NodeServices(name string) (uint64, *structs.NodeServices) {
tx, err := s.tables.StartTxn(true) tx, err := s.tables.StartTxn(true)
if err != nil { if err != nil {
panic(fmt.Errorf("Failed to start txn: %v", err)) panic(fmt.Errorf("Failed to start txn: %v", err))
@ -263,18 +280,24 @@ func (s *StateStore) NodeServices(name string) *structs.NodeServices {
// parseNodeServices is used to get the services belonging to a // parseNodeServices is used to get the services belonging to a
// node, using a given txn // node, using a given txn
func (s *StateStore) parseNodeServices(tx *MDBTxn, name string) *structs.NodeServices { func (s *StateStore) parseNodeServices(tx *MDBTxn, name string) (uint64, *structs.NodeServices) {
ns := &structs.NodeServices{ ns := &structs.NodeServices{
Services: make(map[string]*structs.NodeService), Services: make(map[string]*structs.NodeService),
} }
// Get the maximum index
index, err := s.tables.LastIndexTxn(tx)
if err != nil {
panic(fmt.Errorf("Failed to get last index: %v", err))
}
// Get the node first // Get the node first
res, err := s.nodeTable.GetTxn(tx, "id", name) res, err := s.nodeTable.GetTxn(tx, "id", name)
if err != nil { if err != nil {
panic(fmt.Errorf("Failed to get node: %v", err)) panic(fmt.Errorf("Failed to get node: %v", err))
} }
if len(res) == 0 { if len(res) == 0 {
return ns return index, ns
} }
// Set the address // Set the address
@ -298,51 +321,71 @@ func (s *StateStore) parseNodeServices(tx *MDBTxn, name string) *structs.NodeSer
} }
ns.Services[srv.ID] = srv ns.Services[srv.ID] = srv
} }
return ns return index, ns
} }
// DeleteNodeService is used to delete a node service // DeleteNodeService is used to delete a node service
func (s *StateStore) DeleteNodeService(node, id string) error { func (s *StateStore) DeleteNodeService(index uint64, node, id string) error {
tx, err := s.tables.StartTxn(false) tx, err := s.tables.StartTxn(false)
if err != nil { if err != nil {
panic(fmt.Errorf("Failed to start txn: %v", err)) panic(fmt.Errorf("Failed to start txn: %v", err))
} }
defer tx.Abort() defer tx.Abort()
if _, err := s.serviceTable.DeleteTxn(tx, "id", node, id); err != nil { if n, err := s.serviceTable.DeleteTxn(tx, "id", node, id); err != nil {
return err return err
} else if n > 0 {
if err := s.serviceTable.SetLastIndexTxn(tx, index); err != nil {
return err
}
} }
if _, err := s.checkTable.DeleteTxn(tx, "node", node, id); err != nil { if n, err := s.checkTable.DeleteTxn(tx, "node", node, id); err != nil {
return err return err
} else if n > 0 {
if err := s.checkTable.SetLastIndexTxn(tx, index); err != nil {
return err
}
} }
return tx.Commit() return tx.Commit()
} }
// DeleteNode is used to delete a node and all it's services // DeleteNode is used to delete a node and all it's services
func (s *StateStore) DeleteNode(node string) error { func (s *StateStore) DeleteNode(index uint64, node string) error {
tx, err := s.tables.StartTxn(false) tx, err := s.tables.StartTxn(false)
if err != nil { if err != nil {
panic(fmt.Errorf("Failed to start txn: %v", err)) panic(fmt.Errorf("Failed to start txn: %v", err))
} }
defer tx.Abort() defer tx.Abort()
if _, err := s.serviceTable.DeleteTxn(tx, "id", node); err != nil { if n, err := s.serviceTable.DeleteTxn(tx, "id", node); err != nil {
return err return err
} else if n > 0 {
if err := s.serviceTable.SetLastIndexTxn(tx, index); err != nil {
return err
}
} }
if _, err := s.checkTable.DeleteTxn(tx, "id", node); err != nil { if n, err := s.checkTable.DeleteTxn(tx, "id", node); err != nil {
return err return err
} else if n > 0 {
if err := s.checkTable.SetLastIndexTxn(tx, index); err != nil {
return err
}
} }
if _, err := s.nodeTable.DeleteTxn(tx, "id", node); err != nil { if n, err := s.nodeTable.DeleteTxn(tx, "id", node); err != nil {
return err return err
} else if n > 0 {
if err := s.nodeTable.SetLastIndexTxn(tx, index); err != nil {
return err
}
} }
return tx.Commit() return tx.Commit()
} }
// Services is used to return all the services with a list of associated tags // Services is used to return all the services with a list of associated tags
func (s *StateStore) Services() map[string][]string { func (s *StateStore) Services() (uint64, map[string][]string) {
// TODO: Optimize to not table scan.. We can do a distinct // TODO: Optimize to not table scan.. We can do a distinct
// type of query to avoid this // type of query to avoid this
res, err := s.serviceTable.Get("id") idx, res, err := s.serviceTable.Get("id")
if err != nil { if err != nil {
panic(fmt.Errorf("Failed to get node servicess: %v", err)) panic(fmt.Errorf("Failed to get node servicess: %v", err))
} }
@ -356,31 +399,41 @@ func (s *StateStore) Services() map[string][]string {
services[srv.ServiceName] = tags services[srv.ServiceName] = tags
} }
} }
return services return idx, services
} }
// ServiceNodes returns the nodes associated with a given service // ServiceNodes returns the nodes associated with a given service
func (s *StateStore) ServiceNodes(service string) structs.ServiceNodes { func (s *StateStore) ServiceNodes(service string) (uint64, structs.ServiceNodes) {
tx, err := s.tables.StartTxn(true) tx, err := s.tables.StartTxn(true)
if err != nil { if err != nil {
panic(fmt.Errorf("Failed to start txn: %v", err)) panic(fmt.Errorf("Failed to start txn: %v", err))
} }
defer tx.Abort() defer tx.Abort()
res, err := s.serviceTable.Get("service", service) idx, err := s.tables.LastIndexTxn(tx)
return parseServiceNodes(tx, s.nodeTable, res, err) if err != nil {
panic(fmt.Errorf("Failed to get last index: %v", err))
}
res, err := s.serviceTable.GetTxn(tx, "service", service)
return idx, parseServiceNodes(tx, s.nodeTable, res, err)
} }
// ServiceTagNodes returns the nodes associated with a given service matching a tag // ServiceTagNodes returns the nodes associated with a given service matching a tag
func (s *StateStore) ServiceTagNodes(service, tag string) structs.ServiceNodes { func (s *StateStore) ServiceTagNodes(service, tag string) (uint64, structs.ServiceNodes) {
tx, err := s.tables.StartTxn(true) tx, err := s.tables.StartTxn(true)
if err != nil { if err != nil {
panic(fmt.Errorf("Failed to start txn: %v", err)) panic(fmt.Errorf("Failed to start txn: %v", err))
} }
defer tx.Abort() defer tx.Abort()
res, err := s.serviceTable.Get("service", service, tag) idx, err := s.tables.LastIndexTxn(tx)
return parseServiceNodes(tx, s.nodeTable, res, err) if err != nil {
panic(fmt.Errorf("Failed to get last index: %v", err))
}
res, err := s.serviceTable.GetTxn(tx, "service", service, tag)
return idx, parseServiceNodes(tx, s.nodeTable, res, err)
} }
// parseServiceNodes parses results ServiceNodes and ServiceTagNodes // parseServiceNodes parses results ServiceNodes and ServiceTagNodes
@ -407,7 +460,7 @@ func parseServiceNodes(tx *MDBTxn, table *MDBTable, res []interface{}, err error
} }
// EnsureCheck is used to create a check or updates it's state // EnsureCheck is used to create a check or updates it's state
func (s *StateStore) EnsureCheck(check *structs.HealthCheck) error { func (s *StateStore) EnsureCheck(index uint64, check *structs.HealthCheck) error {
// Ensure we have a status // Ensure we have a status
if check.Status == "" { if check.Status == "" {
check.Status = structs.HealthUnknown check.Status = structs.HealthUnknown
@ -447,33 +500,48 @@ func (s *StateStore) EnsureCheck(check *structs.HealthCheck) error {
if err := s.checkTable.InsertTxn(tx, check); err != nil { if err := s.checkTable.InsertTxn(tx, check); err != nil {
return err return err
} }
if err := s.checkTable.SetLastIndexTxn(tx, index); err != nil {
return err
}
return tx.Commit() return tx.Commit()
} }
// DeleteNodeCheck is used to delete a node health check // DeleteNodeCheck is used to delete a node health check
func (s *StateStore) DeleteNodeCheck(node, id string) error { func (s *StateStore) DeleteNodeCheck(index uint64, node, id string) error {
_, err := s.checkTable.Delete("id", node, id) tx, err := s.checkTable.StartTxn(false, nil)
return err if err != nil {
return err
}
defer tx.Abort()
if n, err := s.checkTable.DeleteTxn(tx, "id", node, id); err != nil {
return err
} else if n > 0 {
if err := s.checkTable.SetLastIndexTxn(tx, index); err != nil {
return err
}
}
return tx.Commit()
} }
// NodeChecks is used to get all the checks for a node // NodeChecks is used to get all the checks for a node
func (s *StateStore) NodeChecks(node string) structs.HealthChecks { func (s *StateStore) NodeChecks(node string) (uint64, structs.HealthChecks) {
return parseHealthChecks(s.checkTable.Get("id", node)) return parseHealthChecks(s.checkTable.Get("id", node))
} }
// ServiceChecks is used to get all the checks for a service // ServiceChecks is used to get all the checks for a service
func (s *StateStore) ServiceChecks(service string) structs.HealthChecks { func (s *StateStore) ServiceChecks(service string) (uint64, structs.HealthChecks) {
return parseHealthChecks(s.checkTable.Get("service", service)) return parseHealthChecks(s.checkTable.Get("service", service))
} }
// CheckInState is used to get all the checks for a service in a given state // CheckInState is used to get all the checks for a service in a given state
func (s *StateStore) ChecksInState(state string) structs.HealthChecks { func (s *StateStore) ChecksInState(state string) (uint64, structs.HealthChecks) {
return parseHealthChecks(s.checkTable.Get("status", state)) return parseHealthChecks(s.checkTable.Get("status", state))
} }
// parseHealthChecks is used to handle the resutls of a Get against // parseHealthChecks is used to handle the resutls of a Get against
// the checkTable // the checkTable
func parseHealthChecks(res []interface{}, err error) structs.HealthChecks { func parseHealthChecks(idx uint64, res []interface{}, err error) (uint64, structs.HealthChecks) {
if err != nil { if err != nil {
panic(fmt.Errorf("Failed to get checks: %v", err)) panic(fmt.Errorf("Failed to get checks: %v", err))
} }
@ -481,33 +549,43 @@ func parseHealthChecks(res []interface{}, err error) structs.HealthChecks {
for i, r := range res { for i, r := range res {
results[i] = r.(*structs.HealthCheck) results[i] = r.(*structs.HealthCheck)
} }
return results return idx, results
} }
// CheckServiceNodes returns the nodes associated with a given service, along // CheckServiceNodes returns the nodes associated with a given service, along
// with any associated check // with any associated check
func (s *StateStore) CheckServiceNodes(service string) structs.CheckServiceNodes { func (s *StateStore) CheckServiceNodes(service string) (uint64, structs.CheckServiceNodes) {
tx, err := s.tables.StartTxn(true) tx, err := s.tables.StartTxn(true)
if err != nil { if err != nil {
panic(fmt.Errorf("Failed to start txn: %v", err)) panic(fmt.Errorf("Failed to start txn: %v", err))
} }
defer tx.Abort() defer tx.Abort()
res, err := s.serviceTable.Get("service", service) idx, err := s.tables.LastIndexTxn(tx)
return s.parseCheckServiceNodes(tx, res, err) if err != nil {
panic(fmt.Errorf("Failed to get last index: %v", err))
}
res, err := s.serviceTable.GetTxn(tx, "service", service)
return idx, s.parseCheckServiceNodes(tx, res, err)
} }
// CheckServiceNodes returns the nodes associated with a given service, along // CheckServiceNodes returns the nodes associated with a given service, along
// with any associated checks // with any associated checks
func (s *StateStore) CheckServiceTagNodes(service, tag string) structs.CheckServiceNodes { func (s *StateStore) CheckServiceTagNodes(service, tag string) (uint64, structs.CheckServiceNodes) {
tx, err := s.tables.StartTxn(true) tx, err := s.tables.StartTxn(true)
if err != nil { if err != nil {
panic(fmt.Errorf("Failed to start txn: %v", err)) panic(fmt.Errorf("Failed to start txn: %v", err))
} }
defer tx.Abort() defer tx.Abort()
res, err := s.serviceTable.Get("service", service, tag) idx, err := s.tables.LastIndexTxn(tx)
return s.parseCheckServiceNodes(tx, res, err) if err != nil {
panic(fmt.Errorf("Failed to get last index: %v", err))
}
res, err := s.serviceTable.GetTxn(tx, "service", service, tag)
return idx, s.parseCheckServiceNodes(tx, res, err)
} }
// parseCheckServiceNodes parses results CheckServiceNodes and CheckServiceTagNodes // parseCheckServiceNodes parses results CheckServiceNodes and CheckServiceTagNodes
@ -527,10 +605,12 @@ func (s *StateStore) parseCheckServiceNodes(tx *MDBTxn, res []interface{}, err e
} }
// Get any associated checks of the service // Get any associated checks of the service
checks := parseHealthChecks(s.checkTable.GetTxn(tx, "node", srv.Node, srv.ServiceID)) res, err := s.checkTable.GetTxn(tx, "node", srv.Node, srv.ServiceID)
_, checks := parseHealthChecks(0, res, err)
// Get any checks of the node, not assciated with any service // Get any checks of the node, not assciated with any service
nodeChecks := parseHealthChecks(s.checkTable.GetTxn(tx, "node", srv.Node, "")) res, err = s.checkTable.GetTxn(tx, "node", srv.Node, "")
_, nodeChecks := parseHealthChecks(0, res, err)
checks = append(checks, nodeChecks...) checks = append(checks, nodeChecks...)
// Setup the node // Setup the node
@ -555,14 +635,27 @@ func (s *StateStore) Snapshot() (*StateSnapshot, error) {
return nil, err return nil, err
} }
// Determine the max index
index, err := s.tables.LastIndexTxn(tx)
if err != nil {
tx.Abort()
return nil, err
}
// Return the snapshot // Return the snapshot
snap := &StateSnapshot{ snap := &StateSnapshot{
store: s, store: s,
tx: tx, tx: tx,
lastIndex: index,
} }
return snap, nil return snap, nil
} }
// LastIndex returns the last index that affects the snapshotted data
func (s *StateSnapshot) LastIndex() uint64 {
return s.lastIndex
}
// Nodes returns all the known nodes, the slice alternates between // Nodes returns all the known nodes, the slice alternates between
// the node name and address // the node name and address
func (s *StateSnapshot) Nodes() structs.Nodes { func (s *StateSnapshot) Nodes() structs.Nodes {
@ -579,10 +672,13 @@ func (s *StateSnapshot) Nodes() structs.Nodes {
// NodeServices is used to return all the services of a given node // NodeServices is used to return all the services of a given node
func (s *StateSnapshot) NodeServices(name string) *structs.NodeServices { func (s *StateSnapshot) NodeServices(name string) *structs.NodeServices {
return s.store.parseNodeServices(s.tx, name) _, res := s.store.parseNodeServices(s.tx, name)
return res
} }
// NodeChecks is used to return all the checks of a given node // NodeChecks is used to return all the checks of a given node
func (s *StateSnapshot) NodeChecks(node string) structs.HealthChecks { func (s *StateSnapshot) NodeChecks(node string) structs.HealthChecks {
return parseHealthChecks(s.store.checkTable.GetTxn(s.tx, "id", node)) res, err := s.store.checkTable.GetTxn(s.tx, "id", node)
_, checks := parseHealthChecks(s.lastIndex, res, err)
return checks
} }

252
consul/state_store_test.go
View File

@ -14,22 +14,22 @@ func TestEnsureNode(t *testing.T) {
} }
defer store.Close() defer store.Close()
if err := store.EnsureNode(structs.Node{"foo", "127.0.0.1"}); err != nil { if err := store.EnsureNode(3, structs.Node{"foo", "127.0.0.1"}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
found, addr := store.GetNode("foo") idx, found, addr := store.GetNode("foo")
if !found || addr != "127.0.0.1" { if idx != 3 || !found || addr != "127.0.0.1" {
t.Fatalf("Bad: %v %v", found, addr) t.Fatalf("Bad: %v %v %v", idx, found, addr)
} }
if err := store.EnsureNode(structs.Node{"foo", "127.0.0.2"}); err != nil { if err := store.EnsureNode(4, structs.Node{"foo", "127.0.0.2"}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
found, addr = store.GetNode("foo") idx, found, addr = store.GetNode("foo")
if !found || addr != "127.0.0.2" { if idx != 4 || !found || addr != "127.0.0.2" {
t.Fatalf("Bad: %v %v", found, addr) t.Fatalf("Bad: %v %v %v", idx, found, addr)
} }
} }
@ -40,15 +40,18 @@ func TestGetNodes(t *testing.T) {
} }
defer store.Close() defer store.Close()
if err := store.EnsureNode(structs.Node{"foo", "127.0.0.1"}); err != nil { if err := store.EnsureNode(40, structs.Node{"foo", "127.0.0.1"}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.EnsureNode(structs.Node{"bar", "127.0.0.2"}); err != nil { if err := store.EnsureNode(41, structs.Node{"bar", "127.0.0.2"}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
nodes := store.Nodes() idx, nodes := store.Nodes()
if idx != 41 {
t.Fatalf("idx: %v", idx)
}
if len(nodes) != 2 { if len(nodes) != 2 {
t.Fatalf("Bad: %v", nodes) t.Fatalf("Bad: %v", nodes)
} }
@ -64,11 +67,11 @@ func BenchmarkGetNodes(b *testing.B) {
} }
defer store.Close() defer store.Close()
if err := store.EnsureNode(structs.Node{"foo", "127.0.0.1"}); err != nil { if err := store.EnsureNode(100, structs.Node{"foo", "127.0.0.1"}); err != nil {
b.Fatalf("err: %v") b.Fatalf("err: %v")
} }
if err := store.EnsureNode(structs.Node{"bar", "127.0.0.2"}); err != nil { if err := store.EnsureNode(101, structs.Node{"bar", "127.0.0.2"}); err != nil {
b.Fatalf("err: %v") b.Fatalf("err: %v")
} }
@ -84,23 +87,26 @@ func TestEnsureService(t *testing.T) {
} }
defer store.Close() defer store.Close()
if err := store.EnsureNode(structs.Node{"foo", "127.0.0.1"}); err != nil { if err := store.EnsureNode(10, structs.Node{"foo", "127.0.0.1"}); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
if err := store.EnsureService("foo", "api", "api", "", 5000); err != nil { if err := store.EnsureService(11, "foo", &structs.NodeService{"api", "api", "", 5000}); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
if err := store.EnsureService("foo", "api", "api", "", 5001); err != nil { if err := store.EnsureService(12, "foo", &structs.NodeService{"api", "api", "", 5001}); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
if err := store.EnsureService("foo", "db", "db", "master", 8000); err != nil { if err := store.EnsureService(13, "foo", &structs.NodeService{"db", "db", "master", 8000}); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
services := store.NodeServices("foo") idx, services := store.NodeServices("foo")
if idx != 13 {
t.Fatalf("bad: %v", idx)
}
entry, ok := services.Services["api"] entry, ok := services.Services["api"]
if !ok { if !ok {
@ -126,23 +132,26 @@ func TestEnsureService_DuplicateNode(t *testing.T) {
} }
defer store.Close() defer store.Close()
if err := store.EnsureNode(structs.Node{"foo", "127.0.0.1"}); err != nil { if err := store.EnsureNode(10, structs.Node{"foo", "127.0.0.1"}); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
if err := store.EnsureService("foo", "api1", "api", "", 5000); err != nil { if err := store.EnsureService(11, "foo", &structs.NodeService{"api1", "api", "", 5000}); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
if err := store.EnsureService("foo", "api2", "api", "", 5001); err != nil { if err := store.EnsureService(12, "foo", &structs.NodeService{"api2", "api", "", 5001}); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
if err := store.EnsureService("foo", "api3", "api", "", 5002); err != nil { if err := store.EnsureService(13, "foo", &structs.NodeService{"api3", "api", "", 5002}); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
services := store.NodeServices("foo") idx, services := store.NodeServices("foo")
if idx != 13 {
t.Fatalf("bad: %v", idx)
}
entry, ok := services.Services["api1"] entry, ok := services.Services["api1"]
if !ok { if !ok {
@ -176,11 +185,11 @@ func TestDeleteNodeService(t *testing.T) {
} }
defer store.Close() defer store.Close()
if err := store.EnsureNode(structs.Node{"foo", "127.0.0.1"}); err != nil { if err := store.EnsureNode(11, structs.Node{"foo", "127.0.0.1"}); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
if err := store.EnsureService("foo", "api", "api", "", 5000); err != nil { if err := store.EnsureService(12, "foo", &structs.NodeService{"api", "api", "", 5000}); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
@ -191,21 +200,27 @@ func TestDeleteNodeService(t *testing.T) {
Status: structs.HealthPassing, Status: structs.HealthPassing,
ServiceID: "api", ServiceID: "api",
} }
if err := store.EnsureCheck(check); err != nil { if err := store.EnsureCheck(13, check); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.DeleteNodeService("foo", "api"); err != nil { if err := store.DeleteNodeService(14, "foo", "api"); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
services := store.NodeServices("foo") idx, services := store.NodeServices("foo")
if idx != 14 {
t.Fatalf("bad: %v", idx)
}
_, ok := services.Services["api"] _, ok := services.Services["api"]
if ok { if ok {
t.Fatalf("has api: %#v", services) t.Fatalf("has api: %#v", services)
} }
checks := store.NodeChecks("foo") idx, checks := store.NodeChecks("foo")
if idx != 14 {
t.Fatalf("bad: %v", idx)
}
if len(checks) != 0 { if len(checks) != 0 {
t.Fatalf("has check: %#v", checks) t.Fatalf("has check: %#v", checks)
} }
@ -218,23 +233,26 @@ func TestDeleteNodeService_One(t *testing.T) {
} }
defer store.Close() defer store.Close()
if err := store.EnsureNode(structs.Node{"foo", "127.0.0.1"}); err != nil { if err := store.EnsureNode(11, structs.Node{"foo", "127.0.0.1"}); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
if err := store.EnsureService("foo", "api", "api", "", 5000); err != nil { if err := store.EnsureService(12, "foo", &structs.NodeService{"api", "api", "", 5000}); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
if err := store.EnsureService("foo", "api2", "api", "", 5001); err != nil { if err := store.EnsureService(13, "foo", &structs.NodeService{"api2", "api", "", 5001}); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
if err := store.DeleteNodeService("foo", "api"); err != nil { if err := store.DeleteNodeService(14, "foo", "api"); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
services := store.NodeServices("foo") idx, services := store.NodeServices("foo")
if idx != 14 {
t.Fatalf("bad: %v", idx)
}
_, ok := services.Services["api"] _, ok := services.Services["api"]
if ok { if ok {
t.Fatalf("has api: %#v", services) t.Fatalf("has api: %#v", services)
@ -252,11 +270,11 @@ func TestDeleteNode(t *testing.T) {
} }
defer store.Close() defer store.Close()
if err := store.EnsureNode(structs.Node{"foo", "127.0.0.1"}); err != nil { if err := store.EnsureNode(20, structs.Node{"foo", "127.0.0.1"}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.EnsureService("foo", "api", "api", "", 5000); err != nil { if err := store.EnsureService(21, "foo", &structs.NodeService{"api", "api", "", 5000}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
@ -267,26 +285,35 @@ func TestDeleteNode(t *testing.T) {
Status: structs.HealthPassing, Status: structs.HealthPassing,
ServiceID: "api", ServiceID: "api",
} }
if err := store.EnsureCheck(check); err != nil { if err := store.EnsureCheck(22, check); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
if err := store.DeleteNode("foo"); err != nil { if err := store.DeleteNode(23, "foo"); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
services := store.NodeServices("foo") idx, services := store.NodeServices("foo")
if idx != 23 {
t.Fatalf("bad: %v", idx)
}
_, ok := services.Services["api"] _, ok := services.Services["api"]
if ok { if ok {
t.Fatalf("has api: %#v", services) t.Fatalf("has api: %#v", services)
} }
checks := store.NodeChecks("foo") idx, checks := store.NodeChecks("foo")
if idx != 23 {
t.Fatalf("bad: %v", idx)
}
if len(checks) > 0 { if len(checks) > 0 {
t.Fatalf("has checks: %v", checks) t.Fatalf("has checks: %v", checks)
} }
found, _ := store.GetNode("foo") idx, found, _ := store.GetNode("foo")
if idx != 23 {
t.Fatalf("bad: %v", idx)
}
if found { if found {
t.Fatalf("found node") t.Fatalf("found node")
} }
@ -299,27 +326,30 @@ func TestGetServices(t *testing.T) {
} }
defer store.Close() defer store.Close()
if err := store.EnsureNode(structs.Node{"foo", "127.0.0.1"}); err != nil { if err := store.EnsureNode(30, structs.Node{"foo", "127.0.0.1"}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.EnsureNode(structs.Node{"bar", "127.0.0.2"}); err != nil { if err := store.EnsureNode(31, structs.Node{"bar", "127.0.0.2"}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.EnsureService("foo", "api", "api", "", 5000); err != nil { if err := store.EnsureService(32, "foo", &structs.NodeService{"api", "api", "", 5000}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.EnsureService("foo", "db", "db", "master", 8000); err != nil { if err := store.EnsureService(33, "foo", &structs.NodeService{"db", "db", "master", 8000}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.EnsureService("bar", "db", "db", "slave", 8000); err != nil { if err := store.EnsureService(34, "bar", &structs.NodeService{"db", "db", "slave", 8000}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
services := store.Services() idx, services := store.Services()
if idx != 34 {
t.Fatalf("bad: %v", idx)
}
tags, ok := services["api"] tags, ok := services["api"]
if !ok { if !ok {
@ -346,35 +376,38 @@ func TestServiceNodes(t *testing.T) {
} }
defer store.Close() defer store.Close()
if err := store.EnsureNode(structs.Node{"foo", "127.0.0.1"}); err != nil { if err := store.EnsureNode(10, structs.Node{"foo", "127.0.0.1"}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.EnsureNode(structs.Node{"bar", "127.0.0.2"}); err != nil { if err := store.EnsureNode(11, structs.Node{"bar", "127.0.0.2"}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.EnsureService("foo", "api", "api", "", 5000); err != nil { if err := store.EnsureService(12, "foo", &structs.NodeService{"api", "api", "", 5000}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.EnsureService("bar", "api", "api", "", 5000); err != nil { if err := store.EnsureService(13, "bar", &structs.NodeService{"api", "api", "", 5000}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.EnsureService("foo", "db", "db", "master", 8000); err != nil { if err := store.EnsureService(14, "foo", &structs.NodeService{"db", "db", "master", 8000}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.EnsureService("bar", "db", "db", "slave", 8000); err != nil { if err := store.EnsureService(15, "bar", &structs.NodeService{"db", "db", "slave", 8000}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.EnsureService("bar", "db2", "db", "slave", 8001); err != nil { if err := store.EnsureService(16, "bar", &structs.NodeService{"db2", "db", "slave", 8001}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
nodes := store.ServiceNodes("db") idx, nodes := store.ServiceNodes("db")
if idx != 16 {
t.Fatalf("bad: %v", 16)
}
if len(nodes) != 3 { if len(nodes) != 3 {
t.Fatalf("bad: %v", nodes) t.Fatalf("bad: %v", nodes)
} }
@ -434,27 +467,30 @@ func TestServiceTagNodes(t *testing.T) {
} }
defer store.Close() defer store.Close()
if err := store.EnsureNode(structs.Node{"foo", "127.0.0.1"}); err != nil { if err := store.EnsureNode(15, structs.Node{"foo", "127.0.0.1"}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.EnsureNode(structs.Node{"bar", "127.0.0.2"}); err != nil { if err := store.EnsureNode(16, structs.Node{"bar", "127.0.0.2"}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.EnsureService("foo", "db", "db", "master", 8000); err != nil { if err := store.EnsureService(17, "foo", &structs.NodeService{"db", "db", "master", 8000}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.EnsureService("foo", "db2", "db", "slave", 8001); err != nil { if err := store.EnsureService(18, "foo", &structs.NodeService{"db2", "db", "slave", 8001}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.EnsureService("bar", "db", "db", "slave", 8000); err != nil { if err := store.EnsureService(19, "bar", &structs.NodeService{"db", "db", "slave", 8000}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
nodes := store.ServiceTagNodes("db", "master") idx, nodes := store.ServiceTagNodes("db", "master")
if idx != 19 {
t.Fatalf("bad: %v", idx)
}
if len(nodes) != 1 { if len(nodes) != 1 {
t.Fatalf("bad: %v", nodes) t.Fatalf("bad: %v", nodes)
} }
@ -479,23 +515,23 @@ func TestStoreSnapshot(t *testing.T) {
} }
defer store.Close() defer store.Close()
if err := store.EnsureNode(structs.Node{"foo", "127.0.0.1"}); err != nil { if err := store.EnsureNode(8, structs.Node{"foo", "127.0.0.1"}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.EnsureNode(structs.Node{"bar", "127.0.0.2"}); err != nil { if err := store.EnsureNode(9, structs.Node{"bar", "127.0.0.2"}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.EnsureService("foo", "db", "db", "master", 8000); err != nil { if err := store.EnsureService(10, "foo", &structs.NodeService{"db", "db", "master", 8000}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.EnsureService("foo", "db2", "db", "slave", 8001); err != nil { if err := store.EnsureService(11, "foo", &structs.NodeService{"db2", "db", "slave", 8001}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.EnsureService("bar", "db", "db", "slave", 8000); err != nil { if err := store.EnsureService(12, "bar", &structs.NodeService{"db", "db", "slave", 8000}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
@ -506,7 +542,7 @@ func TestStoreSnapshot(t *testing.T) {
Status: structs.HealthPassing, Status: structs.HealthPassing,
ServiceID: "db", ServiceID: "db",
} }
if err := store.EnsureCheck(check); err != nil { if err := store.EnsureCheck(13, check); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
@ -517,6 +553,11 @@ func TestStoreSnapshot(t *testing.T) {
} }
defer snap.Close() defer snap.Close()
// Check the last nodes
if idx := snap.LastIndex(); idx != 13 {
t.Fatalf("bad: %v", idx)
}
// Check snapshot has old values // Check snapshot has old values
nodes := snap.Nodes() nodes := snap.Nodes()
if len(nodes) != 2 { if len(nodes) != 2 {
@ -547,13 +588,13 @@ func TestStoreSnapshot(t *testing.T) {
} }
// Make some changes! // Make some changes!
if err := store.EnsureService("foo", "db", "db", "slave", 8000); err != nil { if err := store.EnsureService(14, "foo", &structs.NodeService{"db", "db", "slave", 8000}); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
if err := store.EnsureService("bar", "db", "db", "master", 8000); err != nil { if err := store.EnsureService(15, "bar", &structs.NodeService{"db", "db", "master", 8000}); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
if err := store.EnsureNode(structs.Node{"baz", "127.0.0.3"}); err != nil { if err := store.EnsureNode(16, structs.Node{"baz", "127.0.0.3"}); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
checkAfter := &structs.HealthCheck{ checkAfter := &structs.HealthCheck{
@ -563,7 +604,7 @@ func TestStoreSnapshot(t *testing.T) {
Status: structs.HealthCritical, Status: structs.HealthCritical,
ServiceID: "db", ServiceID: "db",
} }
if err := store.EnsureCheck(checkAfter); err != nil { if err := store.EnsureCheck(17, checkAfter); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
@ -603,10 +644,10 @@ func TestEnsureCheck(t *testing.T) {
} }
defer store.Close() defer store.Close()
if err := store.EnsureNode(structs.Node{"foo", "127.0.0.1"}); err != nil { if err := store.EnsureNode(1, structs.Node{"foo", "127.0.0.1"}); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
if err := store.EnsureService("foo", "db1", "db", "master", 8000); err != nil { if err := store.EnsureService(2, "foo", &structs.NodeService{"db1", "db", "master", 8000}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
check := &structs.HealthCheck{ check := &structs.HealthCheck{
@ -616,7 +657,7 @@ func TestEnsureCheck(t *testing.T) {
Status: structs.HealthPassing, Status: structs.HealthPassing,
ServiceID: "db1", ServiceID: "db1",
} }
if err := store.EnsureCheck(check); err != nil { if err := store.EnsureCheck(3, check); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
@ -626,11 +667,14 @@ func TestEnsureCheck(t *testing.T) {
Name: "memory utilization", Name: "memory utilization",
Status: structs.HealthWarning, Status: structs.HealthWarning,
} }
if err := store.EnsureCheck(check2); err != nil { if err := store.EnsureCheck(4, check2); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
checks := store.NodeChecks("foo") idx, checks := store.NodeChecks("foo")
if idx != 4 {
t.Fatalf("bad: %v", idx)
}
if len(checks) != 2 { if len(checks) != 2 {
t.Fatalf("bad: %v", checks) t.Fatalf("bad: %v", checks)
} }
@ -641,7 +685,10 @@ func TestEnsureCheck(t *testing.T) {
t.Fatalf("bad: %v", checks[1]) t.Fatalf("bad: %v", checks[1])
} }
checks = store.ServiceChecks("db") idx, checks = store.ServiceChecks("db")
if idx != 4 {
t.Fatalf("bad: %v", idx)
}
if len(checks) != 1 { if len(checks) != 1 {
t.Fatalf("bad: %v", checks) t.Fatalf("bad: %v", checks)
} }
@ -649,7 +696,10 @@ func TestEnsureCheck(t *testing.T) {
t.Fatalf("bad: %v", checks[0]) t.Fatalf("bad: %v", checks[0])
} }
checks = store.ChecksInState(structs.HealthPassing) idx, checks = store.ChecksInState(structs.HealthPassing)
if idx != 4 {
t.Fatalf("bad: %v", idx)
}
if len(checks) != 1 { if len(checks) != 1 {
t.Fatalf("bad: %v", checks) t.Fatalf("bad: %v", checks)
} }
@ -657,7 +707,10 @@ func TestEnsureCheck(t *testing.T) {
t.Fatalf("bad: %v", checks[0]) t.Fatalf("bad: %v", checks[0])
} }
checks = store.ChecksInState(structs.HealthWarning) idx, checks = store.ChecksInState(structs.HealthWarning)
if idx != 4 {
t.Fatalf("bad: %v", idx)
}
if len(checks) != 1 { if len(checks) != 1 {
t.Fatalf("bad: %v", checks) t.Fatalf("bad: %v", checks)
} }
@ -673,10 +726,10 @@ func TestDeleteNodeCheck(t *testing.T) {
} }
defer store.Close() defer store.Close()
if err := store.EnsureNode(structs.Node{"foo", "127.0.0.1"}); err != nil { if err := store.EnsureNode(1, structs.Node{"foo", "127.0.0.1"}); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
if err := store.EnsureService("foo", "db1", "db", "master", 8000); err != nil { if err := store.EnsureService(2, "foo", &structs.NodeService{"db1", "db", "master", 8000}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
check := &structs.HealthCheck{ check := &structs.HealthCheck{
@ -686,7 +739,7 @@ func TestDeleteNodeCheck(t *testing.T) {
Status: structs.HealthPassing, Status: structs.HealthPassing,
ServiceID: "db1", ServiceID: "db1",
} }
if err := store.EnsureCheck(check); err != nil { if err := store.EnsureCheck(3, check); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
@ -696,15 +749,18 @@ func TestDeleteNodeCheck(t *testing.T) {
Name: "memory utilization", Name: "memory utilization",
Status: structs.HealthWarning, Status: structs.HealthWarning,
} }
if err := store.EnsureCheck(check2); err != nil { if err := store.EnsureCheck(4, check2); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
if err := store.DeleteNodeCheck("foo", "db"); err != nil { if err := store.DeleteNodeCheck(5, "foo", "db"); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
checks := store.NodeChecks("foo") idx, checks := store.NodeChecks("foo")
if idx != 5 {
t.Fatalf("bad: %v", idx)
}
if len(checks) != 1 { if len(checks) != 1 {
t.Fatalf("bad: %v", checks) t.Fatalf("bad: %v", checks)
} }
@ -720,10 +776,10 @@ func TestCheckServiceNodes(t *testing.T) {
} }
defer store.Close() defer store.Close()
if err := store.EnsureNode(structs.Node{"foo", "127.0.0.1"}); err != nil { if err := store.EnsureNode(1, structs.Node{"foo", "127.0.0.1"}); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
if err := store.EnsureService("foo", "db1", "db", "master", 8000); err != nil { if err := store.EnsureService(2, "foo", &structs.NodeService{"db1", "db", "master", 8000}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
check := &structs.HealthCheck{ check := &structs.HealthCheck{
@ -733,7 +789,7 @@ func TestCheckServiceNodes(t *testing.T) {
Status: structs.HealthPassing, Status: structs.HealthPassing,
ServiceID: "db1", ServiceID: "db1",
} }
if err := store.EnsureCheck(check); err != nil { if err := store.EnsureCheck(3, check); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
check = &structs.HealthCheck{ check = &structs.HealthCheck{
@ -742,11 +798,14 @@ func TestCheckServiceNodes(t *testing.T) {
Name: SerfCheckName, Name: SerfCheckName,
Status: structs.HealthPassing, Status: structs.HealthPassing,
} }
if err := store.EnsureCheck(check); err != nil { if err := store.EnsureCheck(4, check); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
nodes := store.CheckServiceNodes("db") idx, nodes := store.CheckServiceNodes("db")
if idx != 4 {
t.Fatalf("bad: %v", idx)
}
if len(nodes) != 1 { if len(nodes) != 1 {
t.Fatalf("Bad: %v", nodes) t.Fatalf("Bad: %v", nodes)
} }
@ -767,7 +826,10 @@ func TestCheckServiceNodes(t *testing.T) {
t.Fatalf("Bad: %v", nodes[0]) t.Fatalf("Bad: %v", nodes[0])
} }
nodes = store.CheckServiceTagNodes("db", "master") idx, nodes = store.CheckServiceTagNodes("db", "master")
if idx != 4 {
t.Fatalf("bad: %v", idx)
}
if len(nodes) != 1 { if len(nodes) != 1 {
t.Fatalf("Bad: %v", nodes) t.Fatalf("Bad: %v", nodes)
} }
@ -795,10 +857,10 @@ func BenchmarkCheckServiceNodes(t *testing.B) {
} }
defer store.Close() defer store.Close()
if err := store.EnsureNode(structs.Node{"foo", "127.0.0.1"}); err != nil { if err := store.EnsureNode(1, structs.Node{"foo", "127.0.0.1"}); err != nil {
t.Fatalf("err: %v", err) t.Fatalf("err: %v", err)
} }
if err := store.EnsureService("foo", "db1", "db", "master", 8000); err != nil { if err := store.EnsureService(2, "foo", &structs.NodeService{"db1", "db", "master", 8000}); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
check := &structs.HealthCheck{ check := &structs.HealthCheck{
@ -808,7 +870,7 @@ func BenchmarkCheckServiceNodes(t *testing.B) {
Status: structs.HealthPassing, Status: structs.HealthPassing,
ServiceID: "db1", ServiceID: "db1",
} }
if err := store.EnsureCheck(check); err != nil { if err := store.EnsureCheck(3, check); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }
check = &structs.HealthCheck{ check = &structs.HealthCheck{
@ -817,7 +879,7 @@ func BenchmarkCheckServiceNodes(t *testing.B) {
Name: SerfCheckName, Name: SerfCheckName,
Status: structs.HealthPassing, Status: structs.HealthPassing,
} }
if err := store.EnsureCheck(check); err != nil { if err := store.EnsureCheck(4, check); err != nil {
t.Fatalf("err: %v") t.Fatalf("err: %v")
} }