feat: modify pool tests

internal/util/harness.go

@@ -78,6 +78,7 @@ func BuildCabooseHarness(t *testing.T, n int, maxRetries int, opts ...HarnessOpt
 	ch.Caboose = bs
 	ch.CabooseActiveNodes = conf.Harness.ActiveNodes.(*caboose.NodeRing)
 	ch.CabooseAllNodes = conf.Harness.AllNodes.(*caboose.NodeHeap)
+	ch.CaboosePool = conf.Harness.PoolController
 	return ch
 }
 
@@ -87,11 +88,18 @@ type CabooseHarness struct {
 
 	CabooseActiveNodes *caboose.NodeRing
 	CabooseAllNodes    *caboose.NodeHeap
+	CaboosePool state.PoolController
 
 	gol      sync.Mutex
 	goodOrch bool
 }
 
+type NodeStats struct {
+	Start time.Time
+	Latency float64
+	Size float64
+}
+
 func (ch *CabooseHarness) RunFetchesForRandCids(n int) {
 	for i := 0; i < n; i++ {
 		randCid, _ := cid.V1Builder{Codec: uint64(multicodec.Raw), MhType: uint64(multicodec.Sha2_256)}.Sum([]byte{uint8(i)})
@@ -122,6 +130,24 @@ func (ch *CabooseHarness) FetchAndAssertSuccess(t *testing.T, ctx context.Contex
 	require.NotEmpty(t, blk)
 }
 
+func (ch *CabooseHarness) RecordSuccesses(t *testing.T, nodes []*caboose.Node, s NodeStats, n int) {
+	for _, node := range(nodes) {
+		s.Start = time.Now().Add(-time.Second*5)
+		for i := 0; i < n; i++ {
+			node.RecordSuccess(s.Start, s.Latency, s.Size)
+		}
+	}
+}
+
+func (ch *CabooseHarness) RecordFailures(t *testing.T, nodes []*caboose.Node, n int) {
+	for _, node := range(nodes) {
+		for i := 0; i < n; i++ {
+			node.RecordFailure()
+		}
+	}
+}
+
+
 func (ch *CabooseHarness) FailNodesWithCode(t *testing.T, selectorF func(ep *Endpoint) bool, code int) {
 	for _, n := range ch.Endpoints {
 		if selectorF(n) {

node_heap.go

@@ -53,9 +53,14 @@ func (nh *NodeHeap) TopN(n int) []*Node {
 	m := make([]*Node, 0, n)
 	nh.lk.RLock()
 	defer nh.lk.RUnlock()
-	for i := 0; i < n && i < len(nh.Nodes); i++ {
-		node := nh.Nodes[i]
-		m = append(m, node)
+	for i := 0; i < n; i++ {
+		node := heap.Pop(nh)
+		if n, ok := node.(*Node); ok {
+			m = append(m, n)
+		}
+	}
+	for _, n := range(m) {
+		heap.Push(nh, n)
 	}
 	return m
 }

node_ring.go

@@ -8,23 +8,23 @@ import (
 
 // NodeRing represents a set of nodes organized for stable hashing.
 type NodeRing struct {
-	nodes map[string]*Node
+	Nodes map[string]*Node
 	ring  hashring.HashRing
 
 	lk sync.RWMutex
 }
 
 func NewNodeRing() *NodeRing {
 	return &NodeRing{
-		nodes: map[string]*Node{},
+		Nodes: map[string]*Node{},
 		ring:  *hashring.New([]string{}),
 	}
 }
 
 func (nr *NodeRing) updateRing() error {
 	// this method expects that the lk is held when called.
 	rs := make(map[string]int)
-	for _, n := range nr.nodes {
+	for _, n := range nr.Nodes {
 		// TODO: weight multiples
 		rs[n.URL] = 1
 	}
@@ -39,7 +39,7 @@ func (nr *NodeRing) MaybeSubstituteOrAdd(candidate *Node, activationThreshold in
 	_, ok := nr.ring.GetNode(candidate.URL)
 	if !ok {
 		// ring is empty. in this case we always want to add.
-		nr.nodes[candidate.URL] = candidate
+		nr.Nodes[candidate.URL] = candidate
 		return true, nr.updateRing()
 	}
 
@@ -50,7 +50,7 @@ func (nr *NodeRing) MaybeSubstituteOrAdd(candidate *Node, activationThreshold in
 	delta := float64(0)
 
 	for n, v := range overlapEstimate {
-		neighbor = nr.nodes[n]
+		neighbor = nr.Nodes[n]
 		neighborVolume := neighbor.Rate()
 
 		// how much worse is candidate?
@@ -59,7 +59,7 @@ func (nr *NodeRing) MaybeSubstituteOrAdd(candidate *Node, activationThreshold in
 	}
 
 	if delta > float64(activationThreshold) {
-		nr.nodes[candidate.URL] = candidate
+		nr.Nodes[candidate.URL] = candidate
 		return true, nr.updateRing()
 	}
 	return false, nil
@@ -68,16 +68,16 @@ func (nr *NodeRing) MaybeSubstituteOrAdd(candidate *Node, activationThreshold in
 func (nr *NodeRing) Add(n *Node) error {
 	nr.lk.Lock()
 	defer nr.lk.Unlock()
-	nr.nodes[n.URL] = n
+	nr.Nodes[n.URL] = n
 	return nr.updateRing()
 }
 
 func (nr *NodeRing) Remove(n *Node) error {
 	nr.lk.Lock()
 	defer nr.lk.Unlock()
 
-	if _, ok := nr.nodes[n.URL]; ok {
-		delete(nr.nodes, n.URL)
+	if _, ok := nr.Nodes[n.URL]; ok {
+		delete(nr.Nodes, n.URL)
 		return nr.updateRing()
 	}
 	return ErrNoBackend
@@ -87,7 +87,7 @@ func (nr *NodeRing) Contains(n *Node) bool {
 	nr.lk.RLock()
 	defer nr.lk.RUnlock()
 
-	_, ok := nr.nodes[n.URL]
+	_, ok := nr.Nodes[n.URL]
 	return ok
 }
 
@@ -104,7 +104,7 @@ func (nr *NodeRing) GetNodes(key string, number int) ([]*Node, error) {
 	}
 	nodes := make([]*Node, 0, len(keys))
 	for _, k := range keys {
-		if n, ok := nr.nodes[k]; ok {
+		if n, ok := nr.Nodes[k]; ok {
 			nodes = append(nodes, n)
 		}
 	}

pool_dynamics_test.go

@@ -1,13 +1,90 @@
 package caboose_test
 
 import (
+	"context"
+	"fmt"
+	"math/rand"
+	"net/url"
 	"testing"
+	"time"
 
+	"github.com/filecoin-saturn/caboose"
 	"github.com/filecoin-saturn/caboose/internal/util"
+	"github.com/ipfs/go-cid"
+	"github.com/multiformats/go-multicodec"
 )
 
+
+const (
+	nodesSize = 10
+	nodesPoolSize = caboose.PoolConsiderationCount
+)
+
+
 func TestPoolDynamics(t *testing.T) {
-	ch := util.BuildCabooseHarness(t, 3, 3)
 
+	ch := util.BuildCabooseHarness(t, nodesSize , 3)
 	ch.StartOrchestrator()
+	baseStatSize := 100
+	baseStatLatency := 100
+	ctx := context.Background()
+
+	testCid, _ := cid.V1Builder{Codec: uint64(multicodec.Raw), MhType: uint64(multicodec.Sha2_256)}.Sum(testBlock)
+
+	ch.FetchAndAssertSuccess(t, ctx, testCid)
+
+	rand.New(rand.NewSource(0))
+	eps := ch.Endpoints
+	controlGroup := make(map[string]string)
+
+	rand.Shuffle(len(eps), func(i, j int) {
+		eps[i], eps[j] = eps[j], eps[i]
+	})
+
+	for _,ep := range(eps[:nodesPoolSize]) {
+		url, _ := url.Parse(ep.Server.URL)
+		controlGroup[url.Host] = ep.Server.URL
+
+	}
+
+	for i := 0; i < 1; i++ {
+		nodes := ch.CabooseAllNodes
+		goodNodes := make([]*caboose.Node, 0)
+		badNodes := make([]*caboose.Node, 0)
+		goodStats := util.NodeStats{
+			Start: time.Now().Add(-time.Second*6),
+			Latency: float64(baseStatLatency) / float64(10),
+			Size: float64(baseStatSize) * float64(10),
+		}
+		badStats := util.NodeStats{
+			Start: time.Now().Add(-time.Second*6),
+			Latency: float64(baseStatLatency) * float64(10),
+			Size: float64(baseStatSize) / float64(10),
+		}
+		for _,n := range(nodes.Nodes) {
+			_, ok := controlGroup[n.URL]
+			if ok {
+				fmt.Println("Good", n.URL)
+
+				goodNodes = append(goodNodes, n)
+			} else {
+				fmt.Println("Bad", n.URL)
+
+				badNodes = append(badNodes, n)
+			}
+		}
+
+		ch.RecordSuccesses(t, goodNodes, goodStats, 10)
+		ch.RecordSuccesses(t, badNodes, badStats, 10)
+
+	}
+
+
+	ch.CaboosePool.DoRefresh()
+	fmt.Println("Pool", ch.CabooseActiveNodes.Nodes)
+
+	for _,n := range(ch.CabooseAllNodes.Nodes) {
+		fmt.Println("Node", n.URL, "size", n.PredictedThroughput)
+	}
+
 }

pool_tier_promotion.go

@@ -1,19 +1,24 @@
 package caboose
 
+import "fmt"
+
 const (
-	poolConsiderationCount = 30
+	PoolConsiderationCount = 3
 	activationThreshold    = 0
 )
 
 func updateActiveNodes(active *NodeRing, all *NodeHeap) error {
-	candidates := all.TopN(poolConsiderationCount)
+	candidates := all.TopN(PoolConsiderationCount)
+	for _, c := range(candidates) {
+		fmt.Println("Candidates", c.URL, c.PredictedThroughput)
+	}
 	added := 0
 	for _, c := range candidates {
 		if active.Contains(c) {
 			continue
 		}
 		activeSize := active.Len()
-		discount := poolConsiderationCount - activeSize
+		discount := PoolConsiderationCount - activeSize
 		if discount < 0 {
 			discount = 0
 		}