[loadbalancingexporter] Distribution across target endpoints is not approximately uniform

[jamesmoessis]

Component(s)

exporter/loadbalancing

What happened?

Description

In our production workload, the load balancing exporter is sending spans in a fashion that was more widely distributed than I expected, per pod. The highest pod received 3.5x the number of spans that the pod that received the lowest.

This creates a large distribution of CPU usage across endpoints that the LB exporter is sending to, not ideal for managing infrastructure, and right-sizing the allocations for the downstream workload.

Steps to Reproduce

I've written a test that confirms that the distribution is not uniform, over 100k random trace IDs, I tracked the number of resolutions per endpoint:

As you can see the trend is steeply down. I generated this writing this test in consistent_hashing_test.go. Then generated this chart from the CSV.

func TestUniformityOfDistribution(t *testing.T) {
	endpoints := make([]string, 300)
	for i := range endpoints {
		endpoints[i] = fmt.Sprintf("endpoint-%d", i)
	}
	ring := newHashRing(endpoints)

	n := 100_000
	resolutions := make(map[string][]pcommon.TraceID, len(endpoints))

	for i := 0; i < n; i++ {
		id := generateRandomTraceID(t)
		resolved := ring.endpointFor(id[:])
		resolutions[resolved] = append(resolutions[resolved], id)
	}

	timesResolved := make([]int, len(endpoints))
	for i := range timesResolved {
		res, ok := resolutions[endpoints[i]]
		numRes := 0
		if ok {
			numRes = len(res)
		}
		timesResolved[i] = numRes
	}

	f, err := os.Create("resolutions.csv")
	require.NoError(t, err)
	defer f.Close()

	_, err = f.WriteString("endpoint,n\n")
	require.NoError(t, err)

	for i := range endpoints {
		s := fmt.Sprintf("%s,%d\n", endpoints[i], timesResolved[i])
		_, err = f.WriteString(s)
		require.NoError(t, err)
	}
}

Expected Result

Traces are evenly distributed across endpoints.

Actual Result

They are not even close to uniformly distributed. We observe a trend as shown in the screenshot.

Collector version

v0.128.0

Environment information

OpenTelemetry Collector configuration

Log output

Additional context

No response

Tip

_{React with 👍 to help prioritize this issue. Please use comments to provide useful context, avoiding +1 or me too, to help us triage it. Learn more here.}

[github-actions]

Pinging code owners:

• exporter/loadbalancing: @rlankfo

See Adding Labels via Comments if you do not have permissions to add labels yourself.

[jamesmoessis]

@rlankfo can you comment on this behaviour

[rlankfo]

hey @jamesmoessis 👋

The consistent hashing done by the load balancing exporter is an implementation of the algorithm described by Karger et al. in this publication.

My understanding is that we can increase the number of positions per endpoint (defaultWeight) to smooth out the variance in how keys are distributed among endpoints. The current value is hardcoded to 100. In testing, I also increased maxPositions. When we increase the number of positions per endpoint, we also need a larger coordinate space to prevent collisions, which would otherwise reduce the effectiveness of increasing defaultWeight.

I ran your test and increased defaultWeight and maxPositions to the following values, which resulted in a significant improvement in the variance where the most-loaded endpoint was only about 1.5× the least-loaded.

const (
	maxPositions  uint32 = 3600000
	defaultWeight int    = 3000
)

It's important to note that this comes with some trade offs though:

• The main tradeoff would be increase in memory usage (due to more potential positions per endpoint)
• Longer initialization time for the exporter to build the ring
• Rebalancing operations when endpoints are added or removed during scaling events will also be more expensive (since, iiuc, rebuilding the ring is O(n log n), where n is the total number of positions, not just the number of endpoints).

I don’t think endpoint lookup times would be significantly impacted, as they remain O(log n), which is still very fast in practice.

Would you be able to test some different values here and profile the results to see what configuration gives you acceptable variance while maintaining reasonable resource usage? If we can find better defaults for the average use-case that would be great, or perhaps we could at least make these parameters configurable for users who understand the trade-offs.

