Merge pull request #87 from MindHunter86/hofix/balancer_improve_balancer

Hofix/balancer improve balancer

app/handlers.go

@@ -251,6 +251,7 @@ func (m *App) fbHndBlcNodesBalance(ctx *fiber.Ctx) error {
 	sub := m.chunkRegexp.FindSubmatch([]byte(*uri))
 
 	buf := bytes.NewBuffer(sub[utils.ChunkTitleId])
+	buf.Write(sub[utils.ChunkEpisodeId])
 	buf.Write(sub[utils.ChunkQualityLevel])
 
 	_, server, e := m.bareBalancer.BalanceByChunk(buf.String(), string(sub[utils.ChunkName]))

app/middleware.go

@@ -125,10 +125,44 @@ func (m *App) fbMidAppBalance(ctx *fiber.Ctx) (e error) {
 
 	var server *balancer.BalancerServer
 	uri, reqid := []byte(ctx.Locals("uri").(string)), ctx.Locals("requestid").(string)
+	// uri := []byte(ctx.Locals("uri").(string))
 
 	prefixbuf := bytes.NewBuffer(m.chunkRegexp.FindSubmatch(uri)[utils.ChunkTitleId])
+	prefixbuf.Write(m.chunkRegexp.FindSubmatch(uri)[utils.ChunkEpisodeId])
 	prefixbuf.Write(m.chunkRegexp.FindSubmatch(uri)[utils.ChunkQualityLevel])
 
+	// chunkname, prefix := string(m.chunkRegexp.FindSubmatch(uri)[utils.ChunkName]), prefixbuf.String()
+
+	// for _, cluster := range []balancer.Balancer{m.cloudBalancer, m.bareBalancer} {
+	// 	// TODO
+	// 	// ? do we need the failover with RandomBalancing ???
+	// 	// var fallback bool
+
+	// 	// get all servers for balancing
+	// 	var status *balancer.Status
+	// 	if e = cluster.Balance(chunkname, prefix); e == nil {
+	// 		gLog.Error().Msg("there is no status with payload and error from balancer")
+	// 	}
+
+	// 	if errors.As(e, &status) {
+	// 		if e = status.Err(); e != nil {
+	// 			gLog.Error().Err(e).Interface("cluster", status.Cluster()).Msg(status.Descr())
+	// 			continue
+	// 		}
+	// 	} else {
+	// 		gLog.Error().Err(e).Msg("undefined error from balancer")
+	// 	}
+
+	// 	// parse given servers
+	// 	for _, server := range status.Servers {
+	// 		// if all ok (if no errors) - save destination and go to the next fiber handler:
+	// 		ctx.Locals("srv",
+	// 			strings.ReplaceAll(server.Name, "-node", "")+"."+gCli.String("consul-entries-domain"))
+
+	// 		return ctx.Next()
+	// 	}
+	// }
+
 	for _, cluster := range []balancer.Balancer{m.cloudBalancer, m.bareBalancer} {
 		var fallback bool
 

balancer/cbalancer.go

@@ -6,13 +6,13 @@ import (
 	"io"
 	"math/rand"
 	"net"
-	"strconv"
 	"strings"
 	"sync"
 
 	"github.com/MindHunter86/anilibria-hlp-service/utils"
 	"github.com/jedib0t/go-pretty/v6/table"
 	"github.com/rs/zerolog"
+	"github.com/spaolacci/murmur3"
 	"github.com/urfave/cli/v2"
 )
 
@@ -78,14 +78,8 @@ func (m *ClusterBalancer) BalanceByChunk(prefix, chunkname string) (_ string, se
 		return
 	}
 
-	idx, e := strconv.Atoi(prefix + key)
-	if e != nil {
-		m.log.Debug().Err(e).Msgf("chunkname - '%s'; fallback to legacy balancing", chunkname)
-		return
-	}
-
 	var ip *net.IP
-	if ip = m.getServer(idx); ip == nil {
+	if ip = m.getServer(murmur3.Sum128([]byte(prefix + key))); ip == nil {
 		e = ErrUpstreamUnavailable
 		return
 	}
@@ -114,7 +108,7 @@ func (*ClusterBalancer) getKeyFromChunkName(chunkname *string) (key string, e er
 	return
 }
 
-func (m *ClusterBalancer) getServer(idx int) (ip *net.IP) {
+func (m *ClusterBalancer) getServer(idx1, idx2 uint64) (ip *net.IP) {
 	if !m.TryRLock() {
 		m.log.Warn().Msg("could not get lock for reading upstream; fallback to legacy balancing")
 		return
@@ -125,7 +119,11 @@ func (m *ClusterBalancer) getServer(idx int) (ip *net.IP) {
 		return
 	}
 
-	ip = m.ips[idx%int(m.size)]
+	idx3 := idx1 % uint64(m.size)
+	idx4 := idx2 % uint64(m.size)
+	idx0 := idx3 + idx4
+
+	ip = m.ips[idx0%uint64(m.size)]
 	return ip
 }
 

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/hashicorp/consul/api v1.24.0
 	github.com/jedib0t/go-pretty/v6 v6.4.7
 	github.com/rs/zerolog v1.30.0
+	github.com/spaolacci/murmur3 v1.1.0
 	github.com/urfave/cli/v2 v2.25.7
 	golang.org/x/net v0.15.0
 )

go.sum

@@ -172,6 +172,8 @@ github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 h1:nn5Wsu0esKSJiIVhscUt
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
+github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI=
+github.com/spaolacci/murmur3 v1.1.0/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=