Merge pull request #50 from domoslabs/main

Implemented basic Quality Attenuation

6 files changed, 268 insertions, 1 deletions

diff --git a/Makefile b/Makefile
index 564f87b..a024bc2 100644
--- a/Makefile
+++ b/Makefile
@@ -6,7 +6,7 @@ all: build test
 build:
 	go build $(LDFLAGS) networkQuality.go
 test:
-	go test ./timeoutat/ ./traceable/ ./ms/ ./utilities/ ./lgc
+	go test ./timeoutat/ ./traceable/ ./ms/ ./utilities/ ./lgc ./qualityattenuation
 golines:
 	find . -name '*.go' -exec ~/go/bin/golines -w {} \;
 clean:
diff --git a/go.mod b/go.mod
index a448c75..bc9bdf0 100644
--- a/go.mod
+++ b/go.mod
@@ -11,3 +11,14 @@ require (
 	golang.org/x/exp v0.0.0-20230213192124-5e25df0256eb
 	golang.org/x/text v0.7.0 // indirect
 )
+
+require (
+	github.com/influxdata/tdigest v0.0.1
+	github.com/stretchr/testify v1.8.3
+)
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)
diff --git a/go.sum b/go.sum
index 3a4ddff..34c604e 100644
--- a/go.sum
+++ b/go.sum
@@ -1,3 +1,13 @@
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
+github.com/influxdata/tdigest v0.0.1 h1:XpFptwYmnEKUqmkcDjrzffswZ3nvNeevbUSLPP/ZzIY=
+github.com/influxdata/tdigest v0.0.1/go.mod h1:Z0kXnxzbTC2qrx4NaIzYkE1k66+6oEDQTvL95hQFh5Y=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/testify v1.8.3 h1:RP3t2pwF7cMEbC1dqtB6poj3niw/9gnV4Cjg5oW5gtY=
+github.com/stretchr/testify v1.8.3/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+golang.org/x/exp v0.0.0-20180321215751-8460e604b9de/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20230213192124-5e25df0256eb h1:PaBZQdo+iSDyHT053FjUCgZQ/9uqVwPOcl7KSWhKn6w=
 golang.org/x/exp v0.0.0-20230213192124-5e25df0256eb/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc=
 golang.org/x/net v0.7.0 h1:rJrUqqhjsgNp7KqAIc25s9pZnjU7TUcSY7HcVZjdn1g=
@@ -6,3 +16,9 @@ golang.org/x/sys v0.5.0 h1:MUK/U/4lj1t1oPg0HfuXDN/Z1wv31ZJ/YcPiGccS4DU=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/text v0.7.0 h1:4BRB4x83lYWy72KwLD/qYDuTu7q9PjSagHvijDw7cLo=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/tools v0.0.0-20180525024113-a5b4c53f6e8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+gonum.org/v1/gonum v0.0.0-20181121035319-3f7ecaa7e8ca/go.mod h1:Y+Yx5eoAFn32cQvJDxZx5Dpnq+c3wtXuadVZAcxbbBo=
+gonum.org/v1/netlib v0.0.0-20181029234149-ec6d1f5cefe6/go.mod h1:wa6Ws7BG/ESfp6dHfk7C6KdzKA7wR7u/rKwOGE66zvw=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/networkQuality.go b/networkQuality.go
index 309aa94..10ee8d3 100644
--- a/networkQuality.go
+++ b/networkQuality.go
@@ -33,6 +33,7 @@ import (
 	"github.com/network-quality/goresponsiveness/lgc"
 	"github.com/network-quality/goresponsiveness/ms"
 	"github.com/network-quality/goresponsiveness/probe"
+	"github.com/network-quality/goresponsiveness/qualityattenuation"
 	"github.com/network-quality/goresponsiveness/rpm"
 	"github.com/network-quality/goresponsiveness/stabilizer"
 	"github.com/network-quality/goresponsiveness/timeoutat"
@@ -86,6 +87,11 @@ var (
 		false,
 		"Enable the collection and display of extended statistics -- may not be available on certain platforms.",
 	)
+	printQualityAttenuation = flag.Bool(
+		"quality-attenuation",
+		false,
+		"Print quality attenuation information.",
+	)
 	dataLoggerBaseFileName = flag.String(
 		"logger-filename",
 		"",
@@ -469,6 +475,7 @@ func main() {
 	probeStabilizer := stabilizer.NewProbeStabilizer(probeI, K, S, probeStabilizerDebugLevel, probeStabilizerDebugConfig)
 
 	selfRtts := ms.NewInfiniteMathematicalSeries[float64]()
+	selfRttsQualityAttenuation := qualityattenuation.NewSimpleQualityAttenuation()
 	foreignRtts := ms.NewInfiniteMathematicalSeries[float64]()
 
 	// For later debugging output, record the last throughputs on load-generating connectings
@@ -543,6 +550,9 @@ timeout:
 					}
 				} else if probeMeasurement.Type == probe.SelfDown || probeMeasurement.Type == probe.SelfUp {
 					selfRtts.AddElement(probeMeasurement.Duration.Seconds())
+					if *printQualityAttenuation {
+						selfRttsQualityAttenuation.AddSample(probeMeasurement.Duration.Seconds())
+					}
 				}
 
 				if probeMeasurement.Type == probe.Foreign {
@@ -670,6 +680,35 @@ Trimmed Mean Foreign RTT:     %f
 		)
 	}
 
