Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[bugfix] Treat input throughput data as immutable #3360

Merged
merged 5 commits into from
Oct 31, 2021
Merged

[bugfix] Treat input throughput data as immutable #3360

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
e8dfd26
Issue #3349 - enable `-shuffle=on` in tests
rbroggi Oct 28, 2021
4aab3e7
fixing test interdependency due to shared global variable (which is w…
rbroggi Oct 31, 2021
87e4dc1
fixing test interdependency due to shared global variable
rbroggi Oct 31, 2021
b643c12
enforcing method immutability
rbroggi Oct 31, 2021
37d7f34
temporarily disable shuffle
rbroggi Oct 31, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 15 additions & 1 deletion plugin/sampling/strategystore/adaptive/processor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -310,12 +310,26 @@ func (p *Processor) aggregateThroughput(throughputs []*model.Throughput) service
t.Count += throughput.Count
t.Probabilities = merge(t.Probabilities, throughput.Probabilities)
} else {
aggregatedThroughput[service][operation] = throughput
copyThroughput := model.Throughput{
Service: throughput.Service,
Operation: throughput.Operation,
Count: throughput.Count,
Probabilities: copySet(throughput.Probabilities),
}
aggregatedThroughput[service][operation] = &copyThroughput
}
}
return aggregatedThroughput
}

func copySet(in map[string]struct{}) map[string]struct{} {
out := make(map[string]struct{}, len(in))
for key := range in {
out[key] = struct{}{}
}
return out
}

