finish with the gossip up and down statuses

Author: steevehook

mutexes/distributed-db/app/app.go

@@ -16,22 +16,23 @@ import (
 )
 
 func New() (*App, error) {
-	nodes := models.Nodes{Map: map[string]struct{}{}}
+	nodesMap := models.NodesMap{}
 	port := flag.Int("port", 8080, "the port of the running server")
 	dataDir := flag.String("data", "", "the data directory of the running server")
-	flag.Var(&nodes, "node", "the list of nodes to talk to")
+	flag.Var(&nodesMap, "node", "the list of nodes to talk to")
 
 	flag.Parse()
 
-	addr := fmt.Sprintf("localhost:%d", *port)
-	if len(nodes.Map) < 1 {
+	if len(nodesMap) < 1 {
 		return nil, fmt.Errorf("need at least 1 node to talk to")
 	}
+
+	addr := fmt.Sprintf("localhost:%d", *port)
 	if *dataDir == "" {
 		*dataDir = fmt.Sprintf(".data/%s", addr)
 	}
 
-	nodes.CurrentNode = addr
+	nodes := models.NewNodes(addr, nodesMap)
 	tokens := models.NewTokens(nodes, 256)
 	cacheRepo := repositories.NewCache(*dataDir)
 	httpClient := clients.NewHTTP(addr)

mutexes/distributed-db/clients/http.go

@@ -68,25 +68,25 @@ func (c *HTTPClient) Get(node string, keys []string) ([]models.CacheItem, error)
 	return cacheItems, nil
 }
 
-func (c *HTTPClient) Gossip(node string, nodes []string, tokensChecksum string) ([]string, error) {
+func (c *HTTPClient) Gossip(node string, nodes models.NodesMap, tokensChecksum string) (models.NodesMap, error) {
 	body := models.GossipRequest{
 		Nodes:          nodes,
 		TokensChecksum: tokensChecksum,
 	}
 	req, err := c.makeRequest(http.MethodPost, c.url(node, "gossip"), body)
 	if err != nil {
-		return []string{}, err
+		return models.NodesMap{}, err
 	}
 
 	res, err := c.httpClient.Do(req)
 	if err != nil {
-		return []string{}, err
+		return models.NodesMap{}, err
 	}
 
 	var gossipRes models.GossipResponse
 	err = json.NewDecoder(res.Body).Decode(&gossipRes)
 	if err != nil {
-		return []string{}, err
+		return models.NodesMap{}, err
 	}
 
 	return gossipRes.Nodes, nil

mutexes/distributed-db/controllers/gossip.go

@@ -9,7 +9,7 @@ import (
 )
 
 type tokensUpdater interface {
-	UpdateTokens(node string, newNodes []string, tokensChecksum string) ([]string, error)
+	UpdateTokens(node string, newNodes models.NodesMap, tokensChecksum string) (oldNodes models.NodesMap, err error)
 }
 
 func gossip(svc tokensUpdater) http.HandlerFunc {

mutexes/distributed-db/models/flags.go

@@ -1,51 +1,21 @@
 package models
 
 import (
+	"fmt"
 	"strings"
 )
 
-type Nodes struct {
-	Map         map[string]struct{}
-	CurrentNode string
-}
+type NodesMap map[string]int
 
-func (n Nodes) Set(s string) error {
-	n.Add(s)
+func (m NodesMap) Set(element string) error {
+	m[element] = NodeStatusUp
 	return nil
 }
 
-func (n Nodes) String() string {
-	return strings.Join(n.List(len(n.Map)), ",")
-}
-
-func (n Nodes) Add(nodes ...string) {
-	for _, node := range nodes {
-		if node == n.CurrentNode {
-			continue
-		}
-		// update some kind of UpdateAt field for nodes health check
-		// change from map[string]struct{} to map[string]int => states: Up/Down
-		// a node is Down if it hasn't sent a gossip request in 10 seconds
-		n.Map[node] = struct{}{}
-	}
-}
-
-func (n Nodes) WithCurrentNode() map[string]struct{} {
-	nodes := map[string]struct{}{n.CurrentNode: {}}
-	for s := range n.Map {
-		nodes[s] = struct{}{}
-	}
-	return nodes
-}
-
-func (n Nodes) List(a int) []string {
-	i, keys := 0, make([]string, 0, len(n.Map))
-	for k := range n.Map {
-		if i == a {
-			break
-		}
-		keys = append(keys, k)
-		i++
+func (m NodesMap) String() string {
+	list := make([]string, 0, len(m))
+	for node, status := range m {
+		list = append(list, fmt.Sprintf("%s:%d", node, status))
 	}
-	return keys
+	return strings.Join(list, ",")
 }

mutexes/distributed-db/models/nodes.go

@@ -0,0 +1,119 @@
+package models
+
+import (
+	"log"
+	"time"
+)
+
+const (
+	NodeStatusUp   = 1
+	NodeStatusDown = 0
+	gossipDeadline = 15 * time.Second
+	retryPeriod    = time.Minute
+	maxFails       = 5
+)
+
+func NewNodes(currentNode string, nodeMap NodesMap) *Nodes {
+	nodes := &Nodes{
+		current:      currentNode,
+		nodesStatus:  NodesMap{},
+		gossipStatus: map[string]time.Time{},
+		nodeFails:    map[string]int{},
+		nodeRetries:  map[string]time.Time{},
+	}
+	nodes.Set(nodeMap)
+	go nodes.retry()
+	return nodes
+}
+
+type Nodes struct {
+	current      string
+	nodesStatus  NodesMap
+	gossipStatus map[string]time.Time
+	nodeFails    map[string]int
+	nodeRetries  map[string]time.Time
+}
+
+func (n Nodes) Current() string {
+	return n.current
+}
+
+func (n Nodes) Fail(node string) {
+	n.nodeFails[node]++
+	if n.nodeFails[node] >= maxFails {
+		log.Printf("node: %s is down, retryin in: %v", node, retryPeriod)
+		n.nodesStatus[node] = NodeStatusDown
+		n.nodeRetries[node] = time.Now().UTC()
+	}
+}
+
+func (n Nodes) retry() {
+	for {
+		time.Sleep(time.Second)
+		for node, lastTried := range n.nodeRetries {
+			now := time.Now().UTC()
+			if now.Sub(lastTried) >= retryPeriod {
+				log.Printf("retrying gossip on node: %s", node)
+				n.nodeFails[node] = 0
+				n.nodesStatus[node] = NodeStatusUp
+				delete(n.nodeRetries, node)
+				n.gossipStatus[node] = time.Now().UTC()
+			}
+		}
+	}
+}
+
+func (n Nodes) Gossip(node string) {
+	n.gossipStatus[node] = time.Now().UTC()
+}
+
+func (n Nodes) Set(nodesStatus NodesMap) {
+	for node, status := range nodesStatus {
+		if node == n.current {
+			continue
+		}
+		n.nodesStatus[node] = status
+	}
+}
+
+func (n Nodes) Map() NodesMap {
+	nodes := NodesMap{n.current: NodeStatusUp}
+	for node, lastGossip := range n.gossipStatus {
+		now := time.Now().UTC()
+		if now.Sub(lastGossip) < gossipDeadline {
+			nodes[node] = NodeStatusUp
+			continue
+		}
+		nodes[node] = NodeStatusDown
+	}
+	return nodes
+}
+
+func (n Nodes) List(x int) []string {
+	return n.list(x, false)
+}
+
+func (n Nodes) ListAll() []string {
+	return n.list(len(n.nodesStatus), false)
+}
+
+func (n Nodes) ListActive(x int) []string {
+	return n.list(x, true)
+}
+
+func (n Nodes) list(x int, filterByNodeStatusUp bool) []string {
+	i, nodeList := 0, make([]string, 0, len(n.nodesStatus))
+	for node, status := range n.nodesStatus {
+		if i == x {
+			break
+		}
+
+		if filterByNodeStatusUp && status != NodeStatusUp {
+			continue
+		}
+
+		nodeList = append(nodeList, node)
+		i++
+	}
+	return nodeList
+}

mutexes/distributed-db/models/requests.go

@@ -16,6 +16,6 @@ type SetRequest struct {
 }
 
 type GossipRequest struct {
-	Nodes          []string `json:"nodes"`
-	TokensChecksum string   `json:"tokens_checksum"`
+	Nodes          map[string]int `json:"nodes"`
+	TokensChecksum string         `json:"tokens_checksum"`
 }

mutexes/distributed-db/models/responses.go

@@ -1,7 +1,7 @@
 package models
 
 type GossipResponse struct {
-	Nodes []string `json:"nodes"`
+	Nodes map[string]int `json:"nodes"`
 }
 
 type TokensResponse struct {

mutexes/distributed-db/models/tokens.go

@@ -17,13 +17,13 @@ type TokenMappings map[int]string
 
 type Tokens struct {
 	Mappings            TokenMappings
-	Nodes               Nodes
+	Nodes               *Nodes
 	ranges              []int
 	numberOfTokenRanges int
 }
 
-func NewTokens(nodes Nodes, numberOfTokenRanges int) *Tokens {
-	nodeList := append([]string{nodes.CurrentNode}, nodes.List(len(nodes.Map))...)
+func NewTokens(nodes *Nodes, numberOfTokenRanges int) *Tokens {
+	nodeList := append([]string{nodes.Current()}, nodes.ListAll()...)
 	tokenRange := math.MaxInt / len(nodeList) / numberOfTokenRanges
 	ranges := make([]int, 0, len(nodeList)*numberOfTokenRanges)
 	for i := 0; i < len(nodeList); i++ {
@@ -67,18 +67,18 @@ func (t *Tokens) GetNode(key string) string {
 
 func (t *Tokens) Merge(mappings map[int]string) {
 	newMappings := map[int]string{}
-	nodes := t.Nodes.WithCurrentNode()
+	nodes := t.Nodes.Map()
 	ranges := t.ranges
 	m1, m2 := mappings, t.Mappings
 	if len(mappings) < len(t.Mappings) {
 		m1 = t.Mappings
 		m2 = mappings
 	}
 
-	newNodes := map[string]struct{}{}
+	newNodes := map[string]int{}
 	newRanges := make([]int, 0)
 	for _, s := range mappings {
-		newNodes[s] = struct{}{}
+		newNodes[s] = NodeStatusUp
 	}
 	for r := range m1 {
 		newRanges = append(newRanges, r)
@@ -92,14 +92,14 @@ func (t *Tokens) Merge(mappings map[int]string) {
 	}
 
 	for s := range nodes {
-		_, ok := t.Nodes.WithCurrentNode()[s]
+		_, ok := t.Nodes.Map()[s]
 		if ok {
 			delete(nodes, s)
 		}
 	}
 
 	i := 0
-	numberOfNodes := len(nodes) + len(t.Nodes.WithCurrentNode())
+	numberOfNodes := len(nodes) + len(t.Nodes.Map())
 	tokenRange := math.MaxInt / numberOfNodes / t.numberOfTokenRanges
 	m1Nodes := map[string]struct{}{}
 	for r, s := range m1 {

mutexes/distributed-db/services/cache.go

@@ -16,7 +16,7 @@ type CacheRepository interface {
 type HTTPClient interface {
 	Set(node, key, value string) (models.CacheItem, error)
 	Get(node string, keys []string) ([]models.CacheItem, error)
-	Gossip(node string, nodes []string, tokensChecksum string) (oldNodes []string, err error)
+	Gossip(node string, newNodes models.NodesMap, tokensChecksum string) (oldNodes models.NodesMap, err error)
 	Tokens(node string) (models.TokenMappings, error)
 }
 
@@ -55,7 +55,7 @@ func (svc CacheSvc) Get(keys []string) []models.CacheItem {
 
 	cacheItems := make([]models.CacheItem, 0)
 	for node, sums := range nodeToSums {
-		if node == svc.tokens.Nodes.CurrentNode {
+		if node == svc.tokens.Nodes.Current() {
 			items := svc.cacheRepo.Get(sums)
 			for _, item := range items {
 				item.Node = node
@@ -81,7 +81,7 @@ func (svc CacheSvc) Get(keys []string) []models.CacheItem {
 
 func (svc CacheSvc) Set(key, value string) (models.CacheItem, error) {
 	node := svc.tokens.GetNode(key)
-	if node == svc.tokens.Nodes.CurrentNode {
+	if node == svc.tokens.Nodes.Current() {
 		item := svc.cacheRepo.Set(key, value)
 		item.Node = node
 		return item, nil
@@ -91,33 +91,37 @@ func (svc CacheSvc) Set(key, value string) (models.CacheItem, error) {
 }
 
 func (svc CacheSvc) Gossip() {
-	nodes := svc.tokens.Nodes.List(2)
-	log.Println("gossiping to:", strings.Join(nodes, ","))
+	nodes := svc.tokens.Nodes.ListActive(2)
+	if len(nodes) == 0 {
+		return
+	}
 
-	allNodes := append([]string{}, svc.tokens.Nodes.List(len(svc.tokens.Nodes.Map))...)
-	allNodes = append(allNodes, svc.tokens.Nodes.CurrentNode)
+	log.Println("gossiping to:", strings.Join(nodes, ","))
 	for _, node := range nodes {
-		oldNodes, err := svc.httpClient.Gossip(node, allNodes, svc.tokens.Checksum())
+		oldNodes, err := svc.httpClient.Gossip(node, svc.tokens.Nodes.Map(), svc.tokens.Checksum())
 		if err != nil {
 			log.Printf("could not make http call for gossip: %v", err)
+			svc.tokens.Nodes.Fail(node)
 			continue
 		}
 
-		svc.tokens.Nodes.Add(oldNodes...)
+		svc.tokens.Nodes.Set(oldNodes)
 	}
 }
 
-func (svc CacheSvc) UpdateTokens(node string, newNodes []string, tokensChecksum string) ([]string, error) {
-	oldNodes := svc.tokens.Nodes.List(len(svc.tokens.Nodes.Map))
+func (svc CacheSvc) UpdateTokens(node string, newNodes models.NodesMap, tokensChecksum string) (models.NodesMap, error) {
+	svc.tokens.Nodes.Gossip(node)
+	svc.tokens.Nodes.Set(newNodes)
+
 	if svc.tokens.Checksum() != tokensChecksum {
-		svc.tokens.Nodes.Add(newNodes...)
 		tokens, err := svc.httpClient.Tokens(node)
 		if err != nil {
-			return []string{}, err
+			return models.NodesMap{}, err
 		}
 		svc.tokens.Merge(tokens)
 	}
-	return oldNodes, nil
+
+	return svc.tokens.Nodes.Map(), nil
 }
 
 func (svc CacheSvc) GetTokens() map[int]string {