Adding logic to eras to automatically update using the BirdBrain code

Author: Ian Pye

cluster/era.go

@@ -1,5 +1,10 @@
 package cluster
 
+import (
+	"logger"
+	"stash.cloudflare.com/go-stream/util/slog"
+)
+
 type Era interface {
 	GetNodes() []Node
 }
@@ -19,3 +24,60 @@ func (s *SimpleEra) Add(n Node) {
 func (s *SimpleEra) GetNodes() []Node {
 	return s.nodes
 }
+
+type WeightedEra struct {
+	nodes   []Node
+	nodeMap [MAX_WEIGHT]*WeightedNode
+}
+
+func NewWeightedEra() *WeightedEra {
+	return &WeightedEra{make([]Node, 0, 0), [MAX_WEIGHT]*WeightedNode{}}
+}
+
+func (s *WeightedEra) Add(n Node) {
+	s.nodes = append(s.nodes, n)
+}
+
+func (s *WeightedEra) GetNodes() []Node {
+	return s.nodes
+}
+
+func (s *WeightedEra) GetNode(posit int) *WeightedNode {
+	if posit < MAX_WEIGHT {
+		return s.nodeMap[posit]
+	}
+	return s.nodeMap[0]
+}
+
+func (s *WeightedEra) NormalizeAndPopulateMap() {
+	total := uint32(0)
+	scalar := uint32(1)
+	for _, n := range s.nodes {
+		wn := n.(*WeightedNode)
+		total += wn.weight
+	}
+
+	if total == 0 {
+		slog.Logf(logger.Levels.Error, "Total Node Wieght 0")
+		return
+	}
+
+	if total < MAX_WEIGHT {
+		// Scale weights up
+		scalar = MAX_WEIGHT / total
+		total = MAX_WEIGHT
+	}
+
+	lastPosit := 0
+	for _, n := range s.nodes {
+		wn := n.(*WeightedNode)
+		wn.weight = ((wn.weight * scalar) / total) * MAX_WEIGHT
+		slog.Logf(logger.Levels.Debug, "New Weight %d", wn.weight)
+		for i := lastPosit; uint32(i) < wn.weight && i < MAX_WEIGHT; i++ {
+			s.nodeMap[i] = wn
+			lastPosit++
+		}
+	}
+
+	return
+}

cluster/manager.go

@@ -1,6 +1,17 @@
 package cluster
 
-import "time"
+import (
+	"encoding/json"
+	"io/ioutil"
+	"logger"
+	"net/http"
+	"stash.cloudflare.com/go-stream/util/slog"
+	"time"
+)
+
+const (
+	MAX_ERAS_SAVED = 10
+)
 
 /* There should be a separate manager for each service */
 type Manager interface {
@@ -23,3 +34,93 @@ func (c *StaticManager) GetCurrentEra() Era {
 func NewStaticManager(e Era) *StaticManager {
 	return &StaticManager{e}
 }
+
+type DynamicBBManager struct {
+	Eras        map[time.Time]Era
+	ErasAdded   []time.Time
+	BBHosts     []string
+	CurrentTime time.Time
+}
+
+type BBHost struct {
+	Disk_free float32
+	Load      float32
+	Name      string
+	Host      string
+	Port      string
+}
+
+type BBResult struct {
+	Nodes []BBHost
+}
+
+func (c *DynamicBBManager) GetEra(t time.Time) Era {
+	if ct, ok := c.Eras[t]; ok {
+		return ct
+	}
+	return c.Eras[c.CurrentTime]
+}
+
+func (c *DynamicBBManager) GetCurrentEra() Era {
+	return c.Eras[c.CurrentTime]
+}
+
+func (c *DynamicBBManager) pullLatestEra() (err error) {
+	for _, url := range c.BBHosts {
+		if resp, err := http.Get(url); err == nil {
+			if bbbody, err := ioutil.ReadAll(resp.Body); err == nil {
+				// Try parsing this.
+				bbr := BBResult{}
+				if err := json.Unmarshal(bbbody, &bbr); err == nil {
+					ctime := time.Now()
+
+					we := NewWeightedEra()
+					for _, node := range bbr.Nodes {
+						n := NewWeightedNode(node.Name, node.Host, node.Port, node.Disk_free, node.Load)
+						slog.Logf(logger.Levels.Debug, "Trasport LOG INFO %v", n)
+						we.Add(n)
+					}
+
+					// Once all the nodes are in for this era, re-weight the Era
+					we.NormalizeAndPopulateMap()
+					c.Eras[ctime] = we
+					c.CurrentTime = ctime
+					c.ErasAdded = append(c.ErasAdded, ctime)
+
+					// And Remove any super old eras
+					if len(c.ErasAdded) > MAX_ERAS_SAVED {
+						delete(c.Eras, c.ErasAdded[0])
+						c.ErasAdded = append(c.ErasAdded[:1], c.ErasAdded[2:]...)
+					}
+
+					// Once we have hit one BB server with no error, no need to try any others.
+					break
+				}
+			}
+		}
+	}
+	return
+}
+
+func (c *DynamicBBManager) keepErasCurrent() {
+	for {
+		time.Sleep(60 * time.Second)
+		slog.Logf(logger.Levels.Debug, "Updating to new era")
+		err := c.pullLatestEra()
+		if err != nil {
+			slog.Logf(logger.Levels.Error, "Cannot get a valid era %v", err)
+		}
+	}
+}
+
+func NewDynamicBBManager(bbHosts []string) *DynamicBBManager {
+	dm := DynamicBBManager{make(map[time.Time]Era), make([]time.Time, 0, 0), bbHosts, time.Now()}
+	err := dm.pullLatestEra()
+	if err != nil || len(dm.Eras) == 0 {
+		slog.Fatalf("Cannot get a valid era %v", err)
+	}
+
+	// Keep updating with periodic info
+	go dm.keepErasCurrent()
+	return &dm
+}

cluster/node.go

@@ -1,5 +1,12 @@
 package cluster
 
+const (
+	MAX_WEIGHT        = 100
+	MEDIAN_WEIGHT     = 50
+	MIN_DISK_TO_WORRY = 0.000809804
+	DEGRADED_WEIGHT   = 30
+)
+
 type Node interface {
 	Name() string
 }
@@ -31,3 +38,32 @@ func (n *SimpleNode) Ip() string {
 func (n *SimpleNode) Port() string {
 	return n.port
 }
+
+type WeightedNode struct {
+	name   string
+	ip     string
+	port   string
+	weight uint32
+}
+
+func NewWeightedNode(name string, ip string, port string, disk float32, load float32) *WeightedNode {
+	wn := WeightedNode{name, ip, port, MEDIAN_WEIGHT}
+	// For right now, keep it simple. Once a disk falls below a given threshold, trottle it a bit.
+	// Don't pay attention to load now, to avoid flapping.
+	if disk < MIN_DISK_TO_WORRY {
+		wn.weight = DEGRADED_WEIGHT
+	}
+	return &wn
+}
+
+func (n *WeightedNode) Name() string {
+	return n.name
+}
+
+func (n *WeightedNode) Ip() string {
+	return n.ip
+}
+
+func (n *WeightedNode) Port() string {
+	return n.port
+}