cc @jpkrohling as he is the original author of this consistent hashing code and would appreciate any fact checking or additional thoughts here :)

[jamesmoessis]

Thankyou @rlankfo for the great reply. Those results look promising. I like the idea of making it configurable too.

Would you be able to test some different values here and profile the results to see what configuration gives you acceptable variance while maintaining reasonable resource usage?

Let me take a look and I will get back to you here.

[rlankfo]

I played around a bit more with this today and noticed a couple of interesting points.

1. Regardless of what we set default weight to, with the current implementation it is capped at 256. So in my previous comment, when set to 3000, we were only using a max of 256 positions per endpoint.

The 3000 value was somewhat arbitrary for testing as I multiplied the number of endpoints (300) by our current defaultWeight.

The reason this is capped at 256 is because in positionsFor we are only adding the lowest 8 bits of i to the hash:

h.Write([]byte{byte(i)})

When i is > 255 we start using the same 0-255 byte values over again because byte(i) overflows. I made a slight modification to positionsFor to allow encoding up to 4 bytes of i:

func positionsFor(endpoint string, numPoints int) []position {
	res := make([]position, 0, numPoints)
	buf := make([]byte, 4)
	for i := 0; i < numPoints; i++ {
		h := crc32.NewIEEE()
		h.Write([]byte(endpoint))
		binary.LittleEndian.PutUint32(buf, uint32(i))
		h.Write(buf)
		hash := h.Sum32()
		pos := hash % maxPositions
		res = append(res, position(pos))
	}

	return res
}

This allows for much larger defaultWeight and I tested with a few different values, such as 1024, 4096, etc. and wasn't able to improve much on the results I saw with positions per endpoint capped at 256 and 300 endpoints.

2. I do think these values could be influenced by the expected number of endpoints potentially, for example:

const (
	maxPositions  uint32 = 256 * 300 * 10 // defaultWeight * expected number of enpdoints,  increased by order of magnitude to minimize position collisions
	defaultWeight int    = 256            // the number of points in the ring for each entry. For better results, it should be greater than 100.
)

I suspect if we tried with fewer than 300 endpoints, we'd start to see better variance with higher defaultWeight values. I believe this is due to the natural smoothing from many endpoints dividing the ring. Whereas with fewer endpoints, larger default per endpoint share of the ring means each gap matters more and you need a higher weight to maintain lower variance.

In any case, I wanted to point out these findings in case this trips you up in testing. imo, good next steps would be benchmarks and profiling to understand impact of these changes.

[rlankfo]

Oh one other thing I forgot to mention: I believe the reason for the downward trending slope you were seeing in your distribution is due to the position collisions that can happen here:

    // if this position is occupied already, skip this item
    if _, found := positions[pos]; found {
    	continue
    }

With the current defaults we have set and 300 endpoints, we’re trying to fit 30,000 positions into only 36,000 slots. That relatively small coordinate space makes collisions more likely, increasing unevenness.

Because this positions map tracks all positions globally across all endpoints, the further along we are in the endpoint list during ring construction, the more likely it is that positions will collide with those already assigned to earlier endpoints. This results in later endpoints effectively having fewer positions on the ring, which skews the distribution.

[jamesmoessis]

@rlankfo I've been doing some benchmarking, with some curious results. I hope to post something more conclusive later today.

Essentially there is some strange behaviour. Increasing the weights and max positions does not necessarily yield lower variance. More weights = more collisions = left skewed as in my original post. More positions = sparsely populated positions = more room for variance.

This seems to imply there is some kind of sweet spot. Layer this with memory on top, there's multiple variables working here, so it's taken me longer than expected.

An option would be to try again a few times, it wouldn't be a big change, or impact performance - but I would expect it to decrease variance significantly. Something like:

// if this position is occupied already, probe for another free spot close by
actualPos := pos
probeLimit := 3
i := 0
for positions[actualPos] && i < probeLimit {
	actualPos = (actualPos + 1) % position(maxPositions)
	i++
}
if i >= probeLimit {
	continue
}

I will continue to assess the benchmarks I have going at the moment, and come back with a recommendation that's hopefully a bit more conclusive.