Fix advanced metrics performance test

Makefile

@@ -147,7 +147,7 @@ performance-test: ## Run performance tests
 
 test-bench: ## Run benchmark tests
 	cd test/performance && GOWORK=off CGO_ENABLED=0 go test -mod=vendor -count 5 -timeout 2m -bench=. -benchmem metrics_test.go
-	cd test/performance && GOWORK=off CGO_ENABLED=0 go test -mod=vendor -count 1 -bench=. -benchmem userWorkFlow_test.go
+	cd test/performance && GOWORK=off CGO_ENABLED=0 go test -mod=vendor -count 1 -bench=. -benchmem user_workflow_test.go
 	cd test/performance && GOWORK=off CGO_ENABLED=0 go test -mod=vendor -count 5 -timeout 2m -bench=. -benchmem plugins_test.go
 
 build-benchmark-docker: ## Build benchmark test docker image for NGINX Plus, need nginx-repo.crt and nginx-repo.key in build directory

sdk/proto/metrics.proto

@@ -28,7 +28,7 @@ message Dimension {
   string value = 2 [(gogoproto.jsontag) = "value" ];
 }
 
-// StatsEntity a timestamped entry for Dementions and Metrics
+// StatsEntity a timestamped entry for Dimensions and Metrics
 message StatsEntity {
   google.protobuf.Timestamp timestamp = 1;
   // DEPRECATED

test/performance/advanced-metrics/Dockerfile

@@ -1,4 +1,3 @@
 FROM scratch
-
 COPY bin/generator /generator
 ENTRYPOINT ["/generator"]

test/performance/advanced-metrics/README.md

@@ -7,11 +7,11 @@ Benchmarks are run in docker-compose environment. You can start a benchmark with
 make benchmark_run
 ```
 To run benchmarks of old AVR you need to put .deb file with AVR into /bin and name it `avr.deb`
-and do the same with dependencies but with `deps.deb` file name. We are running AVR benchmarks on Ubutnu:xenial.
+and do the same with dependencies but with `deps.deb` file name. We are running AVR benchmarks on Ubuntu:xenial.
 
 ### Configuration
 
-Benchmarks are configurated via .test_env file commited into this repository.
+Benchmarks are configurated via `.test_env` file committed into this repository.
 File should look something like this:
 ```bigquery
 export SOCKET_DIR=/tmp
@@ -40,7 +40,7 @@ export SIMPLE_BENCHMARK=true
 
 ### Setting up external machine for benchmarking
 
-Use setup_external_ubuntu.sh script
+Use `setup_external_ubuntu.sh` script
 
 ### Prometheus and benchmarking metrics
 

test/performance/advanced-metrics/avr_harness/harness.go

@@ -79,22 +79,22 @@ func main() {
 
 	fmt.Println("Nats Subscriber OK")
 	for _, s := range h.natsSubs {
-		go func() {
+		go func(statsChannel chan *Stats) {
 			for {
-				msg := <-s.Stats
+				msg := <-statsChannel
 				messagesProcessed.Inc()
 				if msg.Stats == nil {
 					continue
 				}
 
-				metricsProcessedOnOutput.Add(float64(len(msg.Stats.Metrics)))
+				metricsProcessedOnOutput.Add(float64(len(msg.Stats.Simplemetrics)))
 				for _, d := range msg.Stats.Dimensions {
 					if d.Value == aggregatedValue {
 						aggregatedDimensionValuesProcessedOnOutput.Inc()
 					}
 				}
 			}
-		}()
+		}(s.Stats)
 	}
 	wg.Wait()
 }
@@ -128,12 +128,12 @@ func (h *Harness) SetupNATSSubscriber(subj string, addr string) *NatsConn {
 	out := make(chan *Stats)
 	_, err = nc.Subscribe(subj, func(m *nats.Msg) {
 		size := len(m.Data)
-		avrstat := &metrics.StatsEntity{}
-		err := proto.Unmarshal(m.Data, avrstat)
+		avrStats := &metrics.StatsEntity{}
+		err := proto.Unmarshal(m.Data, avrStats)
 		if err != nil {
 			log.Fatalf("Error proto.Unmarshal: %v, Payload (%d bytes): %q", err, size, string(m.Data))
 		} else {
-			out <- &Stats{Stats: avrstat, Subject: &subj, Size: &size}
+			out <- &Stats{Stats: avrStats, Subject: &subj, Size: &size}
 		}
 	})
 	if err != nil {

test/performance/advanced-metrics/fake_agent/Dockerfile

@@ -1,4 +1,3 @@
 FROM scratch
-
 COPY bin/agent-with-acr /agent-with-acr
 ENTRYPOINT ["/agent-with-acr"]

test/performance/advanced-metrics/fake_agent/agent.go

@@ -1,23 +1,24 @@
-package fake_agent
+package main
 
 import (
 	"context"
 	"fmt"
 	"log"
 	"net/http"
 	"net/http/pprof"
-	_ "net/http/pprof"
 	"os"
 	"os/signal"
 	"syscall"
 	"time"
 
 	"github.com/kelseyhightower/envconfig"
-	"github.com/nginx/agent/v2/src/extensions/advanced-metrics/pkg/schema"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/sirupsen/logrus"
+
+	advanced_metrics "github.com/nginx/agent/v2/src/extensions/advanced-metrics/pkg/advanced-metrics"
+	"github.com/nginx/agent/v2/src/extensions/advanced-metrics/pkg/schema"
 )
 
 const (
@@ -62,10 +63,6 @@ const (
 	proxiedProtocolDimension           = "proxied_protocol"
 	bytesRcvdMetric                    = "bytes_rcvd"
 	bytesSentMetric                    = "bytes_sent"
-
-	aggregationDurrationDimension = "aggregation_duration"
-
-	streamMetricFamilyDimensionValue = "tcp-udp"
 )
 
 var (
@@ -108,20 +105,20 @@ func main() {
 	}()
 
 	builder := schema.NewSchemaBuilder()
-	builder.NewDimension(httpUriDimension, 16000, schema.Lvl(1)).
+	builder.NewDimension(httpUriDimension, 16000).
 		NewIntegerDimension(httpResponseCodeDimension, 600).
-		NewDimension(httpRequestMethodDimension, 16, schema.Lvl(20)).
+		NewDimension(httpRequestMethodDimension, 16).
 		NewMetric(hitcountMetric).
 		NewMetric(httpRequestBytesRcvdMetric).
 		NewMetric(httpRequestBytesSentMetric).
 		NewDimension(environmentDimension, 32).
 		NewDimension(appDimension, 32).
 		NewDimension(componentDimension, 256).
-		NewDimension(countryCodeDimension, 256, schema.Lvl(50)). //TODO should be implemented as GeoIP
-		NewDimension(httpVersionSchemaDimension, 16, schema.Lvl(30)).
-		NewDimension(httpUpstreamAddrDimension, 1024, schema.Lvl(50)).
+		NewDimension(countryCodeDimension, 256). // TODO should be implemented as GeoIP
+		NewDimension(httpVersionSchemaDimension, 16).
+		NewDimension(httpUpstreamAddrDimension, 1024).
 		NewIntegerDimension(upstreamResponseCodeDimension, 600).
-		NewDimension(httpHostnameDimension, 16000, schema.Lvl(10)).
+		NewDimension(httpHostnameDimension, 16000).
 		NewMetric(clientNetworkLatencyMetric).
 		NewMetric(clientTtfbLatencyMetric).
 		NewMetric(clientRequestLatencyMetric).
@@ -130,14 +127,14 @@ func main() {
 		NewMetric(upstreamHeaderLatencyMetric).
 		NewMetric(upstreamResponseLatencyMetric).
 		NewDimension(publishedApiDimension, 256).
-		NewDimension(requestOutcomeDimension, 8, schema.Lvl(80)).
-		NewDimension(requestOutcomeReasonDimension, 32, schema.Lvl(80)).
+		NewDimension(requestOutcomeDimension, 8).
+		NewDimension(requestOutcomeReasonDimension, 32).
 		NewDimension(gatewayDimension, 32).
-		NewDimension(wafSignatureIdsDimension, 16000, schema.Lvl(30)).
-		NewDimension(wafAttackTypesDimension, 8, schema.Lvl(30)).
-		NewDimension(wafViolationRatingDimension, 8, schema.Lvl(30)).
-		NewDimension(wafViolationsDimension, 128, schema.Lvl(30)).
-		NewDimension(wafViolationSubviolationsDimension, 16, schema.Lvl(30)).
+		NewDimension(wafSignatureIdsDimension, 16000).
+		NewDimension(wafAttackTypesDimension, 8).
+		NewDimension(wafViolationRatingDimension, 8).
+		NewDimension(wafViolationsDimension, 128).
+		NewDimension(wafViolationSubviolationsDimension, 16).
 		NewMetric(clientLatencyMetric).
 		NewMetric(upstreamLatencyMetric).
 		NewMetric(connectionDurationMetric).
@@ -148,16 +145,23 @@ func main() {
 
 	config := advanced_metrics.Config{
 		Address: cfg.AdvancedMetricsSocket,
+		AggregatorConfig: advanced_metrics.AggregatorConfig{
+			AggregationPeriod: time.Second,
+			PublishingPeriod:  time.Second * 10,
+		},
+		TableSizesLimits: advanced_metrics.TableSizesLimits{
+			StagingTableMaxSize:    32000,
+			StagingTableThreshold:  32000,
+			PriorityTableMaxSize:   32000,
+			PriorityTableThreshold: 32000,
+		},
 	}
-	config.AggregationPeriod = time.Second * 10
-	config.PublishingPeriod = time.Second * 30
-	config.StagingTableMaxSize = 32000
-	config.StagingTableThreshold = 28000
-	s, err := builder.Build()
+
+	schema, err := builder.Build()
 	if err != nil {
 		log.Fatal(err)
 	}
-	app, err := advanced_metrics.NewAdvancedMetrics(config, s)
+	app, err := advanced_metrics.NewAdvancedMetrics(config, schema)
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -166,7 +170,6 @@ func main() {
 			for _, m := range f {
 				messagesProcessed.Inc()
 				metricsProcessedOnOutput.Add(float64(len(m.Metrics)))
-
 			}
 			if len(f) == 0 {
 				continue

test/performance/advanced-metrics/main.go

@@ -1,4 +1,4 @@
-package advanced_metrics
+package main
 
 import (
 	"context"
@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/kelseyhightower/envconfig"
+	"github.com/nginx/agent/test/performance/advanced-metrics/metric_gen"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -83,7 +84,7 @@ func main() {
 	// wait 3 seconds to make sure socket exists
 	time.Sleep(time.Second * 3)
 
-	pusher := NewPusher(cfg.AvrSocket)
+	pusher := metric_gen.NewPusher(cfg.AvrSocket)
 	err = pusher.Connect()
 	if err != nil {
 		log.Fatal(err)

test/performance/advanced-metrics/socket.go → test/performance/advanced-metrics/metric_gen/socket.go

@@ -1,4 +1,4 @@
-package main
+package metric_gen
 
 import (
 	"net"