+	if *printQualityAttenuation {
+		fmt.Println("Quality Attenuation Statistics:")
+		fmt.Printf(
+			`Number of losses: %d
+Number of samples: %d
+Loss: %f
+Min: %.6f
+Max: %.6f
+Mean: %.6f 
+Variance: %.6f
+Standard Deviation: %.6f
+PDV(90): %.6f
+PDV(99): %.6f
+P(90): %.6f
+P(99): %.6f
+`, selfRttsQualityAttenuation.GetNumberOfLosses(),
+			selfRttsQualityAttenuation.GetNumberOfSamples(),
+			selfRttsQualityAttenuation.GetLossPercentage(),
+			selfRttsQualityAttenuation.GetMinimum(),
+			selfRttsQualityAttenuation.GetMaximum(),
+			selfRttsQualityAttenuation.GetAverage(),
+			selfRttsQualityAttenuation.GetVariance(),
+			selfRttsQualityAttenuation.GetStandardDeviation(),
+			selfRttsQualityAttenuation.GetPDV(90),
+			selfRttsQualityAttenuation.GetPDV(99),
+			selfRttsQualityAttenuation.GetPercentile(90),
+			selfRttsQualityAttenuation.GetPercentile(99))
+	}
+
 	if !testRanToStability {
 		fmt.Printf("Test did not run to stability, these results are estimates:\n")
 	}