func (p *Processor) initializeThroughput(endTime time.Time) {
for i := 0; i < p.AggregationBuckets; i++ {
startTime := endTime.Add(p.CalculationInterval * -1)
Expand Down
53 changes: 33 additions & 20 deletions plugin/sampling/strategystore/adaptive/processor_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,15 +35,17 @@ import (
"github.com/jaegertracing/jaeger/thrift-gen/sampling"
)

var (
testThroughputs = []*model.Throughput{
func testThroughputs() []*model.Throughput {
return []*model.Throughput{
{Service: "svcA", Operation: "GET", Count: 4, Probabilities: map[string]struct{}{"0.1": {}}},
{Service: "svcA", Operation: "GET", Count: 4, Probabilities: map[string]struct{}{"0.2": {}}},
{Service: "svcA", Operation: "PUT", Count: 5, Probabilities: map[string]struct{}{"0.1": {}}},
{Service: "svcB", Operation: "GET", Count: 3, Probabilities: map[string]struct{}{"0.1": {}}},
}
}

testThroughputBuckets = []*throughputBucket{
func testThroughputBuckets() []*throughputBucket {
return []*throughputBucket{
{
throughput: serviceOperationThroughput{
"svcA": map[string]*model.Throughput{
Expand All @@ -69,18 +71,29 @@ var (
interval: 60 * time.Second,
},
}
}

errTestStorage = errors.New("storage error")
func errTestStorage() error {
return errors.New("storage error")
}

testCalculator = calculationstrategy.CalculateFunc(func(targetQPS, qps, oldProbability float64) float64 {
func testCalculator() calculationstrategy.ProbabilityCalculator {
return calculationstrategy.CalculateFunc(func(targetQPS, qps, oldProbability float64) float64 {
factor := targetQPS / qps
return oldProbability * factor
})
)
}

func TestAggregateThroughputInputsImmutability(t *testing.T) {
p := &Processor{}
in := testThroughputs()
_ = p.aggregateThroughput(in)
assert.Equal(t, in, testThroughputs())
}

func TestAggregateThroughput(t *testing.T) {
p := &Processor{}
aggregatedThroughput := p.aggregateThroughput(testThroughputs)
aggregatedThroughput := p.aggregateThroughput(testThroughputs())
require.Len(t, aggregatedThroughput, 2)

throughput, ok := aggregatedThroughput["svcA"]
Expand Down Expand Up @@ -110,7 +123,7 @@ func TestAggregateThroughput(t *testing.T) {
func TestInitializeThroughput(t *testing.T) {
mockStorage := &smocks.Store{}
mockStorage.On("GetThroughput", time.Time{}.Add(time.Minute*19), time.Time{}.Add(time.Minute*20)).
Return(testThroughputs, nil)
Return(testThroughputs(), nil)
mockStorage.On("GetThroughput", time.Time{}.Add(time.Minute*18), time.Time{}.Add(time.Minute*19)).
Return([]*model.Throughput{{Service: "svcA", Operation: "GET", Count: 7}}, nil)
mockStorage.On("GetThroughput", time.Time{}.Add(time.Minute*17), time.Time{}.Add(time.Minute*18)).
Expand All @@ -130,7 +143,7 @@ func TestInitializeThroughput(t *testing.T) {
func TestInitializeThroughputFailure(t *testing.T) {
mockStorage := &smocks.Store{}
mockStorage.On("GetThroughput", time.Time{}.Add(time.Minute*19), time.Time{}.Add(time.Minute*20)).
Return(nil, errTestStorage)
Return(nil, errTestStorage())
p := &Processor{storage: mockStorage, Options: Options{CalculationInterval: time.Minute, AggregationBuckets: 1}}
p.initializeThroughput(time.Time{}.Add(time.Minute * 20))

Expand All @@ -146,7 +159,7 @@ func TestCalculateQPS(t *testing.T) {
}

func TestGenerateOperationQPS(t *testing.T) {
p := &Processor{throughputs: testThroughputBuckets, Options: Options{BucketsForCalculation: 10, AggregationBuckets: 10}}
p := &Processor{throughputs: testThroughputBuckets(), Options: Options{BucketsForCalculation: 10, AggregationBuckets: 10}}
svcOpQPS := p.throughputToQPS()
assert.Len(t, svcOpQPS, 2)

Expand Down Expand Up @@ -194,7 +207,7 @@ func TestGenerateOperationQPS(t *testing.T) {
}

func TestGenerateOperationQPS_UseMostRecentBucketOnly(t *testing.T) {
p := &Processor{throughputs: testThroughputBuckets, Options: Options{BucketsForCalculation: 1, AggregationBuckets: 10}}
p := &Processor{throughputs: testThroughputBuckets(), Options: Options{BucketsForCalculation: 1, AggregationBuckets: 10}}
svcOpQPS := p.throughputToQPS()
assert.Len(t, svcOpQPS, 2)

Expand Down Expand Up @@ -258,7 +271,7 @@ func TestCalculateProbability(t *testing.T) {
p := &Processor{
Options: cfg,
probabilities: probabilities,
probabilityCalculator: testCalculator,
probabilityCalculator: testCalculator(),
throughputs: throughputs,
serviceCache: []SamplingCache{{"svcA": {}, "svcB": {}}},
}
Expand Down Expand Up @@ -302,8 +315,8 @@ func TestCalculateProbabilitiesAndQPS(t *testing.T) {
InitialSamplingProbability: 0.001,
BucketsForCalculation: 10,
},
throughputs: testThroughputBuckets, probabilities: prevProbabilities, qps: qps,
weightVectorCache: NewWeightVectorCache(), probabilityCalculator: testCalculator,
throughputs: testThroughputBuckets(), probabilities: prevProbabilities, qps: qps,
weightVectorCache: NewWeightVectorCache(), probabilityCalculator: testCalculator(),
operationsCalculatedGauge: mets.Gauge(metrics.Options{Name: "test"}),
}
probabilities, qps := p.calculateProbabilitiesAndQPS()
Expand All @@ -324,10 +337,10 @@ func TestRunCalculationLoop(t *testing.T) {
logger := zap.NewNop()
mockStorage := &smocks.Store{}
mockStorage.On("GetThroughput", mock.AnythingOfType("time.Time"), mock.AnythingOfType("time.Time")).
Return(testThroughputs, nil)
mockStorage.On("GetLatestProbabilities").Return(model.ServiceOperationProbabilities{}, errTestStorage)
Return(testThroughputs(), nil)
mockStorage.On("GetLatestProbabilities").Return(model.ServiceOperationProbabilities{}, errTestStorage())
mockStorage.On("InsertProbabilitiesAndQPS", "host", mock.AnythingOfType("model.ServiceOperationProbabilities"),
mock.AnythingOfType("model.ServiceOperationQPS")).Return(errTestStorage)
mock.AnythingOfType("model.ServiceOperationQPS")).Return(errTestStorage())
mockEP := &epmocks.ElectionParticipant{}
mockEP.On("Start").Return(nil)
mockEP.On("Close").Return(nil)
Expand Down Expand Up @@ -366,7 +379,7 @@ func TestRunCalculationLoop_GetThroughputError(t *testing.T) {
logger, logBuffer := testutils.NewLogger()
mockStorage := &smocks.Store{}
mockStorage.On("GetThroughput", mock.AnythingOfType("time.Time"), mock.AnythingOfType("time.Time")).
Return(nil, errTestStorage)
Return(nil, errTestStorage())
mockEP := &epmocks.ElectionParticipant{}
mockEP.On("Start").Return(nil)
mockEP.On("Close").Return(nil)
Expand Down Expand Up @@ -442,7 +455,7 @@ func TestRealisticRunCalculationLoop(t *testing.T) {
t.Skip("Skipped realistic calculation loop test")
logger := zap.NewNop()
// NB: This is an extremely long test since it uses near realistic (1/6th scale) processor config values
testThroughputs = []*model.Throughput{
testThroughputs := []*model.Throughput{
{Service: "svcA", Operation: "GET", Count: 10},
{Service: "svcA", Operation: "POST", Count: 9},
{Service: "svcA", Operation: "PUT", Count: 5},
Expand Down Expand Up @@ -840,7 +853,7 @@ func TestCalculateProbabilitiesAndQPSMultiple(t *testing.T) {

func TestErrors(t *testing.T) {
mockStorage := &smocks.Store{}
mockStorage.On("GetLatestProbabilities").Return(model.ServiceOperationProbabilities{}, errTestStorage)
mockStorage.On("GetLatestProbabilities").Return(model.ServiceOperationProbabilities{}, errTestStorage())
mockStorage.On("GetThroughput", mock.AnythingOfType("time.Time"), mock.AnythingOfType("time.Time")).
Return(nil, nil)

Expand Down