add test for query cost predictors

fe/lbv2/lbv2.go

@@ -6,12 +6,11 @@ import (
 	"math/rand"
 	"net"
 	"sync"
+	"time"
 
 	"github.com/jmcvetta/randutil"
 )
 
-const baseWeight = 1000000
-
 // Node utilization along a specific dimension. Utilization is treated
 // as a vector, the LoadBalancer is opaque to the actual meaning of
 // the dimensions used (though it makes sense for Requests to be the
@@ -47,7 +46,7 @@ func (c NodeScore) SetScore(w float64) NodeScore {
 }
 
 // Damping factor for the utilization query cost model.
-const costAlpha = 0.9
+const costAlpha = 0.8
 
 type costEstimate float64
 
@@ -341,6 +340,8 @@ func NewActiveNodesFilter() NodeFilter {
 	return &activeNodesFilter{}
 }
 
+const baseWeight = 1000000
+
 // Return a random item based on a weighted distribution.
 func weightedPolicyFunc(wnodes []NodeScore) Node {
 	// Need to convert this anyway.
@@ -389,3 +390,9 @@ func highestScorePolicyFunc(wnodes []NodeScore) Node {
 }
 
 var HighestScorePolicy = PolicyFunc(highestScorePolicyFunc)
+
+func init() {
+	// Seed the math/rand PRNG. The current time works just fine,
+	// it doesn't need to be cryptographically robust.
+	rand.Seed(time.Now().Unix())
+}

fe/loadbalancing_test.go

@@ -9,6 +9,7 @@ import (
 	"git.autistici.org/ale/autoradio/fe/lbv2"
 )
 
+// Not a very good way to compare probability distributions.
 func compareDistribution(a, b map[string]int, n int) bool {
 	for k, aval := range a {
 		bval := b[k]
@@ -26,6 +27,7 @@ func runLBTest(t *testing.T, nodes []*autoradio.NodeStatus, ctx lbv2.RequestCont
 		t.Fatalf("Error parsing spec \"%s\": %v", lbspec, err)
 	}
 
+	// We're not testing the time component, disable predictors.
 	lb.DisableUtilizationPredictors()
 
 	// Run lb.Choose() n times and accumulate the results.
@@ -131,3 +133,70 @@ func TestLoadBalancer_PoliciesIgnoreDisabledNodes(t *testing.T) {
 		runLBTest(t, nodes, nil, spec, 1000, map[string]int{"node1": 1000})
 	}
 }
+
+func TestLoadBalancer_UtilizationPredictor(t *testing.T) {
+	cap := 100
+	node := &autoradio.NodeStatus{
+		Name:         "node1",
+		IcecastUp:    true,
+		MaxListeners: cap,
+		Mounts: []autoradio.IcecastMountStatus{
+			{Listeners: 0},
+		},
+	}
+	nodes := []*autoradio.NodeStatus{node}
+
+	lb, err := parseLoadBalancerSpec("listeners_available,listeners_score,best")
+	if err != nil {
+		t.Fatal(err)
+	}
+	lb.Update(nodes)
+
+	// See how many clients can connect successfully until the
+	// predicted utilization goes above 1.
+	okRequests := func() int {
+		for i := 0; i < 110; i++ {
+			result := lb.Choose(nil)
+			if result == nil {
+				return i
+			}
+		}
+		return 111
+	}
+
+	// At first, the query cost function is 1 - a new query will
+	// cause utilization to stay >1 until Update() is called.
+	nr := okRequests()
+	if nr > 1 {
+		t.Fatalf("initial run: ok_requests = %d, want 1", nr)
+	}
+
+	reachedCapacityAt := -1
+	for i := 0; i < 1000; i++ {
+		// Update the LoadBalancer with the current connection count.
+		node.Mounts[0].Listeners = nr
+		lb.Update(nodes)
+
+		newNr := okRequests()
+		if newNr == 0 && nr < cap {
+			t.Errorf("iteration %d: no successful requests but utilization is below 100% (avail=%d, ok_requests=%d)", i, cap-nr, newNr)
+		}
+		if newNr > (cap - nr) {
+			t.Fatalf("iteration %d: over capacity (avail=%d, ok_requests=%d)", i, cap-nr, newNr)
+		}
+		nr += newNr
+		if nr >= cap && reachedCapacityAt < 0 {
+			reachedCapacityAt = i
+		}
+	}
+	// Verify how quickly the query cost converged to the real
+	// value. This depends on the value of the costAlpha constant,
+	// and it will vary from 1 (if the damping factor is zero the
+	// real value immediately kicks in) to 100 (if the query cost
+	// stays at 1, as it is initially).
+	// The default value for costAlpha of 0.8 should cause
+	// convergence at i==20.
+	if reachedCapacityAt > 20 {
+		t.Fatalf("reached capacity at iteration %d, the query cost function did not converge properly", reachedCapacityAt)
+	}
+}