diff --git a/qualityattenuation/qualityattenuation.go b/qualityattenuation/qualityattenuation.go
new file mode 100644
index 0000000..279959e
--- /dev/null
+++ b/qualityattenuation/qualityattenuation.go
@@ -0,0 +1,136 @@
+// Implements a data structure for quality attenuation.
+
+package qualityattenuation
+
+import (
+	"fmt"
+	"math"
+
+	"github.com/influxdata/tdigest"
+)
+
+type SimpleQualityAttenuation struct {
+	empiricalDistribution  *tdigest.TDigest
+	offset                 float64
+	offsetSum              float64
+	offsetSumOfSquares     float64
+	numberOfSamples        int64
+	numberOfLosses         int64
+	latencyEqLossThreshold float64
+	minimumLatency         float64
+	maximumLatency         float64
+}
+
+func NewSimpleQualityAttenuation() *SimpleQualityAttenuation {
+	return &SimpleQualityAttenuation{
+		empiricalDistribution:  tdigest.NewWithCompression(50),
+		offset:                 0.1,
+		offsetSum:              0.0,
+		offsetSumOfSquares:     0.0,
+		numberOfSamples:        0,
+		numberOfLosses:         0,
+		latencyEqLossThreshold: 15.0, // Count latency greater than this value as a loss.
+		minimumLatency:         0.0,
+		maximumLatency:         0.0,
+	}
+}
+
+func (qa *SimpleQualityAttenuation) AddSample(sample float64) error {
+	if sample <= 0.0 {
+		// Ignore zero or negative samples because they cannot be valid.
+		// TODO: This should raise a warning and/or trigger error handling.
+		return fmt.Errorf("sample is zero or negative")
+	}
+	qa.numberOfSamples++
+	if sample > qa.latencyEqLossThreshold {
+		qa.numberOfLosses++
+		return nil
+	} else {
+		if qa.minimumLatency == 0.0 || sample < qa.minimumLatency {
+			qa.minimumLatency = sample
+		}
+		if qa.maximumLatency == 0.0 || sample > qa.maximumLatency {
+			qa.maximumLatency = sample
+		}
+		qa.empiricalDistribution.Add(sample, 1)
+		qa.offsetSum += sample - qa.offset
+		qa.offsetSumOfSquares += (sample - qa.offset) * (sample - qa.offset)
+	}
+	return nil
+}
+
+func (qa *SimpleQualityAttenuation) GetNumberOfLosses() int64 {
+	return qa.numberOfLosses
+}
+
+func (qa *SimpleQualityAttenuation) GetNumberOfSamples() int64 {
+	return qa.numberOfSamples
+}
+
+func (qa *SimpleQualityAttenuation) GetPercentile(percentile float64) float64 {
+	return qa.empiricalDistribution.Quantile(percentile / 100)
+}
+
+func (qa *SimpleQualityAttenuation) GetAverage() float64 {
+	return qa.offsetSum/float64(qa.numberOfSamples-qa.numberOfLosses) + qa.offset
+}
+
+func (qa *SimpleQualityAttenuation) GetVariance() float64 {
+	number_of_latency_samples := float64(qa.numberOfSamples) - float64(qa.numberOfLosses)
+	return (qa.offsetSumOfSquares - (qa.offsetSum * qa.offsetSum / number_of_latency_samples)) / (number_of_latency_samples - 1)
+}
+
+func (qa *SimpleQualityAttenuation) GetStandardDeviation() float64 {
+	return math.Sqrt(qa.GetVariance())
+}
+
+func (qa *SimpleQualityAttenuation) GetMinimum() float64 {
+	return qa.minimumLatency
+}
+
+func (qa *SimpleQualityAttenuation) GetMaximum() float64 {
+	return qa.maximumLatency
+}
+
+func (qa *SimpleQualityAttenuation) GetMedian() float64 {
+	return qa.GetPercentile(50.0)
+}
+
+func (qa *SimpleQualityAttenuation) GetLossPercentage() float64 {
+	return 100 * float64(qa.numberOfLosses) / float64(qa.numberOfSamples)
+}
+
+func (qa *SimpleQualityAttenuation) GetRPM() float64 {
+	return 60.0 / qa.GetAverage()
+}
+
+func (qa *SimpleQualityAttenuation) GetPDV(percentile float64) float64 {
+	return qa.GetPercentile(percentile) - qa.GetMinimum()
+}
+
+// Merge two quality attenuation values. This operation assumes the two samples have the same offset and latency_eq_loss_threshold, and
+// will return an error if they do not.
+// It also assumes that the two quality attenuation values are measurements of the same thing (path, outcome, etc.).
+func (qa *SimpleQualityAttenuation) Merge(other *SimpleQualityAttenuation) error {
+	// Check that offsets are the same
+	if qa.offset != other.offset ||
+		qa.latencyEqLossThreshold != other.latencyEqLossThreshold {
+		return fmt.Errorf("merge quality attenuation values with different offset or latency_eq_loss_threshold")
+	}
+	for _, centroid := range other.empiricalDistribution.Centroids() {
+		mean := centroid.Mean
+		weight := centroid.Weight
+		qa.empiricalDistribution.Add(mean, weight)
+	}
+	qa.offsetSum += other.offsetSum
+	qa.offsetSumOfSquares += other.offsetSumOfSquares
+	qa.numberOfSamples += other.numberOfSamples
+	qa.numberOfLosses += other.numberOfLosses
+	if other.minimumLatency < qa.minimumLatency {
+		qa.minimumLatency = other.minimumLatency
+	}
+	if other.maximumLatency > qa.maximumLatency {
+		qa.maximumLatency = other.maximumLatency
+	}
+	return nil
+}
diff --git a/qualityattenuation/qualityattenuation_test.go b/qualityattenuation/qualityattenuation_test.go
new file mode 100644
index 0000000..4746364
--- /dev/null
+++ b/qualityattenuation/qualityattenuation_test.go
@@ -0,0 +1,65 @@
+package qualityattenuation
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBasicSimpleQualityAttenuation(t *testing.T) {
+	qa := NewSimpleQualityAttenuation()
+	qa.AddSample(1.0)
+	qa.AddSample(2.0)
+	qa.AddSample(3.0)
+	assert.Equal(t, qa.numberOfSamples, int64(3))
+	assert.Equal(t, qa.numberOfLosses, int64(0))
+	assert.InEpsilon(t, 1.0, qa.minimumLatency, 0.000001)
+	assert.InEpsilon(t, 3.0, qa.maximumLatency, 0.000001)
+	assert.InEpsilon(t, 5.7, qa.offsetSum, 0.000001)
+	assert.InEpsilon(t, 12.83, qa.offsetSumOfSquares, 0.000001)
+	assert.InEpsilon(t, 1.0, qa.empiricalDistribution.Quantile(0.1), 0.000001)
+	assert.InEpsilon(t, 2.0, qa.empiricalDistribution.Quantile(0.5), 0.000001)
+	assert.InEpsilon(t, 3.0, qa.empiricalDistribution.Quantile(0.9), 0.000001)
+	//Test the get functions
+	assert.Equal(t, qa.GetNumberOfSamples(), int64(3))
+	assert.Equal(t, qa.GetNumberOfLosses(), int64(0))
+	assert.InEpsilon(t, 1.0, qa.GetMinimum(), 0.000001)
+	assert.InEpsilon(t, 3.0, qa.GetMaximum(), 0.000001)
+	assert.InEpsilon(t, 2.0, qa.GetAverage(), 0.000001)
+	assert.InEpsilon(t, 1.000000, qa.GetVariance(), 0.000001)
+	assert.InEpsilon(t, 1.000000, qa.GetStandardDeviation(), 0.000001)
+	assert.InEpsilon(t, 2.0, qa.GetMedian(), 0.000001)
+	assert.InEpsilon(t, 1.0, qa.GetLossPercentage()+1.000000000, 0.000001)
+	assert.InEpsilon(t, 30, qa.GetRPM(), 0.000001)
+	assert.InEpsilon(t, 1.0, qa.GetPercentile(10.0), 0.000001)
+	assert.InEpsilon(t, 2.0, qa.GetPercentile(50.0), 0.000001)
+	assert.InEpsilon(t, 3.0, qa.GetPercentile(90.0), 0.000001)
+	assert.InEpsilon(t, 2.0, qa.GetPDV(90), 0.000001)
+}
+
+func TestManySamples(t *testing.T) {
+	qa := NewSimpleQualityAttenuation()
+	for i := 1; i < 160000; i++ {
+		qa.AddSample(float64(i) / 10000.0) //Linear ramp from 0.0001 to 16.0
+	}
+	assert.Equal(t, qa.numberOfSamples, int64(160000-1))
+	assert.Equal(t, qa.numberOfLosses, int64(10000-1)) //Samples from 15.0001 to 16.0 are lost
+	assert.InEpsilon(t, 0.0001, qa.minimumLatency, 0.000001)
+	assert.InEpsilon(t, 15.0000, qa.maximumLatency, 0.000001)
+	assert.InEpsilon(t, 1110007.5, qa.offsetSum, 0.000001)
+	assert.InEpsilon(t, 11026611.00024998, qa.offsetSumOfSquares, 0.000001)
+	assert.InEpsilon(t, 1.50005, qa.empiricalDistribution.Quantile(0.1), 0.000001)
+	assert.InEpsilon(t, 7.500049, qa.empiricalDistribution.Quantile(0.5), 0.000001)
+	assert.InEpsilon(t, 13.50005, qa.empiricalDistribution.Quantile(0.9), 0.000001)
+	//Test the get functions
+	assert.Equal(t, qa.GetNumberOfSamples(), int64(160000-1))
+	assert.Equal(t, qa.GetNumberOfLosses(), int64(10000-1))
+	assert.InEpsilon(t, 0.0001, qa.GetMinimum(), 0.000001)
+	assert.InEpsilon(t, 15.0000, qa.GetMaximum(), 0.000001)
+	assert.InEpsilon(t, 7.500049, qa.GetAverage(), 0.000001)
+	assert.InEpsilon(t, 18.750120, qa.GetVariance(), 0.000001)
+	assert.InEpsilon(t, 4.330141, qa.GetStandardDeviation(), 0.000001)
+	assert.InEpsilon(t, 7.500049, qa.GetMedian(), 0.000001)
+	assert.InEpsilon(t, 6.249414, qa.GetLossPercentage(), 0.000001)
+	assert.InEpsilon(t, 7.999947, qa.GetRPM(), 0.000001)
+}